XiaoHongbo-Hope commented on code in PR #278:
URL: https://github.com/apache/paimon-cpp/pull/278#discussion_r3932929363
##########
src/paimon/format/blob/blob_file_batch_reader.cpp:
##########
@@ -39,6 +46,110 @@
#include "paimon/data/blob.h"
namespace paimon::blob {
+namespace {
+
+constexpr int32_t kMapBlobMagicNumber = 0x4D424342;
+constexpr int8_t kMapBlobVersion = 1;
+constexpr int32_t kMapBlobHeaderLength = 9;
+constexpr int32_t kMapBlobIndexLengthsSize = 8;
+constexpr int32_t kMapBlobMinPayloadLength = kMapBlobHeaderLength +
kMapBlobIndexLengthsSize;
+
+template <typename T>
+T ReadLittleEndian(const uint8_t* data) {
+ return arrow::bit_util::FromLittleEndian(arrow::util::SafeLoadAs<T>(data));
+}
+
+Result<int32_t> GetMapBlobFixedKeyLength(const
std::shared_ptr<arrow::DataType>& key_type) {
+ switch (key_type->id()) {
+ case arrow::Type::BOOL:
+ case arrow::Type::INT8:
+ return 1;
+ case arrow::Type::INT16:
+ return 2;
+ case arrow::Type::INT32:
+ case arrow::Type::DATE32:
+ case arrow::Type::TIME32:
+ return 4;
+ case arrow::Type::INT64:
+ return 8;
+ case arrow::Type::DECIMAL128: {
+ const auto& decimal_type = static_cast<const
arrow::Decimal128Type&>(*key_type);
+ return decimal_type.precision() <= 18 ? 8 : -1;
+ }
+ case arrow::Type::STRING:
+ case arrow::Type::BINARY:
+ return -1;
+ default:
+ return Status::Invalid(
+ fmt::format("unsupported MAP<..., BLOB> key type: {}",
key_type->ToString()));
+ }
+}
+
+Status AppendMapBlobKey(const std::shared_ptr<arrow::DataType>& key_type,
const uint8_t* data,
+ int32_t length, arrow::ArrayBuilder* builder) {
+ switch (key_type->id()) {
+ case arrow::Type::BOOL: {
+ if (data[0] != 0 && data[0] != 1) {
+ return Status::Invalid("invalid MAP<..., BLOB> boolean key");
+ }
+ return ToPaimonStatus(
+ checked_cast<arrow::BooleanBuilder*>(builder)->Append(data[0]
== 1));
+ }
+ case arrow::Type::INT8:
+ return ToPaimonStatus(
+
checked_cast<arrow::Int8Builder*>(builder)->Append(static_cast<int8_t>(data[0])));
+ case arrow::Type::INT16:
+ return
ToPaimonStatus(checked_cast<arrow::Int16Builder*>(builder)->Append(
+ ReadLittleEndian<int16_t>(data)));
+ case arrow::Type::INT32:
+ return
ToPaimonStatus(checked_cast<arrow::Int32Builder*>(builder)->Append(
+ ReadLittleEndian<int32_t>(data)));
+ case arrow::Type::INT64:
+ return
ToPaimonStatus(checked_cast<arrow::Int64Builder*>(builder)->Append(
+ ReadLittleEndian<int64_t>(data)));
+ case arrow::Type::DATE32:
+ return
ToPaimonStatus(checked_cast<arrow::Date32Builder*>(builder)->Append(
+ ReadLittleEndian<int32_t>(data)));
+ case arrow::Type::TIME32:
+ return
ToPaimonStatus(checked_cast<arrow::Time32Builder*>(builder)->Append(
+ ReadLittleEndian<int32_t>(data)));
+ case arrow::Type::STRING:
+ if (!arrow::util::ValidateUTF8(data, length)) {
+ return Status::Invalid("invalid UTF-8 in MAP<STRING, BLOB>
key");
+ }
+ return ToPaimonStatus(
Review Comment:
Java's passes the bytes to without an explicit UTF-8 check. I kept the
boundary validation here because Arrow requires valid UTF-8; a corrupt payload
should return instead of constructing an invalid Arrow array.
##########
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_);
Review Comment:
Refactored in 00da44a5. Payload parsing/validation, key appends, and value
appends are now split into , , and ; only coordinates rows and builders.
##########
src/paimon/format/blob/blob_file_batch_reader.cpp:
##########
@@ -39,6 +46,110 @@
#include "paimon/data/blob.h"
namespace paimon::blob {
+namespace {
+
+constexpr int32_t kMapBlobMagicNumber = 0x4D424342;
+constexpr int8_t kMapBlobVersion = 1;
+constexpr int32_t kMapBlobHeaderLength = 9;
+constexpr int32_t kMapBlobIndexLengthsSize = 8;
+constexpr int32_t kMapBlobMinPayloadLength = kMapBlobHeaderLength +
kMapBlobIndexLengthsSize;
+
+template <typename T>
+T ReadLittleEndian(const uint8_t* data) {
+ return arrow::bit_util::FromLittleEndian(arrow::util::SafeLoadAs<T>(data));
+}
Review Comment:
Updated in 00da44a5 to use Paimon's from .
##########
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;
Review Comment:
Fixed in 00da44a5. now rejects on , so all callers fail before filtering
can erase the fallback sentinel. Added a regression test through as used by .
##########
src/paimon/format/blob/blob_file_batch_reader_test.cpp:
##########
@@ -108,6 +139,28 @@ class BlobFileBatchReaderTest : public testing::Test,
public ::testing::WithPara
}
}
+ Result<std::string> ReadMapBlobValue(const
std::shared_ptr<arrow::LargeBinaryArray>& blob_array,
+ int64_t index, bool
blob_as_descriptor,
+ const std::shared_ptr<FileSystem>&
file_system) {
+ std::string stored_value = blob_array->GetString(index);
+ if (!blob_as_descriptor) {
+ return stored_value;
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<Blob> blob,
+ Blob::FromDescriptor(stored_value.data(),
stored_value.size()));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
+ blob->NewInputStream(file_system));
+ PAIMON_ASSIGN_OR_RAISE(int64_t length, input_stream->Length());
+ std::string value(length, '\0');
+ if (length > 0) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t actual_length,
input_stream->Read(value.data(), length));
+ if (actual_length != length) {
+ return Status::IOError("failed to read MAP<..., BLOB>
descriptor content");
+ }
+ }
Review Comment:
Simplified in 00da44a5 with . A direct would read the whole shared file,
while a descriptor refers to an offset/length slice; preserves those slice
semantics.
##########
src/paimon/format/blob/blob_file_batch_reader.cpp:
##########
@@ -39,6 +46,110 @@
#include "paimon/data/blob.h"
namespace paimon::blob {
+namespace {
+
+constexpr int32_t kMapBlobMagicNumber = 0x4D424342;
+constexpr int8_t kMapBlobVersion = 1;
+constexpr int32_t kMapBlobHeaderLength = 9;
+constexpr int32_t kMapBlobIndexLengthsSize = 8;
+constexpr int32_t kMapBlobMinPayloadLength = kMapBlobHeaderLength +
kMapBlobIndexLengthsSize;
+
+template <typename T>
+T ReadLittleEndian(const uint8_t* data) {
+ return arrow::bit_util::FromLittleEndian(arrow::util::SafeLoadAs<T>(data));
+}
+
+Result<int32_t> GetMapBlobFixedKeyLength(const
std::shared_ptr<arrow::DataType>& key_type) {
+ switch (key_type->id()) {
+ case arrow::Type::BOOL:
+ case arrow::Type::INT8:
+ return 1;
+ case arrow::Type::INT16:
+ return 2;
+ case arrow::Type::INT32:
+ case arrow::Type::DATE32:
+ case arrow::Type::TIME32:
+ return 4;
+ case arrow::Type::INT64:
+ return 8;
+ case arrow::Type::DECIMAL128: {
+ const auto& decimal_type = static_cast<const
arrow::Decimal128Type&>(*key_type);
+ return decimal_type.precision() <= 18 ? 8 : -1;
+ }
+ case arrow::Type::STRING:
+ case arrow::Type::BINARY:
+ return -1;
+ default:
+ return Status::Invalid(
+ fmt::format("unsupported MAP<..., BLOB> key type: {}",
key_type->ToString()));
+ }
+}
+
+Status AppendMapBlobKey(const std::shared_ptr<arrow::DataType>& key_type,
const uint8_t* data,
+ int32_t length, arrow::ArrayBuilder* builder) {
+ switch (key_type->id()) {
+ case arrow::Type::BOOL: {
+ if (data[0] != 0 && data[0] != 1) {
+ return Status::Invalid("invalid MAP<..., BLOB> boolean key");
+ }
+ return ToPaimonStatus(
+ checked_cast<arrow::BooleanBuilder*>(builder)->Append(data[0]
== 1));
Review Comment:
Updated in 00da44a5: Arrow builder statuses now use .
##########
src/paimon/format/blob/blob_file_batch_reader_test.cpp:
##########
@@ -132,6 +185,262 @@ TEST_P(BlobFileBatchReaderTest, TestSimple) {
{"blob_9_f54d253c.bin"}, blob_as_descriptor);
}
+TEST_P(BlobFileBatchReaderTest, TestMapBlob) {
+ auto dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ const std::string file_path = dir->Str() + "/map-blob.blob";
+ std::shared_ptr<FileSystem> file_system =
std::make_shared<LocalFileSystem>();
+
+ const std::string file_bytes = MapBlobGoldenBytes();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output_stream,
+ file_system->Create(file_path, /*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(int64_t written,
+ output_stream->Write(file_bytes.data(),
file_bytes.size()));
+ ASSERT_EQ(file_bytes.size(), written);
+ ASSERT_OK(output_stream->Close());
+
+ std::shared_ptr<arrow::Field> blob_item = BlobUtils::ToArrowField("value",
true);
+ auto key_field = arrow::field("key", arrow::utf8(), false);
+ auto map_type = std::make_shared<arrow::MapType>(key_field, blob_item);
+ ASSERT_TRUE(BlobUtils::IsBlobField(map_type->item_field()));
+ auto map_field = arrow::field("blob_map", map_type, true);
+ auto schema = arrow::schema({map_field});
+ ::ArrowSchema c_schema;
+ ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> input_stream,
file_system->Open(file_path));
+ const bool blob_as_descriptor = GetParam();
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<BlobFileBatchReader> reader,
+ BlobFileBatchReader::Create(
+ input_stream, /*batch_size=*/2,
blob_as_descriptor,
+ /*emit_placeholder_sentinel=*/false, pool_,
GetArrowPool(pool_)));
+ ASSERT_OK(reader->SetReadSchema(&c_schema, nullptr, std::nullopt));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> chunked_array,
+
paimon::test::ReadResultCollector::CollectResult(std::move(reader)));
+ std::shared_ptr<arrow::Array> combined_array =
+ arrow::Concatenate(chunked_array->chunks()).ValueOrDie();
+
+ auto struct_array =
std::dynamic_pointer_cast<arrow::StructArray>(combined_array);
+ ASSERT_TRUE(struct_array);
+ auto map_array =
std::dynamic_pointer_cast<arrow::MapArray>(struct_array->field(0));
+ ASSERT_TRUE(map_array);
+ ASSERT_EQ(arrow::Type::LARGE_BINARY,
map_array->map_type()->item_type()->id());
+ ASSERT_EQ(4, map_array->length());
+ ASSERT_EQ(3, map_array->value_length(0));
+ ASSERT_TRUE(map_array->IsNull(1));
+ ASSERT_EQ(0, map_array->value_length(2));
+ ASSERT_EQ(1, map_array->value_length(3));
+
+ auto keys =
std::dynamic_pointer_cast<arrow::StringArray>(map_array->keys());
+ auto values =
std::dynamic_pointer_cast<arrow::LargeBinaryArray>(map_array->items());
+ ASSERT_TRUE(keys);
+ ASSERT_TRUE(values);
+ ASSERT_EQ("alpha", keys->GetString(0));
+ ASSERT_EQ("empty", keys->GetString(1));
+ ASSERT_EQ("missing", keys->GetString(2));
+ ASSERT_EQ("omega", keys->GetString(3));
+ ASSERT_FALSE(values->IsNull(0));
+ ASSERT_FALSE(values->IsNull(1));
+ ASSERT_TRUE(values->IsNull(2));
+ ASSERT_FALSE(values->IsNull(3));
Review Comment:
Updated in 00da44a5. The decoded values are normalized in descriptor mode
and the complete expected Map array is now initialized from JSON and compared
with .
##########
src/paimon/common/types/data_type.cpp:
##########
@@ -112,6 +117,27 @@ std::string DataType::DataTypeToString(const
std::shared_ptr<arrow::DataType>& t
return "BYTES";
case arrow::Type::type::DATE32:
return "DATE";
+ case arrow::Type::type::TIME32: {
+ const auto& time_type = checked_cast<const
arrow::Time32Type&>(*type);
Review Comment:
Removed support from this PR in 00da44a5. now follows the existing
unsupported-TIME path; end-to-end support can be handled separately.
##########
src/paimon/format/blob/blob_file_batch_reader_test.cpp:
##########
@@ -132,6 +185,262 @@ TEST_P(BlobFileBatchReaderTest, TestSimple) {
{"blob_9_f54d253c.bin"}, blob_as_descriptor);
}
+TEST_P(BlobFileBatchReaderTest, TestMapBlob) {
+ auto dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ const std::string file_path = dir->Str() + "/map-blob.blob";
+ std::shared_ptr<FileSystem> file_system =
std::make_shared<LocalFileSystem>();
+
+ const std::string file_bytes = MapBlobGoldenBytes();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<OutputStream> output_stream,
+ file_system->Create(file_path, /*overwrite=*/true));
+ ASSERT_OK_AND_ASSIGN(int64_t written,
+ output_stream->Write(file_bytes.data(),
file_bytes.size()));
+ ASSERT_EQ(file_bytes.size(), written);
+ ASSERT_OK(output_stream->Close());
Review Comment:
Updated all added golden/corruption fixture writes to use in 00da44a5.
##########
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:
Updated in 00da44a5. Key buffers now use pool-backed ; inline value buffers
use the same allocation path.
--
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]
