rustyconover commented on code in PR #928:
URL: https://github.com/apache/arrow-nanoarrow/pull/928#discussion_r4180224254
##########
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:
Gone along with the materialized copies; there are no more `result` chains
in the writer.
##########
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:
Done: dictionaries are now compared by shape and buffer addresses,
recursively. To keep that safe, `WriteArrayStream()` holds the previous batch
until the next one is written, so a released buffer's address can't be reused
for different values mid-comparison. There's a test with a producer that
recycles memory this way. `WriteDictionaryBatch()` only sees a borrowed view
and always writes.
##########
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:
No more `goto`. The only shared cleanup is in the existing outer
`WriteArrayStream()` function.
##########
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:
Agreed. Delta emission is removed from the writer entirely. Applications
that want deltas can write them with `ArrowIpcWriterWriteDictionaryBatch(...,
is_delta = 1, ...)`, so I'm not planning a separate PR for it.
--
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]
