Skip to content

feat(kernel): expose processed results as Arrow IPC streams - #514

Open
rahuls-db wants to merge 1 commit into
mainfrom
feat/kernel-arrow-ipc-streams
Open

rahuls-db wants to merge 1 commit into
mainfrom
feat/kernel-arrow-ipc-streams

Conversation

@rahuls-db

Copy link
Copy Markdown
Collaborator

What did you change, and why?

Why: Consumers such as Go ADBC need a version-neutral way to consume post-processed SEA/kernel results when they use a different Arrow Go major version.

Change: Implement GetArrowIPCStreams for the kernel backend by serializing each processed Arrow record as an independent IPC stream and exposing schema IPC bytes through the existing iterator contract.

How do you know it works?

Testing: Kernel backend unit tests verify IPC round-tripping, iterator behavior, schema framing, and schema-byte ownership. A local Go ADBC end-to-end runner successfully fetched SELECT 1 from warehouse 0000000000e01f56 in both automatic Thrift-to-kernel fallback and explicit kernel modes.

@rahuls-db
rahuls-db marked this pull request as ready for review October 23, 2026 17:22

@peco-review-bot peco-review-bot Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Verdict: 1 Low · 1 Nit

Looks good overall, with two low-severity findings. The IPC adapter correctly reuses the batch iterator's ownership, error and EOF semantics. One finding is that the per-batch schema may not match the advertised SchemaBytes(); the other is that the schema-framing test can't catch a framing bug. Nit: unlike the Thrift iterator, Close() on this iterator closes the underlying rows. That matches the existing kernelBatchIterator behaviour but isn't documented anywhere.

Other findings

  • ⚪ Nit — This SchemaBytes check can't catch a framing bug. It compares the iterator's output with arrowSchemaIPCBytes(schema), which is the same function that produced it.schemaBytes, so if the 8-byte EOS stripping were wrong the test would still pass. The PR description says schema framing is verified, but nothing here tests the actual contract: a schema message with no end-of-stream marker that a consumer can prepend to record-batch messages, or at least decode on its own.

Suggest adding an assertion that ipc.NewReader(bytes.NewReader(schemaBytes)) gives back a schema equal to schema. Also check that the bytes don't end with 0xFFFFFFFF 00000000, or that schemaBytes plus an EOS marker reads cleanly with zero records.

return nil, err
}
defer record.Release()

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 Low — Each per-batch stream is written with record.Schema(), which comes from the per-batch C Data import in nextBatch (cdata.ImportCRecordBatch(&carr, &csch)). SchemaBytes() is built from r.schema, which comes from the separate kernel_result_stream_get_schema call. Nothing checks that the two match. If the kernel ever sends batch schemas that differ from the result schema (different field metadata, nullability, or a later type promotion), the bytes from SchemaBytes() won't describe the streams, which breaks what the interface promises ("schema bytes for the streams"). An ADBC consumer that builds its reader schema from SchemaBytes() would then fail or decode the data wrong.

Suggest checking record.Schema().Equal(it.batches.r.schema), either returning an error or logging when they differ, or picking one schema and using it everywhere. Note that ipc.Writer.Write already rejects a record whose schema differs from the writer's, so simply switching to r.schema would turn a silent mismatch into a hard error. That may be the right call, but it should be a deliberate choice.

@rahuls-db
rahuls-db force-pushed the feat/kernel-arrow-ipc-streams branch from 586ab73 to 909a32c Compare October 23, 2026 17:24

@peco-review-bot peco-review-bot Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Verdict: 1 Low

Looks good: one low test-coverage note. The IPC adapter reuses kernelBatchIterator correctly. Per-batch streams are written with each record's own schema, so the writer's schema-equality check can't fail. Schema bytes drop exactly the 8-byte end-of-stream marker, and SchemaBytes returns a defensive copy.


func TestKernelIPCStreamIterator(t *testing.T) {
schema := arrow.NewSchema([]arrow.Field{{Name: "value", Type: arrow.PrimitiveTypes.Int64}}, nil)
builder := array.NewInt64Builder(memory.DefaultAllocator)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 Low — The new iterator test builds its record with memory.DefaultAllocator, so it can't catch a leaked or double-released batch in kernelIPCStreamIterator.Next, and buffer ownership is the main risk in this adapter. Use memory.NewCheckedAllocator(memory.NewGoAllocator()) and call defer mem.AssertSize(t, 0) after it.Close()/reader.Release() so the test fails if record.Release() is ever dropped or moved.

It would also help to add a case where batches already holds a sticky err (e.g. &kernelBatchIterator{r: rows, err: someErr}). That case should check that Next returns the error once and that HasNext is false afterwards, which pins the forwarding contract for the ADBC consumer.

This branch has not been deployed

No deployments
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