lucasfang commented on code in PR #395:
URL: https://github.com/apache/paimon-cpp/pull/395#discussion_r4128745243
##########
src/paimon/format/blob/blob_file_batch_reader.cpp:
##########
@@ -509,12 +633,119 @@ Result<std::shared_ptr<arrow::Array>>
BlobFileBatchReader::BuildArrayBlobArray(
return Status::Invalid("ARRAY<BLOB> element type must be large
binary");
}
+ std::vector<ArrayBlobPayload> payloads;
+ payloads.reserve(rows_to_read);
+ int64_t value_count = 0;
+ int64_t value_data_length = 0;
+ int64_t descriptor_count = 0;
+ int64_t list_null_count = 0;
+ int64_t value_null_count = 0;
+ for (int32_t k = 0; k < rows_to_read; ++k) {
+ const size_t row_index = current_pos_ + k;
+ if (IsTargetPlaceholder(row_index)) {
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacity(1, &value_count,
"ARRAY<BLOB> element count"));
+
PAIMON_RETURN_NOT_OK(AddBuilderCapacity(BlobDefs::kPlaceholderSentinelLength,
+ &value_data_length,
+ "ARRAY<BLOB> element data
size"));
+ continue;
+ }
+ if (IsTargetNull(row_index)) {
+ ++list_null_count;
+ continue;
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(ArrayBlobPayload payload,
ReadArrayBlobPayload(row_index));
+ PAIMON_RETURN_NOT_OK(
+
AddBuilderCapacity(static_cast<int64_t>(payload.element_lengths.size()),
&value_count,
+ "ARRAY<BLOB> element count"));
+ for (int64_t element_length : payload.element_lengths) {
+ if (element_length == BlobDefs::kNullBinLength) {
+ ++value_null_count;
+ continue;
+ }
+ if (blob_as_descriptor_) {
+ PAIMON_RETURN_NOT_OK(
+ AddBuilderCapacity(1, &descriptor_count, "ARRAY<BLOB>
descriptor count"));
+ } else {
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacity(element_length,
&value_data_length,
+ "ARRAY<BLOB> element
data size"));
+ }
+ }
+ payloads.emplace_back(std::move(payload));
+ }
+ if (descriptor_count > 0) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t descriptor_size,
+ GetSerializedDescriptorSize(file_path_, pool_));
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacityProduct(descriptor_count,
descriptor_size,
+ &value_data_length,
+ "ARRAY<BLOB> descriptor
data size"));
+ }
+
+ if (value_count > std::numeric_limits<int32_t>::max()) {
+ return Status::CapacityError("ARRAY<BLOB> element count exceeds list
offset capacity");
+ }
+ if (!blob_as_descriptor_) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Buffer> list_offsets,
+ AllocateTypedBuffer<int32_t>(rows_to_read + 1,
arrow_pool_));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Buffer> list_null_bitmap,
+ AllocateValidityBitmap(rows_to_read,
list_null_count, arrow_pool_));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<PreallocatedLargeBinaryBuilder>
value_builder,
+ PreallocatedLargeBinaryBuilder::Create(
+ value_count, value_data_length,
value_null_count, arrow_pool_));
+
+ auto* raw_list_offsets =
reinterpret_cast<int32_t*>(list_offsets->mutable_data());
+ raw_list_offsets[0] = 0;
+ size_t payload_index = 0;
+ for (int32_t k = 0; k < rows_to_read; ++k) {
+ const size_t row_index = current_pos_ + k;
+ if (IsTargetNull(row_index)) {
+ arrow::bit_util::ClearBit(list_null_bitmap->mutable_data(), k);
+ } else if (IsTargetPlaceholder(row_index)) {
+ PAIMON_ASSIGN_OR_RAISE(auto* target,
Review Comment:
This violates AGENTS.md: "Use an explicit type, not auto, as the declaration
in PAIMON_ASSIGN_OR_RAISE". The base version of this file has zero occurrences
of ASSIGN_OR_RAISE(auto. Change to uint8_t* target.
##########
src/paimon/format/blob/blob_file_batch_reader.cpp:
##########
@@ -509,12 +633,119 @@ Result<std::shared_ptr<arrow::Array>>
BlobFileBatchReader::BuildArrayBlobArray(
return Status::Invalid("ARRAY<BLOB> element type must be large
binary");
}
+ std::vector<ArrayBlobPayload> payloads;
+ payloads.reserve(rows_to_read);
+ int64_t value_count = 0;
+ int64_t value_data_length = 0;
+ int64_t descriptor_count = 0;
+ int64_t list_null_count = 0;
+ int64_t value_null_count = 0;
+ for (int32_t k = 0; k < rows_to_read; ++k) {
+ const size_t row_index = current_pos_ + k;
+ if (IsTargetPlaceholder(row_index)) {
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacity(1, &value_count,
"ARRAY<BLOB> element count"));
+
PAIMON_RETURN_NOT_OK(AddBuilderCapacity(BlobDefs::kPlaceholderSentinelLength,
+ &value_data_length,
+ "ARRAY<BLOB> element data
size"));
+ continue;
+ }
+ if (IsTargetNull(row_index)) {
+ ++list_null_count;
+ continue;
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(ArrayBlobPayload payload,
ReadArrayBlobPayload(row_index));
+ PAIMON_RETURN_NOT_OK(
+
AddBuilderCapacity(static_cast<int64_t>(payload.element_lengths.size()),
&value_count,
+ "ARRAY<BLOB> element count"));
+ for (int64_t element_length : payload.element_lengths) {
+ if (element_length == BlobDefs::kNullBinLength) {
+ ++value_null_count;
+ continue;
+ }
+ if (blob_as_descriptor_) {
+ PAIMON_RETURN_NOT_OK(
+ AddBuilderCapacity(1, &descriptor_count, "ARRAY<BLOB>
descriptor count"));
+ } else {
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacity(element_length,
&value_data_length,
+ "ARRAY<BLOB> element
data size"));
+ }
+ }
+ payloads.emplace_back(std::move(payload));
+ }
+ if (descriptor_count > 0) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t descriptor_size,
+ GetSerializedDescriptorSize(file_path_, pool_));
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacityProduct(descriptor_count,
descriptor_size,
+ &value_data_length,
+ "ARRAY<BLOB> descriptor
data size"));
+ }
+
+ if (value_count > std::numeric_limits<int32_t>::max()) {
+ return Status::CapacityError("ARRAY<BLOB> element count exceeds list
offset capacity");
+ }
+ if (!blob_as_descriptor_) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Buffer> list_offsets,
+ AllocateTypedBuffer<int32_t>(rows_to_read + 1,
arrow_pool_));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Buffer> list_null_bitmap,
+ AllocateValidityBitmap(rows_to_read,
list_null_count, arrow_pool_));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<PreallocatedLargeBinaryBuilder>
value_builder,
+ PreallocatedLargeBinaryBuilder::Create(
+ value_count, value_data_length,
value_null_count, arrow_pool_));
+
+ auto* raw_list_offsets =
reinterpret_cast<int32_t*>(list_offsets->mutable_data());
+ raw_list_offsets[0] = 0;
+ size_t payload_index = 0;
+ for (int32_t k = 0; k < rows_to_read; ++k) {
+ const size_t row_index = current_pos_ + k;
+ if (IsTargetNull(row_index)) {
+ arrow::bit_util::ClearBit(list_null_bitmap->mutable_data(), k);
+ } else if (IsTargetPlaceholder(row_index)) {
+ PAIMON_ASSIGN_OR_RAISE(auto* target,
+
value_builder->Append(BlobDefs::kPlaceholderSentinelLength));
+ memcpy(target, BlobDefs::kPlaceholderSentinel,
+ BlobDefs::kPlaceholderSentinelLength);
+ } else {
+ const ArrayBlobPayload& payload = payloads[payload_index++];
+ uint8_t* row_target = nullptr;
+ int64_t row_data_length = 0;
+ for (int64_t element_length : payload.element_lengths) {
+ if (element_length == BlobDefs::kNullBinLength) {
+ PAIMON_RETURN_NOT_OK(value_builder->AppendNull());
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(auto* target,
value_builder->Append(element_length));
Review Comment:
same
--
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]
