paleolimbot commented on code in PR #928:
URL: https://github.com/apache/arrow-nanoarrow/pull/928#discussion_r4043551013
##########
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:
I don't think we support views yet for IPC decoding at all yet?
https://github.com/apache/arrow-nanoarrow/issues/824
##########
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:
Can you reuse `enum ArrowType` for this one?
##########
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;
+ if (remainder != 0) {
+ uint8_t mask = (uint8_t)((1U << remainder) - 1U);
+ for (int i = 0; i < NANOARROW_MAX_FIXED_BUFFERS; i++) {
+ if (array_view->layout.element_size_bits[i] == 1) {
+ struct ArrowBuffer* buffer = ArrowArrayBuffer(array, i);
+ if (buffer->size_bytes > 0) {
+ buffer->data[buffer->size_bytes - 1] &= mask;
+ }
+ }
+ }
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->children[i],
array_view->children[i]);
+ }
+
+ if (array->dictionary != NULL) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->dictionary,
array_view->dictionary);
+ }
+}
+
+static ArrowErrorCode ArrowIpcWriterMaterializeArrayView(const struct
ArrowArrayView* src,
+ int64_t offset,
int64_t length,
+ struct ArrowArray*
out,
+ struct ArrowError*
error) {
+ out->release = NULL;
+ if (offset < 0 || length < 0 || offset > src->length || length > src->length
- offset) {
+ ArrowErrorSet(error,
+ "Invalid dictionary slice [%" PRId64 ", %" PRId64
+ ") for array of length %" PRId64,
+ offset, offset + length, src->length);
+ return EINVAL;
+ }
+
+ struct ArrowArrayView slice = *src;
+ slice.offset += offset;
+ slice.length = length;
+ slice.null_count = -1;
+
+ ArrowErrorCode result = ArrowArrayInitFromArrayView(out, src, error);
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayStartAppending(out);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayReserve(out, length);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayAppendStorageFromArrayView(out, &slice, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayFinishBuildingDefault(out, error);
+ }
+ if (result == NANOARROW_OK) {
+ // Arrow bitmaps do not require producers to initialize padding bits.
Clear them so
+ // physical comparisons of two otherwise identical materialized arrays
never read
+ // indeterminate data and do not treat padding as part of dictionary
identity.
+ ArrowIpcWriterCanonicalizeBitmapPadding(out, src);
+ }
+
+ if (result != NANOARROW_OK && out->release != NULL) {
+ ArrowArrayRelease(out);
+ }
+ return result;
+}
+
+static ArrowErrorCode ArrowIpcWriterCompareMaterializedArrays(
+ const struct ArrowArray* lhs, const struct ArrowArray* rhs,
+ const struct ArrowArrayView* shape, int* out, struct ArrowError* error) {
+ struct ArrowArrayView lhs_view;
+ struct ArrowArrayView rhs_view;
+ ArrowArrayViewInitFromType(&lhs_view, NANOARROW_TYPE_UNINITIALIZED);
+ ArrowArrayViewInitFromType(&rhs_view, NANOARROW_TYPE_UNINITIALIZED);
+
+ ArrowErrorCode result = ArrowIpcWriterArrayViewInitLike(&lhs_view, shape);
+ if (result == NANOARROW_OK) {
+ result = ArrowIpcWriterArrayViewInitLike(&rhs_view, shape);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayViewSetArray(&lhs_view, lhs, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayViewSetArray(&rhs_view, rhs, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayViewCompare(&lhs_view, &rhs_view,
NANOARROW_COMPARE_IDENTICAL, out,
+ NULL);
+ }
+
+ ArrowArrayViewReset(&lhs_view);
+ ArrowArrayViewReset(&rhs_view);
+ return result;
+}
+
+static ArrowErrorCode ArrowIpcWriterArrayViewSetMaterialized(
+ struct ArrowArrayView* out, const struct ArrowArrayView* shape,
+ const struct ArrowArray* array, struct ArrowError* error) {
+ ArrowArrayViewInitFromType(out, NANOARROW_TYPE_UNINITIALIZED);
+ NANOARROW_RETURN_NOT_OK(ArrowIpcWriterArrayViewInitLike(out, shape));
+ ArrowErrorCode result = ArrowArrayViewSetArray(out, array, error);
+ if (result != NANOARROW_OK) {
+ ArrowArrayViewReset(out);
+ }
+ return result;
+}
+
+static ArrowErrorCode ArrowIpcWriterWriteDictionaryBatchIfChanged(
+ struct ArrowIpcWriter* writer, int64_t dictionary_id,
+ const struct ArrowArrayView* values_view, int force_emit, int allow_delta,
+ int* emitted, struct ArrowError* error) {
+ struct ArrowIpcWriterPrivate* private =
+ (struct ArrowIpcWriterPrivate*)writer->private_data;
+
+ struct ArrowArray current_values = {.release = NULL};
+ struct ArrowArray prefix_values = {.release = NULL};
+ struct ArrowArray delta_values = {.release = NULL};
+ struct ArrowArrayView encoded_view;
+ ArrowArrayViewInitFromType(&encoded_view, NANOARROW_TYPE_UNINITIALIZED);
+
+ ArrowErrorCode result = ArrowIpcWriterMaterializeArrayView(
+ values_view, 0, values_view->length, ¤t_values, error);
+ if (result != NANOARROW_OK) {
+ return result;
+ }
+
+ struct ArrowIpcWriterDictionaryCacheEntry* cached =
+ ArrowIpcWriterFindDictionaryCacheEntry(private, dictionary_id);
+ int values_equal = 0;
+ char is_delta = 0;
+ int cached_was_added = 0;
+ const struct ArrowArray* values_to_encode = NULL;
+ if (cached != NULL) {
+ result = ArrowIpcWriterCompareMaterializedArrays(&cached->values,
¤t_values,
+ values_view,
&values_equal, error);
+ if (result != NANOARROW_OK) {
+ goto cleanup;
+ }
+ }
+
+ if (!force_emit && cached != NULL && values_equal) {
+ *emitted = 0;
+ result = NANOARROW_OK;
+ goto cleanup;
+ }
+
+ if (allow_delta && !force_emit && cached != NULL &&
+ current_values.length > cached->values.length) {
+ result = ArrowIpcWriterMaterializeArrayView(values_view, 0,
cached->values.length,
+ &prefix_values, error);
+ if (result != NANOARROW_OK) {
+ goto cleanup;
+ }
Review Comment:
I don't think that the writer should emit delta dictionaries in this
way...this is a kind of calculation that application code should do, which may
have access to things like C++ that can do it more efficiently.
In any case, emitting delta dictionaries is a third scope here for a
separate PR
##########
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;
+ if (remainder != 0) {
+ uint8_t mask = (uint8_t)((1U << remainder) - 1U);
+ for (int i = 0; i < NANOARROW_MAX_FIXED_BUFFERS; i++) {
+ if (array_view->layout.element_size_bits[i] == 1) {
+ struct ArrowBuffer* buffer = ArrowArrayBuffer(array, i);
+ if (buffer->size_bytes > 0) {
+ buffer->data[buffer->size_bytes - 1] &= mask;
+ }
+ }
+ }
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->children[i],
array_view->children[i]);
+ }
+
+ if (array->dictionary != NULL) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->dictionary,
array_view->dictionary);
+ }
+}
+
+static ArrowErrorCode ArrowIpcWriterMaterializeArrayView(const struct
ArrowArrayView* src,
+ int64_t offset,
int64_t length,
+ struct ArrowArray*
out,
+ struct ArrowError*
error) {
+ out->release = NULL;
+ if (offset < 0 || length < 0 || offset > src->length || length > src->length
- offset) {
+ ArrowErrorSet(error,
+ "Invalid dictionary slice [%" PRId64 ", %" PRId64
+ ") for array of length %" PRId64,
+ offset, offset + length, src->length);
+ return EINVAL;
+ }
+
+ struct ArrowArrayView slice = *src;
+ slice.offset += offset;
+ slice.length = length;
+ slice.null_count = -1;
+
+ ArrowErrorCode result = ArrowArrayInitFromArrayView(out, src, error);
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayStartAppending(out);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayReserve(out, length);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayAppendStorageFromArrayView(out, &slice, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayFinishBuildingDefault(out, error);
+ }
+ if (result == NANOARROW_OK) {
+ // Arrow bitmaps do not require producers to initialize padding bits.
Clear them so
+ // physical comparisons of two otherwise identical materialized arrays
never read
+ // indeterminate data and do not treat padding as part of dictionary
identity.
+ ArrowIpcWriterCanonicalizeBitmapPadding(out, src);
+ }
+
+ if (result != NANOARROW_OK && out->release != NULL) {
+ ArrowArrayRelease(out);
+ }
+ return result;
+}
+
+static ArrowErrorCode ArrowIpcWriterCompareMaterializedArrays(
+ const struct ArrowArray* lhs, const struct ArrowArray* rhs,
+ const struct ArrowArrayView* shape, int* out, struct ArrowError* error) {
+ struct ArrowArrayView lhs_view;
+ struct ArrowArrayView rhs_view;
+ ArrowArrayViewInitFromType(&lhs_view, NANOARROW_TYPE_UNINITIALIZED);
+ ArrowArrayViewInitFromType(&rhs_view, NANOARROW_TYPE_UNINITIALIZED);
+
+ ArrowErrorCode result = ArrowIpcWriterArrayViewInitLike(&lhs_view, shape);
+ if (result == NANOARROW_OK) {
+ result = ArrowIpcWriterArrayViewInitLike(&rhs_view, shape);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayViewSetArray(&lhs_view, lhs, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayViewSetArray(&rhs_view, rhs, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayViewCompare(&lhs_view, &rhs_view,
NANOARROW_COMPARE_IDENTICAL, out,
+ NULL);
+ }
+
+ ArrowArrayViewReset(&lhs_view);
+ ArrowArrayViewReset(&rhs_view);
+ return result;
+}
+
+static ArrowErrorCode ArrowIpcWriterArrayViewSetMaterialized(
+ struct ArrowArrayView* out, const struct ArrowArrayView* shape,
+ const struct ArrowArray* array, struct ArrowError* error) {
+ ArrowArrayViewInitFromType(out, NANOARROW_TYPE_UNINITIALIZED);
+ NANOARROW_RETURN_NOT_OK(ArrowIpcWriterArrayViewInitLike(out, shape));
+ ArrowErrorCode result = ArrowArrayViewSetArray(out, array, error);
+ if (result != NANOARROW_OK) {
+ ArrowArrayViewReset(out);
+ }
+ return result;
+}
+
+static ArrowErrorCode ArrowIpcWriterWriteDictionaryBatchIfChanged(
+ struct ArrowIpcWriter* writer, int64_t dictionary_id,
+ const struct ArrowArrayView* values_view, int force_emit, int allow_delta,
+ int* emitted, struct ArrowError* error) {
+ struct ArrowIpcWriterPrivate* private =
+ (struct ArrowIpcWriterPrivate*)writer->private_data;
+
+ struct ArrowArray current_values = {.release = NULL};
+ struct ArrowArray prefix_values = {.release = NULL};
+ struct ArrowArray delta_values = {.release = NULL};
+ struct ArrowArrayView encoded_view;
+ ArrowArrayViewInitFromType(&encoded_view, NANOARROW_TYPE_UNINITIALIZED);
+
+ ArrowErrorCode result = ArrowIpcWriterMaterializeArrayView(
+ values_view, 0, values_view->length, ¤t_values, error);
+ if (result != NANOARROW_OK) {
+ return result;
+ }
+
+ struct ArrowIpcWriterDictionaryCacheEntry* cached =
+ ArrowIpcWriterFindDictionaryCacheEntry(private, dictionary_id);
+ int values_equal = 0;
+ char is_delta = 0;
+ int cached_was_added = 0;
+ const struct ArrowArray* values_to_encode = NULL;
+ if (cached != NULL) {
+ result = ArrowIpcWriterCompareMaterializedArrays(&cached->values,
¤t_values,
+ values_view,
&values_equal, error);
+ if (result != NANOARROW_OK) {
+ goto cleanup;
+ }
+ }
+
+ if (!force_emit && cached != NULL && values_equal) {
+ *emitted = 0;
+ result = NANOARROW_OK;
+ goto cleanup;
+ }
+
+ if (allow_delta && !force_emit && cached != NULL &&
+ current_values.length > cached->values.length) {
+ result = ArrowIpcWriterMaterializeArrayView(values_view, 0,
cached->values.length,
+ &prefix_values, error);
+ if (result != NANOARROW_OK) {
+ goto cleanup;
+ }
+
+ int prefix_equal = 0;
+ result = ArrowIpcWriterCompareMaterializedArrays(&cached->values,
&prefix_values,
+ values_view,
&prefix_equal, error);
+ if (result != NANOARROW_OK) {
+ goto cleanup;
+ }
+
+ if (prefix_equal) {
+ result = ArrowIpcWriterMaterializeArrayView(
+ values_view, cached->values.length,
+ current_values.length - cached->values.length, &delta_values, error);
+ if (result != NANOARROW_OK) {
+ goto cleanup;
+ }
+ is_delta = 1;
Review Comment:
You can use the nested function approach for shared cleanup (we don't use
`goto` in nanoarrow, or at least we haven't yet)
##########
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:
The usual streaming loop should be using
`ArrowIpcDecoderDecodeArrayViewWithDictionaries()`, which in a perfect world
never was shared (however, it probably had a shared backing buffer on its first
decode because most decoding happens from a shared buffer).
Rather than trying to unshare, can we check whether this array can be
appended to (and if possibly delay the sharing of it until it is requested as
an array that is not just a view?). Totally ok if not possible, but if so, we
should punt on the trying to unshare piece and implement that 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);
+ }
Review Comment:
For the PR that does contain this, a good pattern is to define a function
that can fail and doesn't own any input pointers:
```c
ArrowErrorCode ArrowIpcDoAppend(struct ArrowArray* combined, struct
ArrowArrayView array_view, struct ArrowError* error) {
NANOARROW_RETURN_NOT_OK(...);
return NANOARROW_OK;
}
```
then here you can just check once and release the temporary
```c
ArrowErrorCode result = ArrowIpcDoAppend(...);
if (result != NANOARROW_OK) {
// release stuff
return result;
}
```
##########
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:
It would help to put the comment about canonicalizing the padding helping to
minimize reemitting a dictionary here.
Other parts of nanoarrow typically avoid this by zeroing out the last byte
of a bitmap when reserving but it is sometimes hard to guarantee that
everywhere.
##########
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;
+ if (remainder != 0) {
+ uint8_t mask = (uint8_t)((1U << remainder) - 1U);
+ for (int i = 0; i < NANOARROW_MAX_FIXED_BUFFERS; i++) {
+ if (array_view->layout.element_size_bits[i] == 1) {
+ struct ArrowBuffer* buffer = ArrowArrayBuffer(array, i);
+ if (buffer->size_bytes > 0) {
+ buffer->data[buffer->size_bytes - 1] &= mask;
+ }
+ }
+ }
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->children[i],
array_view->children[i]);
+ }
+
+ if (array->dictionary != NULL) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->dictionary,
array_view->dictionary);
+ }
+}
+
+static ArrowErrorCode ArrowIpcWriterMaterializeArrayView(const struct
ArrowArrayView* src,
+ int64_t offset,
int64_t length,
+ struct ArrowArray*
out,
+ struct ArrowError*
error) {
+ out->release = NULL;
+ if (offset < 0 || length < 0 || offset > src->length || length > src->length
- offset) {
+ ArrowErrorSet(error,
+ "Invalid dictionary slice [%" PRId64 ", %" PRId64
+ ") for array of length %" PRId64,
+ offset, offset + length, src->length);
+ return EINVAL;
+ }
+
+ struct ArrowArrayView slice = *src;
+ slice.offset += offset;
+ slice.length = length;
+ slice.null_count = -1;
+
+ ArrowErrorCode result = ArrowArrayInitFromArrayView(out, src, error);
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayStartAppending(out);
+ }
Review Comment:
The same nested function strategy for avoiding these repeated result checks
applies here, too
##########
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;
+ if (remainder != 0) {
+ uint8_t mask = (uint8_t)((1U << remainder) - 1U);
+ for (int i = 0; i < NANOARROW_MAX_FIXED_BUFFERS; i++) {
+ if (array_view->layout.element_size_bits[i] == 1) {
+ struct ArrowBuffer* buffer = ArrowArrayBuffer(array, i);
+ if (buffer->size_bytes > 0) {
+ buffer->data[buffer->size_bytes - 1] &= mask;
+ }
+ }
+ }
+ }
+
+ for (int64_t i = 0; i < array->n_children; i++) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->children[i],
array_view->children[i]);
+ }
+
+ if (array->dictionary != NULL) {
+ ArrowIpcWriterCanonicalizeBitmapPadding(array->dictionary,
array_view->dictionary);
+ }
+}
+
+static ArrowErrorCode ArrowIpcWriterMaterializeArrayView(const struct
ArrowArrayView* src,
+ int64_t offset,
int64_t length,
+ struct ArrowArray*
out,
+ struct ArrowError*
error) {
+ out->release = NULL;
+ if (offset < 0 || length < 0 || offset > src->length || length > src->length
- offset) {
+ ArrowErrorSet(error,
+ "Invalid dictionary slice [%" PRId64 ", %" PRId64
+ ") for array of length %" PRId64,
+ offset, offset + length, src->length);
+ return EINVAL;
+ }
+
+ struct ArrowArrayView slice = *src;
+ slice.offset += offset;
+ slice.length = length;
+ slice.null_count = -1;
+
+ ArrowErrorCode result = ArrowArrayInitFromArrayView(out, src, error);
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayStartAppending(out);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayReserve(out, length);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayAppendStorageFromArrayView(out, &slice, error);
+ }
+ if (result == NANOARROW_OK) {
+ result = ArrowArrayFinishBuildingDefault(out, error);
+ }
+ if (result == NANOARROW_OK) {
+ // Arrow bitmaps do not require producers to initialize padding bits.
Clear them so
+ // physical comparisons of two otherwise identical materialized arrays
never read
+ // indeterminate data and do not treat padding as part of dictionary
identity.
+ ArrowIpcWriterCanonicalizeBitmapPadding(out, src);
+ }
+
+ if (result != NANOARROW_OK && out->release != NULL) {
+ ArrowArrayRelease(out);
+ }
+ return result;
+}
+
+static ArrowErrorCode ArrowIpcWriterCompareMaterializedArrays(
+ const struct ArrowArray* lhs, const struct ArrowArray* rhs,
+ const struct ArrowArrayView* shape, int* out, struct ArrowError* error) {
Review Comment:
Rather than compare by value (possibly slower than just writing another
dictionary to the stream), probably comparing the buffer pointers makes more
sense. If it's a shared array it should be pointing to the same data.
##########
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:
The same pattern applies here
##########
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:
This is a great and super useful change, but a separate one from the
decoding or encoding of delta dictionaries
##########
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:
I think we can skip this test unless there's something special about
rejecting a type in the decoder implementation
##########
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:
Just curious: what feature is needed to roundtrip the generated extension
beyond what's here?
--
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]
