Skip to content

[PureGo] Add Arrow IPC payload - #758

Open
zlata-stefanovic-db wants to merge 3 commits into
mainfrom
purego-arrow-ipc-payload
Open

[PureGo] Add Arrow IPC payload#758
zlata-stefanovic-db wants to merge 3 commits into
mainfrom
purego-arrow-ipc-payload

Conversation

@zlata-stefanovic-db

@zlata-stefanovic-db zlata-stefanovic-db commented Aug 18, 2026

Copy link
Copy Markdown
Contributor

What changes are proposed in this pull request?

Adds internal/arrowproto with the Arrow IPC payload type and the typed encoding
path that produces it. This is the first of five changes split out of #732,
which carried the whole encoder at +3386 lines. Nothing in the package is
reachable from a public API yet.

A payload has to survive a reconnect, and a caller's RecordBatch cannot: it
points at buffers the caller still owns and may release the moment ingest
returns. So a payload materializes when it is built, as a canonical
self-contained Arrow IPC stream carrying schema, dictionaries, exactly one
non-empty batch, and the end marker. The core can then hold it across a
reconnect without pinning anything of the user's.

The two pieces worth review attention:

  • Schema ownership goes through IPC. New serializes the caller's schema and
    reads it back rather than copying fields, because a shallow field copy still
    aliases mutable pointer-backed types such as DictionaryType — a caller
    mutating one after construction would change what the stream encodes.
  • Slice is a genuine row-range slice. A partially acknowledged batch
    replays only its unacknowledged suffix rather than duplicating an acknowledged
    prefix, which is the same client-slicing choice the Rust SDK made, so the core
    does not depend on the server deduplicating a replayed prefix. This became
    expressible only because [PureGo] Generalize stream durability model #682 generalized the core from "one offset = one
    atomic ack" to protocol-neutral durability units. Slice reconciles the
    payload's row count against the decoded batch before slicing, since a bound
    past the decoded extent panics inside Arrow and that panic would escape
    recovery into the caller's goroutine.

Admission sizes a batch from the rows it covers, not the whole buffers it
points at.
A slice shares its parent's buffers, so measuring buffer lengths
charges a ten-row slice of a 20,000-row batch for the entire parent and rejects
it as too large. Layouts with a derivable per-row extent (fixed-width, and
variable-width via the offsets buffer) are charged for their own rows; the rest
fall back to whole buffers, an over-estimate that the reconciliation against the
materialized payload corrects in a later change.

Dependencies. Adds github.com/apache/arrow-go/v18, the first dependency
this module has taken beyond grpc and protobuf. arrow-go requires
google.golang.org/grpc v1.82.0, which raises this module's grpc minimum from
v1.81.1.

Deliberately not here, each following as its own change:

  1. Raw IPC bytes as an input, with the flatbuffer preflight that sums declared
    uncompressed buffer sizes so a compression bomb is rejected at admission
    instead of expanding past the limit once Arrow materializes it.
  2. Flight frame encoding: the 2 MiB chunk plan measured by binary searching
    actual encoded protobuf size, and the frame emitter.
  3. Flight acknowledgment and batch metadata parsing in internal/transport.
  4. stream.EncoderHooks and the stream-core wiring.

EncodeSchemaIPC/DecodeSchemaIPC are exported in #732 but consumed only by
tests there, so this change keeps an unexported encodeSchemaIPC for the
admission base and they get exported by the change that consumes them.

How is this tested?

8 unit tests in internal/arrowproto, none of which need a server or a network —
every test runs in-process against Arrow. go test -race ./... passes across the
module (561 tests, 10 packages), as does CGO_ENABLED=0 go test ./..., and
gofmt, go vet, and go mod tidy -diff are clean.

