This is an automated email from the ASF dual-hosted git repository.
paleolimbot pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-nanoarrow.git
The following commit(s) were added to refs/heads/main by this push:
new 3f824063 feat(ipc): Support custom_metadata on Message for IPC (#924)
3f824063 is described below
commit 3f824063f59848e05692ab520de8ab4d9ebb1880
Author: Rusty Conover <[email protected]>
AuthorDate: Sat Aug 22 21:52:09 2026 -0400
feat(ipc): Support custom_metadata on Message for IPC (#924)
Closes #923.
Adds encoder and decoder support for `Message.custom_metadata` — the
per-message
key/value metadata that is separate from the Schema/Field metadata
carried in the schema.
### Encoder
```c
ArrowErrorCode ArrowIpcEncoderSetMessageMetadata(struct ArrowIpcEncoder*
encoder,
struct ArrowBuffer*
metadata,
struct ArrowError* error);
```
`metadata` holds nanoarrow's packed representation (the same as
`ArrowSchema.metadata`, so
`ArrowMetadataBuilder*` can produce it and no new representation is
introduced). The
encoder takes ownership by moving the buffer, so the common case doesn't
copy. It applies
to the *next* message encoded by `ArrowIpcEncoderEncodeSchema()` or
`ArrowIpcEncoderEncodeSimpleRecordBatch()` and is cleared once that
message is encoded.
`NULL` (or metadata with no keys) clears it, and no `custom_metadata`
field is written at
all in that case, so the encoded bytes are unchanged for callers that
don't use this.
`ArrowIpcEncodeMetadata()` now takes the packed `const char* metadata`
instead of a
`const struct ArrowSchema*` so it is reusable from the Message path;
behaviour at the
Schema/Field call sites is unchanged.
### Decoder
```c
// Copy, using the same packed representation
ArrowErrorCode ArrowIpcDecoderGetMessageMetadata(struct ArrowIpcDecoder*
decoder,
struct ArrowBuffer* out,
struct ArrowError* error);
// Zero copy: both borrow from the message header until the next header is
decoded
ArrowErrorCode ArrowIpcDecoderGetMessageMetadataValue(struct
ArrowIpcDecoder* decoder,
struct
ArrowStringView key,
struct
ArrowStringView* value_out,
struct ArrowError*
error);
ArrowErrorCode ArrowIpcDecoderVisitMessageMetadata(struct ArrowIpcDecoder*
decoder,
ArrowIpcMetadataVisitFunction visit,
void* private_data,
struct ArrowError*
error);
```
Per @paleolimbot's review comment on the issue, the copying accessor
writes to a
`struct ArrowBuffer*` (what `ArrowMetadataBuilder*` builds), and the
get-by-name and
visit-in-place variants avoid the copy for large metadata. All three are
implemented on
top of one internal KeyValue-vector visitor, which also backs the
packing used for Schema
and Field metadata. If the message has no
`custom_metadata`, `out` is left empty so `(const char*)out->data` is
`NULL` and can be
passed straight to `ArrowSchemaSetMetadata()` /
`ArrowMetadataReaderInit()`. A missing key
leaves `value_out` unmodified, matching `ArrowMetadataGetValue()`.
The metadata is recorded by both `ArrowIpcDecoderVerifyHeader()` and
`ArrowIpcDecoderDecodeHeader()` and applies to any message type, since
`Message.custom_metadata` is a field of `Message` rather than of
`RecordBatch`.
### Tests
- `encoder_test.cc`: round trip of message metadata through encoder →
decoder for both
RecordBatch and Schema messages, that Schema-message metadata stays
distinct from the
contained schema's own metadata, that the metadata applies to exactly
one message, that
empty/`NULL`/cleared metadata produces byte-identical output to not
calling the setter,
and that an erroring visitor stops the visit and propagates its error.
- `decoder_test.cc`: interop with Arrow C++ in both directions —
`arrow::ipc::ReadMessage()`
reads back what nanoarrow writes, and nanoarrow reads the metadata
written by
`RecordBatchWriter::WriteRecordBatch(batch, custom_metadata)`.
Verified locally against Arrow C++ 25.0.1
(`-DNANOARROW_BUILD_TESTS_WITH_ARROW=ON`).
### Also fixed here: a crash decoding metadata with an absent key
`KeyValue.key` and `KeyValue.value` are optional fields in `Schema.fbs`
(the generated
verifier agrees: `flatcc_verify_string_field(td, 0, 0)`), so a
flatbuffer-verified message
may contain a `KeyValue` with no key at all. The decoder used `strlen()`
on it, so
`ArrowIpcDecoderDecodeSchema()` segfaulted (SIGSEGV, confirmed on
`main`) on input that had
passed verification — reachable from any untrusted stream via
`ArrowIpcArrayStreamReader`.
Consolidating the iteration replaced `strlen()` with
`flatbuffers_string_len()`, which is
NULL-safe, so an absent key or value now decodes as an empty string. The
same change stops
metadata values containing embedded nulls from being silently truncated,
which the packed
representation round-trips fine. `NanoarrowIpcDecodeMetadataWithoutKey`
covers it and
crashes without the fix.
Happy to split this into its own PR if you'd rather keep this one purely
additive — it
landed here because it's the same lines the consolidation rewrites.
### Out of scope
`ArrowIpcArrayStreamReader` / `ArrowIpcWriter` are unchanged:
`ArrowArrayStream` has
nowhere to put per-batch metadata, so as discussed in the issue this
first pass exposes the
feature only at the encoder/decoder layer.
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
src/nanoarrow/ipc/decoder.c | 172 ++++++++++++++++----
src/nanoarrow/ipc/decoder_test.cc | 194 ++++++++++++++++++++++
src/nanoarrow/ipc/encoder.c | 78 ++++++++-
src/nanoarrow/ipc/encoder_test.cc | 329 ++++++++++++++++++++++++++++++++++++++
src/nanoarrow/nanoarrow_ipc.h | 86 ++++++++++
5 files changed, 824 insertions(+), 35 deletions(-)
diff --git a/src/nanoarrow/ipc/decoder.c b/src/nanoarrow/ipc/decoder.c
index c9687f0c..b13d0bd6 100644
--- a/src/nanoarrow/ipc/decoder.c
+++ b/src/nanoarrow/ipc/decoder.c
@@ -81,6 +81,10 @@ struct ArrowIpcDecoderPrivate {
int64_t n_union_fields;
// A pointer to the last flatbuffers message.
const void* last_message;
+ // The custom_metadata of the last flatbuffers Message (i.e.,
Message.custom_metadata,
+ // which is distinct from the metadata of the Schema or Field it may
contain). This is
+ // NULL if the last message had no custom_metadata.
+ ns(KeyValue_vec_t) last_message_metadata;
// Storage for a DictionaryBatch
struct ArrowIpcDictionaryBatch dictionary;
// Storage for a Footer
@@ -640,50 +644,90 @@ static inline int32_t ArrowIpcReadInt32LE(struct
ArrowBufferView* data, int swap
return value;
}
-static int ArrowIpcDecoderSetMetadata(struct ArrowSchema* schema,
- ns(KeyValue_vec_t) kv_vec,
- struct ArrowError* error) {
+// Returned by an internal ArrowIpcMetadataVisitFunction to stop iterating
early.
+// This is never returned to a caller of the public API.
+#define _NANOARROW_IPC_VISIT_STOP (-1)
+
+// Visits each key/value pair in a flatbuffers vector of KeyValue.
+//
+// Keys and values point into the message and are passed with an explicit size
because
+// both are optional fields whose content may contain embedded nulls; a
KeyValue with no
+// key or no value is visited with an empty string view.
+static ArrowErrorCode ArrowIpcDecoderVisitMetadata(ns(KeyValue_vec_t) kv_vec,
+
ArrowIpcMetadataVisitFunction visit,
+ void* private_data,
+ struct ArrowError* error) {
int64_t n_pairs = ns(KeyValue_vec_len(kv_vec));
- if (n_pairs == 0) {
- return NANOARROW_OK;
+
+ for (int64_t i = 0; i < n_pairs; i++) {
+ ns(KeyValue_table_t) kv = ns(KeyValue_vec_at(kv_vec, i));
+ flatbuffers_string_t key = ns(KeyValue_key(kv));
+ flatbuffers_string_t value = ns(KeyValue_value(kv));
+
+ struct ArrowStringView key_view = {key ? key : "",
+ (int64_t)flatbuffers_string_len(key)};
+ struct ArrowStringView value_view = {value ? value : "",
+
(int64_t)flatbuffers_string_len(value)};
+ NANOARROW_RETURN_NOT_OK(visit(key_view, value_view, private_data, error));
+ }
+
+ return NANOARROW_OK;
+}
+
+static ArrowErrorCode ArrowIpcDecoderAppendMetadata(struct ArrowStringView key,
+ struct ArrowStringView
value,
+ void* private_data,
+ struct ArrowError* error) {
+ int result = ArrowMetadataBuilderAppend((struct ArrowBuffer*)private_data,
key, value);
+ if (result != NANOARROW_OK) {
+ ArrowErrorSet(error, "ArrowMetadataBuilderAppend() failed");
+ }
+
+ return result;
+}
+
+// Packs a flatbuffers vector of KeyValue into nanoarrow's metadata
representation.
+// out is initialized by this function and will be empty if kv_vec contains no
pairs.
+static int ArrowIpcDecoderBuildMetadata(ns(KeyValue_vec_t) kv_vec,
+ struct ArrowBuffer* out,
+ struct ArrowError* error) {
+ int result = ArrowMetadataBuilderInit(out, NULL);
+ if (result != NANOARROW_OK) {
+ ArrowBufferReset(out);
+ ArrowErrorSet(error, "ArrowMetadataBuilderInit() failed");
+ return result;
}
+ int64_t n_pairs = ns(KeyValue_vec_len(kv_vec));
if (n_pairs > 2147483647) {
+ ArrowBufferReset(out);
ArrowErrorSet(error,
"Expected between 0 and 2147483647 key/value pairs but found
%" PRId64,
n_pairs);
return EINVAL;
}
- struct ArrowBuffer buf;
- struct ArrowStringView key;
- struct ArrowStringView value;
- ns(KeyValue_table_t) kv;
-
- int result = ArrowMetadataBuilderInit(&buf, NULL);
+ result =
+ ArrowIpcDecoderVisitMetadata(kv_vec, &ArrowIpcDecoderAppendMetadata,
out, error);
if (result != NANOARROW_OK) {
- ArrowBufferReset(&buf);
- ArrowErrorSet(error, "ArrowMetadataBuilderInit() failed");
+ ArrowBufferReset(out);
return result;
}
- for (int64_t i = 0; i < n_pairs; i++) {
- kv = ns(KeyValue_vec_at(kv_vec, i));
-
- key.data = ns(KeyValue_key(kv));
- key.size_bytes = strlen(key.data);
- value.data = ns(KeyValue_value(kv));
- value.size_bytes = strlen(value.data);
+ return NANOARROW_OK;
+}
- result = ArrowMetadataBuilderAppend(&buf, key, value);
- if (result != NANOARROW_OK) {
- ArrowBufferReset(&buf);
- ArrowErrorSet(error, "ArrowMetadataBuilderAppend() failed");
- return result;
- }
+static int ArrowIpcDecoderSetMetadata(struct ArrowSchema* schema,
+ ns(KeyValue_vec_t) kv_vec,
+ struct ArrowError* error) {
+ if (ns(KeyValue_vec_len(kv_vec)) == 0) {
+ return NANOARROW_OK;
}
- result = ArrowSchemaSetMetadata(schema, (const char*)buf.data);
+ struct ArrowBuffer buf;
+ NANOARROW_RETURN_NOT_OK(ArrowIpcDecoderBuildMetadata(kv_vec, &buf, error));
+
+ int result = ArrowSchemaSetMetadata(schema, (const char*)buf.data);
ArrowBufferReset(&buf);
if (result != NANOARROW_OK) {
ArrowErrorSet(error, "ArrowSchemaSetMetadata() failed");
@@ -1434,6 +1478,7 @@ static inline void ArrowIpcDecoderResetHeaderInfo(struct
ArrowIpcDecoder* decode
decoder->footer = NULL;
ArrowIpcFooterReset(&private_data->footer);
private_data->last_message = NULL;
+ private_data->last_message_metadata = NULL;
}
// Returns NANOARROW_OK if data is large enough to read the first 8 bytes
@@ -1536,6 +1581,7 @@ ArrowErrorCode ArrowIpcDecoderVerifyHeader(struct
ArrowIpcDecoder* decoder,
decoder->body_size_bytes = ns(Message_bodyLength(message));
private_data->last_message = ns(Message_header_get(message));
+ private_data->last_message_metadata = ns(Message_custom_metadata(message));
return NANOARROW_OK;
}
@@ -1692,9 +1738,81 @@ ArrowErrorCode ArrowIpcDecoderDecodeHeader(struct
ArrowIpcDecoder* decoder,
}
private_data->last_message = message_header;
+ private_data->last_message_metadata = ns(Message_custom_metadata(message));
return NANOARROW_OK;
}
+ArrowErrorCode ArrowIpcDecoderGetMessageMetadata(struct ArrowIpcDecoder*
decoder,
+ struct ArrowBuffer* out,
+ struct ArrowError* error) {
+ NANOARROW_DCHECK(decoder != NULL && decoder->private_data != NULL && out !=
NULL);
+ struct ArrowIpcDecoderPrivate* private_data =
+ (struct ArrowIpcDecoderPrivate*)decoder->private_data;
+
+ return ArrowIpcDecoderBuildMetadata(private_data->last_message_metadata,
out, error);
+}
+
+struct ArrowIpcMetadataValueLookup {
+ struct ArrowStringView key;
+ struct ArrowStringView* value_out;
+};
+
+static ArrowErrorCode ArrowIpcDecoderMatchMetadataKey(struct ArrowStringView
key,
+ struct ArrowStringView
value,
+ void* private_data,
+ struct ArrowError*
error) {
+ NANOARROW_UNUSED(error);
+ struct ArrowIpcMetadataValueLookup* lookup =
+ (struct ArrowIpcMetadataValueLookup*)private_data;
+
+ if (key.size_bytes != lookup->key.size_bytes) {
+ return NANOARROW_OK;
+ }
+
+ if (key.size_bytes > 0 &&
+ memcmp(key.data, lookup->key.data, (size_t)key.size_bytes) != 0) {
+ return NANOARROW_OK;
+ }
+
+ *lookup->value_out = value;
+ return _NANOARROW_IPC_VISIT_STOP;
+}
+
+ArrowErrorCode ArrowIpcDecoderGetMessageMetadataValue(struct ArrowIpcDecoder*
decoder,
+ struct ArrowStringView
key,
+ struct ArrowStringView*
value_out,
+ struct ArrowError*
error) {
+ NANOARROW_DCHECK(decoder != NULL && decoder->private_data != NULL &&
value_out != NULL);
+ struct ArrowIpcDecoderPrivate* private_data =
+ (struct ArrowIpcDecoderPrivate*)decoder->private_data;
+
+ struct ArrowIpcMetadataValueLookup lookup;
+ lookup.key = key;
+ lookup.value_out = value_out;
+
+ int result =
+ ArrowIpcDecoderVisitMetadata(private_data->last_message_metadata,
+ &ArrowIpcDecoderMatchMetadataKey, &lookup,
error);
+ if (result == _NANOARROW_IPC_VISIT_STOP) {
+ // key was found and value_out was set
+ return NANOARROW_OK;
+ }
+
+ return result;
+}
+
+ArrowErrorCode ArrowIpcDecoderVisitMessageMetadata(struct ArrowIpcDecoder*
decoder,
+
ArrowIpcMetadataVisitFunction visit,
+ void* private_data,
+ struct ArrowError* error) {
+ NANOARROW_DCHECK(decoder != NULL && decoder->private_data != NULL && visit
!= NULL);
+ struct ArrowIpcDecoderPrivate* decoder_private =
+ (struct ArrowIpcDecoderPrivate*)decoder->private_data;
+
+ return ArrowIpcDecoderVisitMetadata(decoder_private->last_message_metadata,
visit,
+ private_data, error);
+}
+
static ArrowErrorCode ArrowIpcDecoderDecodeSchemaImpl(
ns(Schema_table_t) schema, struct ArrowSchema* out,
struct ArrowIpcDictionaryEncodings* dictionary_encodings_out,
diff --git a/src/nanoarrow/ipc/decoder_test.cc
b/src/nanoarrow/ipc/decoder_test.cc
index 900bc32f..6adf1e36 100644
--- a/src/nanoarrow/ipc/decoder_test.cc
+++ b/src/nanoarrow/ipc/decoder_test.cc
@@ -22,6 +22,7 @@
#include <arrow/array.h>
#include <arrow/c/bridge.h>
#include <arrow/extension/uuid.h>
+#include <arrow/io/memory.h>
#include <arrow/ipc/api.h>
#include <arrow/util/key_value_metadata.h>
#endif
@@ -1381,6 +1382,62 @@ void AssertArrayViewIdentical(const struct
ArrowArrayView* actual,
}
}
+// A Message whose Message.custom_metadata and Schema.custom_metadata each
contain a
+// KeyValue with no key. Both `key` and `value` are optional fields of
KeyValue in the
+// IPC format, so this passes flatbuffer verification and must not crash the
decoder.
+// Generated with flatcc: a Schema message with no fields, whose two metadata
vectors
+// each contain KeyValue{value: "message_value" / "schema_value"} and no key.
+alignas(8) static uint8_t kKeylessMetadataSchema[] = {
+ 0xff, 0xff, 0xff, 0xff, 0x90, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00,
0x86, 0xff,
+ 0xff, 0xff, 0x04, 0x00, 0x01, 0x00, 0x2c, 0x00, 0x00, 0x00, 0x04, 0x00,
0x00, 0x00,
+ 0x01, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0xb0, 0xff, 0xff, 0xff,
0x04, 0x00,
+ 0x00, 0x00, 0x0d, 0x00, 0x00, 0x00, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67,
0x65, 0x5f,
+ 0x76, 0x61, 0x6c, 0x75, 0x65, 0x00, 0x00, 0x00, 0xc4, 0xff, 0xff, 0xff,
0x2c, 0x00,
+ 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x04, 0x00,
0x00, 0x00,
+ 0xe0, 0xff, 0xff, 0xff, 0x04, 0x00, 0x00, 0x00, 0x0c, 0x00, 0x00, 0x00,
0x73, 0x63,
+ 0x68, 0x65, 0x6d, 0x61, 0x5f, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x00, 0x00,
0x00, 0x00,
+ 0x00, 0x00, 0x00, 0x00, 0x08, 0x00, 0x08, 0x00, 0x00, 0x00, 0x04, 0x00,
0x0a, 0x00,
+ 0x0c, 0x00, 0x00, 0x00, 0x04, 0x00, 0x08, 0x00, 0x0e, 0x00, 0x10, 0x00,
0x04, 0x00,
+ 0x06, 0x00, 0x08, 0x00, 0x00, 0x00, 0x0c, 0x00, 0x00, 0x00, 0x00, 0x00};
+
+TEST(NanoarrowIpcTest, NanoarrowIpcDecodeMetadataWithoutKey) {
+ struct ArrowBufferView data;
+ data.data.as_uint8 = kKeylessMetadataSchema;
+ data.size_bytes = sizeof(kKeylessMetadataSchema);
+
+ struct ArrowError error;
+ nanoarrow::ipc::UniqueDecoder decoder;
+ ASSERT_EQ(ArrowIpcDecoderInit(decoder.get()), NANOARROW_OK);
+ ASSERT_EQ(ArrowIpcDecoderVerifyHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(ArrowIpcDecoderDecodeHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+
+ // A missing key is decoded as an empty key
+ nanoarrow::UniqueBuffer message_metadata;
+ ASSERT_EQ(
+ ArrowIpcDecoderGetMessageMetadata(decoder.get(), message_metadata.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(
+ ArrowSchemaMetadataToString(reinterpret_cast<const
char*>(message_metadata->data)),
+ "=message_value");
+
+ struct ArrowStringView value = ArrowCharView(nullptr);
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadataValue(decoder.get(),
ArrowCharView(""),
+ &value, &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(std::string(value.data, value.size_bytes), "message_value");
+
+ // ...including the metadata of the schema the message contains
+ nanoarrow::UniqueSchema schema;
+ ASSERT_EQ(ArrowIpcDecoderDecodeSchema(decoder.get(), schema.get(), &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(ArrowSchemaMetadataToString(schema->metadata), "=schema_value");
+}
+
#if defined(NANOARROW_BUILD_TESTS_WITH_ARROW)
TEST_P(ArrowTypeParameterizedTestFixture, NanoarrowIpcNanoarrowArrayRoundtrip)
{
if (GetParam()->id() == arrow::Type::DICTIONARY) {
@@ -1706,6 +1763,143 @@ INSTANTIATE_TEST_SUITE_P(
// arrow::KeyValueMetadata::Make({"key1", "key2"}, {"value1",
"value2"}))})
));
+// Advance data past the message whose header was just decoded
+static void AdvancePastMessage(struct ArrowIpcDecoder* decoder,
+ struct ArrowBufferView* data) {
+ int64_t message_size_bytes =
_ArrowRoundUpToMultipleOf8(decoder->header_size_bytes) +
+
_ArrowRoundUpToMultipleOf8(decoder->body_size_bytes);
+ ASSERT_LE(message_size_bytes, data->size_bytes);
+ data->data.as_uint8 += message_size_bytes;
+ data->size_bytes -= message_size_bytes;
+}
+
+TEST(NanoarrowIpcTest, NanoarrowIpcMessageMetadataArrowInterop) {
+ auto arrow_schema = arrow::schema({arrow::field("some_name",
arrow::int32())});
+ auto custom_metadata =
+ arrow::KeyValueMetadata::Make({"key1", "key2"}, {"value1", "value2"});
+
+ auto maybe_batch = arrow::RecordBatch::MakeEmpty(arrow_schema);
+ ASSERT_TRUE(maybe_batch.ok()) << maybe_batch.status();
+ auto batch = maybe_batch.ValueUnsafe();
+
+ struct ArrowError error;
+ nanoarrow::UniqueSchema schema;
+ nanoarrow::UniqueArray array;
+ nanoarrow::UniqueArrayView array_view;
+ ASSERT_TRUE(arrow::ExportSchema(*arrow_schema, schema.get()).ok());
+ ASSERT_TRUE(arrow::ExportRecordBatch(*batch, array.get()).ok());
+ ASSERT_EQ(ArrowArrayViewInitFromSchema(array_view.get(), schema.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(ArrowArrayViewSetArray(array_view.get(), array.get(), &error),
NANOARROW_OK)
+ << error.message;
+
+ // What nanoarrow writes, Arrow C++ can read
+ nanoarrow::UniqueBuffer metadata;
+ ASSERT_EQ(ArrowMetadataBuilderInit(metadata.get(), nullptr), NANOARROW_OK);
+ for (int64_t i = 0; i < custom_metadata->size(); i++) {
+ ASSERT_EQ(ArrowMetadataBuilderAppend(
+ metadata.get(),
ArrowCharView(custom_metadata->key(i).c_str()),
+ ArrowCharView(custom_metadata->value(i).c_str())),
+ NANOARROW_OK);
+ }
+
+ nanoarrow::ipc::UniqueEncoder encoder;
+ ASSERT_EQ(ArrowIpcEncoderInit(encoder.get()), NANOARROW_OK);
+ ASSERT_EQ(ArrowIpcEncoderSetMessageMetadata(encoder.get(), metadata.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+
+ nanoarrow::UniqueBuffer nanoarrow_message;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleRecordBatch(encoder.get(),
array_view.get(),
+ nanoarrow_message.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ {
+ // Prepend the encapsulated header to the body written above
+ nanoarrow::UniqueBuffer body;
+ ArrowBufferMove(nanoarrow_message.get(), body.get());
+ ASSERT_EQ(ArrowIpcEncoderFinalizeBuffer(encoder.get(),
/*encapsulate=*/true,
+ nanoarrow_message.get()),
+ NANOARROW_OK);
+ ASSERT_EQ(
+ ArrowBufferAppendBufferView(nanoarrow_message.get(),
+ ArrowBufferView{{body->data},
body->size_bytes}),
+ NANOARROW_OK);
+ }
+
+ arrow::io::BufferReader message_reader(
+ arrow::Buffer::Wrap(nanoarrow_message->data,
nanoarrow_message->size_bytes));
+ auto maybe_message = arrow::ipc::ReadMessage(&message_reader);
+ ASSERT_TRUE(maybe_message.ok()) << maybe_message.status();
+ ASSERT_NE(maybe_message.ValueUnsafe()->custom_metadata(), nullptr);
+
EXPECT_TRUE(maybe_message.ValueUnsafe()->custom_metadata()->Equals(*custom_metadata));
+
+ // What Arrow C++ writes, nanoarrow can read
+ auto maybe_out = arrow::io::BufferOutputStream::Create();
+ ASSERT_TRUE(maybe_out.ok()) << maybe_out.status();
+ auto out = maybe_out.ValueUnsafe();
+
+ auto maybe_writer = arrow::ipc::MakeStreamWriter(out, arrow_schema);
+ ASSERT_TRUE(maybe_writer.ok()) << maybe_writer.status();
+ auto writer = maybe_writer.ValueUnsafe();
+ ASSERT_TRUE(writer->WriteRecordBatch(*batch, custom_metadata).ok());
+ ASSERT_TRUE(writer->Close().ok());
+
+ auto maybe_stream = out->Finish();
+ ASSERT_TRUE(maybe_stream.ok()) << maybe_stream.status();
+ auto stream = maybe_stream.ValueUnsafe();
+
+ struct ArrowBufferView data;
+ data.data.data = stream->data();
+ data.size_bytes = stream->size();
+
+ nanoarrow::ipc::UniqueDecoder decoder;
+ ASSERT_EQ(ArrowIpcDecoderInit(decoder.get()), NANOARROW_OK);
+
+ // The Schema message has no metadata of its own
+ ASSERT_EQ(ArrowIpcDecoderVerifyHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(ArrowIpcDecoderDecodeHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(decoder->message_type, NANOARROW_IPC_MESSAGE_TYPE_SCHEMA);
+ nanoarrow::UniqueBuffer schema_message_metadata;
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadata(decoder.get(),
+ schema_message_metadata.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(schema_message_metadata->size_bytes, 0);
+ ASSERT_EQ(ArrowIpcDecoderSetSchema(decoder.get(), schema.get(), &error),
NANOARROW_OK)
+ << error.message;
+
+ // ...but the RecordBatch message does
+ ASSERT_NO_FATAL_FAILURE(AdvancePastMessage(decoder.get(), &data));
+ ASSERT_EQ(ArrowIpcDecoderVerifyHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(ArrowIpcDecoderDecodeHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(decoder->message_type, NANOARROW_IPC_MESSAGE_TYPE_RECORD_BATCH);
+
+ nanoarrow::UniqueBuffer batch_message_metadata;
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadata(decoder.get(),
batch_message_metadata.get(),
+ &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(ArrowSchemaMetadataToString(
+ reinterpret_cast<const char*>(batch_message_metadata->data)),
+ "key1=value1, key2=value2");
+
+ for (int64_t i = 0; i < custom_metadata->size(); i++) {
+ struct ArrowStringView value = ArrowCharView(nullptr);
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadataValue(
+ decoder.get(),
ArrowCharView(custom_metadata->key(i).c_str()), &value,
+ &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(std::string(value.data, value.size_bytes),
custom_metadata->value(i));
+ }
+}
+
class ArrowTypeIdParameterizedTestFixture
: public ::testing::TestWithParam<enum ArrowType> {
protected:
diff --git a/src/nanoarrow/ipc/encoder.c b/src/nanoarrow/ipc/encoder.c
index d959bfa9..2c12a1b5 100644
--- a/src/nanoarrow/ipc/encoder.c
+++ b/src/nanoarrow/ipc/encoder.c
@@ -46,6 +46,9 @@ struct ArrowIpcEncoderPrivate {
struct ArrowBuffer nodes;
int encoding_footer;
struct ArrowIpcDictionaryEncodings dictionary_encodings;
+ // Metadata to attach to the next encoded Message (in nanoarrow's packed
+ // representation), or an empty buffer if the next Message has no metadata.
+ struct ArrowBuffer message_metadata;
};
ArrowErrorCode ArrowIpcEncoderInit(struct ArrowIpcEncoder* encoder) {
@@ -65,6 +68,7 @@ ArrowErrorCode ArrowIpcEncoderInit(struct ArrowIpcEncoder*
encoder) {
ArrowBufferInit(&private->buffers);
ArrowBufferInit(&private->nodes);
ArrowIpcDictionaryEncodingsInit(&private->dictionary_encodings);
+ ArrowBufferInit(&private->message_metadata);
return NANOARROW_OK;
}
@@ -77,11 +81,44 @@ void ArrowIpcEncoderReset(struct ArrowIpcEncoder* encoder) {
ArrowBufferReset(&private->nodes);
ArrowBufferReset(&private->buffers);
ArrowIpcDictionaryEncodingsReset(&private->dictionary_encodings);
+ ArrowBufferReset(&private->message_metadata);
ArrowFree(private);
}
memset(encoder, 0, sizeof(struct ArrowIpcEncoder));
}
+ArrowErrorCode ArrowIpcEncoderSetMessageMetadata(struct ArrowIpcEncoder*
encoder,
+ struct ArrowBuffer* metadata,
+ struct ArrowError* error) {
+ NANOARROW_DCHECK(encoder != NULL && encoder->private_data != NULL);
+ struct ArrowIpcEncoderPrivate* private =
+ (struct ArrowIpcEncoderPrivate*)encoder->private_data;
+
+ // Any previously set metadata that was not yet encoded is discarded
+ ArrowBufferReset(&private->message_metadata);
+
+ if (metadata != NULL) {
+ ArrowBufferMove(metadata, &private->message_metadata);
+ }
+
+ // Metadata that can't contain a key count is empty; metadata with no keys is
+ // equivalent to no metadata at all. In both cases no custom_metadata is
encoded.
+ if (private->message_metadata.size_bytes < (int64_t)sizeof(int32_t)) {
+ ArrowBufferReset(&private->message_metadata);
+ return NANOARROW_OK;
+ }
+
+ struct ArrowMetadataReader reader;
+ NANOARROW_RETURN_NOT_OK_WITH_ERROR(
+ ArrowMetadataReaderInit(&reader, (const
char*)private->message_metadata.data),
+ error);
+ if (reader.remaining_keys <= 0) {
+ ArrowBufferReset(&private->message_metadata);
+ }
+
+ return NANOARROW_OK;
+}
+
static ArrowErrorCode ArrowIpcEncoderWriteContinuationAndSize(struct
ArrowBuffer* out,
size_t size) {
_NANOARROW_CHECK_UPPER_LIMIT(size, INT32_MAX);
@@ -346,13 +383,13 @@ static ArrowErrorCode ArrowIpcEncodeField(
struct ArrowError* error);
static ArrowErrorCode ArrowIpcEncodeMetadata(flatcc_builder_t* builder,
- const struct ArrowSchema* schema,
+ const char* packed_metadata,
int
(*push_start)(flatcc_builder_t*),
ns(KeyValue_ref_t) *
(*push_end)(flatcc_builder_t*),
struct ArrowError* error) {
struct ArrowMetadataReader metadata;
- NANOARROW_RETURN_NOT_OK_WITH_ERROR(ArrowMetadataReaderInit(&metadata,
schema->metadata),
+ NANOARROW_RETURN_NOT_OK_WITH_ERROR(ArrowMetadataReaderInit(&metadata,
packed_metadata),
error);
while (metadata.remaining_keys > 0) {
struct ArrowStringView key, value;
@@ -368,6 +405,27 @@ static ArrowErrorCode
ArrowIpcEncodeMetadata(flatcc_builder_t* builder,
return NANOARROW_OK;
}
+// Encodes any metadata set by ArrowIpcEncoderSetMessageMetadata() into the
Message
+// currently under construction and clears it, such that it applies to exactly
one
+// Message.
+static ArrowErrorCode ArrowIpcEncodeMessageMetadata(
+ struct ArrowIpcEncoderPrivate* private, struct ArrowError* error) {
+ if (private->message_metadata.size_bytes == 0) {
+ return NANOARROW_OK;
+ }
+
+ flatcc_builder_t* builder = &private->builder;
+ FLATCC_RETURN_UNLESS_0(Message_custom_metadata_start(builder), error);
+ NANOARROW_RETURN_NOT_OK(
+ ArrowIpcEncodeMetadata(builder, (const
char*)private->message_metadata.data,
+ &ns(Message_custom_metadata_push_start),
+ &ns(Message_custom_metadata_push_end), error));
+ FLATCC_RETURN_UNLESS_0(Message_custom_metadata_end(builder), error);
+
+ ArrowBufferReset(&private->message_metadata);
+ return NANOARROW_OK;
+}
+
static ArrowErrorCode ArrowIpcEncodeFields(
flatcc_builder_t* builder, const struct ArrowSchema* schema,
int (*push_start)(flatcc_builder_t*),
@@ -476,9 +534,9 @@ static ArrowErrorCode ArrowIpcEncodeField(
if (schema->metadata) {
FLATCC_RETURN_UNLESS_0(Field_custom_metadata_start(builder), error);
- NANOARROW_RETURN_NOT_OK(
- ArrowIpcEncodeMetadata(builder, schema,
&ns(Field_custom_metadata_push_start),
- &ns(Field_custom_metadata_push_end), error));
+ NANOARROW_RETURN_NOT_OK(ArrowIpcEncodeMetadata(
+ builder, schema->metadata, &ns(Field_custom_metadata_push_start),
+ &ns(Field_custom_metadata_push_end), error));
FLATCC_RETURN_UNLESS_0(Field_custom_metadata_end(builder), error);
}
return NANOARROW_OK;
@@ -512,9 +570,9 @@ static ArrowErrorCode ArrowIpcEncodeSchema(
FLATCC_RETURN_UNLESS_0(Schema_custom_metadata_start(builder), error);
if (schema->metadata) {
- NANOARROW_RETURN_NOT_OK(
- ArrowIpcEncodeMetadata(builder, schema,
&ns(Schema_custom_metadata_push_start),
- &ns(Schema_custom_metadata_push_end), error));
+ NANOARROW_RETURN_NOT_OK(ArrowIpcEncodeMetadata(
+ builder, schema->metadata, &ns(Schema_custom_metadata_push_start),
+ &ns(Schema_custom_metadata_push_end), error));
}
FLATCC_RETURN_UNLESS_0(Schema_custom_metadata_end(builder), error);
@@ -554,6 +612,8 @@ ArrowErrorCode ArrowIpcEncoderEncodeSchema(struct
ArrowIpcEncoder* encoder,
FLATCC_RETURN_UNLESS_0(Message_header_Schema_end(builder), error);
+ NANOARROW_RETURN_NOT_OK(ArrowIpcEncodeMessageMetadata(private, error));
+
FLATCC_RETURN_UNLESS_0(Message_bodyLength_add(builder, 0), error);
FLATCC_RETURN_IF_NULL(ns(Message_end_as_root(builder)), error);
@@ -700,6 +760,8 @@ static ArrowErrorCode ArrowIpcEncoderEncodeRecordBatch(
FLATCC_RETURN_UNLESS_0(Message_header_RecordBatch_end(builder), error);
+ NANOARROW_RETURN_NOT_OK(ArrowIpcEncodeMessageMetadata(private, error));
+
FLATCC_RETURN_UNLESS_0(Message_bodyLength_add(builder,
buffer_encoder->body_length),
error);
FLATCC_RETURN_IF_NULL(ns(Message_end_as_root(builder)), error);
diff --git a/src/nanoarrow/ipc/encoder_test.cc
b/src/nanoarrow/ipc/encoder_test.cc
index 2e4bcca9..3eeff1ca 100644
--- a/src/nanoarrow/ipc/encoder_test.cc
+++ b/src/nanoarrow/ipc/encoder_test.cc
@@ -17,6 +17,10 @@
#include <gtest/gtest.h>
+#include <string>
+#include <utility>
+#include <vector>
+
#include "flatcc/flatcc_builder.h"
#include "nanoarrow/nanoarrow.hpp"
#include "nanoarrow/nanoarrow_ipc.hpp"
@@ -101,3 +105,328 @@ TEST(NanoarrowIpcTest, NanoarrowIpcFooterEncoding) {
EXPECT_GT(footer_buffer->size_bytes, raw_schema_buffer->size_bytes);
}
+
+using KeyValues = std::vector<std::pair<std::string, std::string>>;
+
+// Unpack nanoarrow's metadata representation into something comparable
+static KeyValues UnpackMetadata(const char* metadata) {
+ struct ArrowMetadataReader reader;
+ NANOARROW_THROW_NOT_OK(ArrowMetadataReaderInit(&reader, metadata));
+
+ KeyValues out;
+ while (reader.remaining_keys > 0) {
+ struct ArrowStringView key, value;
+ NANOARROW_THROW_NOT_OK(ArrowMetadataReaderRead(&reader, &key, &value));
+ out.emplace_back(std::string(key.data, key.size_bytes),
+ std::string(value.data, value.size_bytes));
+ }
+ return out;
+}
+
+static nanoarrow::UniqueBuffer PackMetadata(const KeyValues& key_values) {
+ nanoarrow::UniqueBuffer metadata;
+ NANOARROW_THROW_NOT_OK(ArrowMetadataBuilderInit(metadata.get(), nullptr));
+ for (const auto& kv : key_values) {
+ NANOARROW_THROW_NOT_OK(ArrowMetadataBuilderAppend(metadata.get(),
+
ArrowCharView(kv.first.c_str()),
+
ArrowCharView(kv.second.c_str())));
+ }
+ return metadata;
+}
+
+static ArrowErrorCode CollectKeyValue(struct ArrowStringView key,
+ struct ArrowStringView value, void*
private_data,
+ struct ArrowError* error) {
+ NANOARROW_UNUSED(error);
+ static_cast<KeyValues*>(private_data)
+ ->emplace_back(std::string(key.data, key.size_bytes),
+ std::string(value.data, value.size_bytes));
+ return NANOARROW_OK;
+}
+
+// Decodes the header of an encapsulated message and returns its
Message.custom_metadata
+static KeyValues DecodeMessageMetadata(struct ArrowBuffer* message,
+ struct ArrowIpcDecoder* decoder) {
+ struct ArrowBufferView view;
+ view.data.data = message->data;
+ view.size_bytes = message->size_bytes;
+
+ struct ArrowError error;
+ NANOARROW_THROW_NOT_OK(ArrowIpcDecoderVerifyHeader(decoder, view, &error));
+ NANOARROW_THROW_NOT_OK(ArrowIpcDecoderDecodeHeader(decoder, view, &error));
+
+ nanoarrow::UniqueBuffer metadata;
+ NANOARROW_THROW_NOT_OK(
+ ArrowIpcDecoderGetMessageMetadata(decoder, metadata.get(), &error));
+ KeyValues out = UnpackMetadata(reinterpret_cast<const
char*>(metadata->data));
+
+ // The visitor should see exactly the same pairs, in the same order
+ KeyValues visited;
+ NANOARROW_THROW_NOT_OK(
+ ArrowIpcDecoderVisitMessageMetadata(decoder, &CollectKeyValue, &visited,
&error));
+ EXPECT_EQ(visited, out);
+
+ return out;
+}
+
+// A struct array view with no columns and no rows: the smallest valid
RecordBatch
+class SimpleRecordBatch {
+ public:
+ SimpleRecordBatch() {
+ NANOARROW_THROW_NOT_OK(ArrowSchemaInitFromType(schema_.get(),
NANOARROW_TYPE_STRUCT));
+ NANOARROW_THROW_NOT_OK(
+ ArrowArrayInitFromSchema(array_.get(), schema_.get(), nullptr));
+ NANOARROW_THROW_NOT_OK(
+ ArrowArrayViewInitFromSchema(array_view_.get(), schema_.get(),
nullptr));
+ NANOARROW_THROW_NOT_OK(
+ ArrowArrayViewSetArray(array_view_.get(), array_.get(), nullptr));
+ }
+
+ struct ArrowSchema* schema() { return schema_.get(); }
+ const struct ArrowArrayView* array_view() { return array_view_.get(); }
+
+ private:
+ nanoarrow::UniqueSchema schema_;
+ nanoarrow::UniqueArray array_;
+ nanoarrow::UniqueArrayView array_view_;
+};
+
+TEST(NanoarrowIpcTest, NanoarrowIpcEncoderMessageMetadataRoundtrip) {
+ nanoarrow::ipc::UniqueEncoder encoder;
+ ASSERT_EQ(ArrowIpcEncoderInit(encoder.get()), NANOARROW_OK);
+ nanoarrow::ipc::UniqueDecoder decoder;
+ ASSERT_EQ(ArrowIpcDecoderInit(decoder.get()), NANOARROW_OK);
+
+ SimpleRecordBatch batch;
+ struct ArrowError error;
+ ASSERT_EQ(ArrowIpcDecoderSetSchema(decoder.get(), batch.schema(), &error),
NANOARROW_OK)
+ << error.message;
+
+ KeyValues key_values{{"message_type", "data"}, {"cache-control",
"no-store"}};
+ auto metadata = PackMetadata(key_values);
+ ASSERT_EQ(ArrowIpcEncoderSetMessageMetadata(encoder.get(), metadata.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+
+ // The encoder took ownership of the metadata
+ EXPECT_EQ(metadata->data, nullptr);
+ EXPECT_EQ(metadata->size_bytes, 0);
+
+ nanoarrow::UniqueBuffer message, body;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleRecordBatch(encoder.get(),
batch.array_view(),
+ body.get(), &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcEncoderFinalizeBuffer(encoder.get(), /*encapsulate=*/true,
message.get()),
+ NANOARROW_OK);
+
+ EXPECT_EQ(DecodeMessageMetadata(message.get(), decoder.get()), key_values);
+
+ // Values can also be read in place, without copying
+ struct ArrowStringView value = ArrowCharView(nullptr);
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadataValue(
+ decoder.get(), ArrowCharView("cache-control"), &value, &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(std::string(value.data, value.size_bytes), "no-store");
+
+ // A key that isn't present leaves value_out untouched
+ value = ArrowCharView(nullptr);
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadataValue(
+ decoder.get(), ArrowCharView("not-a-key"), &value, &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(value.data, nullptr);
+
+ // A key which is a prefix of a present key is not a match
+ value = ArrowCharView(nullptr);
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadataValue(decoder.get(),
ArrowCharView("cache"),
+ &value, &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(value.data, nullptr);
+
+ // The metadata applied to exactly one message: the next one has none
+ message->size_bytes = 0;
+ body->size_bytes = 0;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleRecordBatch(encoder.get(),
batch.array_view(),
+ body.get(), &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcEncoderFinalizeBuffer(encoder.get(), /*encapsulate=*/true,
message.get()),
+ NANOARROW_OK);
+ EXPECT_EQ(DecodeMessageMetadata(message.get(), decoder.get()), KeyValues{});
+}
+
+TEST(NanoarrowIpcTest, NanoarrowIpcEncoderSchemaMessageMetadata) {
+ nanoarrow::ipc::UniqueEncoder encoder;
+ ASSERT_EQ(ArrowIpcEncoderInit(encoder.get()), NANOARROW_OK);
+ nanoarrow::ipc::UniqueDecoder decoder;
+ ASSERT_EQ(ArrowIpcDecoderInit(decoder.get()), NANOARROW_OK);
+
+ SimpleRecordBatch batch;
+ struct ArrowError error;
+
+ // Message metadata is distinct from the metadata of the Schema it contains
+ auto schema_metadata = PackMetadata({{"schema_key", "schema_value"}});
+ ASSERT_EQ(ArrowSchemaSetMetadata(batch.schema(),
+ reinterpret_cast<const
char*>(schema_metadata->data)),
+ NANOARROW_OK);
+
+ KeyValues message_key_values{{"message_key", "message_value"}};
+ auto message_metadata = PackMetadata(message_key_values);
+ ASSERT_EQ(
+ ArrowIpcEncoderSetMessageMetadata(encoder.get(), message_metadata.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+
+ nanoarrow::UniqueBuffer message;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSchema(encoder.get(), batch.schema(), &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcEncoderFinalizeBuffer(encoder.get(), /*encapsulate=*/true,
message.get()),
+ NANOARROW_OK);
+
+ EXPECT_EQ(DecodeMessageMetadata(message.get(), decoder.get()),
message_key_values);
+
+ nanoarrow::UniqueSchema roundtripped;
+ ASSERT_EQ(ArrowIpcDecoderDecodeSchema(decoder.get(), roundtripped.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(UnpackMetadata(roundtripped->metadata),
+ (KeyValues{{"schema_key", "schema_value"}}));
+}
+
+TEST(NanoarrowIpcTest, NanoarrowIpcEncoderMessageMetadataEmpty) {
+ nanoarrow::ipc::UniqueEncoder encoder;
+ ASSERT_EQ(ArrowIpcEncoderInit(encoder.get()), NANOARROW_OK);
+ nanoarrow::ipc::UniqueDecoder decoder;
+ ASSERT_EQ(ArrowIpcDecoderInit(decoder.get()), NANOARROW_OK);
+
+ SimpleRecordBatch batch;
+ struct ArrowError error;
+ ASSERT_EQ(ArrowIpcDecoderSetSchema(decoder.get(), batch.schema(), &error),
NANOARROW_OK)
+ << error.message;
+
+ // Metadata with no keys, NULL metadata, and no call at all are all
equivalent
+ auto empty_metadata = PackMetadata({});
+ auto keyless_metadata = PackMetadata({{"key", "value"}});
+ ASSERT_EQ(ArrowMetadataBuilderRemove(keyless_metadata.get(),
ArrowCharView("key")),
+ NANOARROW_OK);
+
+ nanoarrow::UniqueBuffer baseline, body;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleRecordBatch(encoder.get(),
batch.array_view(),
+ body.get(), &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcEncoderFinalizeBuffer(encoder.get(), /*encapsulate=*/true,
baseline.get()),
+ NANOARROW_OK);
+
+ for (struct ArrowBuffer* metadata :
+ {empty_metadata.get(), keyless_metadata.get(), (struct
ArrowBuffer*)nullptr}) {
+ ASSERT_EQ(ArrowIpcEncoderSetMessageMetadata(encoder.get(), metadata,
&error),
+ NANOARROW_OK)
+ << error.message;
+
+ nanoarrow::UniqueBuffer message;
+ body->size_bytes = 0;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleRecordBatch(encoder.get(),
batch.array_view(),
+ body.get(), &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcEncoderFinalizeBuffer(encoder.get(), /*encapsulate=*/true,
message.get()),
+ NANOARROW_OK);
+
+ // No custom_metadata field is written at all
+ ASSERT_EQ(message->size_bytes, baseline->size_bytes);
+ EXPECT_EQ(memcmp(message->data, baseline->data, message->size_bytes), 0);
+
+ EXPECT_EQ(DecodeMessageMetadata(message.get(), decoder.get()),
KeyValues{});
+ }
+
+ // Setting metadata and then clearing it encodes nothing
+ auto metadata = PackMetadata({{"key", "value"}});
+ ASSERT_EQ(ArrowIpcEncoderSetMessageMetadata(encoder.get(), metadata.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(ArrowIpcEncoderSetMessageMetadata(encoder.get(), nullptr, &error),
+ NANOARROW_OK)
+ << error.message;
+
+ nanoarrow::UniqueBuffer message;
+ body->size_bytes = 0;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleRecordBatch(encoder.get(),
batch.array_view(),
+ body.get(), &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcEncoderFinalizeBuffer(encoder.get(), /*encapsulate=*/true,
message.get()),
+ NANOARROW_OK);
+ ASSERT_EQ(message->size_bytes, baseline->size_bytes);
+ EXPECT_EQ(memcmp(message->data, baseline->data, message->size_bytes), 0);
+
+ // Reading in place from a message without metadata finds nothing
+ struct ArrowStringView value = ArrowCharView(nullptr);
+ ASSERT_EQ(ArrowIpcDecoderGetMessageMetadataValue(decoder.get(),
ArrowCharView("key"),
+ &value, &error),
+ NANOARROW_OK)
+ << error.message;
+ EXPECT_EQ(value.data, nullptr);
+}
+
+TEST(NanoarrowIpcTest, NanoarrowIpcVisitMessageMetadataError) {
+ nanoarrow::ipc::UniqueEncoder encoder;
+ ASSERT_EQ(ArrowIpcEncoderInit(encoder.get()), NANOARROW_OK);
+ nanoarrow::ipc::UniqueDecoder decoder;
+ ASSERT_EQ(ArrowIpcDecoderInit(decoder.get()), NANOARROW_OK);
+
+ SimpleRecordBatch batch;
+ struct ArrowError error;
+ ASSERT_EQ(ArrowIpcDecoderSetSchema(decoder.get(), batch.schema(), &error),
NANOARROW_OK)
+ << error.message;
+
+ auto metadata = PackMetadata({{"key1", "value1"}, {"key2", "value2"}});
+ ASSERT_EQ(ArrowIpcEncoderSetMessageMetadata(encoder.get(), metadata.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+
+ nanoarrow::UniqueBuffer message, body;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleRecordBatch(encoder.get(),
batch.array_view(),
+ body.get(), &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcEncoderFinalizeBuffer(encoder.get(), /*encapsulate=*/true,
message.get()),
+ NANOARROW_OK);
+
+ struct ArrowBufferView view;
+ view.data.data = message->data;
+ view.size_bytes = message->size_bytes;
+ ASSERT_EQ(ArrowIpcDecoderDecodeHeader(decoder.get(), view, &error),
NANOARROW_OK)
+ << error.message;
+
+ // A visitor which errors stops the visit and its error code is returned
+ struct Visitor {
+ static ArrowErrorCode Visit(struct ArrowStringView key, struct
ArrowStringView value,
+ void* private_data, struct ArrowError* error) {
+ NANOARROW_UNUSED(value);
+ auto* visited = static_cast<KeyValues*>(private_data);
+ ArrowErrorSet(error, "visitor stopped at %.*s",
static_cast<int>(key.size_bytes),
+ key.data);
+ NANOARROW_RETURN_NOT_OK(CollectKeyValue(key, value, private_data,
nullptr));
+ return visited->size() == 1 ? ENOTSUP : NANOARROW_OK;
+ }
+ };
+
+ KeyValues visited;
+ EXPECT_EQ(ArrowIpcDecoderVisitMessageMetadata(decoder.get(),
&Visitor::Visit, &visited,
+ &error),
+ ENOTSUP);
+ EXPECT_EQ(visited, (KeyValues{{"key1", "value1"}}));
+ EXPECT_STREQ(error.message, "visitor stopped at key1");
+}
diff --git a/src/nanoarrow/nanoarrow_ipc.h b/src/nanoarrow/nanoarrow_ipc.h
index 72327730..fb38750c 100644
--- a/src/nanoarrow/nanoarrow_ipc.h
+++ b/src/nanoarrow/nanoarrow_ipc.h
@@ -66,6 +66,12 @@
NANOARROW_SYMBOL(NANOARROW_NAMESPACE,
ArrowIpcDecoderSetSchemaWithDictionaries)
#define ArrowIpcDecoderSetEndianness \
NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcDecoderSetEndianness)
+#define ArrowIpcDecoderGetMessageMetadata \
+ NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcDecoderGetMessageMetadata)
+#define ArrowIpcDecoderGetMessageMetadataValue \
+ NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcDecoderGetMessageMetadataValue)
+#define ArrowIpcDecoderVisitMessageMetadata \
+ NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcDecoderVisitMessageMetadata)
#define ArrowIpcDecoderPeekFooter \
NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcDecoderPeekFooter)
#define ArrowIpcDecoderVerifyFooter \
@@ -84,6 +90,8 @@
#define ArrowIpcEncoderReset NANOARROW_SYMBOL(NANOARROW_NAMESPACE,
ArrowIpcEncoderReset)
#define ArrowIpcEncoderFinalizeBuffer \
NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcEncoderFinalizeBuffer)
+#define ArrowIpcEncoderSetMessageMetadata \
+ NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcEncoderSetMessageMetadata)
#define ArrowIpcEncoderEncodeSchema \
NANOARROW_SYMBOL(NANOARROW_NAMESPACE, ArrowIpcEncoderEncodeSchema)
#define ArrowIpcEncoderEncodeSimpleRecordBatch \
@@ -461,6 +469,20 @@ NANOARROW_DLL void ArrowIpcDecoderReset(struct
ArrowIpcDecoder* decoder);
NANOARROW_DLL ArrowErrorCode ArrowIpcDecoderSetDecompressor(
struct ArrowIpcDecoder* decoder, struct ArrowIpcDecompressor*
decompressor);
+/// \brief Callback invoked for each key/value pair by
+/// ArrowIpcDecoderVisitMessageMetadata()
+///
+/// Both key and value are borrowed and are passed with an explicit size
because
+/// neither is required to be present nor free of embedded nulls: a key or
value that
+/// the message omits is passed as an empty string view.
+///
+/// Returning any value other than NANOARROW_OK will stop the visit and cause
that
+/// value to be returned by ArrowIpcDecoderVisitMessageMetadata().
+typedef ArrowErrorCode (*ArrowIpcMetadataVisitFunction)(struct ArrowStringView
key,
+ struct ArrowStringView
value,
+ void* private_data,
+ struct ArrowError*
error);
+
/// \brief Peek at a message header
///
/// The first 8 bytes of an Arrow IPC message are 0xFFFFFFFF followed by the
size
@@ -510,6 +532,52 @@ NANOARROW_DLL ArrowErrorCode
ArrowIpcDecoderDecodeHeader(struct ArrowIpcDecoder*
struct
ArrowBufferView data,
struct ArrowError*
error);
+/// \brief Get the custom metadata of the most recently decoded message
+///
+/// After a successful call to ArrowIpcDecoderVerifyHeader() or
+/// ArrowIpcDecoderDecodeHeader(), copy the message's custom_metadata into out
using
+/// the same representation as ArrowSchema::metadata. Note that this is the
metadata
+/// attached to the message itself and is distinct from the metadata of a
Schema or
+/// Field that the message may contain.
+///
+/// out is initialized by this call (i.e., it must not contain data) and the
caller
+/// is responsible for calling ArrowBufferReset(). If the message had no custom
+/// metadata, out will be empty (i.e., out->data will be NULL) such that
+/// (const char*)out->data can be passed to ArrowSchemaSetMetadata() or
+/// ArrowMetadataReaderInit().
+///
+/// Returns ENOMEM if allocation fails, EINVAL if the metadata cannot be
decoded, or
+/// NANOARROW_OK otherwise.
+NANOARROW_DLL ArrowErrorCode ArrowIpcDecoderGetMessageMetadata(
+ struct ArrowIpcDecoder* decoder, struct ArrowBuffer* out, struct
ArrowError* error);
+
+/// \brief Get one value from the custom metadata of the most recently decoded
message
+///
+/// Unlike ArrowIpcDecoderGetMessageMetadata(), this does not copy: the value
returned
+/// points into the message header passed to ArrowIpcDecoderVerifyHeader() or
+/// ArrowIpcDecoderDecodeHeader() and is only valid until that data is
invalidated or
+/// another message header is decoded.
+///
+/// If key occurs more than once, the first value is returned. If key does not
occur,
+/// value_out is left unmodified: initialize it with ArrowCharView(NULL) and
check
+/// value_out->data for NULL to detect a missing key.
+NANOARROW_DLL ArrowErrorCode ArrowIpcDecoderGetMessageMetadataValue(
+ struct ArrowIpcDecoder* decoder, struct ArrowStringView key,
+ struct ArrowStringView* value_out, struct ArrowError* error);
+
+/// \brief Visit each key/value pair in the most recently decoded message's
metadata
+///
+/// Like ArrowIpcDecoderGetMessageMetadataValue(), the keys and values passed
to visit
+/// point into the message header passed to ArrowIpcDecoderVerifyHeader() or
+/// ArrowIpcDecoderDecodeHeader() and must not be retained beyond the lifetime
of that
+/// data. private_data and error are passed to each invocation of visit.
+///
+/// Returns the first non-NANOARROW_OK value returned by visit, or
NANOARROW_OK if all
+/// pairs were visited.
+NANOARROW_DLL ArrowErrorCode ArrowIpcDecoderVisitMessageMetadata(
+ struct ArrowIpcDecoder* decoder, ArrowIpcMetadataVisitFunction visit,
+ void* private_data, struct ArrowError* error);
+
/// \brief Decode an ArrowSchema
///
/// After a successful call to ArrowIpcDecoderDecodeHeader(), retrieve an
ArrowSchema.
@@ -778,6 +846,24 @@ NANOARROW_DLL void ArrowIpcEncoderReset(struct
ArrowIpcEncoder* encoder);
NANOARROW_DLL ArrowErrorCode ArrowIpcEncoderFinalizeBuffer(
struct ArrowIpcEncoder* encoder, char encapsulate, struct ArrowBuffer*
out);
+/// \brief Set the custom metadata of the next encoded message
+///
+/// Attaches metadata to the next message encoded by
ArrowIpcEncoderEncodeSchema() or
+/// ArrowIpcEncoderEncodeSimpleRecordBatch() (i.e., Message::custom_metadata,
which is
+/// distinct from the metadata of the Schema or Field that the message may
contain).
+/// The metadata applies to exactly one message: after a message is encoded the
+/// encoder's message metadata is cleared. Any metadata that was set but not
yet
+/// encoded is replaced by this call; pass NULL to clear it.
+///
+/// metadata contains the same representation as ArrowSchema::metadata, as
built by
+/// ArrowMetadataBuilderInit()/ArrowMetadataBuilderAppend(). The encoder takes
ownership
+/// of it: unless metadata is NULL it is moved into the encoder and left
empty. Metadata
+/// containing no keys is equivalent to no metadata at all and no
custom_metadata is
+/// encoded for it.
+NANOARROW_DLL ArrowErrorCode
+ArrowIpcEncoderSetMessageMetadata(struct ArrowIpcEncoder* encoder,
+ struct ArrowBuffer* metadata, struct
ArrowError* error);
+
/// \brief Encode an ArrowSchema
///
/// Returns ENOMEM if allocation fails, NANOARROW_OK otherwise.
