lxy-9602 commented on code in PR #278:
URL: https://github.com/apache/paimon-cpp/pull/278#discussion_r3931684976
##########
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:
There are similar endianness-conversion utilities in cpp; you can refer to
`math.h`.
##########
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:
If we want to support the `TIME` type, I believe the required changes would
be quite extensive. The full `paimon-cpp` end-to-end flow would likely need
updates, including append, PK, spill, and compaction. You can refer to the
vector type PR (#197) as a reference. So I’d suggest handling `TIME` type
support in a separate PR rather than combining it with the current blob changes.
##########
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:
May use `PAIMON_RETURN_NOT_OK_FROM_ARROW`?
##########
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:
I’m not sure whether Java also validates UTF-8 here.
##########
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:
Consider use `file_system->ReadFile`.
##########
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:
This function seems too long. Please break it down into smaller, more
fine-grained pieces.
##########
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:
Consider use `file_system->WriteFile`
##########
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:
Please use JSON to initialize the expected array. The current way of
verification is not very intuitive, and it takes quite a lot of lines.
--
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]
