[PureGo] Add Arrow IPC and Flight frame encoder - #732
Draft
zlata-stefanovic-db wants to merge 2 commits into
Draft
[PureGo] Add Arrow IPC and Flight frame encoder#732zlata-stefanovic-db wants to merge 2 commits into
zlata-stefanovic-db wants to merge 2 commits into
Conversation
Fills the stream core's encoder seam for Arrow, in its own package, so the Arrow protocol can be reviewed and tested without a server and without touching internal/stream. Two hooks carry the weight that proto and JSON do not have: unitCount reports row counts rather than one, and slice performs a real row-range slice so a partially acknowledged batch replays only its unacknowledged suffix. Both were made possible by the generalized durability model this branch builds on; nothing in the core changes here. Payloads hold only self-contained IPC bytes, so caller-owned Arrow arrays are materialized synchronously and never retained by the core. Compressed IPC input is inspected for its declared uncompressed buffer sizes before Arrow decodes it, so a highly compressible payload cannot pass admission and then expand past the buffered-bytes limit. The Flight wire adapter, the acknowledgment model, and the public Arrow API follow in separate changes. Signed-off-by: Zlata Stefanovic <zlata.stefanovic@databricks.com>
zlata-stefanovic-db
force-pushed
the
purego-arrow-encoder
branch
from
August 17, 2026 09:39
1f00a8f to
f92df40
Compare
Charging the buffered-bytes limit from the input length let a compressed Arrow payload be admitted for a fraction of the memory it retains: 34 KB charged against 8.4 MB held, so the limit was defeated by the caller's compression ratio and the result was an OOM rather than a rejection. Give the encoding seam an optional hook reporting what a payload retains once it exists, and reconcile the reservation against that. Proto and JSON keep charging their existing estimate. Size a RecordBatch from the rows it covers instead of the whole buffers it points at. A slice shares its parent's buffers, so a ten-row slice of a 200,000-row batch reserved 413 MB for an 11 KB payload and was rejected as too large. Four smaller defects: - Slice checked the acknowledged prefix against the cached header count but sliced the decoded batch, so a drift panicked inside Arrow instead of returning an error, escaping into the caller's goroutine. - The decode entry points skipped the metadata preflight, letting Arrow allocate against a declared length no input backs. - An explicit null for the optional close_stream_duration_ms field failed the whole parse, turning a valid acknowledgment into a stream failure. - A dictionary over the frame target was reported as a one-row overflow, blaming the row for a property of the batch. Add tests for the strict Flight metadata parser, which had none. Signed-off-by: Zlata Stefanovic <zlata.stefanovic@databricks.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes are proposed in this pull request?
Adds
internal/arrowproto: the Arrow IPC payload type and the encoder that turnsit into Arrow Flight frames. This is the first of several changes building the
Arrow ingestion path, and it fills exactly one seam —
stream.EncoderHooks— withno changes to
internal/stream.Stacked on #682, which generalized the core from "one offset = one atomic ack"
to protocol-neutral durability units. Two of the eight encoder hooks are trivial
for proto and JSON but carry the real weight for Arrow, and both only became
expressible because of that change:
UnitCountreturns the batch's row count rather than 1.Sliceperforms a genuine row-range slice, so a partially acknowledged batchreplays only its unacknowledged suffix instead of duplicating an acknowledged
prefix. This is the same client-slicing choice the Rust SDK made, so the core
does not depend on the server deduplicating a replayed prefix.
Why this is its own package and its own PR. The Arrow work was originally
drafted as one large branch mixing the encoder, the Flight transport, the ack
model, and the public API. Splitting it let the risky durability semantics be
reviewed on their own in #682, and lets this change be reviewed and tested
entirely without a server: every test here runs in-process against Arrow and
protobuf, with no Flight connection and no test double standing in for one.
Other things worth knowing:
Payloadholds only a canonical,self-contained IPC stream. Caller-owned Arrow arrays are materialized
synchronously at encode time, so the core can hold a payload across a reconnect
without pinning the user's
RecordBatch.ipc_preflight.gowalks IPC flatbuffermetadata and sums declared uncompressed buffer sizes without constructing any
Arrow array. Without it a highly compressible payload passes the buffered-bytes
limit as a small input, then expands past it once Arrow materializes it.
searching actual encoded protobuf size inside an exponentially grown bracket,
because compressed bytes per row cannot be predicted from row count.
StampOffsetis deliberately a no-op. Arrow's wire offset lives in eachframe's
app_metadataand is stamped by the frame emitter, not on the payload.A payload is therefore offset-independent and needs no re-stamp when replayed on
a new connection.
Dependencies. Adds
github.com/apache/arrow-go/v18andgithub.com/google/flatbuffers. arrow-go requiresgoogle.golang.org/grpcv1.82.0, which raises this module's grpc minimum from v1.81.1. These are the first
dependencies this module has taken beyond grpc and protobuf.
Deliberately not here, each to follow as its own change: the Flight
wireStreamand transport handshake, theAckModelHooksmappingack_up_to_recordsonto submitted ranges, and the publiczerobusArrow API.Nothing in this PR is reachable from a public API yet.
How is this tested?
19 unit tests in
internal/arrowproto, none of which need a server or a network.go test -race ./...passes across the module (571 tests, 10 packages), andgofmtandgo vetare clean.Coverage worth calling out:
RecordBatchand raw IPC bytes,asserting the payload holds no reference to the caller's batch, columns, or
buffers. Several tests run on
memory.NewCheckedAllocatorand assert zerooutstanding allocations at the end.
Sliceproduces the correct suffix, leaves the original payload unchanged, andrejects both a fully acknowledged prefix and a zero prefix. The core only ever
slices a partially acknowledged payload, so either bound means its accounting is
wrong and should surface rather than quietly yield an empty or unchanged payload.
rejected at admission with
ErrPayloadTooLarge, and the test asserts the Arrowdecoder never ran and nothing was allocated.
target, rejects a single row that cannot fit, bounds probe work across many
chunks, and emits an unchanged dictionary only once per writer.
TestFlightEmitterReportsSubmittedRowPrefixOnPartialSendfails the sink on thesecond record frame of a chunked payload and asserts the submission receipt
reports a row prefix strictly inside
(0, rows). That pins the multi-framesubmission contract from [PureGo] Generalize stream durability model #682: when one logical send dies partway, the rows
actually submitted are fewer than the rows in the item, and the core has to be
told which.
Not covered here, because it needs the transport that follows: end-to-end DoPut
against a live or mock Flight server, and the connection-scoped dictionary cache
across multiple logical payloads.