feat: Support IPC Message custom_metadata in ArrowStreamWriter and ArrowStreamReader - #432
Open
CurtHagenlocher wants to merge 6 commits into
Open
feat: Support IPC Message custom_metadata in ArrowStreamWriter and ArrowStreamReader#432CurtHagenlocher wants to merge 6 commits into
CurtHagenlocher wants to merge 6 commits into
Conversation
The Arrow IPC format supports custom_metadata on each Message (RecordBatch), but the C# implementation currently ignores it on read. This adds a LastBatchCustomMetadata property to ArrowStreamReader that exposes the key-value pairs from the most recently read batch's Message. This is the read-side counterpart to pyarrow's read_next_batch_with_custom_metadata() and enables use cases like RPC frameworks that embed method routing or log metadata in per-batch custom_metadata fields. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Adds WriteRecordBatch(batch, customMetadata) and its async counterpart to ArrowStreamWriter, allowing callers to attach per-message custom_metadata key-value pairs when writing IPC streams. The Arrow IPC flatbuf Message already defines a custom_metadata field, and pyarrow supports writing it via write_batch(batch, custom_metadata). This brings the C# writer to parity. Includes round-trip tests verifying custom_metadata survives write → read through ArrowStreamWriter/ArrowStreamReader. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
- C# writes IPC stream with custom_metadata → Python reads via read_next_batch_with_custom_metadata() - Python writes IPC stream with custom_metadata → C# reads via LastBatchCustomMetadata
- FlightDataStream: override the customMetadata-aware WriteMessageAsync
overload (the one WriteRecordBatchInternalAsync now actually calls)
instead of the old 4-arg overload, so Flight writes aren't silently
routed through the base implementation and don't bypass DataHeader
capture.
- ArrowStreamWriter: validate that caller-supplied custom metadata has
no null keys/values before building FlatBuffer offsets, so failures
are reported as a clear ArgumentException rather than an opaque
FlatBufferBuilder exception.
- ArrowStreamReader: correct the LastBatchCustomMetadata XML doc to
describe its actual update semantics (it's left unchanged when a
read call returns null, e.g. at end of stream).
- CustomMetadataPythonTests: use the repo's shared PythonNetFixture +
[Collection("PythonNet")] instead of a private per-class Python.NET
init/shutdown, avoiding double-Initialize/premature-Shutdown races
with other Python.NET tests; this also fixes the missing Py.GIL()
guard around the Windows sys.path append, since the shared fixture
already wraps that in using (Py.GIL()).
Verified with:
- dotnet build Apache.Arrow.sln — 0 warnings/errors
- dotnet test test/Apache.Arrow.Tests/Apache.Arrow.Tests.csproj —
1876 passed, 30 skipped, 0 failed
- dotnet format Apache.Arrow.sln --exclude src/Apache.Arrow/Flatbuf/FlatBuffers/ --verify-no-changes — clean
The new WriteRecordBatch(batch, customMetadata) overloads went straight to WriteRecordBatchInternal, but ArrowFileWriter relied on overriding each public WriteRecordBatch to call WriteStart() first. Calling the new overload on an ArrowFileWriter therefore skipped the ARROW1 file magic and silently produced a file that ArrowFileReader rejects with "Invalid magic at offset <6>". Rather than adding two more overrides that a future overload could again forget, move the WriteStart()/WriteStartAsync() call into WriteRecordBatchInternal, where every write path must pass through it. Both are idempotent, so the byte output is unchanged for the stream writer, the file writer and Flight. ArrowFileWriter's WriteRecordBatch/WriteRecordBatchAsync overrides are now redundant and removed. Also: - Remove the second virtual WriteMessageAsync overload. Two virtual overloads where one forwards to the other is the trap that already routed Flight's record batch writes past FlightDataStream's override; the sync WriteMessage has always been a single method with a defaulted customMetadataOffset. Callers that do not supply metadata now pass default explicitly. - Remove the private protected WriteRecordBatchInternal/WriteRecordBatchInternalAsync forwarding overloads. They carry no backwards-compatibility obligation, and fewer near-identical overloads means fewer places for the metadata argument to get silently dropped. - Validate custom metadata before anything is written instead of part-way through building the message, so a rejected dictionary leaves the writer usable, and hoist the duplicated offset-building block into GetCustomMetadataOffset. - Read Message.custom_metadata the same way schema and field metadata are already read in MessageSerializer, rather than skipping null keys and rewriting null values as "". Tests: ArrowFileWriter round-trips with custom metadata (sync and async) and still emits the file magic, custom metadata after an explicit WriteStart, empty dictionary, null key and null value rejection, and writer reuse after a rejected dictionary. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XhMo3XSWYHo1apHd9PzZTb
Replace the ArrowStreamReader.LastBatchCustomMetadata property with a
RecordBatchWithMetadata result type, mirroring pyarrow's
read_next_batch_with_custom_metadata() and the equivalent Arrow C++ struct:
RecordBatchWithMetadata ReadNextRecordBatchWithCustomMetadata();
ValueTask<RecordBatchWithMetadata> ReadNextRecordBatchWithCustomMetadataAsync(...);
A property that has to be read at exactly the right moment is easy to get out
of step with the batch in hand, and it had no sensible value at the end of the
stream. Pairing the two in the return value removes both problems and reads the
same in the sync and async APIs. The struct deconstructs, so callers who want
the pair can write `var (batch, metadata) = ...`.
ArrowFileReader gains ReadRecordBatchWithCustomMetadataAsync(int index) so the
indexed read has the same capability as the sequential one.
The transient state on ArrowReaderImplementation stays, but it is internal and
consumed immediately by the two new methods rather than being public surface.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XhMo3XSWYHo1apHd9PzZTb
Contributor
|
Thanks @CurtHagenlocher I've just been away from this for a little while. |
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
3 participants
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.
Adds IPC
Message.custom_metadatasupport to the stream and file readers and writers.Credit
The original implementation is @cmettler's work in #283 (closing #282). @rustyconover rebased it onto a
mainthat had drifted ~76 commits ahead and addressed the first round of review comments in #424. Both of those PRs are branches on personal forks; this one moves the work onto a branch in the base repository so that any committer can push to it, and supersedes them. The individual commits here retain their original authorship.What this adds
ArrowStreamReader.ReadNextRecordBatchWithCustomMetadata()and…Async()return aRecordBatchWithMetadatapairing the batch with its IPCMessage.custom_metadata, mirroring pyarrow'sread_next_batch_with_custom_metadata()and the equivalent Arrow C++ struct. The struct deconstructs, so callers can writevar (batch, metadata) = ….ArrowFileReader.ReadRecordBatchWithCustomMetadataAsync(int index)gives the indexed read the same capability as the sequential one.ArrowStreamWriter.WriteRecordBatch(batch, customMetadata)and the async counterpart attach per-messagecustom_metadatawhen writing, matching pyarrow'swrite_batch(batch, custom_metadata).PYTHONNET_PYDLLis set, consistent with the existingCDataSchemaPythonTestpattern.Changes on top of #424
LastBatchCustomMetadataproperty. A property that has to be read at exactly the right moment is easy to get out of step with the batch in hand, and it had no sensible value at the end of the stream, so it was replaced with theRecordBatchWithMetadatareturn type above. The transient state onArrowReaderImplementationremains, but it is internal and consumed immediately.WriteRecordBatchInternal, whileArrowFileWriterrelied on overriding each publicWriteRecordBatchto callWriteStart()first — so the new overload on anArrowFileWriterskipped the ARROW1 magic and silently produced a file thatArrowFileReaderrejects.WriteStart()/WriteStartAsync()moved intoWriteRecordBatchInternal, where every write path must pass through it; both are idempotent, so byte output is unchanged for the stream writer, the file writer and Flight.WriteMessageAsyncoverload. Two virtual overloads where one forwards to the other is the trap that already routed Flight's record batch writes pastFlightDataStream's override.Message.custom_metadatais read the same way schema and field metadata already are inMessageSerializer, rather than skipping null keys and rewriting null values as"".Verification
dotnet build Apache.Arrow.sln— succeeds, 0 warnings, 0 errors.dotnet test test/Apache.Arrow.Tests— 1893 passed / 30 skipped on net8.0, 1849 passed / 30 skipped on net462 and net472, 0 failed. The skips are the pre-existing Python-dependent tests.Closes #282
Supersedes #283
Supersedes #424
🤖 Generated with Claude Code
https://claude.ai/code/session_01DT86mdkGm3XseKwiUeGUcx
Co-Authored-By: Christoph Mettler 116812500+cmettler@users.noreply.github.com
Co-Authored-By: Rusty Conover rusty@conover.me