diff --git a/src/nanoarrow/ipc/decoder.c b/src/nanoarrow/ipc/decoder.c index 8953cb297..80e53ec8d 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 3fb2bf6e0..c7ac78ce5 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 +#include +#include +#include +#include #include #include #include #include +#include #include #endif #include @@ -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(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(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 {}; + +static std::shared_ptr FinishDeltaBuilder(arrow::ArrayBuilder* builder) { + std::shared_ptr out; + EXPECT_TRUE(builder->Finish(&out).ok()); + return out; +} + +static std::shared_ptr 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::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(); + auto string_builder = std::make_shared(); + 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::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 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& dictionary_array1, + const std::shared_ptr& 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