[PureGo] Add Arrow IPC payload - #758
Open
zlata-stefanovic-db wants to merge 3 commits into
Open
Conversation
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>
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/arrowprotowith the Arrow IPC payload type and the typed encodingpath 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
RecordBatchcannot: itpoints at buffers the caller still owns and may release the moment
ingestreturns. 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:
Newserializes the caller's schema andreads it back rather than copying fields, because a shallow field copy still
aliases mutable pointer-backed types such as
DictionaryType— a callermutating one after construction would change what the stream encodes.
Sliceis a genuine row-range slice. A partially acknowledged batchreplays 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.
Slicereconciles thepayload'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 dependencythis module has taken beyond grpc and protobuf. arrow-go requires
google.golang.org/grpcv1.82.0, which raises this module's grpc minimum fromv1.81.1.
Deliberately not here, each following as its own change:
uncompressed buffer sizes so a compression bomb is rejected at admission
instead of expanding past the limit once Arrow materializes it.
actual encoded protobuf size, and the frame emitter.
internal/transport.stream.EncoderHooksand the stream-core wiring.EncodeSchemaIPC/DecodeSchemaIPCare exported in #732 but consumed only bytests there, so this change keeps an unexported
encodeSchemaIPCfor theadmission 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 themodule (561 tests, 10 packages), as does
CGO_ENABLED=0 go test ./..., andgofmt,go vet, andgo mod tidy -diffare clean.Coverage worth calling out:
memory.NewCheckedAllocatorand asserts zerooutstanding 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
IPCByteshands back a copy rather than the retained slice.differing in schema metadata is rejected rather than silently encoded.
DictionaryTypeafterNewdoes not change the owned schema.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. A payload whose row count drifted from its bytes returns an error
instead of panicking.
uncompressed; an unsupported compression value is rejected at construction.
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.