lxy-9602 commented on code in PR #278:
URL: https://github.com/apache/paimon-cpp/pull/278#discussion_r3932011798
##########
src/paimon/format/blob/blob_file_batch_reader.cpp:
##########
@@ -252,9 +368,219 @@ Result<std::shared_ptr<arrow::Array>>
BlobFileBatchReader::BuildContentArray(
return std::make_shared<arrow::StructArray>(struct_array_data);
}
+Result<std::shared_ptr<arrow::Array>> BlobFileBatchReader::BuildMapBlobArray(
+ int32_t rows_to_read) const {
+ const auto& struct_type = static_cast<const
arrow::StructType&>(*target_type_);
+ const std::shared_ptr<arrow::Field>& map_field = struct_type.field(0);
+ auto map_type = checked_pointer_cast<arrow::MapType>(map_field->type());
+ const std::shared_ptr<arrow::DataType>& key_type = map_type->key_type();
+ if (key_type->id() == arrow::Type::STRING) {
+ arrow::util::InitializeUTF8();
+ }
+ 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,
+ arrow::MakeBuilder(key_type,
arrow_pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder>
item_builder_unique,
+
arrow::MakeBuilder(map_type->item_type(), arrow_pool_.get()));
+ std::shared_ptr<arrow::ArrayBuilder>
key_builder(std::move(key_builder_unique));
+ std::shared_ptr<arrow::ArrayBuilder>
item_builder(std::move(item_builder_unique));
+ if (!item_builder || !item_builder->type() ||
+ item_builder->type()->id() != arrow::Type::LARGE_BINARY) {
+ return Status::Invalid("cast MAP<..., BLOB> item builder to large
binary builder failed");
+ }
+ auto* blob_builder =
checked_cast<arrow::LargeBinaryBuilder*>(item_builder.get());
+ arrow::MapBuilder map_builder(arrow_pool_.get(), key_builder,
item_builder, map_type);
+
+ for (int32_t k = 0; k < rows_to_read; ++k) {
+ const size_t row_index = current_pos_ + k;
+ if (IsTargetNull(row_index)) {
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.AppendNull());
+ continue;
+ }
+ if (IsTargetPlaceholder(row_index)) {
+ // Duplicate map keys cannot occur in a valid Paimon map, so two
empty/default keys
+ // with null values form an unambiguous, Arrow-valid internal
sentinel.
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Append());
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->AppendEmptyValues(2));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(blob_builder->AppendNulls(2));
+ continue;
+ }
+ if (target_blob_lengths_[row_index] < 0) {
+ return Status::Invalid(fmt::format("unsupported MAP<..., BLOB>
record length: {}",
+
target_blob_lengths_[row_index]));
+ }
+
+ const int64_t payload_offset = GetTargetContentOffset(row_index);
+ const int64_t payload_length = GetTargetContentLength(row_index);
+ if (payload_length < kMapBlobMinPayloadLength) {
+ return Status::Invalid(
+ fmt::format("invalid MAP<..., BLOB> payload length: {}",
payload_length));
+ }
+
+ std::array<uint8_t, kMapBlobHeaderLength> header;
+ PAIMON_RETURN_NOT_OK(ReadBlobContentAt(payload_offset, header.size(),
header.data()));
+ const auto magic_number = ReadLittleEndian<int32_t>(header.data());
+ if (magic_number != kMapBlobMagicNumber) {
+ return Status::Invalid(
+ fmt::format("invalid MAP<..., BLOB> payload magic number: {}",
magic_number));
+ }
+ const auto version = static_cast<int8_t>(header[4]);
+ if (version != kMapBlobVersion) {
+ return Status::NotImplemented(
+ fmt::format("unsupported MAP<..., BLOB> payload version: {}",
version));
+ }
+ const auto entry_count = ReadLittleEndian<int32_t>(header.data() + 5);
+ if (entry_count < 0) {
+ return Status::Invalid(
+ fmt::format("invalid MAP<..., BLOB> entry count: {}",
entry_count));
+ }
+
+ const int64_t index_lengths_offset =
+ payload_offset + payload_length - kMapBlobIndexLengthsSize;
+ std::array<uint8_t, kMapBlobIndexLengthsSize> index_lengths;
+ PAIMON_RETURN_NOT_OK(
+ ReadBlobContentAt(index_lengths_offset, index_lengths.size(),
index_lengths.data()));
+ const auto key_index_length =
ReadLittleEndian<int32_t>(index_lengths.data());
+ const auto value_index_length =
+ ReadLittleEndian<int32_t>(index_lengths.data() + sizeof(int32_t));
+ const int64_t maximum_indexes_length = payload_length -
kMapBlobMinPayloadLength;
+ if (key_index_length < 0 || key_index_length > maximum_indexes_length)
{
+ return Status::Invalid(
+ fmt::format("invalid MAP<..., BLOB> key index length: {}",
key_index_length));
+ }
+ if (value_index_length < 0 || value_index_length >
maximum_indexes_length) {
+ return Status::Invalid(
+ fmt::format("invalid MAP<..., BLOB> value index length: {}",
value_index_length));
+ }
+ if (static_cast<int64_t>(key_index_length) + value_index_length >
maximum_indexes_length) {
+ return Status::Invalid("MAP<..., BLOB> indexes exceed the payload
length");
+ }
+ if (entry_count > key_index_length || entry_count >
value_index_length) {
+ return Status::Invalid("MAP<..., BLOB> entry count exceeds index
length");
+ }
+
+ const int64_t value_index_offset = index_lengths_offset -
value_index_length;
+ const int64_t key_index_offset = value_index_offset - key_index_length;
+ std::vector<char> key_index_bytes(key_index_length);
+ std::vector<char> value_index_bytes(value_index_length);
+ PAIMON_RETURN_NOT_OK(ReadBlobContentAt(key_index_offset,
key_index_length,
+
reinterpret_cast<uint8_t*>(key_index_bytes.data())));
+ PAIMON_RETURN_NOT_OK(
+ ReadBlobContentAt(value_index_offset, value_index_length,
+
reinterpret_cast<uint8_t*>(value_index_bytes.data())));
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int64_t> key_lengths,
+
DeltaVarintCompressor::Decompress(key_index_bytes));
+ PAIMON_ASSIGN_OR_RAISE(std::vector<int64_t> value_lengths,
+
DeltaVarintCompressor::Decompress(value_index_bytes));
+ if (key_lengths.size() != static_cast<size_t>(entry_count)) {
+ return Status::Invalid("MAP<..., BLOB> entry count does not match
key index length");
+ }
+ if (value_lengths.size() != static_cast<size_t>(entry_count)) {
+ return Status::Invalid("MAP<..., BLOB> entry count does not match
value index length");
+ }
+
+ const int64_t data_offset = payload_offset + kMapBlobHeaderLength;
+ const int64_t data_length = key_index_offset - data_offset;
+ int64_t key_data_length = 0;
+ for (int64_t key_length : key_lengths) {
+ if (key_length < 0) {
+ return Status::Invalid("MAP<..., BLOB> keys cannot be null");
+ }
+ if (key_length > std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid(
+ fmt::format("MAP<..., BLOB> key is too large: {}",
key_length));
+ }
+ if (fixed_key_length >= 0 && key_length != fixed_key_length) {
+ return Status::Invalid(
+ fmt::format("invalid MAP<..., BLOB> fixed-width key
length: {}", key_length));
+ }
+ if (key_length > data_length - key_data_length) {
+ return Status::Invalid("MAP<..., BLOB> key lengths exceed the
payload data length");
+ }
+ key_data_length += key_length;
+ }
+
+ const int64_t maximum_value_data_length = data_length -
key_data_length;
+ int64_t value_data_length = 0;
+ for (int64_t value_length : value_lengths) {
+ if (value_length == BlobDefs::kNullBinLength) {
+ continue;
+ }
+ if (value_length < 0) {
+ return Status::Invalid(
+ fmt::format("invalid MAP<..., BLOB> value length: {}",
value_length));
+ }
+ if (!blob_as_descriptor_ && value_length >
std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid(
+ fmt::format("MAP<..., BLOB> inline value is too large:
{}", value_length));
+ }
+ if (value_length > maximum_value_data_length - value_data_length) {
+ return Status::Invalid(
+ "MAP<..., BLOB> value lengths exceed the payload data
length");
+ }
+ value_data_length += value_length;
+ }
+ if (value_data_length != maximum_value_data_length) {
+ return Status::Invalid(
+ "MAP<..., BLOB> key/value lengths do not match the payload
data length");
+ }
+
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(map_builder.Append());
+ int64_t key_offset = data_offset;
+ std::set<std::string> serialized_keys;
+ for (int32_t entry = 0; entry < entry_count; ++entry) {
+ const auto key_length = static_cast<int32_t>(key_lengths[entry]);
+ std::vector<uint8_t> key_bytes(key_length);
+ PAIMON_RETURN_NOT_OK(ReadBlobContentAt(key_offset, key_length,
key_bytes.data()));
Review Comment:
I’m not sure how large `key_bytes` can get. If it can be large, it would be
better to allocate it from the pool, for example via `Bytes`.
--
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]
