Skip to content

[PureGo] Add Arrow IPC and Flight frame encoder - #732

Draft
zlata-stefanovic-db wants to merge 2 commits into
mainfrom
purego-arrow-encoder
Draft

[PureGo] Add Arrow IPC and Flight frame encoder#732
zlata-stefanovic-db wants to merge 2 commits into
mainfrom
purego-arrow-encoder

Conversation

@zlata-stefanovic-db

Copy link
Copy Markdown
Contributor

What changes are proposed in this pull request?

Adds internal/arrowproto: the Arrow IPC payload type and the encoder that turns
it into Arrow Flight frames. This is the first of several changes building the
Arrow ingestion path, and it fills exactly one seam — stream.EncoderHooks — with
no 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:

  • UnitCount returns the batch's row count rather than 1.
  • Slice performs a genuine row-range slice, so a partially acknowledged batch
    replays 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:

  • Payloads never retain caller memory. A Payload holds 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.
  • Admission happens before decoding. ipc_preflight.go walks IPC flatbuffer
    metadata 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.
  • Chunking is measured, not estimated. Frames are capped at 2 MiB by binary
    searching actual encoded protobuf size inside an exponentially grown bracket,
    because compressed bytes per row cannot be predicted from row count.
  • StampOffset is deliberately a no-op. Arrow's wire offset lives in each
    frame's app_metadata and 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/v18 and
github.com/google/flatbuffers. arrow-go requires google.golang.org/grpc
v1.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
wireStream and transport handshake, the AckModelHooks mapping
ack_up_to_records onto submitted ranges, and the public zerobus Arrow 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), and
gofmt and go vet are clean.

Coverage worth calling out:

  • Round trips through both entry points, typed RecordBatch and raw IPC bytes,
    asserting the payload holds no reference to the caller's batch, columns, or
    buffers. Several tests run on memory.NewCheckedAllocator and assert zero
    outstanding allocations at the end.
  • Slice produces the correct suffix, leaves the original payload unchanged, and
    rejects 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.
  • Compression bombs: a payload compressing to a fraction of its declared size is
    rejected at admission with ErrPayloadTooLarge, and the test asserts the Arrow
    decoder never ran and nothing was allocated.
  • Chunking emits sequential frame offsets, holds every frame under the 2 MiB
    target, rejects a single row that cannot fit, bounds probe work across many
    chunks, and emits an unchanged dictionary only once per writer.
  • TestFlightEmitterReportsSubmittedRowPrefixOnPartialSend fails the sink on the
    second record frame of a chunked payload and asserts the submission receipt
    reports a row prefix strictly inside (0, rows). That pins the multi-frame
    submission 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.

@zlata-stefanovic-db zlata-stefanovic-db self-assigned this Aug 14, 2026
Base automatically changed from purego-stream-durability-model to main August 17, 2026 09:05
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>
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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant