rustyconover commented on code in PR #928:
URL: https://github.com/apache/arrow-nanoarrow/pull/928#discussion_r4180223508
##########
src/nanoarrow/ipc/decoder.c:
##########
@@ -337,17 +352,117 @@ static ArrowErrorCode ArrowIpcDictionaryReplace(struct
ArrowIpcDictionary* dicti
return NANOARROW_OK;
}
+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;
+}
+
+static void ArrowIpcArrayPrepareForAppend(struct ArrowArray* array,
+ const struct ArrowArrayView*
array_view) {
+ // Finishing a view array materializes its variadic-buffer sizes. Appending
may
+ // extend the last variadic buffer or add another one, so force the sizes
buffer
+ // to be regenerated by the next ArrowArrayFinishBuildingDefault().
+ if (array_view->storage_type == NANOARROW_TYPE_BINARY_VIEW ||
+ array_view->storage_type == NANOARROW_TYPE_STRING_VIEW) {
+ ArrowBufferReset(ArrowArrayBuffer(array, array->n_buffers - 1));
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcArrayPrepareForAppend(array->children[i], array_view->children[i]);
+ }
+}
Review Comment:
Removed. The delta decoding moved to #955, and there is no view-specific
handling there.
##########
src/nanoarrow/ipc/decoder.c:
##########
@@ -337,17 +352,117 @@ static ArrowErrorCode ArrowIpcDictionaryReplace(struct
ArrowIpcDictionary* dicti
return NANOARROW_OK;
}
+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;
+}
+
+static void ArrowIpcArrayPrepareForAppend(struct ArrowArray* array,
+ const struct ArrowArrayView*
array_view) {
+ // Finishing a view array materializes its variadic-buffer sizes. Appending
may
+ // extend the last variadic buffer or add another one, so force the sizes
buffer
+ // to be regenerated by the next ArrowArrayFinishBuildingDefault().
+ if (array_view->storage_type == NANOARROW_TYPE_BINARY_VIEW ||
+ array_view->storage_type == NANOARROW_TYPE_STRING_VIEW) {
+ ArrowBufferReset(ArrowArrayBuffer(array, array->n_buffers - 1));
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcArrayPrepareForAppend(array->children[i], array_view->children[i]);
+ }
+}
+
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);
+ }
+
+ // In the usual streaming loop, the previously returned batch has been
released
+ // before the next one is requested. Recover the mutable backing array and
append
+ // directly so a sequence of small deltas grows geometrically instead of
copying
+ // the complete dictionary for every message. If an older batch is still
alive,
+ // keep the copy-on-write path below to preserve its dictionary snapshot.
+ if (ArrowArrayInternalTryUnshare(&dictionary->current_value)) {
+ struct ArrowArray combined;
+ ArrowArrayMove(&dictionary->current_value, &combined);
+
+ ArrowIpcArrayPrepareForAppend(&combined, array_view);
+ ArrowErrorCode result = ArrowArrayReserve(&combined, value->length);
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayAppendStorageFromArrayView(&combined, array_view,
error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowIpcArraySetDictionaries(&combined, value);
+ }
Review Comment:
Done in #955: `ArrowIpcDictionaryConcatenate()` can fail and doesn't own its
inputs, and the caller checks once and releases the temporary.
##########
src/nanoarrow/ipc/decoder_test.cc:
##########
@@ -2099,4 +2164,279 @@ INSTANTIATE_TEST_SUITE_P(NanoarrowIpcTest,
ArrowTypeIdParameterizedTestFixture,
NANOARROW_TYPE_DECIMAL128,
NANOARROW_TYPE_DECIMAL256,
NANOARROW_TYPE_INTERVAL_MONTH_DAY_NANO));
+
+enum class DeltaDictionaryValueCase {
+ kBoolean,
+ kInt64,
+ kInt64WithNull,
+ kUInt64,
+ kDouble,
+ kString,
+ kBinary,
+ kDecimal128,
+ kList,
+ kStruct,
+ kFixedSizeList
+};
Review Comment:
Done in #955: the test parameters are now `enum ArrowType` values.
##########
src/nanoarrow/ipc/decoder_test.cc:
##########
@@ -2099,4 +2164,279 @@ INSTANTIATE_TEST_SUITE_P(NanoarrowIpcTest,
ArrowTypeIdParameterizedTestFixture,
NANOARROW_TYPE_DECIMAL128,
NANOARROW_TYPE_DECIMAL256,
NANOARROW_TYPE_INTERVAL_MONTH_DAY_NANO));
+
+enum class DeltaDictionaryValueCase {
+ kBoolean,
+ kInt64,
+ kInt64WithNull,
+ kUInt64,
+ kDouble,
+ kString,
+ kBinary,
+ kDecimal128,
+ kList,
+ kStruct,
+ kFixedSizeList
+};
+
+class DeltaDictionaryTypeTest
+ : public ::testing::TestWithParam<DeltaDictionaryValueCase> {};
+
+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(
+ DeltaDictionaryValueCase value_case) {
+ switch (value_case) {
+ case DeltaDictionaryValueCase::kBoolean: {
+ 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 DeltaDictionaryValueCase::kInt64:
+ case DeltaDictionaryValueCase::kInt64WithNull: {
+ arrow::Int64Builder builder;
+ EXPECT_TRUE(builder.Append(1).ok());
+ if (value_case == DeltaDictionaryValueCase::kInt64WithNull) {
+ EXPECT_TRUE(builder.AppendNull().ok());
+ } else {
+ EXPECT_TRUE(builder.Append(-2).ok());
+ }
+ EXPECT_TRUE(builder.Append(3).ok());
+ EXPECT_TRUE(builder.Append(4).ok());
+ return FinishDeltaBuilder(&builder);
+ }
+ case DeltaDictionaryValueCase::kUInt64: {
+ 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 DeltaDictionaryValueCase::kDouble: {
+ 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 DeltaDictionaryValueCase::kString: {
+ 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 DeltaDictionaryValueCase::kBinary: {
+ 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 DeltaDictionaryValueCase::kDecimal128: {
+ 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 DeltaDictionaryValueCase::kList: {
+ 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 DeltaDictionaryValueCase::kStruct: {
+ 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 DeltaDictionaryValueCase::kFixedSizeList: {
+ 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);
+ }
+ }
+
+ ADD_FAILURE() << "Unknown dictionary value case";
+ 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,
+ const char* expected_error = nullptr) {
+ 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;
+ int result = ArrowArrayStreamGetNext(reader.get(), roundtrip2.get(), &error);
+ if (expected_error != nullptr) {
+ EXPECT_EQ(result, ENOTSUP);
+ EXPECT_STREQ(error.message, expected_error);
+ return;
+ }
+ ASSERT_EQ(result, 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());
+}
+
+TEST(NanoarrowIpcTest, RejectsArrowCppDenseUnionDictionaryDelta) {
+ arrow::Int8Builder type_ids_builder;
+ arrow::Int32Builder offsets_builder;
Review Comment:
Removed in #955.
##########
src/nanoarrow/ipc/decoder.c:
##########
@@ -337,17 +352,117 @@ static ArrowErrorCode ArrowIpcDictionaryReplace(struct
ArrowIpcDictionary* dicti
return NANOARROW_OK;
}
+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;
+}
+
+static void ArrowIpcArrayPrepareForAppend(struct ArrowArray* array,
+ const struct ArrowArrayView*
array_view) {
+ // Finishing a view array materializes its variadic-buffer sizes. Appending
may
+ // extend the last variadic buffer or add another one, so force the sizes
buffer
+ // to be regenerated by the next ArrowArrayFinishBuildingDefault().
+ if (array_view->storage_type == NANOARROW_TYPE_BINARY_VIEW ||
+ array_view->storage_type == NANOARROW_TYPE_STRING_VIEW) {
+ ArrowBufferReset(ArrowArrayBuffer(array, array->n_buffers - 1));
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcArrayPrepareForAppend(array->children[i], array_view->children[i]);
+ }
+}
+
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);
+ }
+
+ // In the usual streaming loop, the previously returned batch has been
released
+ // before the next one is requested. Recover the mutable backing array and
append
+ // directly so a sequence of small deltas grows geometrically instead of
copying
+ // the complete dictionary for every message. If an older batch is still
alive,
+ // keep the copy-on-write path below to preserve its dictionary snapshot.
+ if (ArrowArrayInternalTryUnshare(&dictionary->current_value)) {
+ struct ArrowArray combined;
+ ArrowArrayMove(&dictionary->current_value, &combined);
Review Comment:
Punted. #955 drops the unshare helper (and the `common/utils.c` change) and
always builds a new combined array, so each delta copies the dictionary.
Appending in place when the dictionary isn't shared can come in a follow-up.
##########
src/nanoarrow/ipc/decoder.c:
##########
@@ -337,17 +352,117 @@ static ArrowErrorCode ArrowIpcDictionaryReplace(struct
ArrowIpcDictionary* dicti
return NANOARROW_OK;
}
+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;
+}
+
+static void ArrowIpcArrayPrepareForAppend(struct ArrowArray* array,
+ const struct ArrowArrayView*
array_view) {
+ // Finishing a view array materializes its variadic-buffer sizes. Appending
may
+ // extend the last variadic buffer or add another one, so force the sizes
buffer
+ // to be regenerated by the next ArrowArrayFinishBuildingDefault().
+ if (array_view->storage_type == NANOARROW_TYPE_BINARY_VIEW ||
+ array_view->storage_type == NANOARROW_TYPE_STRING_VIEW) {
+ ArrowBufferReset(ArrowArrayBuffer(array, array->n_buffers - 1));
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcArrayPrepareForAppend(array->children[i], array_view->children[i]);
+ }
+}
+
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);
+ }
+
+ // In the usual streaming loop, the previously returned batch has been
released
+ // before the next one is requested. Recover the mutable backing array and
append
+ // directly so a sequence of small deltas grows geometrically instead of
copying
+ // the complete dictionary for every message. If an older batch is still
alive,
+ // keep the copy-on-write path below to preserve its dictionary snapshot.
+ if (ArrowArrayInternalTryUnshare(&dictionary->current_value)) {
+ struct ArrowArray combined;
+ ArrowArrayMove(&dictionary->current_value, &combined);
+
+ ArrowIpcArrayPrepareForAppend(&combined, array_view);
+ ArrowErrorCode result = ArrowArrayReserve(&combined, value->length);
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayAppendStorageFromArrayView(&combined, array_view,
error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowIpcArraySetDictionaries(&combined, value);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayFinishBuildingDefault(&combined, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowIpcDictionaryReplace(dictionary, &combined, error);
+ }
+
+ if (combined.release != NULL) {
+ ArrowArrayRelease(&combined);
+ }
+ if (result == NANOARROW_OK) {
+ ArrowArrayRelease(value);
+ }
+ return result;
}
- NANOARROW_RETURN_NOT_OK(ArrowIpcDictionaryReplace(dictionary, value, error));
- return NANOARROW_OK;
+ struct ArrowArray combined;
+ combined.release = NULL;
+ NANOARROW_RETURN_NOT_OK(ArrowArrayInitFromArrayView(&combined, array_view,
error));
+ ArrowErrorCode result = ArrowArrayStartAppending(&combined);
+ if (result == NANOARROW_OK) {
+ result =
+ ArrowArrayReserve(&combined, dictionary->current_value.length +
value->length);
+ }
Review Comment:
Same pattern applied in #955 (the reserve is inside the helper).
##########
src/nanoarrow/ipc/encoder.c:
##########
@@ -216,6 +218,14 @@ ArrowErrorCode ArrowIpcEncoderSetCompression(
return ArrowIpcEncoderSetCompressor(encoder, &compressor);
}
+void ArrowIpcEncoderSetDictionaryReplacement(struct ArrowIpcEncoder* encoder,
+ char enabled) {
+ NANOARROW_DCHECK(encoder != NULL && encoder->private_data != NULL);
+ struct ArrowIpcEncoderPrivate* private =
+ (struct ArrowIpcEncoderPrivate*)encoder->private_data;
+ private->dictionary_replacement = enabled != 0;
+}
Review Comment:
Agreed. It stays in this PR, which is now only the non-delta writer and
encoder changes.
##########
src/nanoarrow/ipc/files_test.cc:
##########
@@ -608,10 +614,10 @@ INSTANTIATE_TEST_SUITE_P(
TestFile::OK("cpp-21.0.0/generated_primitive_zerolength.stream"),
TestFile::OK("cpp-21.0.0/generated_recursive_nested.stream"),
TestFile::OK("cpp-21.0.0/generated_union.stream"),
- TestFile::ReadOnly("cpp-21.0.0/generated_dictionary.stream"),
- TestFile::ReadOnly("cpp-21.0.0/generated_dictionary_unsigned.stream"),
+ TestFile::OK("cpp-21.0.0/generated_dictionary.stream"),
+ TestFile::OK("cpp-21.0.0/generated_dictionary_unsigned.stream"),
TestFile::ReadOnly("cpp-21.0.0/generated_extension.stream"),
- TestFile::ReadOnly("cpp-21.0.0/generated_nested_dictionary.stream"),
+ TestFile::OK("cpp-21.0.0/generated_nested_dictionary.stream"),
Review Comment:
Nothing beyond this PR: the file's `dict_exts` column is a
dictionary-encoded extension type, and the test previously wrote with
`ArrowIpcWriterWriteArrayView()`, which doesn't write dictionaries. Now that
the roundtrip goes through `ArrowIpcWriterWriteArrayStream()` it passes, so I
switched it (and the shared-dictionary file) to `OK`.
##########
src/nanoarrow/ipc/writer.c:
##########
@@ -357,57 +430,386 @@ ArrowErrorCode ArrowIpcWriterWriteDictionaryBatch(
&private->buffer),
error);
- if (private->writing_file) {
- _NANOARROW_CHECK_RANGE(private->buffer.size_bytes, 0, INT32_MAX);
- struct ArrowIpcFileBlock block = {
- .offset = private->bytes_written,
- .metadata_length = (int32_t) private->buffer.size_bytes,
- .body_length = private->body_buffer.size_bytes,
- };
- NANOARROW_RETURN_NOT_OK_WITH_ERROR(
- ArrowBufferAppend(&private->footer.dictionary_blocks, &block,
sizeof(block)),
- error);
+ return ArrowIpcWriterWriteEncodedDictionaryBatch(writer, error);
+}
+
+static struct ArrowIpcWriterDictionaryCacheEntry*
ArrowIpcWriterFindDictionaryCacheEntry(
+ struct ArrowIpcWriterPrivate* private, int64_t dictionary_id) {
+ int64_t n_cached_dictionaries =
+ private->dictionary_cache.size_bytes /
+ (int64_t)sizeof(struct ArrowIpcWriterDictionaryCacheEntry);
+ struct ArrowIpcWriterDictionaryCacheEntry* cached_dictionaries =
+ (struct
ArrowIpcWriterDictionaryCacheEntry*)private->dictionary_cache.data;
+ for (int64_t i = 0; i < n_cached_dictionaries; i++) {
+ if (cached_dictionaries[i].dictionary_id == dictionary_id) {
+ return &cached_dictionaries[i];
+ }
+ }
+
+ return NULL;
+}
+
+static ArrowErrorCode ArrowIpcWriterArrayViewInitLike(struct ArrowArrayView*
out,
+ const struct
ArrowArrayView* src) {
+ ArrowArrayViewInitFromType(out, src->storage_type);
+ out->layout = src->layout;
+
+ ArrowErrorCode result = ArrowArrayViewAllocateChildren(out, src->n_children);
+ if (result != NANOARROW_OK) {
+ ArrowArrayViewReset(out);
+ return result;
+ }
+
+ for (int64_t i = 0; i < src->n_children; i++) {
+ result = ArrowIpcWriterArrayViewInitLike(out->children[i],
src->children[i]);
+ if (result != NANOARROW_OK) {
+ ArrowArrayViewReset(out);
+ return result;
+ }
+ }
+
+ if (src->dictionary != NULL) {
+ result = ArrowArrayViewAllocateDictionary(out);
+ if (result != NANOARROW_OK) {
+ ArrowArrayViewReset(out);
+ return result;
+ }
+
+ result = ArrowIpcWriterArrayViewInitLike(out->dictionary, src->dictionary);
+ if (result != NANOARROW_OK) {
+ ArrowArrayViewReset(out);
+ return result;
+ }
}
- private->bytes_written += private->buffer.size_bytes;
- private->bytes_written += private->body_buffer.size_bytes;
- NANOARROW_RETURN_NOT_OK(ArrowIpcOutputStreamWrite(
- &private->output_stream, ArrowBufferToBufferView(&private->buffer),
error));
- NANOARROW_RETURN_NOT_OK(ArrowIpcOutputStreamWrite(
- &private->output_stream, ArrowBufferToBufferView(&private->body_buffer),
error));
return NANOARROW_OK;
}
+static void ArrowIpcWriterCanonicalizeBitmapPadding(
+ struct ArrowArray* array, const struct ArrowArrayView* array_view) {
+ int64_t remainder = array->length % 8;
Review Comment:
This code is gone. With comparison by buffer address there is no need to
canonicalize bitmap padding.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
