lucasfang commented on code in PR #395:
URL: https://github.com/apache/paimon-cpp/pull/395#discussion_r4128747678
##########
src/paimon/format/blob/blob_file_batch_reader.cpp:
##########
@@ -724,19 +947,141 @@ Result<std::shared_ptr<arrow::Array>>
BlobFileBatchReader::BuildMapBlobArray(
}
PAIMON_ASSIGN_OR_RAISE(int32_t fixed_key_length,
GetMapBlobFixedKeyLength(key_type));
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder>
key_builder_unique,
+ std::vector<MapBlobPayload> payloads;
+ payloads.reserve(rows_to_read);
+ int64_t entry_count = 0;
+ int64_t key_data_length = 0;
+ int64_t value_data_length = 0;
+ int64_t descriptor_count = 0;
+ int64_t map_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(2, &entry_count, "MAP<...,
BLOB> entry count"));
+ value_null_count += 2;
+ continue;
+ }
+ if (IsTargetNull(row_index)) {
+ ++map_null_count;
+ continue;
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(MapBlobPayload payload,
+ ReadMapBlobPayload(row_index,
fixed_key_length));
+
PAIMON_RETURN_NOT_OK(AddBuilderCapacity(static_cast<int64_t>(payload.value_lengths.size()),
+ &entry_count, "MAP<..., BLOB>
entry count"));
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacity(payload.key_data_length,
&key_data_length,
+ "MAP<..., BLOB> key data
size"));
+ for (int64_t value_length : payload.value_lengths) {
+ if (value_length == BlobDefs::kNullBinLength) {
+ ++value_null_count;
+ continue;
+ }
+ if (blob_as_descriptor_) {
+ PAIMON_RETURN_NOT_OK(
+ AddBuilderCapacity(1, &descriptor_count, "MAP<..., BLOB>
descriptor count"));
+ } else {
+ PAIMON_RETURN_NOT_OK(AddBuilderCapacity(value_length,
&value_data_length,
+ "MAP<..., BLOB> value
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,
+ "MAP<..., BLOB>
descriptor data size"));
+ }
+
+ if (entry_count > std::numeric_limits<int32_t>::max()) {
+ return Status::CapacityError("MAP<..., BLOB> entry count exceeds map
offset capacity");
+ }
+
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::ArrayBuilder>
key_builder,
arrow::MakeBuilder(key_type,
arrow_pool_.get()));
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder>
item_builder_unique,
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Reserve(entry_count));
+ if (key_type->id() == arrow::Type::STRING) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+
checked_cast<arrow::StringBuilder*>(key_builder.get())->ReserveData(key_data_length));
+ } else if (key_type->id() == arrow::Type::BINARY) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+
checked_cast<arrow::BinaryBuilder*>(key_builder.get())->ReserveData(key_data_length));
+ }
+
+ if (!blob_as_descriptor_) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Buffer> map_offsets,
+ AllocateTypedBuffer<int32_t>(rows_to_read + 1,
arrow_pool_));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Buffer> map_null_bitmap,
+ AllocateValidityBitmap(rows_to_read,
map_null_count, arrow_pool_));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<PreallocatedLargeBinaryBuilder>
value_builder,
+ PreallocatedLargeBinaryBuilder::Create(
+ entry_count, value_data_length,
value_null_count, arrow_pool_));
+
+ auto* raw_map_offsets =
reinterpret_cast<int32_t*>(map_offsets->mutable_data());
+ raw_map_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(map_null_bitmap->mutable_data(), k);
+ } else if (IsTargetPlaceholder(row_index)) {
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->AppendEmptyValues(2));
+ PAIMON_RETURN_NOT_OK(value_builder->AppendNull());
+ PAIMON_RETURN_NOT_OK(value_builder->AppendNull());
+ } else {
+ const MapBlobPayload& payload = payloads[payload_index++];
+ PAIMON_RETURN_NOT_OK(AppendMapBlobKeys(payload, key_type,
key_builder.get()));
+ uint8_t* row_target = nullptr;
+ int64_t row_data_length = 0;
+ for (int64_t value_length : payload.value_lengths) {
+ if (value_length == BlobDefs::kNullBinLength) {
+ PAIMON_RETURN_NOT_OK(value_builder->AppendNull());
+ continue;
+ }
+ PAIMON_ASSIGN_OR_RAISE(auto* target,
value_builder->Append(value_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]