Coverage worth calling out:

  • The typed round trip runs on memory.NewCheckedAllocator and asserts zero
    outstanding allocations after the caller's batch is released, so the payload
    demonstrably holds no reference to the batch, its columns, or their buffers.
    It also asserts IPCBytes hands back a copy rather than the retained slice.
  • Schema exactness includes metadata, so a batch matching field-for-field but
    differing in schema metadata is rejected rather than silently encoded.
  • Mutating a DictionaryType after New does not change the owned schema.
  • 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. A payload whose row count drifted from its bytes returns an error
    instead of panicking.
  • LZ4 and Zstd both round-trip 2,000 rows and each come out smaller than
    uncompressed; an unsupported compression value is rejected at construction.
  • Admission covers the materialized payload in every case tested, and a ten-row
    slice of a 20,000-row batch is charged under a hundredth of its parent.

Not covered here because it needs the changes that follow: Flight frame
chunking, compression-bomb rejection at admission, and anything end-to-end
against a Flight server.

The Arrow ingestion path needs a payload the stream core can hold across a
reconnect. A caller's RecordBatch is unusable for that: it points at buffers
the caller still owns and may release the moment ingest returns. So a payload
materializes when it is built, as a canonical self-contained Arrow IPC stream
carrying schema, dictionaries, one non-empty batch, and the end marker. The
protocol copies the caller's schema through IPC rather than keeping the Schema
object, because a shallow field copy still aliases mutable pointer-backed types
such as DictionaryType.

Slice drops an acknowledged row prefix and reserializes the suffix, so a
partially acknowledged batch replays only what the server has not confirmed.
It reconciles the payload's row count against the decoded batch first, since
slicing past a decoded bound panics inside Arrow and that panic would escape
recovery into the caller's goroutine.

Size a batch for admission from the rows it covers rather than the whole
buffers it points at. A slice shares its parent's buffers, so measuring buffer
lengths charges a ten-row slice of a 20,000-row batch for the entire parent and
rejects it as too large. Layouts with a derivable per-row extent are charged
for their own rows; the rest fall back to whole buffers, an over-estimate a
later reconciliation against the materialized payload corrects.

Nothing here is reachable from a public API yet. Flight frame encoding, raw IPC
input with its compression-expansion guard, and the stream-core wiring follow
in separate changes.

Signed-off-by: Zlata Stefanovic <zlata.stefanovic@databricks.com>
A sliced RecordBatch shares its parent's buffers, and slicing rebases only
the top-level node: a struct's fields stay parallel to the parent's offset
and a list's values are reached through its offsets buffer. The admission
estimate recursed into those children as-is, so a ten-row slice of a
20,000-row list batch was charged its parent's whole extent — 2,834,200
bytes for a 1,600-byte payload.

That is not merely conservative. The core reserves the estimate before
encoding, and reserve rejects any single weight above the buffered-bytes
limit, so reconciliation against the materialized payload never runs. Past
the 64 MiB default, a small slice of a large parent was refused with
ErrPayloadTooLarge outright.

Thread an explicit row window through the sizing recursion instead of
reading each node's own length. childWindow maps a parent's covered rows
onto a struct, list, map, large-list, or fixed-size-list child, and
ownedBufferSize gains per-row rules for those node buffers. Scaling by
offset+length would not work: a head slice has offset 0, so that
denominator equals the slice length and charges the whole buffer again.

List drops to 67,392 and struct to 66,412; int32 and dictionary are
byte-identical, so the derived paths do not regress. List-view, union, and
run-end-encoded still fall back to whole buffers.

Signed-off-by: Zlata Stefanovic <zlata.stefanovic@databricks.com>
Neither constant said what it stands in for. The slop covers IPC framing a
buffer sum cannot see — batch metadata, buffer padding, end markers — which
tracks schema width rather than row count, hence a flat allowance.

It bounds the estimate rather than what an admitted payload occupies, since
the core reconciles a reservation down to the payload's real size after
encoding. The consequence worth naming is the floor: a
MaxBufferedPayloadBytes below it rejects every batch.

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

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant