diff --git a/include/paimon/catalog/identifier.h b/include/paimon/catalog/identifier.h index dd86c9175..44207434e 100644 --- a/include/paimon/catalog/identifier.h +++ b/include/paimon/catalog/identifier.h @@ -36,7 +36,7 @@ class PAIMON_EXPORT Identifier { explicit Identifier(const std::string& table); Identifier(const std::string& database, const std::string& table); - bool operator==(const Identifier& other); + bool operator==(const Identifier& other) const; const std::string& GetDatabaseName() const; const std::string& GetTableName() const; Result GetDataTableName() const; @@ -44,7 +44,12 @@ class PAIMON_EXPORT Identifier { Result GetBranchNameOrDefault() const; Result> GetSystemTableName() const; Result IsSystemTable() const; + std::string GetFullName() const; std::string ToString() const; + int32_t HashCode() const; + + public: + static Result FromString(const std::string& full_name); private: Status SplitTableName() const; @@ -58,3 +63,13 @@ class PAIMON_EXPORT Identifier { }; } // namespace paimon + +namespace std { +template <> +struct hash { + size_t operator()(const paimon::Identifier& identifier) const { + return identifier.HashCode(); + } +}; + +} // namespace std diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 0d47aef95..a4e919030 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -25,6 +25,7 @@ set(PAIMON_COMMON_SRCS common/data/binary_string.cpp common/data/blob.cpp common/data/blob_descriptor.cpp + common/data/blob_view_struct.cpp common/data/blob_utils.cpp common/data/columnar/columnar_array.cpp common/data/columnar/columnar_map.cpp @@ -110,6 +111,7 @@ set(PAIMON_COMMON_SRCS common/reader/predicate_batch_reader.cpp common/reader/prefetch_file_batch_reader_impl.cpp common/reader/reader_utils.cpp + common/reader/blob_view_resolving_batch_reader.cpp common/reader/complete_row_kind_batch_reader.cpp common/reader/data_evolution_file_reader.cpp common/sst/block_handle.cpp @@ -332,6 +334,7 @@ set(PAIMON_CORE_SRCS core/table/system/system_table_schema.cpp core/tag/tag.cpp core/utils/branch_manager.cpp + core/utils/blob_view_lookup.cpp core/utils/consumer_manager.cpp core/utils/field_mapping.cpp core/utils/file_store_path_factory.cpp @@ -408,6 +411,7 @@ if(PAIMON_BUILD_TESTS) common/data/timestamp_test.cpp common/data/blob_test.cpp common/data/blob_descriptor_test.cpp + common/data/blob_view_struct_test.cpp common/data/blob_utils_test.cpp common/executor/default_executor_test.cpp common/format/column_stats_test.cpp @@ -470,6 +474,7 @@ if(PAIMON_BUILD_TESTS) common/reader/prefetch_file_batch_reader_impl_test.cpp common/reader/reader_utils_test.cpp common/reader/complete_row_kind_batch_reader_test.cpp + common/reader/blob_view_resolving_batch_reader_test.cpp common/reader/data_evolution_file_reader_test.cpp common/reader/data_evolution_array_test.cpp common/reader/data_evolution_row_test.cpp @@ -715,6 +720,7 @@ if(PAIMON_BUILD_TESTS) core/table/source/table_scan_test.cpp core/table/system/system_table_test.cpp core/tag/tag_test.cpp + core/utils/blob_view_lookup_test.cpp core/utils/branch_manager_test.cpp core/utils/consumer_manager_test.cpp core/utils/file_store_path_factory_cache_test.cpp diff --git a/src/paimon/common/catalog/catalog_context.h b/src/paimon/common/catalog/catalog_context.h new file mode 100644 index 000000000..03b7d0918 --- /dev/null +++ b/src/paimon/common/catalog/catalog_context.h @@ -0,0 +1,35 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once +#include +#include + +#include "paimon/fs/file_system.h" + +namespace paimon { +struct CatalogContext { + CatalogContext(const std::string& _root_path, + const std::map& _options, + const std::shared_ptr& _file_system) + : root_path(_root_path), options(_options), file_system(_file_system) {} + + std::string root_path; + std::map options; + std::shared_ptr file_system; +}; + +} // namespace paimon diff --git a/src/paimon/common/data/blob_descriptor.cpp b/src/paimon/common/data/blob_descriptor.cpp index aff60eef9..4b0701090 100644 --- a/src/paimon/common/data/blob_descriptor.cpp +++ b/src/paimon/common/data/blob_descriptor.cpp @@ -50,7 +50,7 @@ Result> BlobDescriptor::Create(int8_t version, PAIMON_UNIQUE_PTR BlobDescriptor::Serialize(const std::shared_ptr& pool) const { MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); out.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN); - out.WriteValue(version_); + out.WriteValue(kCurrentVersion); out.WriteValue(kMagic); out.WriteValue(static_cast(uri_.size())); diff --git a/src/paimon/common/data/blob_descriptor_test.cpp b/src/paimon/common/data/blob_descriptor_test.cpp index c3c1ad08f..32ea9cf36 100644 --- a/src/paimon/common/data/blob_descriptor_test.cpp +++ b/src/paimon/common/data/blob_descriptor_test.cpp @@ -56,6 +56,10 @@ TEST_F(BlobDescriptorTest, TestDeserializeCompatibilityForJavaWithVersion1) { ASSERT_EQ(descriptor->Uri(), "test_uri"); ASSERT_EQ(descriptor->Offset(), 1024); ASSERT_EQ(descriptor->Length(), 2048); + + PAIMON_UNIQUE_PTR cpp_serialized = descriptor->Serialize(pool_); + auto cpp_serialized_string = std::string(cpp_serialized->data(), cpp_serialized->size()); + ASSERT_NE(cpp_serialized_string, java_serialized); } TEST_F(BlobDescriptorTest, TestDeserializeCompatibilityForJavaWithVersion2) { diff --git a/src/paimon/common/data/blob_view_struct.cpp b/src/paimon/common/data/blob_view_struct.cpp new file mode 100644 index 000000000..1a21b1136 --- /dev/null +++ b/src/paimon/common/data/blob_view_struct.cpp @@ -0,0 +1,113 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "paimon/common/data/blob_view_struct.h" + +#include + +#include "fmt/format.h" +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/memory/memory_segment_utils.h" +#include "paimon/common/utils/murmurhash_utils.h" +#include "paimon/io/byte_array_input_stream.h" +#include "paimon/io/byte_order.h" +#include "paimon/io/data_input_stream.h" +#include "paimon/memory/bytes.h" +#include "paimon/status.h" + +namespace paimon { +PAIMON_UNIQUE_PTR BlobViewStruct::Serialize(const std::shared_ptr& pool) const { + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + out.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN); + + out.WriteValue(kCurrentVersion); + out.WriteValue(kMagic); + std::string identifier = identifier_.GetFullName(); + out.WriteValue(static_cast(identifier.size())); + auto uri_bytes = std::make_shared(identifier, pool.get()); + out.WriteBytes(uri_bytes); + out.WriteValue(field_id_); + out.WriteValue(row_id_); + return MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); +} + +Result> BlobViewStruct::Deserialize(const char* buffer, + uint64_t size) { + auto input_stream = std::make_shared(buffer, size); + DataInputStream in(std::move(input_stream)); + in.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN); + + PAIMON_ASSIGN_OR_RAISE(int8_t version, in.ReadValue()); + if (version != kCurrentVersion) { + return Status::Invalid(fmt::format( + "Expecting BlobViewStruct version to be {}, but found {}.", kCurrentVersion, version)); + } + PAIMON_ASSIGN_OR_RAISE(int64_t magic, in.ReadValue()); + if (kMagic != magic) { + return Status::Invalid( + fmt::format("Invalid BlobViewStruct: missing magic header. Expected magic: {}, " + "but found {}", + kMagic, magic)); + } + PAIMON_ASSIGN_OR_RAISE(int32_t length, in.ReadValue()); + std::string identifier_str(length, '\0'); + PAIMON_RETURN_NOT_OK(in.Read(identifier_str.data(), identifier_str.size())); + PAIMON_ASSIGN_OR_RAISE(int32_t field_id, in.ReadValue()); + PAIMON_ASSIGN_OR_RAISE(int64_t row_id, in.ReadValue()); + PAIMON_ASSIGN_OR_RAISE(Identifier identifier, Identifier::FromString(identifier_str)); + return std::make_unique(identifier, field_id, row_id); +} + +Result BlobViewStruct::IsBlobViewStruct(const char* buffer, uint64_t size) { + if (size < kMinViewLength) { + return false; + } + auto input_stream = std::make_shared(buffer, size); + DataInputStream in(std::move(input_stream)); + in.SetOrder(ByteOrder::PAIMON_LITTLE_ENDIAN); + + PAIMON_ASSIGN_OR_RAISE(int8_t version, in.ReadValue()); + if (version != kCurrentVersion) { + return false; + } + PAIMON_ASSIGN_OR_RAISE(int64_t magic, in.ReadValue()); + return kMagic == magic; +} + +std::string BlobViewStruct::ToString() const { + return fmt::format("BlobViewStruct{{identifier={}, fieldId={}, rowId={}}}", + identifier_.GetFullName(), field_id_, row_id_); +} + +bool BlobViewStruct::operator==(const BlobViewStruct& other) const { + if (this == &other) { + return true; + } + return field_id_ == other.field_id_ && row_id_ == other.row_id_ && + identifier_ == other.identifier_; +} + +bool BlobViewStruct::operator!=(const BlobViewStruct& other) const { + return !(*this == other); +} + +int32_t BlobViewStruct::HashCode() const { + int32_t hash = + MurmurHashUtils::HashUnsafeBytes(reinterpret_cast(&field_id_), + /*offset=*/0, sizeof(field_id_), identifier_.HashCode()); + return MurmurHashUtils::HashUnsafeBytes(reinterpret_cast(&row_id_), + /*offset=*/0, sizeof(row_id_), hash); +} +} // namespace paimon diff --git a/src/paimon/common/data/blob_view_struct.h b/src/paimon/common/data/blob_view_struct.h new file mode 100644 index 000000000..f0ac2cb32 --- /dev/null +++ b/src/paimon/common/data/blob_view_struct.h @@ -0,0 +1,85 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include + +#include "paimon/catalog/identifier.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" + +namespace paimon { +/// Serialized metadata for a BLOB view field. +/// A blob view only stores the coordinates needed to locate the original blob value in the +/// upstream table: identifier, field_id and row_id. The actual blob data is +/// resolved at read time by scanning the upstream table. +class BlobViewStruct { + public: + BlobViewStruct(const Identifier& identifier, int32_t field_id, int64_t row_id) + : identifier_(identifier), field_id_(field_id), row_id_(row_id) {} + + const Identifier& GetIdentifier() const { + return identifier_; + } + + int32_t FieldId() const { + return field_id_; + } + + int64_t RowId() const { + return row_id_; + } + + static Result> Deserialize(const char* buffer, uint64_t size); + static Result IsBlobViewStruct(const char* buffer, uint64_t size); + PAIMON_UNIQUE_PTR Serialize(const std::shared_ptr& pool) const; + std::string ToString() const; + int32_t HashCode() const; + + bool operator==(const BlobViewStruct& other) const; + bool operator!=(const BlobViewStruct& other) const; + + private: + static constexpr int64_t kMagic = 0x424C4F4256494557l; + static constexpr int8_t kCurrentVersion = 1; + /// one byte for version, eight bytes for magic number. + static constexpr uint64_t kMinViewLength = 9; + + Identifier identifier_; + int32_t field_id_; + int64_t row_id_; +}; + +/// Resolves a BlobViewStruct into the serialized BlobDescriptor bytes stored in the upstream +/// table. Returns nullptr when the referenced source-table cell is null. +using BlobViewResolver = std::function>(const BlobViewStruct&)>; + +} // namespace paimon + +namespace std { +template <> +struct hash { + size_t operator()(const paimon::BlobViewStruct& blob_view_struct) const { + return blob_view_struct.HashCode(); + } +}; + +} // namespace std diff --git a/src/paimon/common/data/blob_view_struct_test.cpp b/src/paimon/common/data/blob_view_struct_test.cpp new file mode 100644 index 000000000..58c68fa3d --- /dev/null +++ b/src/paimon/common/data/blob_view_struct_test.cpp @@ -0,0 +1,109 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "paimon/common/data/blob_view_struct.h" + +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/catalog/identifier.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/status.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +class BlobViewStructTest : public testing::Test { + public: + std::shared_ptr pool_ = GetDefaultPool(); + Identifier identifier_ = Identifier("test_db", "test_table"); + BlobViewStruct view_struct_ = BlobViewStruct(identifier_, /*field_id=*/7, /*row_id=*/1024); +}; + +TEST_F(BlobViewStructTest, TestConstructorAndGetters) { + ASSERT_EQ(view_struct_.GetIdentifier().GetDatabaseName(), "test_db"); + ASSERT_EQ(view_struct_.GetIdentifier().GetTableName(), "test_table"); + ASSERT_EQ(view_struct_.FieldId(), 7); + ASSERT_EQ(view_struct_.RowId(), 1024); +} + +TEST_F(BlobViewStructTest, TestSerializeDeserializeRoundTrip) { + auto serialized = view_struct_.Serialize(pool_); + ASSERT_NE(serialized, nullptr); + ASSERT_GT(serialized->size(), 0u); + + ASSERT_OK_AND_ASSIGN(auto restored, + BlobViewStruct::Deserialize(serialized->data(), serialized->size())); + ASSERT_EQ(restored->GetIdentifier().GetDatabaseName(), "test_db"); + ASSERT_EQ(restored->GetIdentifier().GetTableName(), "test_table"); + ASSERT_EQ(restored->FieldId(), 7); + ASSERT_EQ(restored->RowId(), 1024); +} + +TEST_F(BlobViewStructTest, TestDeserializeWithInvalidVersion) { + auto serialized = view_struct_.Serialize(pool_); + (*serialized)[0] = '\x02'; // invalid version (current is 1). + ASSERT_NOK_WITH_MSG(BlobViewStruct::Deserialize(serialized->data(), serialized->size()), + "Expecting BlobViewStruct version to be 1, but found 2"); +} + +TEST_F(BlobViewStructTest, TestDeserializeWithInvalidMagic) { + auto serialized = view_struct_.Serialize(pool_); + (*serialized)[1] = '\x00'; + ASSERT_NOK_WITH_MSG(BlobViewStruct::Deserialize(serialized->data(), serialized->size()), + "missing magic header"); +} + +TEST_F(BlobViewStructTest, TestToString) { + std::string debug_str = view_struct_.ToString(); + ASSERT_EQ(debug_str, "BlobViewStruct{identifier=test_db.test_table, fieldId=7, rowId=1024}"); +} + +TEST_F(BlobViewStructTest, TestEqual) { + { + // test equal itself + ASSERT_EQ(view_struct_, view_struct_); + } + { + // test equal + BlobViewStruct other_view_struct = + BlobViewStruct(identifier_, /*field_id=*/7, /*row_id=*/1024); + ASSERT_EQ(view_struct_, other_view_struct); + } + { + // test wrong identifier + Identifier wrong_identifier = Identifier("db", "table"); + BlobViewStruct wrong_view_struct = + BlobViewStruct(wrong_identifier, /*field_id=*/7, /*row_id=*/1024); + ASSERT_NE(view_struct_, wrong_view_struct); + } + { + // test wrong field_id + BlobViewStruct wrong_view_struct = + BlobViewStruct(identifier_, /*field_id=*/8, /*row_id=*/1024); + ASSERT_NE(view_struct_, wrong_view_struct); + } + { + // test wrong row_id + BlobViewStruct wrong_view_struct = + BlobViewStruct(identifier_, /*field_id=*/7, /*row_id=*/1000); + ASSERT_NE(view_struct_, wrong_view_struct); + } +} + +} // namespace paimon::test diff --git a/src/paimon/common/reader/blob_view_resolving_batch_reader.cpp b/src/paimon/common/reader/blob_view_resolving_batch_reader.cpp new file mode 100644 index 000000000..9a876c349 --- /dev/null +++ b/src/paimon/common/reader/blob_view_resolving_batch_reader.cpp @@ -0,0 +1,126 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/common/reader/blob_view_resolving_batch_reader.h" + +#include +#include + +#include "arrow/api.h" +#include "arrow/array/array_base.h" +#include "arrow/array/array_binary.h" +#include "arrow/array/array_nested.h" +#include "arrow/array/builder_binary.h" +#include "arrow/c/abi.h" +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "paimon/common/data/blob_view_struct.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/memory/bytes.h" +#include "paimon/status.h" + +namespace paimon { +BlobViewResolvingBatchReader::BlobViewResolvingBatchReader( + std::unique_ptr&& reader, std::vector read_blob_view_fields, + BlobViewResolver resolver, const std::shared_ptr& pool) + : pool_(pool), + arrow_pool_(GetArrowPool(pool)), + reader_(std::move(reader)), + read_blob_view_fields_(std::make_move_iterator(read_blob_view_fields.begin()), + std::make_move_iterator(read_blob_view_fields.end())), + resolver_(std::move(resolver)) {} + +Result BlobViewResolvingBatchReader::NextBatch() { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch, reader_->NextBatch()); + if (BatchReader::IsEofBatch(batch)) { + return batch; + } + if (read_blob_view_fields_.empty()) { + return batch; + } + + auto& [c_array, c_schema] = batch; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr arrow_array, + arrow::ImportArray(c_array.get(), c_schema.get())); + auto struct_array = std::dynamic_pointer_cast(arrow_array); + if (struct_array == nullptr) { + return Status::Invalid( + "invalid batch, BlobViewResolvingBatchReader expects a StructArray batch."); + } + const auto struct_type = struct_array->struct_type(); + + arrow::ArrayVector new_fields = struct_array->fields(); + std::vector field_names; + field_names.reserve(struct_type->num_fields()); + + for (int32_t field_idx = 0; field_idx < struct_type->num_fields(); ++field_idx) { + const auto& field = struct_type->field(field_idx); + field_names.push_back(field->name()); + if (read_blob_view_fields_.find(field->name()) == read_blob_view_fields_.end()) { + continue; + } + const auto& column = struct_array->field(field_idx); + if (auto large_binary_array = std::dynamic_pointer_cast(column)) { + PAIMON_ASSIGN_OR_RAISE(new_fields[field_idx], ResolveBinaryColumn(large_binary_array)); + } else { + return Status::Invalid(fmt::format( + "BlobViewResolvingBatchReader expects blob-view column {} to be LargeBinaryArray.", + field->name())); + } + } + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr resolved_struct_array, + arrow::StructArray::Make(new_fields, field_names)); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + arrow::ExportArray(*resolved_struct_array, c_array.get(), c_schema.get())); + return batch; +} + +Result> BlobViewResolvingBatchReader::ResolveBinaryColumn( + const std::shared_ptr& blob_view_struct_array) { + arrow::LargeBinaryBuilder builder(arrow_pool_.get()); + PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(blob_view_struct_array->length())); + for (int64_t row = 0; row < blob_view_struct_array->length(); ++row) { + if (blob_view_struct_array->IsNull(row)) { + PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendNull()); + continue; + } + auto view = blob_view_struct_array->GetView(row); + PAIMON_ASSIGN_OR_RAISE(bool is_view_struct, + BlobViewStruct::IsBlobViewStruct(view.data(), view.size())); + if (!is_view_struct) { + return Status::Invalid( + "BlobViewResolvingBatchReader expects a serialized BlobViewStruct in the " + "blob-view column."); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr view_struct, + BlobViewStruct::Deserialize(view.data(), view.size())); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr descriptor_bytes, resolver_(*view_struct)); + if (descriptor_bytes == nullptr) { + // null in source table + PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.AppendNull()); + continue; + } + PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Append( + reinterpret_cast(descriptor_bytes->data()), descriptor_bytes->size())); + } + std::shared_ptr blob_descriptor_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&blob_descriptor_array)); + return blob_descriptor_array; +} + +} // namespace paimon diff --git a/src/paimon/common/reader/blob_view_resolving_batch_reader.h b/src/paimon/common/reader/blob_view_resolving_batch_reader.h new file mode 100644 index 000000000..f85c5458a --- /dev/null +++ b/src/paimon/common/reader/blob_view_resolving_batch_reader.h @@ -0,0 +1,62 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "arrow/memory_pool.h" +#include "paimon/common/data/blob_view_struct.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/reader/batch_reader.h" +#include "paimon/result.h" + +namespace paimon { +class BlobViewResolvingBatchReader : public BatchReader { + public: + BlobViewResolvingBatchReader(std::unique_ptr&& reader, + std::vector read_blob_view_fields, + BlobViewResolver resolver, + const std::shared_ptr& pool); + + Result NextBatch() override; + + void Close() override { + reader_->Close(); + } + + std::shared_ptr GetReaderMetrics() const override { + return reader_->GetReaderMetrics(); + } + + private: + Result> ResolveBinaryColumn( + const std::shared_ptr& blob_view_struct_array); + + private: + std::shared_ptr pool_; + std::unique_ptr arrow_pool_; + std::unique_ptr reader_; + std::set read_blob_view_fields_; + BlobViewResolver resolver_; +}; + +} // namespace paimon diff --git a/src/paimon/common/reader/blob_view_resolving_batch_reader_test.cpp b/src/paimon/common/reader/blob_view_resolving_batch_reader_test.cpp new file mode 100644 index 000000000..facb0308f --- /dev/null +++ b/src/paimon/common/reader/blob_view_resolving_batch_reader_test.cpp @@ -0,0 +1,211 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/common/reader/blob_view_resolving_batch_reader.h" + +#include +#include +#include +#include +#include + +#include "arrow/api.h" +#include "arrow/array/array_binary.h" +#include "arrow/array/array_nested.h" +#include "arrow/array/builder_binary.h" +#include "arrow/c/abi.h" +#include "arrow/c/bridge.h" +#include "gtest/gtest.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/data/blob_descriptor.h" +#include "paimon/common/data/blob_utils.h" +#include "paimon/common/data/blob_view_struct.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/reader/batch_reader.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/testing/utils/read_result_collector.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { + +class BlobViewResolvingBatchReaderTest : public ::testing::Test { + public: + void SetUp() override { + pool_ = GetDefaultPool(); + } + + void TearDown() override { + pool_.reset(); + } + + class InMemoryBatchReader : public BatchReader { + public: + explicit InMemoryBatchReader(const std::shared_ptr& struct_array) + : struct_array_(struct_array) { + if (!struct_array_) { + exhausted_ = true; + } + } + + Result NextBatch() override { + if (exhausted_) { + return MakeEofBatch(); + } + exhausted_ = true; + auto c_array = std::make_unique(); + auto c_schema = std::make_unique(); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + arrow::ExportArray(*struct_array_, c_array.get(), c_schema.get())); + return std::make_pair(std::move(c_array), std::move(c_schema)); + } + + std::shared_ptr GetReaderMetrics() const override { + return std::make_shared(); + } + + void Close() override {} + + private: + std::shared_ptr struct_array_; + bool exhausted_ = false; + }; + + std::string MakeBlobViewStructBytes(const std::string& database, const std::string& table, + int32_t field_id, int64_t row_id) const { + Identifier identifier(database, table); + BlobViewStruct view_struct(identifier, field_id, row_id); + auto bytes = view_struct.Serialize(pool_); + return std::string(bytes->data(), bytes->size()); + } + + Result MakeBlobDescriptorBytes(const std::string& uri, int64_t offset, + int64_t length) const { + PAIMON_ASSIGN_OR_RAISE(auto descriptor, BlobDescriptor::Create(uri, offset, length)); + auto bytes = descriptor->Serialize(pool_); + return std::string(bytes->data(), bytes->size()); + } + + std::shared_ptr BuildStructArray(const std::vector& values, + const std::vector& valid) const { + arrow::LargeBinaryBuilder builder; + EXPECT_TRUE(builder.Reserve(static_cast(values.size())).ok()); + for (size_t i = 0; i < values.size(); ++i) { + if (!valid[i]) { + EXPECT_TRUE(builder.AppendNull().ok()); + } else { + EXPECT_TRUE(builder + .Append(reinterpret_cast(values[i].data()), + static_cast(values[i].size())) + .ok()); + } + } + std::shared_ptr array; + EXPECT_TRUE(builder.Finish(&array).ok()); + + arrow::FieldVector fields = {BlobUtils::ToArrowField("blob_col", /*nullable=*/true)}; + arrow::ArrayVector arrays = {array}; + auto result = arrow::StructArray::Make(arrays, fields).ValueOrDie(); + return result; + } + + private: + std::shared_ptr pool_; +}; + +TEST_F(BlobViewResolvingBatchReaderTest, TestEofBatch) { + auto inner_reader = std::make_unique(nullptr); + auto resolver = BlobViewResolver([](const BlobViewStruct&) -> Result> { + return std::shared_ptr(); + }); + BlobViewResolvingBatchReader reader(std::move(inner_reader), {"blob_col"}, std::move(resolver), + pool_); + ASSERT_OK_AND_ASSIGN(auto batch, reader.NextBatch()); + ASSERT_TRUE(BatchReader::IsEofBatch(batch)); +} + +TEST_F(BlobViewResolvingBatchReaderTest, TestEmptyReadBlobViewFields) { + std::string view_bytes = MakeBlobViewStructBytes("db", "table", /*field_id=*/1, /*row_id=*/7); + std::shared_ptr struct_array = BuildStructArray({view_bytes}, {true}); + + bool resolver_called = false; + auto resolver = BlobViewResolver( + [&resolver_called](const BlobViewStruct&) -> Result> { + resolver_called = true; + return std::shared_ptr(); + }); + + auto inner_reader = std::make_unique(struct_array); + BlobViewResolvingBatchReader reader(std::move(inner_reader), /*read_blob_view_fields=*/{}, + std::move(resolver), pool_); + ASSERT_OK_AND_ASSIGN(auto result_array, ReadResultCollector::CollectResult(&reader)); + auto expected_array = std::make_shared(struct_array); + ASSERT_TRUE(expected_array->Equals(*result_array)); + ASSERT_FALSE(resolver_called); +} + +TEST_F(BlobViewResolvingBatchReaderTest, TestResolvesBlobViewColumn) { + auto row0_view = MakeBlobViewStructBytes("db", "tbl", /*field_id=*/3, /*row_id=*/100); + auto row1_view = MakeBlobViewStructBytes("db", "tbl", /*field_id=*/3, /*row_id=*/200); + std::shared_ptr src_struct = + BuildStructArray({row0_view, row1_view}, {true, true}); + + ASSERT_OK_AND_ASSIGN(auto expected_row0_descriptor, + MakeBlobDescriptorBytes("/path/a", /*offset=*/0, /*length=*/8)); + ASSERT_OK_AND_ASSIGN(auto expected_row1_descriptor, + MakeBlobDescriptorBytes("/path/b", /*offset=*/16, /*length=*/32)); + + auto resolver = + BlobViewResolver([&](const BlobViewStruct& view_struct) -> Result> { + if (view_struct.RowId() == 100) { + return std::make_shared(expected_row0_descriptor, pool_.get()); + } + if (view_struct.RowId() == 200) { + return std::make_shared(expected_row1_descriptor, pool_.get()); + } + return Status::Invalid("unexpected view struct"); + }); + + auto inner_reader = std::make_unique(src_struct); + BlobViewResolvingBatchReader reader(std::move(inner_reader), {"blob_col"}, std::move(resolver), + pool_); + ASSERT_OK_AND_ASSIGN(auto result_array, ReadResultCollector::CollectResult(&reader)); + auto struct_array = std::dynamic_pointer_cast(result_array->chunk(0)); + + auto result_blob_column = + std::dynamic_pointer_cast(struct_array->field(0)); + ASSERT_FALSE(result_blob_column->IsNull(0)); + ASSERT_FALSE(result_blob_column->IsNull(1)); + ASSERT_EQ(result_blob_column->GetString(0), expected_row0_descriptor); + ASSERT_EQ(result_blob_column->GetString(1), expected_row1_descriptor); +} + +TEST_F(BlobViewResolvingBatchReaderTest, TestResolverError) { + auto view_bytes = MakeBlobViewStructBytes("db", "tbl", /*field_id=*/1, /*row_id=*/5); + std::shared_ptr src_struct = BuildStructArray({view_bytes}, {true}); + auto resolver = BlobViewResolver([](const BlobViewStruct&) -> Result> { + return Status::Invalid("cache miss"); + }); + auto inner_reader = std::make_unique(src_struct); + BlobViewResolvingBatchReader reader(std::move(inner_reader), {"blob_col"}, std::move(resolver), + pool_); + ASSERT_NOK_WITH_MSG(reader.NextBatch(), "cache miss"); +} + +} // namespace paimon::test diff --git a/src/paimon/core/catalog/identifier.cpp b/src/paimon/core/catalog/identifier.cpp index 3ec998c83..658fec2ad 100644 --- a/src/paimon/core/catalog/identifier.cpp +++ b/src/paimon/core/catalog/identifier.cpp @@ -21,6 +21,7 @@ #include #include "fmt/format.h" +#include "paimon/common/utils/murmurhash_utils.h" #include "paimon/common/utils/string_utils.h" #include "paimon/result.h" #include "paimon/status.h" @@ -38,8 +39,11 @@ Identifier::Identifier(const std::string& table) Identifier::Identifier(const std::string& database, const std::string& table) : database_(database), table_(table) {} -bool Identifier::operator==(const Identifier& other) { - return (database_ == other.database_ && table_ == other.table_); +bool Identifier::operator==(const Identifier& other) const { + if (this == &other) { + return true; + } + return database_ == other.database_ && table_ == other.table_; } const std::string& Identifier::GetDatabaseName() const { @@ -79,6 +83,35 @@ std::string Identifier::ToString() const { return fmt::format("Identifier{{database='{}', table='{}'}}", database_, table_); } +int32_t Identifier::HashCode() const { + int32_t hash = MurmurHashUtils::HashUnsafeBytes(reinterpret_cast(database_.data()), + /*offset=*/0, database_.size()); + return MurmurHashUtils::HashUnsafeBytes(reinterpret_cast(table_.data()), + /*offset=*/0, table_.size(), hash); +} + +std::string Identifier::GetFullName() const { + if (database_ == kUnknownDatabase) { + return table_; + } + return fmt::format("{}.{}", database_, table_); +} + +Result Identifier::FromString(const std::string& full_name) { + if (StringUtils::IsNullOrWhitespaceOnly(full_name)) { + return Status::Invalid("full name cannot be empty or whitespace only"); + } + // TODO(lisizhuo.lsz): deal with kUnknownDatabase to be done + const auto dot_pos = full_name.find('.'); + if (dot_pos == std::string::npos || dot_pos == 0 || dot_pos == full_name.size() - 1) { + return Status::Invalid( + fmt::format("cannot get splits from '{}' to get database and table", full_name)); + } + std::string database = full_name.substr(0, dot_pos); + std::string table = full_name.substr(dot_pos + 1); + return Identifier(database, table); +} + Status Identifier::SplitTableName() const { if (parsed_) { return Status::OK(); diff --git a/src/paimon/core/catalog/identifier_test.cpp b/src/paimon/core/catalog/identifier_test.cpp index 836ecd4c6..efc6ab16a 100644 --- a/src/paimon/core/catalog/identifier_test.cpp +++ b/src/paimon/core/catalog/identifier_test.cpp @@ -111,6 +111,47 @@ TEST(IdentifierTest, ParseBranchSystemTable) { ASSERT_TRUE(is_system_table); } +TEST(IdentifierTest, TestGetFullName) { + { + // test unknown database + std::string table = "tbl$branch_dev$options"; + Identifier id(table); + ASSERT_EQ(table, id.GetFullName()); + } + { + // test normal database + std::string db = "database"; + std::string table = "tbl$branch_dev$options"; + Identifier id(db, table); + ASSERT_EQ("database.tbl$branch_dev$options", id.GetFullName()); + } +} + +TEST(IdentifierTest, TestFromString) { + { + // test invalid identifier string + ASSERT_NOK_WITH_MSG(Identifier::FromString(""), + "full name cannot be empty or whitespace only"); + ASSERT_NOK_WITH_MSG(Identifier::FromString(" "), + "full name cannot be empty or whitespace only"); + ASSERT_NOK_WITH_MSG(Identifier::FromString("abcd"), "cannot get splits from 'abcd'"); + ASSERT_NOK_WITH_MSG(Identifier::FromString(".abcd"), "cannot get splits from '.abcd'"); + ASSERT_NOK_WITH_MSG(Identifier::FromString("abcd."), "cannot get splits from 'abcd.'"); + } + { + // test normal database + ASSERT_OK_AND_ASSIGN(Identifier identifier, Identifier::FromString("ab.cd")); + Identifier expected("ab", "cd"); + ASSERT_EQ(identifier, expected); + } + { + // test normal database + ASSERT_OK_AND_ASSIGN(Identifier identifier, Identifier::FromString("ab.cd.ef")); + Identifier expected("ab", "cd.ef"); + ASSERT_EQ(identifier, expected); + } +} + TEST(IdentifierTest, InvalidSystemTableName) { Identifier invalid_middle("db", "tbl$bad$options"); ASSERT_NOK_WITH_MSG(invalid_middle.IsSystemTable(), diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index 61a6e4c86..fef2d94b6 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -17,18 +17,32 @@ #include "paimon/core/operation/data_evolution_split_read.h" #include +#include +#include +#include #include +#include "arrow/array/array_base.h" +#include "arrow/array/array_binary.h" +#include "arrow/array/array_nested.h" +#include "arrow/c/bridge.h" +#include "paimon/common/catalog/catalog_context.h" #include "paimon/common/data/blob_utils.h" +#include "paimon/common/data/blob_view_struct.h" #include "paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader.h" #include "paimon/common/global_index/complete_index_score_batch_reader.h" +#include "paimon/common/reader/blob_view_resolving_batch_reader.h" #include "paimon/common/reader/complete_row_kind_batch_reader.h" #include "paimon/common/reader/concat_batch_reader.h" #include "paimon/common/table/special_fields.h" +#include "paimon/common/types/data_field.h" +#include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/object_utils.h" +#include "paimon/common/utils/path_util.h" #include "paimon/common/utils/range_helper.h" #include "paimon/core/core_options.h" #include "paimon/core/global_index/indexed_split_impl.h" +#include "paimon/core/utils/blob_view_lookup.h" namespace paimon { Status DataEvolutionSplitRead::BlobBunch::Add(const std::shared_ptr& file) { if (!BlobUtils::IsBlobFile(file->file_name)) { @@ -113,28 +127,157 @@ bool DataEvolutionSplitRead::HasIndexScoreField(const std::shared_ptrGetFieldIndex(SpecialFields::IndexScore().Name()) != -1; } +std::vector DataEvolutionSplitRead::HasBlobViewField( + const CoreOptions& options, const std::shared_ptr& read_schema) { + std::vector read_blob_view_fields; + std::vector blob_view_fields = options.GetBlobViewFields(); + for (const auto& blob : blob_view_fields) { + if (read_schema->GetFieldByName(blob)) { + read_blob_view_fields.push_back(blob); + } + } + return read_blob_view_fields; +} + Result> DataEvolutionSplitRead::CreateReader( const std::shared_ptr& split) { if (auto indexed_split = std::dynamic_pointer_cast(split)) { PAIMON_RETURN_NOT_OK(indexed_split->Validate()); - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr batch_reader, - InnerCreateReader(indexed_split->GetDataSplit(), indexed_split->RowRanges())); + const auto& data_split = indexed_split->GetDataSplit(); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr batch_reader, + InnerCreateReader(data_split, indexed_split->RowRanges())); if (HasIndexScoreField(raw_read_schema_)) { - return std::make_unique(std::move(batch_reader), - indexed_split->Scores(), pool_); + batch_reader = std::make_unique( + std::move(batch_reader), indexed_split->Scores(), pool_); } - return batch_reader; + return WrapWithBlobViewResolverIfNeeded(data_split, std::move(batch_reader)); } else if (auto data_split = std::dynamic_pointer_cast(split)) { if (HasIndexScoreField(raw_read_schema_)) { return Status::Invalid( "Invalid read schema, read _INDEX_SCORE while split cannot cast to IndexedSplit"); } - return InnerCreateReader(data_split, /*row_ranges=*/std::nullopt); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr inner_reader, + InnerCreateReader(data_split, /*row_ranges=*/std::nullopt)); + return WrapWithBlobViewResolverIfNeeded(data_split, std::move(inner_reader)); } return Status::Invalid("Invalid Split, cannot cast to IndexedSplit or DataSplit"); } +Result> DataEvolutionSplitRead::WrapWithBlobViewResolverIfNeeded( + const std::shared_ptr& data_split, + std::unique_ptr&& inner_reader) const { + std::vector read_blob_view_fields = HasBlobViewField(options_, raw_read_schema_); + if (read_blob_view_fields.empty()) { + return std::move(inner_reader); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr pre_reader, + CreateBlobViewReader(data_split, read_blob_view_fields)); + PAIMON_ASSIGN_OR_RAISE(std::unordered_set blob_view_structs, + ExtractBlobViewStructs(pre_reader.get())); + std::string warehouse_path = + PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(context_->GetPath())); + auto catalog_context = std::make_shared(warehouse_path, options_.ToMap(), + options_.GetFileSystem()); + PAIMON_ASSIGN_OR_RAISE( + BlobViewResolver resolver, + BlobViewLookup::CreateResolver(blob_view_structs, catalog_context, pool_)); + return std::make_unique( + std::move(inner_reader), std::move(read_blob_view_fields), std::move(resolver), pool_); +} + +Result> DataEvolutionSplitRead::CreateBlobViewReader( + const std::shared_ptr& data_split, + const std::vector& read_blob_view_fields) const { + auto split_impl = dynamic_cast(data_split.get()); + if (split_impl == nullptr) { + return Status::Invalid("unexpected error, split cast to impl failed"); + } + assert(raw_read_schema_->num_fields() > 0); + + arrow::FieldVector blob_view_arrow_fields; + blob_view_arrow_fields.reserve(read_blob_view_fields.size()); + for (const auto& field_name : read_blob_view_fields) { + auto field = raw_read_schema_->GetFieldByName(field_name); + if (field == nullptr) { + return Status::Invalid( + fmt::format("Blob view field {} not found in read schema.", field_name)); + } + blob_view_arrow_fields.push_back(std::move(field)); + } + auto blob_view_schema = arrow::schema(std::move(blob_view_arrow_fields)); + + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr data_file_path_factory, + path_factory_->CreateDataFilePathFactory(split_impl->Partition(), split_impl->Bucket())); + + // skip blob files: they only contain blob payloads, not the blob-view columns. + std::vector> data_files; + data_files.reserve(split_impl->DataFiles().size()); + for (const auto& file : split_impl->DataFiles()) { + if (!BlobUtils::IsBlobFile(file->file_name)) { + data_files.push_back(file); + } + } + PAIMON_ASSIGN_OR_RAISE( + std::vector> raw_file_readers, + CreateRawFileReaders(split_impl->Partition(), data_files, blob_view_schema, + /*predicate=*/nullptr, /*dv_factory=*/nullptr, + /*row_ranges=*/std::nullopt, data_file_path_factory)); + + auto batch_readers = + ObjectUtils::MoveVector>(std::move(raw_file_readers)); + return std::make_unique(std::move(batch_readers), pool_); +} + +Result> DataEvolutionSplitRead::ExtractBlobViewStructs( + BatchReader* reader) { + if (reader == nullptr) { + return Status::Invalid("invalid reader in ExtractBlobViewStructs, reader is nullptr"); + } + std::unordered_set blob_view_structs; + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch, reader->NextBatch()); + if (BatchReader::IsEofBatch(batch)) { + break; + } + auto& [c_array, c_schema] = batch; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr arrow_array, + arrow::ImportArray(c_array.get(), c_schema.get())); + auto struct_array = std::dynamic_pointer_cast(arrow_array); + if (struct_array == nullptr) { + return Status::Invalid( + "invalid array in ExtractBlobViewStructs, batch array is not a StructArray."); + } + + for (int32_t field_idx = 0; field_idx < struct_array->num_fields(); ++field_idx) { + auto binary_array = + std::dynamic_pointer_cast(struct_array->field(field_idx)); + if (binary_array == nullptr) { + return Status::Invalid( + "invalid array in ExtractBlobViewStructs, blob view column is not a " + "LargeBinaryArray."); + } + for (int64_t row = 0; row < binary_array->length(); ++row) { + if (binary_array->IsNull(row)) { + continue; + } + std::string_view bytes = binary_array->GetView(row); + PAIMON_ASSIGN_OR_RAISE( + bool is_view, BlobViewStruct::IsBlobViewStruct(bytes.data(), bytes.size())); + if (!is_view) { + return Status::Invalid( + "blob-view-field requires blob field value to be a serialized " + "BlobViewStruct."); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr view_struct, + BlobViewStruct::Deserialize(bytes.data(), bytes.size())); + blob_view_structs.insert(*view_struct); + } + } + } + return blob_view_structs; +} + Result> DataEvolutionSplitRead::InnerCreateReader( const std::shared_ptr& data_split, const std::optional>& row_ranges) const { @@ -174,6 +317,7 @@ Result> DataEvolutionSplitRead::InnerCreateReader( ApplyPredicateFilterIfNeeded(std::move(concat_batch_reader), context_->GetPredicate())); return std::make_unique(std::move(batch_reader), pool_); } + Result> DataEvolutionSplitRead::ApplyIndexAndDvReaderIfNeeded( std::unique_ptr&& file_reader, const std::shared_ptr& file, const std::shared_ptr& data_schema, diff --git a/src/paimon/core/operation/data_evolution_split_read.h b/src/paimon/core/operation/data_evolution_split_read.h index 6f571f706..6e4ce6954 100644 --- a/src/paimon/core/operation/data_evolution_split_read.h +++ b/src/paimon/core/operation/data_evolution_split_read.h @@ -21,8 +21,10 @@ #include #include #include +#include #include +#include "paimon/common/data/blob_view_struct.h" #include "paimon/common/reader/data_evolution_file_reader.h" #include "paimon/core/io/data_file_meta.h" #include "paimon/core/operation/abstract_split_read.h" @@ -51,7 +53,8 @@ struct DeletionFile; /// otherwise, it must be present in the read path. /// /// Readers Overview: (ConcatBatchReader across -/// splits)->(CompleteIndexScoreBatchReader)->CompleteRowKindBatchReader->(PredicateBatchReader) +/// splits)->(BlobViewResolvingBatchReader)->(CompleteIndexScoreBatchReader)-> +/// CompleteRowKindBatchReader->(PredicateBatchReader) /// ->ConcatBatchReader across files->DataEvolutionFileReader->(ConcatBatchReader across blob files) /// ->FieldMappingReader->(CompleteRowTrackingFieldsBatchReader) /// ->(DelegatingPrefetchReader)->(PrefetchFileBatchReader)->FormatReader @@ -141,6 +144,19 @@ class DataEvolutionSplitRead : public AbstractSplitRead { static bool HasIndexScoreField(const std::shared_ptr& read_schema); + static std::vector HasBlobViewField( + const CoreOptions& options, const std::shared_ptr& read_schema); + + static Result> ExtractBlobViewStructs(BatchReader* reader); + + Result> CreateBlobViewReader( + const std::shared_ptr& data_split, + const std::vector& read_blob_view_fields) const; + + Result> WrapWithBlobViewResolverIfNeeded( + const std::shared_ptr& data_split, + std::unique_ptr&& inner_reader) const; + private: Result> CreateUnionReader( const BinaryRow& partition, diff --git a/src/paimon/core/utils/blob_view_lookup.cpp b/src/paimon/core/utils/blob_view_lookup.cpp new file mode 100644 index 000000000..dff77a637 --- /dev/null +++ b/src/paimon/core/utils/blob_view_lookup.cpp @@ -0,0 +1,241 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/core/utils/blob_view_lookup.h" + +#include +#include + +#include "arrow/array.h" +#include "arrow/c/bridge.h" +#include "fmt/format.h" +#include "paimon/catalog/catalog.h" +#include "paimon/common/data/blob_descriptor.h" +#include "paimon/common/table/special_fields.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/defs.h" +#include "paimon/global_index/bitmap_global_index_result.h" +#include "paimon/memory/bytes.h" +#include "paimon/read_context.h" +#include "paimon/scan_context.h" +#include "paimon/table/source/table_read.h" +#include "paimon/table/source/table_scan.h" +#include "paimon/utils/special_field_ids.h" + +namespace paimon { +BlobViewLookup::TableReadPlan::TableReadPlan(const BlobViewStruct& view_struct) + : identifier_(view_struct.GetIdentifier()) { + references_by_field_id_.insert(view_struct.FieldId()); + row_ranges_.push_back(view_struct.RowId()); +} + +void BlobViewLookup::TableReadPlan::Add(const BlobViewStruct& view_struct) { + references_by_field_id_.insert(view_struct.FieldId()); + row_ranges_.push_back(view_struct.RowId()); +} + +std::vector BlobViewLookup::TableReadPlan::GetFieldIds() const { + return std::vector(references_by_field_id_.begin(), references_by_field_id_.end()); +} + +std::vector BlobViewLookup::TableReadPlan::GetSortedDistinctRanges() const { + if (row_ranges_.empty()) { + return {}; + } + std::vector sorted = row_ranges_; + std::sort(sorted.begin(), sorted.end()); + std::vector ranges; + int64_t range_start = sorted[0]; + int64_t range_end = range_start; + for (size_t i = 1; i < sorted.size(); ++i) { + const int64_t row_id = sorted[i]; + if (row_id == range_end) { + continue; + } + if (row_id != range_end + 1) { + ranges.emplace_back(range_start, range_end); + range_start = row_id; + } + range_end = row_id; + } + ranges.emplace_back(range_start, range_end); + return ranges; +} + +Result BlobViewLookup::CreateResolver( + const std::unordered_set& view_structs, + const std::shared_ptr& catalog_context, + const std::shared_ptr& pool) { + PAIMON_ASSIGN_OR_RAISE(DescriptorMapping mapping, + PreloadDescriptors(view_structs, catalog_context, pool)); + return BlobViewResolver([cached = std::move(mapping)](const BlobViewStruct& view_struct) + -> Result> { + auto iter = cached.find(view_struct); + if (iter == cached.end()) { + return Status::Invalid(fmt::format("BlobViewStruct not found in preloaded cache: {}", + view_struct.ToString())); + } + return iter->second; + }); +} + +Result BlobViewLookup::PreloadDescriptors( + const std::unordered_set& view_structs, + const std::shared_ptr& catalog_context, + const std::shared_ptr& pool) { + std::unordered_map plan_by_identifier = + GroupByIdentifier(view_structs); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr catalog, + Catalog::Create(catalog_context->root_path, catalog_context->options, + catalog_context->file_system)); + DescriptorMapping mapping; + for (const auto& [identifier, table_read_plan] : plan_by_identifier) { + std::string source_table_path = catalog->GetTableLocation(identifier); + PAIMON_ASSIGN_OR_RAISE(std::optional branch, identifier.GetBranchName()); + ScanContextBuilder scan_builder(source_table_path); + auto global_index_result = + BitmapGlobalIndexResult::FromRanges(table_read_plan.GetSortedDistinctRanges()); + scan_builder.SetGlobalIndexResult(global_index_result) + .WithMemoryPool(pool) + .WithFileSystem(catalog_context->file_system); + if (branch) { + scan_builder.AddOption(Options::BRANCH, branch.value()); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr scan_context, scan_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_scan, + TableScan::Create(std::move(scan_context))); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr plan, table_scan->CreatePlan()); + + ReadContextBuilder read_builder(source_table_path); + std::vector field_ids = table_read_plan.GetFieldIds(); + field_ids.push_back(SpecialFieldIds::ROW_ID); + read_builder.SetReadFieldIds(field_ids) + .AddOption(Options::BLOB_AS_DESCRIPTOR, "true") + .EnablePrefetch(true) + .WithMemoryPool(pool) + .WithFileSystem(catalog_context->file_system); + if (branch) { + read_builder.WithBranch(branch.value()); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr read_context, read_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_read, + TableRead::Create(std::move(read_context))); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr reader, + table_read->CreateReader(plan->Splits())); + PAIMON_RETURN_NOT_OK( + ExtractBlobDescriptors(identifier, field_ids, pool, reader.get(), &mapping)); + } + return mapping; +} + +Status BlobViewLookup::ExtractBlobDescriptors(const Identifier& identifier, + const std::vector& field_ids, + const std::shared_ptr& pool, + BatchReader* reader, DescriptorMapping* mapping) { + if (reader == nullptr) { + return Status::Invalid("invalid reader in ExtractBlobDescriptors, reader is nullptr"); + } + while (true) { + PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatch batch, reader->NextBatch()); + if (BatchReader::IsEofBatch(batch)) { + break; + } + auto& [c_array, c_schema] = batch; + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr arrow_array, + arrow::ImportArray(c_array.get(), c_schema.get())); + auto struct_array = std::dynamic_pointer_cast(arrow_array); + if (struct_array == nullptr) { + return Status::Invalid( + "invalid array in ExtractBlobDescriptors, batch array is not a StructArray."); + } + // skip the _VALUE_KIND column + if (static_cast(struct_array->num_fields()) - 1 != field_ids.size()) { + return Status::Invalid( + fmt::format("invalid array in ExtractBlobDescriptors, batch array fields(exclude " + "_VALUE_KIND) {} mismatch read field ids {}.", + struct_array->num_fields() - 1, field_ids.size())); + } + // get _VALUE_KIND + if (struct_array->struct_type()->field(0)->name() != SpecialFields::ValueKind().Name()) { + return Status::Invalid( + "invalid array in ExtractBlobDescriptors, expected _VALUE_KIND as the first " + "column"); + } + + // get _ROW_ID + if (struct_array->struct_type()->field(struct_array->num_fields() - 1)->name() != + SpecialFields::RowId().Name()) { + return Status::Invalid( + "invalid array in ExtractBlobDescriptors, expected _ROW_ID as the last column"); + } + auto row_id_array = struct_array->field(struct_array->num_fields() - 1); + auto typed_row_id_array = std::dynamic_pointer_cast(row_id_array); + if (!typed_row_id_array) { + return Status::Invalid( + fmt::format("invalid array does not contain {} field, or it cannot be casted to " + "Int64Array in ExtractBlobDescriptors.", + SpecialFields::RowId().Name())); + } + + // skip _VALUE_KIND + for (int32_t idx = 1; idx < struct_array->num_fields() - 1; ++idx) { + auto binary_array = + std::dynamic_pointer_cast(struct_array->field(idx)); + if (binary_array == nullptr) { + return Status::Invalid( + "invalid array in ExtractBlobDescriptors, column is not a LargeBinaryArray."); + } + for (int64_t row = 0; row < binary_array->length(); ++row) { + BlobViewStruct blob_view_struct(identifier, field_ids[idx - 1], + typed_row_id_array->Value(row)); + if (binary_array->IsNull(row)) { + // null in source table + (*mapping)[blob_view_struct] = nullptr; + continue; + } + std::string_view bytes = binary_array->GetView(row); + PAIMON_ASSIGN_OR_RAISE(bool is_descriptor, BlobDescriptor::IsBlobDescriptor( + bytes.data(), bytes.size())); + if (!is_descriptor) { + return Status::Invalid( + "requires blob field value to be a serialized BlobDescriptor in source " + "table."); + } + auto descriptor_bytes = std::make_shared(bytes.size(), pool.get()); + std::memcpy(descriptor_bytes->data(), bytes.data(), bytes.size()); + (*mapping)[blob_view_struct] = std::move(descriptor_bytes); + } + } + } + return Status::OK(); +} + +std::unordered_map BlobViewLookup::GroupByIdentifier( + const std::unordered_set& view_structs) { + std::unordered_map grouped; + for (const auto& view_struct : view_structs) { + auto identifier = view_struct.GetIdentifier(); + auto iter = grouped.find(identifier); + if (iter != grouped.end()) { + iter->second.Add(view_struct); + } else { + grouped.emplace(identifier, BlobViewLookup::TableReadPlan(view_struct)); + } + } + return grouped; +} + +} // namespace paimon diff --git a/src/paimon/core/utils/blob_view_lookup.h b/src/paimon/core/utils/blob_view_lookup.h new file mode 100644 index 000000000..3b3b3d026 --- /dev/null +++ b/src/paimon/core/utils/blob_view_lookup.h @@ -0,0 +1,80 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once +#include +#include +#include +#include +#include +#include +#include + +#include "paimon/catalog/identifier.h" +#include "paimon/common/catalog/catalog_context.h" +#include "paimon/common/data/blob_view_struct.h" +#include "paimon/memory/bytes.h" +#include "paimon/result.h" +#include "paimon/utils/range.h" + +namespace paimon { + +class BatchReader; + +/// Provide a function for converting {@link BlobViewStruct}s to {@link BlobDescriptor}s by scanning +/// the upstream tables in row-range chunks. +class BlobViewLookup { + public: + using DescriptorMapping = std::unordered_map>; + + BlobViewLookup() = delete; + ~BlobViewLookup() = delete; + + static Result CreateResolver( + const std::unordered_set& view_structs, + const std::shared_ptr& catalog_context, + const std::shared_ptr& pool); + + private: + class TableReadPlan { + public: + explicit TableReadPlan(const BlobViewStruct& view_struct); + + void Add(const BlobViewStruct& view_struct); + std::vector GetFieldIds() const; + std::vector GetSortedDistinctRanges() const; + + private: + Identifier identifier_; + std::set references_by_field_id_; + std::vector row_ranges_; + }; + + static Result PreloadDescriptors( + const std::unordered_set& view_structs, + const std::shared_ptr& catalog_context, + const std::shared_ptr& pool); + + static Status ExtractBlobDescriptors(const Identifier& identifier, + const std::vector& field_ids, + const std::shared_ptr& pool, + BatchReader* reader, DescriptorMapping* mapping); + + static std::unordered_map GroupByIdentifier( + const std::unordered_set& view_structs); +}; + +} // namespace paimon diff --git a/src/paimon/core/utils/blob_view_lookup_test.cpp b/src/paimon/core/utils/blob_view_lookup_test.cpp new file mode 100644 index 000000000..d5cbf7ab3 --- /dev/null +++ b/src/paimon/core/utils/blob_view_lookup_test.cpp @@ -0,0 +1,191 @@ +/* + * Copyright 2026-present Alibaba Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "paimon/core/utils/blob_view_lookup.h" + +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/data/blob_view_struct.h" +#include "paimon/utils/range.h" + +namespace paimon::test { + +class BlobViewLookupTest : public testing::Test { + public: + static Identifier MakeIdentifier(const std::string& database, const std::string& table) { + return Identifier(database, table); + } + + static BlobViewStruct MakeView(const std::string& database, const std::string& table, + int32_t field_id, int64_t row_id) { + return BlobViewStruct(MakeIdentifier(database, table), field_id, row_id); + } +}; + +TEST_F(BlobViewLookupTest, TestConstruct) { + BlobViewStruct view = MakeView("db", "t", /*field_id=*/7, /*row_id=*/42); + BlobViewLookup::TableReadPlan plan(view); + + ASSERT_EQ(plan.identifier_, MakeIdentifier("db", "t")); + const auto& references = plan.references_by_field_id_; + ASSERT_EQ(references.size(), 1U); + auto iter = references.find(7); + ASSERT_NE(iter, references.end()); + + const auto& row_ranges = plan.row_ranges_; + ASSERT_EQ(row_ranges.size(), 1U); + ASSERT_EQ(row_ranges[0], 42); + + ASSERT_EQ(plan.GetFieldIds(), std::vector{7}); +} + +TEST_F(BlobViewLookupTest, TestAdd) { + BlobViewLookup::TableReadPlan plan(MakeView("db", "t", /*field_id=*/7, /*row_id=*/10)); + + plan.Add(MakeView("db", "t", /*field_id=*/7, /*row_id=*/12)); + plan.Add(MakeView("db", "t", /*field_id=*/9, /*row_id=*/11)); + plan.Add(MakeView("db", "t", /*field_id=*/9, /*row_id=*/13)); + + const auto& references = plan.references_by_field_id_; + ASSERT_EQ(references.size(), 2U); + auto iter = references.find(7); + ASSERT_NE(iter, references.end()); + iter = references.find(9); + ASSERT_NE(iter, references.end()); + + ASSERT_EQ(plan.row_ranges_, (std::vector{10, 12, 11, 13})); +} + +TEST_F(BlobViewLookupTest, TestGetFieldIds) { + BlobViewLookup::TableReadPlan plan(MakeView("db", "t", /*field_id=*/3, /*row_id=*/0)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/1)); + plan.Add(MakeView("db", "t", /*field_id=*/3, /*row_id=*/2)); + plan.Add(MakeView("db", "t", /*field_id=*/2, /*row_id=*/3)); + + ASSERT_EQ(plan.GetFieldIds(), (std::vector{1, 2, 3})); +} + +TEST_F(BlobViewLookupTest, TestGetSortedDistinctRangesMergesTwoAdjacentRowIds) { + BlobViewLookup::TableReadPlan plan(MakeView("db", "t", /*field_id=*/1, /*row_id=*/5)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/6)); + + auto ranges = plan.GetSortedDistinctRanges(); + ASSERT_EQ(ranges.size(), 1U); + ASSERT_EQ(ranges[0], Range(5, 6)); +} + +TEST_F(BlobViewLookupTest, TestGetSortedDistinctRangesMergesContiguousAndGaps) { + BlobViewLookup::TableReadPlan plan(MakeView("db", "t", /*field_id=*/1, /*row_id=*/5)); + // Out of order, with duplicates and a gap so we get two output ranges. + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/6)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/5)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/7)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/10)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/11)); + + auto ranges = plan.GetSortedDistinctRanges(); + ASSERT_EQ(ranges.size(), 2U); + ASSERT_EQ(ranges[0].from, 5); + ASSERT_EQ(ranges[0].to, 7); + ASSERT_EQ(ranges[1].from, 10); + ASSERT_EQ(ranges[1].to, 11); +} + +TEST_F(BlobViewLookupTest, TestGetSortedDistinctRangesWithNonContiguous) { + BlobViewLookup::TableReadPlan plan(MakeView("db", "t", /*field_id=*/1, /*row_id=*/1)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/100)); + plan.Add(MakeView("db", "t", /*field_id=*/1, /*row_id=*/50)); + + const auto ranges = plan.GetSortedDistinctRanges(); + ASSERT_EQ(ranges.size(), 3U); + ASSERT_EQ(ranges[0].from, 1); + ASSERT_EQ(ranges[0].to, 1); + ASSERT_EQ(ranges[1].from, 50); + ASSERT_EQ(ranges[1].to, 50); + ASSERT_EQ(ranges[2].from, 100); + ASSERT_EQ(ranges[2].to, 100); +} + +TEST_F(BlobViewLookupTest, TestEmptyInputProducesEmptyOutput) { + auto grouped = BlobViewLookup::GroupByIdentifier({}); + ASSERT_TRUE(grouped.empty()); +} + +TEST_F(BlobViewLookupTest, TestSingleViewStructProducesSingleGroup) { + std::unordered_set views; + views.emplace(MakeView("db", "t", /*field_id=*/3, /*row_id=*/42)); + + auto grouped = BlobViewLookup::GroupByIdentifier(views); + ASSERT_EQ(grouped.size(), 1U); + + auto iter = grouped.find(MakeIdentifier("db", "t")); + ASSERT_NE(iter, grouped.end()); + const auto& plan = iter->second; + ASSERT_EQ(plan.GetFieldIds(), std::vector{3}); + ASSERT_EQ(plan.row_ranges_, std::vector{42}); +} + +TEST_F(BlobViewLookupTest, TestMultipleViewStructsOfSameTableAreMergedIntoOnePlan) { + std::unordered_set views; + views.emplace(MakeView("db", "t", /*field_id=*/3, /*row_id=*/1)); + views.emplace(MakeView("db", "t", /*field_id=*/3, /*row_id=*/2)); + views.emplace(MakeView("db", "t", /*field_id=*/4, /*row_id=*/1)); + + auto grouped = BlobViewLookup::GroupByIdentifier(views); + ASSERT_EQ(grouped.size(), 1U); + + const auto& plan = grouped.at(MakeIdentifier("db", "t")); + ASSERT_EQ(plan.GetFieldIds(), (std::vector{3, 4})); + + const auto& references = plan.references_by_field_id_; + ASSERT_EQ(references.size(), 2U); + auto iter = references.find(3); + ASSERT_NE(iter, references.end()); + iter = references.find(4); + ASSERT_NE(iter, references.end()); + + ASSERT_EQ(plan.row_ranges_.size(), 3U); +} + +TEST_F(BlobViewLookupTest, TestViewStructsOfDifferentTablesAreSplitIntoDistinctPlans) { + std::unordered_set views; + views.emplace(MakeView("db1", "t1", /*field_id=*/3, /*row_id=*/1)); + views.emplace(MakeView("db1", "t2", /*field_id=*/3, /*row_id=*/1)); + views.emplace(MakeView("db2", "t1", /*field_id=*/3, /*row_id=*/1)); + views.emplace(MakeView("db1", "t1", /*field_id=*/4, /*row_id=*/2)); + + auto grouped = BlobViewLookup::GroupByIdentifier(views); + ASSERT_EQ(grouped.size(), 3U); + + const auto& plan_db1_t1 = grouped.at(MakeIdentifier("db1", "t1")); + ASSERT_EQ(plan_db1_t1.GetFieldIds(), (std::vector{3, 4})); + ASSERT_EQ(plan_db1_t1.row_ranges_.size(), 2U); + + const auto& plan_db1_t2 = grouped.at(MakeIdentifier("db1", "t2")); + ASSERT_EQ(plan_db1_t2.GetFieldIds(), std::vector{3}); + ASSERT_EQ(plan_db1_t2.row_ranges_.size(), 1U); + + const auto& plan_db2_t1 = grouped.at(MakeIdentifier("db2", "t1")); + ASSERT_EQ(plan_db2_t1.GetFieldIds(), std::vector{3}); + ASSERT_EQ(plan_db2_t1.row_ranges_.size(), 1U); +} + +} // namespace paimon::test diff --git a/test/inte/blob_table_inte_test.cpp b/test/inte/blob_table_inte_test.cpp index 75010ebc0..507b31bc9 100644 --- a/test/inte/blob_table_inte_test.cpp +++ b/test/inte/blob_table_inte_test.cpp @@ -38,6 +38,7 @@ #include "paimon/common/data/binary_array_writer.h" #include "paimon/common/data/binary_row.h" #include "paimon/common/data/binary_row_writer.h" +#include "paimon/common/data/blob_view_struct.h" #include "paimon/common/factories/io_hook.h" #include "paimon/common/table/special_fields.h" #include "paimon/common/utils/path_util.h" @@ -2384,4 +2385,166 @@ TEST_P(BlobTableInteTest, TestBlobDescriptorFieldWriteRawBytesDirectly) { "to be a BlobDescriptor or BlobViewStruct."); } +TEST_P(BlobTableInteTest, TestBlobViewFieldWithUpstreamTable) { + auto file_format = GetParam(); + if (file_format != "orc" && file_format != "parquet") { + return; + } + + const std::string upstream_db_name = "append_table_with_multi_blob"; + const std::string upstream_table_name = "append_table_with_multi_blob"; + std::string src_db_path = paimon::test::GetDataDir() + file_format + "/" + upstream_db_name + + ".db/" + upstream_table_name; + std::string dst_db_path = + PathUtil::JoinPath(dir_->Str(), upstream_db_name + ".db/" + upstream_table_name); + ASSERT_TRUE(TestUtil::CopyDirectory(src_db_path, dst_db_path)); + + arrow::FieldVector fields = {arrow::field("f0", arrow::int32()), + BlobUtils::ToArrowField("view", true)}; + std::map options = {{Options::MANIFEST_FORMAT, "orc"}, + {Options::FILE_FORMAT, file_format}, + {Options::BUCKET, "-1"}, + {Options::ROW_TRACKING_ENABLED, "true"}, + {Options::DATA_EVOLUTION_ENABLED, "true"}, + {Options::BLOB_VIEW_FIELD, "view"}, + {Options::FILE_SYSTEM, "local"}}; + CreateTable(fields, /*partition_keys=*/{}, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + + // src array + Identifier upstream_identifier(upstream_db_name, upstream_table_name); + arrow::LargeBinaryBuilder view_builder; + for (int32_t i = 0; i < 8; ++i) { + if (i < 6) { + BlobViewStruct view_struct(upstream_identifier, /*field_id=*/6, + /*row_id=*/static_cast(i)); + auto serialized = view_struct.Serialize(GetDefaultPool()); + ASSERT_TRUE(view_builder + .Append(reinterpret_cast(serialized->data()), + serialized->size()) + .ok()); + } else { + ASSERT_TRUE(view_builder.AppendNull().ok()); + } + } + std::shared_ptr write_view_array; + ASSERT_TRUE(view_builder.Finish(&write_view_array).ok()); + auto write_f0_array = arrow::ipc::internal::json::ArrayFromJSON( + arrow::int32(), R"([100,101,102,103,104,105,106,107])") + .ValueOrDie(); + auto write_struct = std::dynamic_pointer_cast( + arrow::StructArray::Make(arrow::ArrayVector({write_f0_array, write_view_array}), + std::vector({"f0", "view"})) + .ValueOrDie()); + + // write & commit + auto schema = arrow::schema(fields); + ASSERT_OK_AND_ASSIGN(auto commit_msgs, + WriteArray(table_path, {}, schema->field_names(), {write_struct})); + ASSERT_OK(Commit(table_path, commit_msgs)); + + std::string padding_b(2048, 'b'); + std::string padding_d(2048, 'd'); + std::string padding_e(2048, 'e'); + std::string padding_f(2048, 'f'); + + // scan & read + ASSERT_OK_AND_ASSIGN(auto plan, ScanTable(table_path)); + ASSERT_OK_AND_ASSIGN(auto result, + ReadTable(table_path, schema->field_names(), plan, /*predicate=*/nullptr)); + ASSERT_TRUE(result.chunked_array); + auto read_concat = arrow::Concatenate(result.chunked_array->chunks()).ValueOrDie(); + auto read_struct = std::dynamic_pointer_cast(read_concat); + ASSERT_EQ(read_struct->length(), 8); + ASSERT_OK_AND_ASSIGN(auto result_array, ConvertDescriptorToRawBlob(read_struct, {"view"})); + + // clang-format off + std::string expected_json = R"([ +[100, null], +[101, ")" + padding_b + R"("], +[102, null], +[103, ")" + padding_d + R"("], +[104, ")" + padding_e + R"("], +[105, ")" + padding_f + R"("], +[106, null], +[107, null] +])"; + // clang-format on + auto expected_struct = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), expected_json) + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto expected_with_rk, PrependRowKindColumn(expected_struct)); + + ASSERT_TRUE(result_array->Equals(expected_with_rk)) + << "result_array:" << result_array->ToString() << std::endl + << "expected:" << expected_with_rk->ToString(); + + // Sub-case 1: scan with row_ranges. + { + ASSERT_OK_AND_ASSIGN(auto range_plan, ScanTable(table_path, /*predicate=*/nullptr, + /*row_ranges=*/{Range(1, 3), Range(5, 5)})); + ASSERT_OK_AND_ASSIGN(auto range_result, ReadTable(table_path, schema->field_names(), + range_plan, /*predicate=*/nullptr)); + ASSERT_TRUE(range_result.chunked_array); + auto range_concat = arrow::Concatenate(range_result.chunked_array->chunks()).ValueOrDie(); + auto range_struct = std::dynamic_pointer_cast(range_concat); + ASSERT_EQ(range_struct->length(), 4); + ASSERT_OK_AND_ASSIGN(auto range_resolved, + ConvertDescriptorToRawBlob(range_struct, {"view"})); + + // clang-format off + std::string range_json = R"([ +[101, ")" + padding_b + R"("], +[102, null], +[103, ")" + padding_d + R"("], +[105, ")" + padding_f + R"("] +])"; + // clang-format on + auto range_expected_struct = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), range_json) + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto range_expected_with_rk, + PrependRowKindColumn(range_expected_struct)); + ASSERT_TRUE(range_resolved->Equals(range_expected_with_rk)) + << "range_resolved:" << range_resolved->ToString() << std::endl + << "expected:" << range_expected_with_rk->ToString(); + } + + // Sub-case 2: scan with predicate (f0 > 102), data evolution split read will ignore format push + // down + { + auto predicate = PredicateBuilder::GreaterThan(/*field_index=*/0, /*field_name=*/"f0", + FieldType::INT, Literal(102)); + ASSERT_OK_AND_ASSIGN(auto pred_plan, ScanTable(table_path, predicate, /*row_ranges=*/{})); + ASSERT_OK_AND_ASSIGN(auto pred_result, + ReadTable(table_path, schema->field_names(), pred_plan, predicate)); + ASSERT_TRUE(pred_result.chunked_array); + auto pred_concat = arrow::Concatenate(pred_result.chunked_array->chunks()).ValueOrDie(); + auto pred_struct = std::dynamic_pointer_cast(pred_concat); + ASSERT_EQ(pred_struct->length(), 8); + ASSERT_OK_AND_ASSIGN(auto pred_resolved, ConvertDescriptorToRawBlob(pred_struct, {"view"})); + + // clang-format off + std::string pred_json = R"([ +[100, null], +[101, ")" + padding_b + R"("], +[102, null], +[103, ")" + padding_d + R"("], +[104, ")" + padding_e + R"("], +[105, ")" + padding_f + R"("], +[106, null], +[107, null] +])"; + // clang-format on + auto pred_expected_struct = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), pred_json) + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN(auto pred_expected_with_rk, + PrependRowKindColumn(pred_expected_struct)); + ASSERT_TRUE(pred_resolved->Equals(pred_expected_with_rk)) + << "pred_resolved:" << pred_resolved->ToString() << std::endl + << "expected:" << pred_expected_with_rk->ToString(); + } +} + } // namespace paimon::test