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 b8a0d3ca feat: Decode dictionary delta batches (#955)
b8a0d3ca is described below
commit b8a0d3cad6a79c7dcfa4271930c6b16f938a768c
Author: Rusty Conover <[email protected]>
AuthorDate: Fri Oct 16 10:19:02 2026 -0400
feat: Decode dictionary delta batches (#955)
Split out of #928 as requested in review. This PR only adds decoding of
dictionary delta batches; the writer changes remain in #928.
Previously, `ArrowIpcDecoderDecodeDictionary()` failed with "Dictionary
concatenation is not yet supported" when it got a DictionaryBatch with
`isDelta` set and the current dictionary was non-empty. Now the delta
values are appended: a new array is built from the current dictionary
value and the delta using `ArrowArrayAppendStorageFromArrayView()`
(#930), and it replaces the stored dictionary. Arrays decoded earlier
keep referencing their previous dictionary value. Value types that
`ArrowArrayAppendStorageFromArrayView()` can't append (e.g., unions)
fail with `ENOTSUP`.
Changes from the review of #928:
- Removed the string/binary-view-specific handling (IPC doesn't support
views yet, #824).
- Replaced the `if (result == NANOARROW_OK)` chains with a helper that
doesn't own its inputs, plus a single check and release.
- Dropped the "try to unshare" optimization and the internal
`common/utils.c` helper. Each delta copies the dictionary; appending in
place when nothing else references the dictionary can come in a
follow-up.
- Tests use `enum ArrowType` for parameters, and the dense union
rejection test is removed.
This also fixes a double release on the error path of
`ArrowIpcDecoderDecodeDictionaryInternal()`: `ArrowArrayMoveShared()`
already releases its input on failure, so the temporary is now only
released if it is still live.
Tests: `NanoarrowIpcDecodeDictionaryBatch` covers deltas, including a
clone held across a delta keeping its old value. With
`NANOARROW_BUILD_TESTS_WITH_ARROW`, streams written by Arrow C++ with
`emit_dictionary_deltas = true` are read back for bool, int64 (with
nulls), uint64, double, string, binary, decimal128, list, struct and
fixed-size list values. Locally (aarch64 Linux, Debug, ZSTD/LZ4), ctest
passes with and without Arrow C++ 21.0.0, and the decoder tests are
clean under valgrind.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
src/nanoarrow/ipc/decoder.c | 82 ++++++++-
src/nanoarrow/ipc/decoder_test.cc | 366 +++++++++++++++++++++++++++++++++++++-
2 files changed, 434 insertions(+), 14 deletions(-)
diff --git a/src/nanoarrow/ipc/decoder.c b/src/nanoarrow/ipc/decoder.c
index 8953cb29..80e53ec8 100644
--- a/src/nanoarrow/ipc/decoder.c
+++ b/src/nanoarrow/ipc/decoder.c
@@ -337,16 +337,82 @@ static ArrowErrorCode ArrowIpcDictionaryReplace(struct
ArrowIpcDictionary* dicti
return NANOARROW_OK;
}
+// Point dictionaries of dst (and its children) at those of src, which reflect
+// the most recent value of any nested dictionaries.
+static ArrowErrorCode ArrowIpcArraySetDictionaries(struct ArrowArray* dst,
+ const struct ArrowArray*
src) {
+ if (src->dictionary != NULL) {
+ NANOARROW_DCHECK(dst->dictionary != NULL);
+ if (dst->dictionary->release != NULL) {
+ ArrowArrayRelease(dst->dictionary);
+ }
+ NANOARROW_RETURN_NOT_OK(ArrowArrayCloneShared(src->dictionary,
dst->dictionary));
+ }
+
+ for (int64_t i = 0; i < src->n_children; i++) {
+ NANOARROW_RETURN_NOT_OK(
+ ArrowIpcArraySetDictionaries(dst->children[i], src->children[i]));
+ }
+
+ return NANOARROW_OK;
+}
+
+// Append the values of current and then value into combined, which must have
been
+// initialized from array_view. array_view is used as scratch space to view
current
+// and value.
+static ArrowErrorCode ArrowIpcDictionaryConcatenate(struct ArrowArray*
combined,
+ const struct ArrowArray*
current,
+ const struct ArrowArray*
value,
+ struct ArrowArrayView*
array_view,
+ struct ArrowError* error) {
+ if (current->length > INT64_MAX - value->length) {
+ ArrowErrorSet(error,
+ "Expected dictionary length plus delta length to fit in
int64 "
+ "but found %" PRId64 " and %" PRId64,
+ current->length, value->length);
+ return EOVERFLOW;
+ }
+
+ NANOARROW_RETURN_NOT_OK_WITH_ERROR(ArrowArrayStartAppending(combined),
error);
+ NANOARROW_RETURN_NOT_OK_WITH_ERROR(
+ ArrowArrayReserve(combined, current->length + value->length), error);
+
+ NANOARROW_RETURN_NOT_OK(ArrowArrayViewSetArray(array_view, current, error));
+ NANOARROW_RETURN_NOT_OK(
+ ArrowArrayAppendStorageFromArrayView(combined, array_view, error));
+ NANOARROW_RETURN_NOT_OK(ArrowArrayViewSetArray(array_view, value, error));
+ NANOARROW_RETURN_NOT_OK(
+ ArrowArrayAppendStorageFromArrayView(combined, array_view, error));
+
+ NANOARROW_RETURN_NOT_OK_WITH_ERROR(ArrowIpcArraySetDictionaries(combined,
value),
+ error);
+ return ArrowArrayFinishBuildingDefault(combined, error);
+}
+
+// On success, value is released and the dictionary's current value is replaced
+// with the concatenation of its previous value and value. Previously exported
+// clones of the dictionary keep referencing the previous value.
static ArrowErrorCode ArrowIpcDictionaryAppend(struct ArrowIpcDictionary*
dictionary,
struct ArrowArray* value,
+ struct ArrowArrayView*
array_view,
struct ArrowError* error) {
- if (dictionary->current_value.release != NULL &&
- dictionary->current_value.length != 0) {
- ArrowErrorSet(error, "Dictionary concatenation is not yet supported");
- return ENOTSUP;
+ if (dictionary->current_value.release == NULL ||
+ dictionary->current_value.length == 0) {
+ return ArrowIpcDictionaryReplace(dictionary, value, error);
}
- NANOARROW_RETURN_NOT_OK(ArrowIpcDictionaryReplace(dictionary, value, error));
+ struct ArrowArray combined;
+ NANOARROW_RETURN_NOT_OK(ArrowArrayInitFromArrayView(&combined, array_view,
error));
+ int result = ArrowIpcDictionaryConcatenate(&combined,
&dictionary->current_value, value,
+ array_view, error);
+ if (result != NANOARROW_OK) {
+ ArrowArrayRelease(&combined);
+ return result;
+ }
+
+ // On failure, ArrowIpcDictionaryReplace() releases combined
+ NANOARROW_RETURN_NOT_OK(ArrowIpcDictionaryReplace(dictionary, &combined,
error));
+ ArrowArrayRelease(value);
return NANOARROW_OK;
}
@@ -2824,13 +2890,15 @@ static ArrowErrorCode
ArrowIpcDecoderDecodeDictionaryInternal(
}
if (decoder->dictionary->is_delta) {
- result = ArrowIpcDictionaryAppend(dictionary, &tmp, error);
+ result = ArrowIpcDictionaryAppend(dictionary, &tmp, array_view, error);
} else {
result = ArrowIpcDictionaryReplace(dictionary, &tmp, error);
}
if (result != NANOARROW_OK) {
- ArrowArrayRelease(&tmp);
+ if (tmp.release != NULL) {
+ ArrowArrayRelease(&tmp);
+ }
return result;
}
diff --git a/src/nanoarrow/ipc/decoder_test.cc
b/src/nanoarrow/ipc/decoder_test.cc
index 3fb2bf6e..c7ac78ce 100644
--- a/src/nanoarrow/ipc/decoder_test.cc
+++ b/src/nanoarrow/ipc/decoder_test.cc
@@ -20,10 +20,15 @@
#if defined(NANOARROW_BUILD_TESTS_WITH_ARROW)
#include <arrow/array.h>
+#include <arrow/array/builder_binary.h>
+#include <arrow/array/builder_decimal.h>
+#include <arrow/array/builder_nested.h>
+#include <arrow/array/builder_primitive.h>
#include <arrow/c/bridge.h>
#include <arrow/extension/uuid.h>
#include <arrow/io/memory.h>
#include <arrow/ipc/api.h>
+#include <arrow/util/decimal.h>
#include <arrow/util/key_value_metadata.h>
#endif
#include <gmock/gmock-matchers.h>
@@ -851,12 +856,42 @@ TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryBatch)
{
ASSERT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 1), "one"_asv);
ASSERT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 2), "two"_asv);
- // If we try to decode a delta dictionary, we should fail with a reasonable
message
+ // A delta dictionary appends its values to the current dictionary
const_cast<struct ArrowIpcDictionaryBatch*>(decoder.dictionary)->is_delta =
1;
ASSERT_EQ(ArrowIpcDecoderDecodeDictionary(
&decoder, body, NANOARROW_VALIDATION_LEVEL_FULL,
&dictionaries, &error),
- ENOTSUP);
- ASSERT_STREQ(error.message, "Dictionary concatenation is not yet supported");
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcDictionariesFindCurrentValue(&dictionaries, 0,
&dictionary_value, &error),
+ NANOARROW_OK);
+ ASSERT_EQ(ArrowArrayViewSetArray(&array_view, dictionary_value, &error),
NANOARROW_OK);
+ ASSERT_EQ(array_view.length, 6);
+ EXPECT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 0), "zero"_asv);
+ EXPECT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 3), "zero"_asv);
+ EXPECT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 5), "two"_asv);
+
+ // A clone of the dictionary held across another delta keeps its previous
value
+ struct ArrowArray snapshot;
+ ASSERT_EQ(
+ ArrowArrayCloneShared(const_cast<struct ArrowArray*>(dictionary_value),
&snapshot),
+ NANOARROW_OK);
+ ASSERT_EQ(ArrowIpcDecoderDecodeDictionary(
+ &decoder, body, NANOARROW_VALIDATION_LEVEL_FULL,
&dictionaries, &error),
+ NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(
+ ArrowIpcDictionariesFindCurrentValue(&dictionaries, 0,
&dictionary_value, &error),
+ NANOARROW_OK);
+ ASSERT_EQ(ArrowArrayViewSetArray(&array_view, dictionary_value, &error),
NANOARROW_OK);
+ ASSERT_EQ(array_view.length, 9);
+ EXPECT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 6), "zero"_asv);
+ EXPECT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 8), "two"_asv);
+
+ ASSERT_EQ(ArrowArrayViewSetArray(&array_view, &snapshot, &error),
NANOARROW_OK);
+ ASSERT_EQ(array_view.length, 6);
+ EXPECT_EQ(ArrowArrayViewGetStringUnsafe(&array_view, 5), "two"_asv);
+ ArrowArrayRelease(&snapshot);
// After all of this, we should be able to actually decode a RecordBatch
ASSERT_EQ(ArrowIpcDecoderSetSchemaWithDictionaries(&decoder, &schema,
@@ -868,6 +903,14 @@ TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryBatch) {
data.data.as_uint8 += decoder.header_size_bytes;
data.size_bytes -= decoder.header_size_bytes;
+ // The convenience APIs without a dictionary memo must reject
dictionary-encoded
+ // fields instead of returning an array with an unresolved dictionary.
+ struct ArrowArrayView* unresolved_view;
+ ASSERT_EQ(ArrowIpcDecoderDecodeArrayView(&decoder, data, 0,
&unresolved_view, &error),
+ ENOTSUP);
+ EXPECT_STREQ(error.message,
+ "Can't decode a dictionary-encoded field without
ArrowIpcDictionaries");
+
// Decode the entire batch and check the dictionary
struct ArrowArrayView* batch_view;
ASSERT_EQ(ArrowIpcDecoderDecodeArrayViewWithDictionaries(
@@ -876,7 +919,7 @@ TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryBatch) {
<< error.message;
ASSERT_NE(batch_view->children[0]->dictionary, nullptr);
- ASSERT_EQ(batch_view->children[0]->dictionary->length, 3);
+ ASSERT_EQ(batch_view->children[0]->dictionary->length, 9);
ASSERT_EQ(ArrowArrayViewGetStringUnsafe(batch_view->children[0]->dictionary,
0),
"zero"_asv);
@@ -888,7 +931,7 @@ TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryBatch) {
<< error.message;
ASSERT_NE(column_view->dictionary, nullptr);
- ASSERT_EQ(column_view->dictionary->length, 3);
+ ASSERT_EQ(column_view->dictionary->length, 9);
ASSERT_EQ(ArrowArrayViewGetStringUnsafe(column_view->dictionary, 0),
"zero"_asv);
// Decode the array from the ArrowBufferView
@@ -899,7 +942,7 @@ TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryBatch) {
NANOARROW_OK)
<< error.message;
ASSERT_NE(batch.children[0]->dictionary, nullptr);
- ASSERT_EQ(batch.children[0]->dictionary->length, 3);
+ ASSERT_EQ(batch.children[0]->dictionary->length, 9);
ArrowArrayRelease(&batch);
// Decode the array from a shared buffer
@@ -917,7 +960,7 @@ TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryBatch) {
NANOARROW_OK)
<< error.message;
ASSERT_NE(batch.children[0]->dictionary, nullptr);
- ASSERT_EQ(batch.children[0]->dictionary->length, 3);
+ ASSERT_EQ(batch.children[0]->dictionary->length, 9);
ArrowArrayRelease(&batch);
ArrowBufferReset(&record_batch_shared);
@@ -928,6 +971,107 @@ TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryBatch)
{
ArrowIpcDecoderReset(&decoder);
}
+TEST(NanoarrowIpcTest, NanoarrowIpcDecodeDictionaryDeltaOverflow) {
+ struct ArrowError error;
+ nanoarrow::UniqueSchema schema;
+ ASSERT_EQ(ArrowSchemaInitFromType(schema.get(), NANOARROW_TYPE_INT8),
NANOARROW_OK);
+ ASSERT_EQ(ArrowSchemaAllocateDictionary(schema.get()), NANOARROW_OK);
+ ASSERT_EQ(ArrowSchemaInitFromType(schema->dictionary, NANOARROW_TYPE_NA),
NANOARROW_OK);
+
+ nanoarrow::ipc::UniqueDictionaryEncodings encodings;
+ struct ArrowIpcDictionaryEncoding encoding;
+ encoding.id = 0;
+ encoding.kind = NANOARROW_IPC_DICTIONARY_KIND_DENSE_ARRAY;
+ encoding.schema = schema.get();
+ ASSERT_EQ(ArrowIpcDictionaryEncodingsAppend(encodings.get(), encoding),
NANOARROW_OK);
+
+ nanoarrow::ipc::UniqueEncoder encoder;
+ ASSERT_EQ(ArrowIpcEncoderInit(encoder.get()), NANOARROW_OK);
+ nanoarrow::ipc::UniqueDecoder decoder;
+ ASSERT_EQ(ArrowIpcDecoderInit(decoder.get()), NANOARROW_OK);
+
+ // Null dictionaries can have enormous lengths without allocating data
buffers.
+ const int64_t lengths[][2] = {{INT64_MAX, 1}, {1, INT64_MAX}};
+ for (bool use_shared : {false, true}) {
+ for (const auto& length : lengths) {
+ SCOPED_TRACE(::testing::Message() << "shared=" << use_shared << ",
current="
+ << length[0] << ", delta=" <<
length[1]);
+ nanoarrow::ipc::UniqueDictionaries dictionaries;
+ ASSERT_EQ(ArrowIpcDictionariesInit(dictionaries.get(), encodings.get(),
&error),
+ NANOARROW_OK);
+
+ auto decode_nulls = [&](int64_t size, bool is_delta, ArrowErrorCode
expected) {
+ nanoarrow::UniqueArray array;
+ ASSERT_EQ(ArrowArrayInitFromType(array.get(), NANOARROW_TYPE_NA),
NANOARROW_OK);
+ ASSERT_EQ(ArrowArrayStartAppending(array.get()), NANOARROW_OK);
+ ASSERT_EQ(ArrowArrayAppendNull(array.get(), size), NANOARROW_OK);
+ ASSERT_EQ(ArrowArrayFinishBuildingDefault(array.get(), &error),
NANOARROW_OK);
+ nanoarrow::UniqueArrayView values;
+ ArrowArrayViewInitFromType(values.get(), NANOARROW_TYPE_NA);
+ ASSERT_EQ(ArrowArrayViewSetArray(values.get(), array.get(), &error),
+ NANOARROW_OK);
+ nanoarrow::UniqueBuffer body;
+ ASSERT_EQ(ArrowIpcEncoderEncodeSimpleDictionaryBatch(
+ encoder.get(), 0, is_delta, values.get(), body.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ nanoarrow::UniqueBuffer message;
+ ASSERT_EQ(ArrowIpcEncoderFinalizeBuffer(encoder.get(), true,
message.get()),
+ NANOARROW_OK);
+ struct ArrowBufferView data;
+ data.data.as_uint8 = message->data;
+ data.size_bytes = message->size_bytes;
+ ASSERT_EQ(ArrowIpcDecoderVerifyHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(ArrowIpcDecoderDecodeHeader(decoder.get(), data, &error),
NANOARROW_OK)
+ << error.message;
+
+ if (use_shared) {
+ nanoarrow::UniqueBuffer shared_body;
+ ASSERT_EQ(ArrowSharedBufferInit(shared_body.get(), body.get()),
NANOARROW_OK);
+ ASSERT_EQ(ArrowIpcDecoderDecodeDictionaryFromShared(
+ decoder.get(), shared_body.get(),
NANOARROW_VALIDATION_LEVEL_FULL,
+ dictionaries.get(), &error),
+ expected)
+ << error.message;
+ } else {
+ data.data.as_uint8 = body->data;
+ data.size_bytes = body->size_bytes;
+ ASSERT_EQ(ArrowIpcDecoderDecodeDictionary(decoder.get(), data,
+
NANOARROW_VALIDATION_LEVEL_FULL,
+ dictionaries.get(),
&error),
+ expected)
+ << error.message;
+ }
+ };
+
+ ASSERT_NO_FATAL_FAILURE(decode_nulls(length[0], false, NANOARROW_OK));
+ ASSERT_NO_FATAL_FAILURE(decode_nulls(length[1], true, EOVERFLOW));
+ std::string expected =
+ "Expected dictionary length plus delta length to fit in int64 but
found " +
+ std::to_string(length[0]) + " and " + std::to_string(length[1]);
+ EXPECT_STREQ(error.message, expected.c_str());
+
+ // Rejecting the delta must leave the current dictionary intact.
+ const struct ArrowArray* current;
+ ASSERT_EQ(
+ ArrowIpcDictionariesFindCurrentValue(dictionaries.get(), 0,
¤t, &error),
+ NANOARROW_OK);
+ EXPECT_EQ(current->length, length[0]);
+ EXPECT_EQ(current->null_count, length[0]);
+
+ // The same decoder and dictionary memo can still replace and append
values.
+ ASSERT_NO_FATAL_FAILURE(decode_nulls(1, false, NANOARROW_OK));
+ ASSERT_NO_FATAL_FAILURE(decode_nulls(1, true, NANOARROW_OK));
+ ASSERT_EQ(
+ ArrowIpcDictionariesFindCurrentValue(dictionaries.get(), 0,
¤t, &error),
+ NANOARROW_OK);
+ EXPECT_EQ(current->length, 2);
+ EXPECT_EQ(current->null_count, 2);
+ }
+ }
+}
+
TEST(NanoarrowIpcTest, NanoarrowIpcSetSchema) {
struct ArrowIpcDecoder decoder;
struct ArrowSchema schema;
@@ -2338,4 +2482,212 @@ INSTANTIATE_TEST_SUITE_P(NanoarrowIpcTest,
ArrowTypeIdParameterizedTestFixture,
NANOARROW_TYPE_DECIMAL128,
NANOARROW_TYPE_DECIMAL256,
NANOARROW_TYPE_INTERVAL_MONTH_DAY_NANO));
+
+class DeltaDictionaryTypeTest : public ::testing::TestWithParam<enum
ArrowType> {};
+
+static std::shared_ptr<arrow::Array> FinishDeltaBuilder(arrow::ArrayBuilder*
builder) {
+ std::shared_ptr<arrow::Array> out;
+ EXPECT_TRUE(builder->Finish(&out).ok());
+ return out;
+}
+
+static std::shared_ptr<arrow::Array> MakeDeltaDictionaryValues(enum ArrowType
type) {
+ switch (type) {
+ case NANOARROW_TYPE_BOOL: {
+ arrow::BooleanBuilder builder;
+ EXPECT_TRUE(builder.Append(true).ok());
+ EXPECT_TRUE(builder.Append(false).ok());
+ EXPECT_TRUE(builder.Append(true).ok());
+ EXPECT_TRUE(builder.Append(false).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_INT64: {
+ arrow::Int64Builder builder;
+ EXPECT_TRUE(builder.Append(1).ok());
+ EXPECT_TRUE(builder.AppendNull().ok());
+ EXPECT_TRUE(builder.Append(3).ok());
+ EXPECT_TRUE(builder.Append(4).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_UINT64: {
+ arrow::UInt64Builder builder;
+ EXPECT_TRUE(builder.Append(1).ok());
+ EXPECT_TRUE(builder.Append(2).ok());
+ EXPECT_TRUE(builder.Append(3).ok());
+ EXPECT_TRUE(builder.Append(4).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_DOUBLE: {
+ arrow::DoubleBuilder builder;
+ EXPECT_TRUE(builder.Append(1.5).ok());
+ EXPECT_TRUE(builder.Append(-2.25).ok());
+ EXPECT_TRUE(builder.Append(3.5).ok());
+ EXPECT_TRUE(builder.Append(4.75).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_STRING: {
+ arrow::StringBuilder builder;
+ EXPECT_TRUE(builder.Append("one", 3).ok());
+ EXPECT_TRUE(builder.Append("two", 3).ok());
+ EXPECT_TRUE(builder.Append("three", 5).ok());
+ EXPECT_TRUE(builder.Append("four", 4).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_BINARY: {
+ arrow::BinaryBuilder builder;
+ EXPECT_TRUE(builder.Append("one", 3).ok());
+ EXPECT_TRUE(builder.Append("two", 3).ok());
+ EXPECT_TRUE(builder.Append("three", 5).ok());
+ EXPECT_TRUE(builder.Append("four", 4).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_DECIMAL128: {
+ arrow::Decimal128Builder builder(arrow::decimal128(10, 2));
+ EXPECT_TRUE(builder.Append(arrow::Decimal128(125)).ok());
+ EXPECT_TRUE(builder.Append(arrow::Decimal128(-250)).ok());
+ EXPECT_TRUE(builder.Append(arrow::Decimal128(375)).ok());
+ EXPECT_TRUE(builder.Append(arrow::Decimal128(400)).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_LIST: {
+ auto value_builder = std::make_shared<arrow::Int32Builder>();
+ arrow::ListBuilder builder(arrow::default_memory_pool(), value_builder);
+ EXPECT_TRUE(builder.Append().ok());
+ EXPECT_TRUE(value_builder->Append(1).ok());
+ EXPECT_TRUE(value_builder->Append(2).ok());
+ EXPECT_TRUE(builder.Append().ok());
+ EXPECT_TRUE(builder.Append().ok());
+ EXPECT_TRUE(value_builder->Append(3).ok());
+ EXPECT_TRUE(value_builder->AppendNull().ok());
+ EXPECT_TRUE(builder.Append().ok());
+ EXPECT_TRUE(value_builder->Append(4).ok());
+ EXPECT_TRUE(value_builder->Append(5).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_STRUCT: {
+ auto int_builder = std::make_shared<arrow::Int32Builder>();
+ auto string_builder = std::make_shared<arrow::StringBuilder>();
+ auto type = arrow::struct_(
+ {arrow::field("i", arrow::int32()), arrow::field("s",
arrow::utf8())});
+ arrow::StructBuilder builder(type, arrow::default_memory_pool(),
+ {int_builder, string_builder});
+ for (int32_t i = 1; i <= 4; i++) {
+ EXPECT_TRUE(builder.Append().ok());
+ EXPECT_TRUE(int_builder->Append(i).ok());
+ if (i == 3) {
+ EXPECT_TRUE(string_builder->AppendNull().ok());
+ } else {
+ EXPECT_TRUE(string_builder->Append(std::to_string(i)).ok());
+ }
+ }
+ return FinishDeltaBuilder(&builder);
+ }
+ case NANOARROW_TYPE_FIXED_SIZE_LIST: {
+ auto value_builder = std::make_shared<arrow::Int16Builder>();
+ arrow::FixedSizeListBuilder builder(arrow::default_memory_pool(),
value_builder, 2);
+ for (int16_t value = 1; value <= 7; value += 2) {
+ EXPECT_TRUE(builder.Append().ok());
+ EXPECT_TRUE(value_builder->Append(value).ok());
+ EXPECT_TRUE(value_builder->Append(value + 1).ok());
+ }
+ return FinishDeltaBuilder(&builder);
+ }
+ default:
+ break;
+ }
+
+ ADD_FAILURE() << "Unexpected type " << ArrowTypeString(type);
+ return nullptr;
+}
+
+static std::shared_ptr<arrow::Array> MakeDeltaDictionaryIndices(bool extended)
{
+ arrow::Int32Builder builder;
+ if (extended) {
+ EXPECT_TRUE(builder.Append(2).ok());
+ EXPECT_TRUE(builder.Append(3).ok());
+ EXPECT_TRUE(builder.Append(1).ok());
+ EXPECT_TRUE(builder.AppendNull().ok());
+ } else {
+ EXPECT_TRUE(builder.Append(0).ok());
+ EXPECT_TRUE(builder.Append(1).ok());
+ EXPECT_TRUE(builder.AppendNull().ok());
+ EXPECT_TRUE(builder.Append(0).ok());
+ }
+ return FinishDeltaBuilder(&builder);
+}
+
+static void AssertReadsArrowCppDeltaStream(
+ const std::shared_ptr<arrow::Array>& dictionary_array1,
+ const std::shared_ptr<arrow::Array>& dictionary_array2) {
+ auto schema = arrow::schema({arrow::field("dictionary",
dictionary_array1->type())});
+ auto expected1 = arrow::RecordBatch::Make(schema, 4, {dictionary_array1});
+ auto expected2 = arrow::RecordBatch::Make(schema, 4, {dictionary_array2});
+
+ auto maybe_sink = arrow::io::BufferOutputStream::Create();
+ ASSERT_TRUE(maybe_sink.ok()) << maybe_sink.status();
+ auto sink = maybe_sink.ValueUnsafe();
+ auto options = arrow::ipc::IpcWriteOptions::Defaults();
+ options.emit_dictionary_deltas = true;
+ auto maybe_writer = arrow::ipc::MakeStreamWriter(sink, schema, options);
+ ASSERT_TRUE(maybe_writer.ok()) << maybe_writer.status();
+ auto writer = maybe_writer.ValueUnsafe();
+ ASSERT_TRUE(writer->WriteRecordBatch(*expected1).ok());
+ ASSERT_TRUE(writer->WriteRecordBatch(*expected2).ok());
+ ASSERT_TRUE(writer->Close().ok());
+ auto maybe_buffer = sink->Finish();
+ ASSERT_TRUE(maybe_buffer.ok()) << maybe_buffer.status();
+ auto buffer = maybe_buffer.ValueUnsafe();
+
+ nanoarrow::UniqueBuffer ipc_buffer;
+ ASSERT_EQ(ArrowBufferAppend(ipc_buffer.get(), buffer->data(),
buffer->size()),
+ NANOARROW_OK);
+ struct ArrowIpcInputStream input;
+ ASSERT_EQ(ArrowIpcInputStreamInitBuffer(&input, ipc_buffer.get()),
NANOARROW_OK);
+ nanoarrow::UniqueArrayStream reader;
+ ASSERT_EQ(ArrowIpcArrayStreamReaderInit(reader.get(), &input, nullptr),
NANOARROW_OK);
+
+ struct ArrowError error;
+ nanoarrow::UniqueSchema roundtrip_schema;
+ ASSERT_EQ(ArrowArrayStreamGetSchema(reader.get(), roundtrip_schema.get(),
&error),
+ NANOARROW_OK)
+ << error.message;
+ auto maybe_arrow_schema = arrow::ImportSchema(roundtrip_schema.get());
+ ASSERT_TRUE(maybe_arrow_schema.ok()) << maybe_arrow_schema.status();
+ auto arrow_schema = maybe_arrow_schema.ValueUnsafe();
+
+ nanoarrow::UniqueArray roundtrip1;
+ nanoarrow::UniqueArray roundtrip2;
+ ASSERT_EQ(ArrowArrayStreamGetNext(reader.get(), roundtrip1.get(), &error),
NANOARROW_OK)
+ << error.message;
+ ASSERT_EQ(ArrowArrayStreamGetNext(reader.get(), roundtrip2.get(), &error),
NANOARROW_OK)
+ << error.message;
+ auto maybe_roundtrip1 = arrow::ImportRecordBatch(roundtrip1.get(),
arrow_schema);
+ auto maybe_roundtrip2 = arrow::ImportRecordBatch(roundtrip2.get(),
arrow_schema);
+ ASSERT_TRUE(maybe_roundtrip1.ok()) << maybe_roundtrip1.status();
+ ASSERT_TRUE(maybe_roundtrip2.ok()) << maybe_roundtrip2.status();
+ EXPECT_TRUE(maybe_roundtrip1.ValueUnsafe()->Equals(*expected1));
+ EXPECT_TRUE(maybe_roundtrip2.ValueUnsafe()->Equals(*expected2));
+}
+
+TEST_P(DeltaDictionaryTypeTest, ReadsArrowCppDeltaStream) {
+ auto values2 = MakeDeltaDictionaryValues(GetParam());
+ ASSERT_NE(values2, nullptr);
+ auto values1 = values2->Slice(0, 2);
+ auto dictionary_type = arrow::dictionary(arrow::int32(), values2->type());
+ auto maybe_array1 = arrow::DictionaryArray::FromArrays(
+ dictionary_type, MakeDeltaDictionaryIndices(false), values1);
+ auto maybe_array2 = arrow::DictionaryArray::FromArrays(
+ dictionary_type, MakeDeltaDictionaryIndices(true), values2);
+ ASSERT_TRUE(maybe_array1.ok()) << maybe_array1.status();
+ ASSERT_TRUE(maybe_array2.ok()) << maybe_array2.status();
+ AssertReadsArrowCppDeltaStream(maybe_array1.ValueUnsafe(),
maybe_array2.ValueUnsafe());
+}
+
+INSTANTIATE_TEST_SUITE_P(NanoarrowIpcDecoder, DeltaDictionaryTypeTest,
+ ::testing::Values(NANOARROW_TYPE_BOOL,
NANOARROW_TYPE_INT64,
+ NANOARROW_TYPE_UINT64,
NANOARROW_TYPE_DOUBLE,
+ NANOARROW_TYPE_STRING,
NANOARROW_TYPE_BINARY,
+ NANOARROW_TYPE_DECIMAL128,
NANOARROW_TYPE_LIST,
+ NANOARROW_TYPE_STRUCT,
+ NANOARROW_TYPE_FIXED_SIZE_LIST));
#endif
