From af1257471d1ce183f0367c5c46de78d335bee3f1 Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Mon, 29 Jun 2026 14:39:02 +0800 Subject: [PATCH 1/6] Support chain data split --- src/paimon/CMakeLists.txt | 2 + .../core/io/chain_data_file_path_factory.cpp | 45 ++++ .../core/io/chain_data_file_path_factory.h | 39 +++ src/paimon/core/io/data_file_path_factory.h | 2 +- .../core/operation/abstract_split_read.cpp | 17 ++ .../core/operation/abstract_split_read.h | 3 + .../operation/data_evolution_split_read.cpp | 9 +- .../core/operation/merge_file_split_read.cpp | 5 +- .../core/operation/raw_file_split_read.cpp | 17 +- .../core/operation/raw_file_split_read.h | 4 +- .../core/table/source/chain_data_split_impl.h | 70 ++++++ .../table/source/chain_data_split_test.cpp | 223 ++++++++++++++++++ .../core/table/source/data_split_impl.h | 14 +- src/paimon/core/table/source/split.cpp | 122 +++++++++- 14 files changed, 550 insertions(+), 22 deletions(-) create mode 100644 src/paimon/core/io/chain_data_file_path_factory.cpp create mode 100644 src/paimon/core/io/chain_data_file_path_factory.h create mode 100644 src/paimon/core/table/source/chain_data_split_impl.h create mode 100644 src/paimon/core/table/source/chain_data_split_test.cpp diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index b044badea..e12908c71 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -220,6 +220,7 @@ set(PAIMON_CORE_SRCS core/io/data_file_meta_first_row_id_legacy_serializer.cpp core/io/data_file_meta.cpp core/io/data_file_meta_serializer.cpp + core/io/chain_data_file_path_factory.cpp core/io/data_file_path_factory.cpp core/io/append_data_file_writer_factory.cpp core/io/blob_data_file_writer_factory.cpp @@ -735,6 +736,7 @@ if(PAIMON_BUILD_TESTS) core/table/source/table_read_test.cpp core/table/source/append_count_reader_test.cpp core/table/source/pk_count_reader_test.cpp + core/table/source/chain_data_split_test.cpp core/table/source/data_split_test.cpp core/table/source/deletion_file_test.cpp core/table/source/split_generator_test.cpp diff --git a/src/paimon/core/io/chain_data_file_path_factory.cpp b/src/paimon/core/io/chain_data_file_path_factory.cpp new file mode 100644 index 000000000..c5419899f --- /dev/null +++ b/src/paimon/core/io/chain_data_file_path_factory.cpp @@ -0,0 +1,45 @@ +/* + * 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/io/chain_data_file_path_factory.h" + +#include + +#include "paimon/common/utils/path_util.h" +#include "paimon/core/io/data_file_meta.h" + +namespace paimon { + +ChainDataFilePathFactory::ChainDataFilePathFactory( + std::shared_ptr fallback, + std::unordered_map file_bucket_path_mapping) + : fallback_(std::move(fallback)), + file_bucket_path_mapping_(std::move(file_bucket_path_mapping)) {} + +std::string ChainDataFilePathFactory::ToPath(const std::shared_ptr& file_meta) const { + if (file_meta->external_path) { + return file_meta->external_path.value(); + } + + auto it = file_bucket_path_mapping_.find(file_meta->file_name); + if (it != file_bucket_path_mapping_.end()) { + return PathUtil::JoinPath(it->second, file_meta->file_name); + } + + return fallback_->ToPath(file_meta); +} + +} // namespace paimon diff --git a/src/paimon/core/io/chain_data_file_path_factory.h b/src/paimon/core/io/chain_data_file_path_factory.h new file mode 100644 index 000000000..30270a30f --- /dev/null +++ b/src/paimon/core/io/chain_data_file_path_factory.h @@ -0,0 +1,39 @@ +/* + * 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 "paimon/core/io/data_file_path_factory.h" + +namespace paimon { + +class ChainDataFilePathFactory : public DataFilePathFactory { + public: + ChainDataFilePathFactory(std::shared_ptr fallback, + std::unordered_map file_bucket_path_mapping); + + std::string ToPath(const std::shared_ptr& file_meta) const override; + + private: + std::shared_ptr fallback_; + std::unordered_map file_bucket_path_mapping_; +}; + +} // namespace paimon diff --git a/src/paimon/core/io/data_file_path_factory.h b/src/paimon/core/io/data_file_path_factory.h index 110a7035e..e071c42c4 100644 --- a/src/paimon/core/io/data_file_path_factory.h +++ b/src/paimon/core/io/data_file_path_factory.h @@ -70,7 +70,7 @@ class DataFilePathFactory : public PathFactory { } std::string ToPath(const std::string& file_name) const override; - std::string ToPath(const std::shared_ptr& file_meta) const; + virtual std::string ToPath(const std::shared_ptr& file_meta) const; const std::string& GetUUID() const { return uuid_; diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp index 1e70c0554..46d5cc86d 100644 --- a/src/paimon/core/operation/abstract_split_read.cpp +++ b/src/paimon/core/operation/abstract_split_read.cpp @@ -33,6 +33,7 @@ #include "paimon/common/table/special_fields.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/object_utils.h" +#include "paimon/core/io/chain_data_file_path_factory.h" #include "paimon/core/io/complete_row_tracking_fields_reader.h" #include "paimon/core/io/data_file_meta.h" #include "paimon/core/io/data_file_path_factory.h" @@ -40,6 +41,7 @@ #include "paimon/core/operation/internal_read_context.h" #include "paimon/core/partition/partition_info.h" #include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/source/chain_data_split_impl.h" #include "paimon/core/table/source/data_split_impl.h" #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/nested_projection_utils.h" @@ -116,6 +118,21 @@ Result> AbstractSplitRead::ApplyPredicateFilterIfNe return PredicateBatchReader::Create(std::move(reader), predicate, pool_); } +Result> AbstractSplitRead::CreateDataFilePathFactory( + const std::shared_ptr& data_split) const { + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr base_factory, + path_factory_->CreateDataFilePathFactory(data_split->Partition(), data_split->Bucket())); + + auto chain_split = std::dynamic_pointer_cast(data_split); + if (!chain_split) { + return base_factory; + } + + return std::make_shared(base_factory, + chain_split->FileBucketPathMapping()); +} + Result> AbstractSplitRead::PrepareReaderBuilder( const std::string& format_identifier) const { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr file_format, diff --git a/src/paimon/core/operation/abstract_split_read.h b/src/paimon/core/operation/abstract_split_read.h index 4d49d78bc..22a481041 100644 --- a/src/paimon/core/operation/abstract_split_read.h +++ b/src/paimon/core/operation/abstract_split_read.h @@ -76,6 +76,9 @@ class AbstractSplitRead : public SplitRead { Result> ApplyPredicateFilterIfNeeded( std::unique_ptr&& reader, const std::shared_ptr& predicate) const; + Result> CreateDataFilePathFactory( + const std::shared_ptr& data_split) const; + protected: // return nullptr if file is skipped by index or dv virtual Result> ApplyIndexAndDvReaderIfNeeded( diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index fab21ac01..47f81b2d7 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -289,14 +289,13 @@ Result> DataEvolutionSplitRead::ExtractBlobVi Result> DataEvolutionSplitRead::InnerCreateReader( const std::shared_ptr& data_split, const std::optional>& row_ranges) const { - auto split_impl = dynamic_cast(data_split.get()); - if (split_impl == nullptr) { + auto split_impl = std::dynamic_pointer_cast(data_split); + if (!split_impl) { return Status::Invalid("unexpected error, split cast to impl failed"); } assert(raw_read_schema_->num_fields() > 0); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr data_file_path_factory, - path_factory_->CreateDataFilePathFactory(split_impl->Partition(), split_impl->Bucket())); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_file_path_factory, + CreateDataFilePathFactory(split_impl)); auto metas = split_impl->DataFiles(); PAIMON_ASSIGN_OR_RAISE(std::vector>> split_by_row_id, MergeRangesAndSort(std::move(metas))); diff --git a/src/paimon/core/operation/merge_file_split_read.cpp b/src/paimon/core/operation/merge_file_split_read.cpp index e794a403d..8c295fb36 100644 --- a/src/paimon/core/operation/merge_file_split_read.cpp +++ b/src/paimon/core/operation/merge_file_split_read.cpp @@ -141,9 +141,8 @@ Result> MergeFileSplitRead::CreateReader( if (!data_split->BeforeFiles().empty()) { return Status::Invalid("this read cannot accept split with before files."); } - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr data_file_path_factory, - path_factory_->CreateDataFilePathFactory(data_split->Partition(), data_split->Bucket())); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_file_path_factory, + CreateDataFilePathFactory(data_split)); std::unique_ptr batch_reader; if (data_split->IsStreaming() || data_split->Bucket() == BucketModeDefine::POSTPONE_BUCKET) { PAIMON_ASSIGN_OR_RAISE( diff --git a/src/paimon/core/operation/raw_file_split_read.cpp b/src/paimon/core/operation/raw_file_split_read.cpp index 79755487e..42af0f78d 100644 --- a/src/paimon/core/operation/raw_file_split_read.cpp +++ b/src/paimon/core/operation/raw_file_split_read.cpp @@ -66,17 +66,26 @@ Result> RawFileSplitRead::CreateReader( if (!data_split) { return Status::Invalid("cannot cast split to data_split in RawFileSplitRead"); } + auto dv_factory = DeletionVector::CreateFactory( + options_.GetFileSystem(), + DeletionVector::CreateDeletionFileMap(data_split->DataFiles(), data_split->DeletionFiles()), + pool_); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_file_path_factory, + CreateDataFilePathFactory(data_split)); return CreateReader(data_split->Partition(), data_split->Bucket(), data_split->DataFiles(), - data_split->DeletionFiles()); + dv_factory, data_file_path_factory); } Result> RawFileSplitRead::CreateReader( const BinaryRow& partition, int32_t bucket, const std::vector>& data_files, - DeletionVector::Factory dv_factory) { + DeletionVector::Factory dv_factory, + std::shared_ptr data_file_path_factory) { const auto& predicate = context_->GetPredicate(); - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_file_path_factory, - path_factory_->CreateDataFilePathFactory(partition, bucket)); + if (!data_file_path_factory) { + PAIMON_ASSIGN_OR_RAISE(data_file_path_factory, + path_factory_->CreateDataFilePathFactory(partition, bucket)); + } PAIMON_ASSIGN_OR_RAISE( std::vector> raw_file_readers, diff --git a/src/paimon/core/operation/raw_file_split_read.h b/src/paimon/core/operation/raw_file_split_read.h index ebcbc2ef0..eb6924ad2 100644 --- a/src/paimon/core/operation/raw_file_split_read.h +++ b/src/paimon/core/operation/raw_file_split_read.h @@ -70,8 +70,8 @@ class RawFileSplitRead : public AbstractSplitRead { Result> CreateReader( const BinaryRow& partition, int32_t bucket, - const std::vector>& files, - DeletionVector::Factory dv_factory); + const std::vector>& files, DeletionVector::Factory dv_factory, + std::shared_ptr data_file_path_factory = nullptr); Result Match(const std::shared_ptr& split, bool force_keep_delete) const override; diff --git a/src/paimon/core/table/source/chain_data_split_impl.h b/src/paimon/core/table/source/chain_data_split_impl.h new file mode 100644 index 000000000..7588a3bfc --- /dev/null +++ b/src/paimon/core/table/source/chain_data_split_impl.h @@ -0,0 +1,70 @@ +/* + * 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 "paimon/common/data/binary_row.h" +#include "paimon/core/table/source/data_split_impl.h" + +namespace paimon { + +class ChainDataSplitImpl : public DataSplitImpl { + public: + static constexpr const char* VIRTUAL_BUCKET_PATH = "placeholder::virtual-bucket-path"; + + ChainDataSplitImpl(const std::shared_ptr& base_split, bool all_snapshot_split, + const BinaryRow& read_partition, + std::unordered_map&& file_bucket_path_mapping, + std::unordered_map&& file_branch_mapping) + : DataSplitImpl(base_split->Partition(), base_split->Bucket(), base_split->BucketPath(), + std::vector>(base_split->DataFiles())), + all_snapshot_split_(all_snapshot_split), + read_partition_(read_partition), + file_bucket_path_mapping_(std::move(file_bucket_path_mapping)), + file_branch_mapping_(std::move(file_branch_mapping)) { + CopyMetadataFrom(*base_split); + } + + bool AllSnapshotSplit() const { + return all_snapshot_split_; + } + + const BinaryRow& ReadPartition() const { + return read_partition_; + } + + const std::unordered_map& FileBucketPathMapping() const { + return file_bucket_path_mapping_; + } + + const std::unordered_map& FileBranchMapping() const { + return file_branch_mapping_; + } + + private: + bool all_snapshot_split_; + BinaryRow read_partition_; + std::unordered_map file_bucket_path_mapping_; + std::unordered_map file_branch_mapping_; +}; + +} // namespace paimon diff --git a/src/paimon/core/table/source/chain_data_split_test.cpp b/src/paimon/core/table/source/chain_data_split_test.cpp new file mode 100644 index 000000000..82f27d451 --- /dev/null +++ b/src/paimon/core/table/source/chain_data_split_test.cpp @@ -0,0 +1,223 @@ +/* + * 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 +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/memory/memory_segment_utils.h" +#include "paimon/common/utils/serialization_utils.h" +#include "paimon/core/io/chain_data_file_path_factory.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/io/data_file_meta_serializer.h" +#include "paimon/core/io/data_file_path_factory.h" +#include "paimon/core/manifest/file_source.h" +#include "paimon/core/stats/simple_stats.h" +#include "paimon/core/table/source/chain_data_split_impl.h" +#include "paimon/core/table/source/data_split_impl.h" +#include "paimon/core/table/source/deletion_file.h" +#include "paimon/data/timestamp.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/table/source/data_split.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { +namespace { +constexpr const char* kFileName = "data-b446f78a-2cfb-4b3b-add8-31295d24a277-0.parquet"; +constexpr const char* kBucketPath = "data/parquet/append_09.db/append_09/f1=20/bucket-0"; +constexpr const char* kBranch = "delta"; +constexpr const char* kFallbackBucketPath = "data/parquet/append_09.db/append_09/f1=10/bucket-1"; +constexpr const char* kSingleSplitBucketPath = + "data/parquet/append_table_with_append_pt_branch.db/append_table_with_append_pt_branch/" + "pt=2/bucket-0"; +constexpr const char* kSingleSplitFileName0 = "data-39204ff8-55b2-497b-8e87-a1c736799eab-0.parquet"; +constexpr const char* kSingleSplitFileName1 = "data-625b3277-84d3-4320-80b9-89a5075bf5fd-0.parquet"; + +std::shared_ptr CreateDataFileMeta( + const std::string& file_name, std::optional external_path = std::nullopt) { + return std::make_shared( + file_name, /*file_size=*/1024, /*row_count=*/7, BinaryRow::EmptyRow(), + BinaryRow::EmptyRow(), SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), + /*min_sequence_number=*/0, /*max_sequence_number=*/6, /*schema_id=*/0, + DataFileMeta::DUMMY_LEVEL, std::vector>(), Timestamp(0, 0), + /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(), + /*value_stats_cols=*/std::nullopt, external_path, + /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt); +} + +void WriteStringMap(MemorySegmentOutputStream* out, + const std::unordered_map& values) { + out->WriteValue(values.size()); + for (const auto& [key, value] : values) { + out->WriteString(key); + out->WriteString(value); + } +} + +std::string ToString(const MemorySegmentOutputStream& out, + const std::shared_ptr& pool) { + PAIMON_UNIQUE_PTR bytes = + MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); + return std::string(bytes->data(), bytes->size()); +} + +std::string SerializeVersion7DataSplit(const std::shared_ptr& split, + const std::shared_ptr& pool) { + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + out.WriteValue(DataSplitImpl::MAGIC); + out.WriteValue(7); + out.WriteValue(split->SnapshotId()); + EXPECT_OK(SerializationUtils::SerializeBinaryRow(split->Partition(), &out)); + out.WriteValue(split->Bucket()); + out.WriteString(split->BucketPath()); + + out.WriteValue(split->TotalBuckets().has_value()); + if (split->TotalBuckets().has_value()) { + out.WriteValue(split->TotalBuckets().value()); + } + + DataFileMetaSerializer serializer(pool); + EXPECT_OK(serializer.SerializeList(split->BeforeFiles(), &out)); + DeletionFile::SerializeList(split->BeforeDeletionFiles(), &out); + EXPECT_OK(serializer.SerializeList(split->DataFiles(), &out)); + DeletionFile::SerializeList(split->DeletionFiles(), &out); + out.WriteValue(split->IsStreaming()); + out.WriteValue(split->RawConvertible()); + return ToString(out, pool); +} + +std::shared_ptr CreateBaseSplit(const std::shared_ptr& pool) { + DataSplitImpl::Builder builder(BinaryRow::EmptyRow(), /*bucket=*/118, + ChainDataSplitImpl::VIRTUAL_BUCKET_PATH, + {CreateDataFileMeta(kFileName)}); + return builder.WithSnapshot(42) + .WithTotalBuckets(256) + .IsStreaming(false) + .RawConvertible(true) + .Build() + .value(); +} + +std::string AppendChainDataSplitTail( + const std::string& base_bytes, const std::shared_ptr& pool, + const std::unordered_map& bucket_paths, + const std::unordered_map& branches) { + MemorySegmentOutputStream tail(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + tail.WriteValue(true); + EXPECT_OK(SerializationUtils::SerializeBinaryRow(BinaryRow::EmptyRow(), &tail)); + WriteStringMap(&tail, bucket_paths); + WriteStringMap(&tail, branches); + return base_bytes + ToString(tail, pool); +} + +std::string AppendMalformedChainDataSplitTail(const std::string& base_bytes, + const std::shared_ptr& pool) { + MemorySegmentOutputStream tail(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + tail.WriteValue(true); + return base_bytes + ToString(tail, pool); +} + +} // namespace + +TEST(ChainDataSplitTest, DeserializeChainDataSplitTail) { + auto pool = GetDefaultPool(); + auto base_split = CreateBaseSplit(pool); + ASSERT_OK_AND_ASSIGN(std::string base_bytes, Split::Serialize(base_split, pool)); + + std::string chain_bytes = AppendChainDataSplitTail(base_bytes, pool, {{kFileName, kBucketPath}}, + {{kFileName, kBranch}}); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool)); + auto chain_split = std::dynamic_pointer_cast(split); + ASSERT_TRUE(chain_split); + EXPECT_TRUE(chain_split->AllSnapshotSplit()); + EXPECT_EQ(chain_split->BucketPath(), ChainDataSplitImpl::VIRTUAL_BUCKET_PATH); + EXPECT_EQ(chain_split->FileBucketPathMapping().at(kFileName), kBucketPath); + EXPECT_EQ(chain_split->FileBranchMapping().at(kFileName), kBranch); +} + +TEST(ChainDataSplitTest, DeserializeVersion7SingleSplitWithOriginalBucketPath) { + auto pool = GetDefaultPool(); + DataSplitImpl::Builder builder( + BinaryRow::EmptyRow(), /*bucket=*/3, kSingleSplitBucketPath, + {CreateDataFileMeta(kSingleSplitFileName0), CreateDataFileMeta(kSingleSplitFileName1)}); + ASSERT_OK_AND_ASSIGN(auto base_split, builder.WithSnapshot(42) + .WithTotalBuckets(256) + .IsStreaming(false) + .RawConvertible(true) + .Build()); + std::string base_bytes = SerializeVersion7DataSplit(base_split, pool); + std::string split_bytes = AppendChainDataSplitTail( + base_bytes, pool, + {{kSingleSplitFileName0, kSingleSplitBucketPath}, + {kSingleSplitFileName1, kSingleSplitBucketPath}}, + {{kSingleSplitFileName0, "snapshot"}, {kSingleSplitFileName1, "snapshot"}}); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(split_bytes.data(), split_bytes.size(), pool)); + auto chain_split = std::dynamic_pointer_cast(split); + ASSERT_TRUE(chain_split); + EXPECT_TRUE(chain_split->AllSnapshotSplit()); + EXPECT_EQ(chain_split->BucketPath(), kSingleSplitBucketPath); + EXPECT_EQ(chain_split->FileBucketPathMapping().at(kSingleSplitFileName0), + kSingleSplitBucketPath); + EXPECT_EQ(chain_split->FileBucketPathMapping().at(kSingleSplitFileName1), + kSingleSplitBucketPath); + EXPECT_EQ(chain_split->FileBranchMapping().at(kSingleSplitFileName0), "snapshot"); + EXPECT_EQ(chain_split->FileBranchMapping().at(kSingleSplitFileName1), "snapshot"); +} + +TEST(ChainDataSplitTest, ChainDataFilePathFactoryUsesPerFileBucketPath) { + auto fallback = std::make_shared(); + ASSERT_OK(fallback->Init(kFallbackBucketPath, "parquet", "data-", + /*external_path_provider=*/nullptr)); + ChainDataFilePathFactory factory(fallback, {{kFileName, kBucketPath}}); + + auto file_meta = CreateDataFileMeta(kFileName); + EXPECT_EQ(factory.ToPath(file_meta), std::string(kBucketPath) + "/" + kFileName); +} + +TEST(ChainDataSplitTest, ChainDataFilePathFactoryPreservesExternalPath) { + auto fallback = std::make_shared(); + ASSERT_OK(fallback->Init(kFallbackBucketPath, "parquet", "data-", + /*external_path_provider=*/nullptr)); + ChainDataFilePathFactory factory(fallback, {{kFileName, kBucketPath}}); + + const std::string external_path = "hdfs://external/path/" + std::string(kFileName); + auto file_meta = CreateDataFileMeta(kFileName, external_path); + EXPECT_EQ(factory.ToPath(file_meta), external_path); +} + +TEST(ChainDataSplitTest, MalformedChainDataSplitTailReturnsContextualError) { + auto pool = GetDefaultPool(); + auto base_split = CreateBaseSplit(pool); + ASSERT_OK_AND_ASSIGN(std::string base_bytes, Split::Serialize(base_split, pool)); + + std::string chain_bytes = AppendMalformedChainDataSplitTail(base_bytes, pool); + + ASSERT_NOK_WITH_MSG(Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool), + "ChainDataSplit"); +} + +} // namespace paimon::test diff --git a/src/paimon/core/table/source/data_split_impl.h b/src/paimon/core/table/source/data_split_impl.h index a90ca55dc..e8bb7ebd7 100644 --- a/src/paimon/core/table/source/data_split_impl.h +++ b/src/paimon/core/table/source/data_split_impl.h @@ -177,7 +177,7 @@ class DataSplitImpl : public DataSplit { std::string ToString() const; - private: + protected: DataSplitImpl(const BinaryRow& partition, int32_t bucket, const std::string& bucket_path, std::vector>&& data_files) : partition_(partition), @@ -185,11 +185,21 @@ class DataSplitImpl : public DataSplit { bucket_path_(bucket_path), data_files_(std::move(data_files)) {} + void CopyMetadataFrom(const DataSplitImpl& other) { + total_buckets_ = other.total_buckets_; + snapshot_id_ = other.snapshot_id_; + before_files_ = other.before_files_; + before_deletion_files_ = other.before_deletion_files_; + data_deletion_files_ = other.data_deletion_files_; + is_streaming_ = other.is_streaming_; + raw_convertible_ = other.raw_convertible_; + } + + private: Result> RawMergedRowCount(DeletionVector::Factory dv_factory) const; bool DataEvolutionRowCountAvailable() const; Result DataEvolutionMergedRowCount() const; - private: int64_t snapshot_id_ = 0; BinaryRow partition_ = BinaryRow::EmptyRow(); int32_t bucket_ = -1; diff --git a/src/paimon/core/table/source/split.cpp b/src/paimon/core/table/source/split.cpp index 2d188cd53..c3059796b 100644 --- a/src/paimon/core/table/source/split.cpp +++ b/src/paimon/core/table/source/split.cpp @@ -14,6 +14,7 @@ * limitations under the License. */ +#include #include #include "fmt/format.h" @@ -22,7 +23,9 @@ #include "paimon/common/memory/memory_segment_utils.h" #include "paimon/common/utils/serialization_utils.h" #include "paimon/core/global_index/indexed_split_impl.h" +#include "paimon/core/io/data_file_meta_first_row_id_legacy_serializer.h" #include "paimon/core/io/data_file_meta_serializer.h" +#include "paimon/core/table/source/chain_data_split_impl.h" #include "paimon/core/table/source/data_split_impl.h" #include "paimon/core/table/source/deletion_file.h" #include "paimon/core/table/source/fallback_data_split.h" @@ -37,6 +40,65 @@ namespace paimon { struct DataFileMeta; namespace { +Result>> ReadVersion7DataFileMetaList( + DataInputStream* in, const std::shared_ptr& pool) { + PAIMON_ASSIGN_OR_RAISE(int32_t size, in->ReadValue()); + if (size < 0) { + return Status::Invalid(fmt::format("invalid data file meta list size: {}", size)); + } + + DataFileMetaFirstRowIdLegacySerializer legacy_serializer(pool); + DataFileMetaSerializer current_serializer(pool); + std::vector> result; + result.reserve(size); + for (int32_t i = 0; i < size; ++i) { + PAIMON_ASSIGN_OR_RAISE(int32_t row_size, in->ReadValue()); + if (row_size < BinaryRow::CalculateFixPartSizeInBytes(19)) { + return Status::Invalid( + fmt::format("invalid version 7 data file meta row size: {}", row_size)); + } + std::shared_ptr bytes = Bytes::AllocateBytes(row_size, pool.get()); + PAIMON_RETURN_NOT_OK(in->ReadBytes(bytes.get())); + + MemorySegment segment = MemorySegment::Wrap(bytes); + int64_t file_name_offset_and_size = + segment.GetValue(BinaryRow::CalculateBitSetWidthInBytes(/*arity=*/19)); + if ((file_name_offset_and_size & BinarySection::HIGHEST_FIRST_BIT) != 0) { + return Status::Invalid( + "cannot determine version 7 data file meta format from inline file name"); + } + int32_t variable_part_offset = static_cast(file_name_offset_and_size >> 32); + + int32_t arity = 0; + const ObjectSerializer>* serializer = nullptr; + if (variable_part_offset == BinaryRow::CalculateFixPartSizeInBytes(/*arity=*/19)) { + arity = 19; + serializer = &legacy_serializer; + } else if (variable_part_offset == BinaryRow::CalculateFixPartSizeInBytes(/*arity=*/20)) { + arity = 20; + serializer = ¤t_serializer; + } else { + return Status::Invalid(fmt::format( + "invalid version 7 data file meta variable part offset: {}", variable_part_offset)); + } + + BinaryRow row(arity); + row.PointTo(segment, /*offset=*/0, row_size); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr meta, serializer->FromRow(row)); + result.emplace_back(std::move(meta)); + } + return result; +} + +Result>> ReadDataFileMetaList( + int32_t version, const ObjectSerializer>* data_file_serializer, + DataInputStream* in, const std::shared_ptr& pool) { + if (version == 7) { + return ReadVersion7DataFileMetaList(in, pool); + } + return data_file_serializer->DeserializeList(in); +} + Status WriteDataSplit(const std::shared_ptr& data_split_impl, MemorySegmentOutputStream* out, const std::shared_ptr& pool) { out->WriteValue(DataSplitImpl::MAGIC); @@ -98,13 +160,15 @@ Result> ReadDataSplitWithoutMagicNumber( std::unique_ptr>> data_file_serializer, DataSplitImpl::GetFileMetaSerializer(version, pool)); std::vector> before_files; - PAIMON_ASSIGN_OR_RAISE(before_files, data_file_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(before_files, + ReadDataFileMetaList(version, data_file_serializer.get(), in, pool)); // compatible for deletion file std::vector> before_deletion_files; PAIMON_ASSIGN_OR_RAISE(before_deletion_files, DeletionFile::DeserializeList(in, version)); std::vector> data_files; - PAIMON_ASSIGN_OR_RAISE(data_files, data_file_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(data_files, + ReadDataFileMetaList(version, data_file_serializer.get(), in, pool)); // compatible for deletion file std::vector> data_deletion_files; PAIMON_ASSIGN_OR_RAISE(data_deletion_files, DeletionFile::DeserializeList(in, version)); @@ -129,6 +193,44 @@ Result> ReadDataSplitWithoutMagicNumber( return builder.Build(); } +Result> ReadStringMap(DataInputStream* in) { + PAIMON_ASSIGN_OR_RAISE(int32_t size, in->ReadValue()); + if (size < 0) { + return Status::Invalid(fmt::format("invalid string map size: {}", size)); + } + + std::unordered_map result; + result.reserve(size); + for (int32_t i = 0; i < size; ++i) { + PAIMON_ASSIGN_OR_RAISE(std::string key, in->ReadString()); + PAIMON_ASSIGN_OR_RAISE(std::string value, in->ReadString()); + result.emplace(std::move(key), std::move(value)); + } + return result; +} + +Result> ReadChainDataSplitTail( + const std::shared_ptr& base_split, DataInputStream* in, + const std::shared_ptr& pool) { + PAIMON_ASSIGN_OR_RAISE(bool all_snapshot_split, in->ReadValue()); + PAIMON_ASSIGN_OR_RAISE(BinaryRow read_partition, + SerializationUtils::DeserializeBinaryRow(in, pool.get())); + PAIMON_ASSIGN_OR_RAISE(auto file_bucket_path_mapping, ReadStringMap(in)); + PAIMON_ASSIGN_OR_RAISE(auto file_branch_mapping, ReadStringMap(in)); + + PAIMON_ASSIGN_OR_RAISE(int64_t pos, in->GetPos()); + PAIMON_ASSIGN_OR_RAISE(int64_t stream_length, in->Length()); + if (pos != stream_length) { + return Status::Invalid(fmt::format( + "invalid ChainDataSplit byte stream, remaining {} bytes after deserializing", + stream_length - pos)); + } + + return std::make_shared(base_split, all_snapshot_split, read_partition, + std::move(file_bucket_path_mapping), + std::move(file_branch_mapping)); +} + } // namespace Result Split::Serialize(const std::shared_ptr& split, @@ -224,13 +326,23 @@ Result> Split::Deserialize(const char* buffer, size_t len PAIMON_ASSIGN_OR_RAISE(int64_t stream_length, in.Length()); if (pos == stream_length) { return data_split; + } else if (data_split->BucketPath() == ChainDataSplitImpl::VIRTUAL_BUCKET_PATH) { + auto chain_split = ReadChainDataSplitTail(data_split, &in, pool); + if (!chain_split.ok()) { + return Status::Invalid(fmt::format("invalid ChainDataSplit byte stream: {}", + chain_split.status().ToString())); + } + return std::static_pointer_cast(chain_split.value()); } else if (pos == stream_length - 1) { PAIMON_ASSIGN_OR_RAISE(bool is_fallback, in.ReadValue()); return std::make_shared(data_split, is_fallback); } else { - return Status::Invalid(fmt::format( - "invalid data split byte stream, remaining {} bytes after deserializing", - stream_length - pos)); + auto chain_split = ReadChainDataSplitTail(data_split, &in, pool); + if (!chain_split.ok()) { + return Status::Invalid(fmt::format("invalid ChainDataSplit byte stream: {}", + chain_split.status().ToString())); + } + return std::static_pointer_cast(chain_split.value()); } } return Status::Invalid("invalid split, must be DataSplit or IndexedSplit"); From 8332390defe51115311ef72ff07d65040f8a4174 Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Wed, 1 Jul 2026 11:54:38 +0800 Subject: [PATCH 2/6] Make raw file path factory explicit --- src/paimon/core/operation/raw_file_split_read.cpp | 13 +++++++++---- src/paimon/core/operation/raw_file_split_read.h | 10 ++++++++-- 2 files changed, 17 insertions(+), 6 deletions(-) diff --git a/src/paimon/core/operation/raw_file_split_read.cpp b/src/paimon/core/operation/raw_file_split_read.cpp index 42af0f78d..2b01caf43 100644 --- a/src/paimon/core/operation/raw_file_split_read.cpp +++ b/src/paimon/core/operation/raw_file_split_read.cpp @@ -76,16 +76,21 @@ Result> RawFileSplitRead::CreateReader( dv_factory, data_file_path_factory); } +Result> RawFileSplitRead::CreateReader( + const BinaryRow& partition, int32_t bucket, + const std::vector>& data_files, + DeletionVector::Factory dv_factory) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_file_path_factory, + path_factory_->CreateDataFilePathFactory(partition, bucket)); + return CreateReader(partition, bucket, data_files, dv_factory, data_file_path_factory); +} + Result> RawFileSplitRead::CreateReader( const BinaryRow& partition, int32_t bucket, const std::vector>& data_files, DeletionVector::Factory dv_factory, std::shared_ptr data_file_path_factory) { const auto& predicate = context_->GetPredicate(); - if (!data_file_path_factory) { - PAIMON_ASSIGN_OR_RAISE(data_file_path_factory, - path_factory_->CreateDataFilePathFactory(partition, bucket)); - } PAIMON_ASSIGN_OR_RAISE( std::vector> raw_file_readers, diff --git a/src/paimon/core/operation/raw_file_split_read.h b/src/paimon/core/operation/raw_file_split_read.h index eb6924ad2..04a71c2a2 100644 --- a/src/paimon/core/operation/raw_file_split_read.h +++ b/src/paimon/core/operation/raw_file_split_read.h @@ -70,8 +70,8 @@ class RawFileSplitRead : public AbstractSplitRead { Result> CreateReader( const BinaryRow& partition, int32_t bucket, - const std::vector>& files, DeletionVector::Factory dv_factory, - std::shared_ptr data_file_path_factory = nullptr); + const std::vector>& files, + DeletionVector::Factory dv_factory); Result Match(const std::shared_ptr& split, bool force_keep_delete) const override; @@ -82,6 +82,12 @@ class RawFileSplitRead : public AbstractSplitRead { const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, const std::optional>& ranges, const std::shared_ptr& data_file_path_factory) const override; + + private: + Result> CreateReader( + const BinaryRow& partition, int32_t bucket, + const std::vector>& files, DeletionVector::Factory dv_factory, + std::shared_ptr data_file_path_factory); }; } // namespace paimon From ad8fa331fee21d7e5922f62c63e18fc9a5c569d2 Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Thu, 9 Jul 2026 15:09:34 +0800 Subject: [PATCH 3/6] Reorder ChainDataSplit tail parameter --- src/paimon/core/table/source/split.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/paimon/core/table/source/split.cpp b/src/paimon/core/table/source/split.cpp index c3059796b..6fbf8a14d 100644 --- a/src/paimon/core/table/source/split.cpp +++ b/src/paimon/core/table/source/split.cpp @@ -210,8 +210,8 @@ Result> ReadStringMap(DataInputStre } Result> ReadChainDataSplitTail( - const std::shared_ptr& base_split, DataInputStream* in, - const std::shared_ptr& pool) { + const std::shared_ptr& base_split, const std::shared_ptr& pool, + DataInputStream* in) { PAIMON_ASSIGN_OR_RAISE(bool all_snapshot_split, in->ReadValue()); PAIMON_ASSIGN_OR_RAISE(BinaryRow read_partition, SerializationUtils::DeserializeBinaryRow(in, pool.get())); @@ -327,7 +327,7 @@ Result> Split::Deserialize(const char* buffer, size_t len if (pos == stream_length) { return data_split; } else if (data_split->BucketPath() == ChainDataSplitImpl::VIRTUAL_BUCKET_PATH) { - auto chain_split = ReadChainDataSplitTail(data_split, &in, pool); + auto chain_split = ReadChainDataSplitTail(data_split, pool, &in); if (!chain_split.ok()) { return Status::Invalid(fmt::format("invalid ChainDataSplit byte stream: {}", chain_split.status().ToString())); @@ -337,7 +337,7 @@ Result> Split::Deserialize(const char* buffer, size_t len PAIMON_ASSIGN_OR_RAISE(bool is_fallback, in.ReadValue()); return std::make_shared(data_split, is_fallback); } else { - auto chain_split = ReadChainDataSplitTail(data_split, &in, pool); + auto chain_split = ReadChainDataSplitTail(data_split, pool, &in); if (!chain_split.ok()) { return Status::Invalid(fmt::format("invalid ChainDataSplit byte stream: {}", chain_split.status().ToString())); From 98d0811cd07fa389f33aa74c609cef865763dcaf Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Thu, 9 Jul 2026 15:30:04 +0800 Subject: [PATCH 4/6] Use assign-or-raise for chain split parsing --- src/paimon/core/table/source/split.cpp | 18 ++++++------------ 1 file changed, 6 insertions(+), 12 deletions(-) diff --git a/src/paimon/core/table/source/split.cpp b/src/paimon/core/table/source/split.cpp index 6fbf8a14d..462059b5a 100644 --- a/src/paimon/core/table/source/split.cpp +++ b/src/paimon/core/table/source/split.cpp @@ -327,22 +327,16 @@ Result> Split::Deserialize(const char* buffer, size_t len if (pos == stream_length) { return data_split; } else if (data_split->BucketPath() == ChainDataSplitImpl::VIRTUAL_BUCKET_PATH) { - auto chain_split = ReadChainDataSplitTail(data_split, pool, &in); - if (!chain_split.ok()) { - return Status::Invalid(fmt::format("invalid ChainDataSplit byte stream: {}", - chain_split.status().ToString())); - } - return std::static_pointer_cast(chain_split.value()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr chain_split, + ReadChainDataSplitTail(data_split, pool, &in)); + return std::static_pointer_cast(chain_split); } else if (pos == stream_length - 1) { PAIMON_ASSIGN_OR_RAISE(bool is_fallback, in.ReadValue()); return std::make_shared(data_split, is_fallback); } else { - auto chain_split = ReadChainDataSplitTail(data_split, pool, &in); - if (!chain_split.ok()) { - return Status::Invalid(fmt::format("invalid ChainDataSplit byte stream: {}", - chain_split.status().ToString())); - } - return std::static_pointer_cast(chain_split.value()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr chain_split, + ReadChainDataSplitTail(data_split, pool, &in)); + return std::static_pointer_cast(chain_split); } } return Status::Invalid("invalid split, must be DataSplit or IndexedSplit"); From 2e87f754f411a2a75cb24f9a9bf4c8d62e0ec523 Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Thu, 9 Jul 2026 15:48:45 +0800 Subject: [PATCH 5/6] Align chain split parser with community format --- src/paimon/CMakeLists.txt | 4 +- .../core/io/chain_data_file_path_factory.cpp | 45 --- .../core/io/chain_split_file_path_factory.cpp | 66 ++++ ...tory.h => chain_split_file_path_factory.h} | 15 +- src/paimon/core/io/data_file_path_factory.h | 4 +- .../core/operation/abstract_split_read.cpp | 19 +- .../operation/data_evolution_split_read.cpp | 9 +- .../table/source/chain_data_split_test.cpp | 223 -------------- ...n_data_split_impl.h => chain_split_impl.h} | 10 +- .../core/table/source/chain_split_test.cpp | 283 ++++++++++++++++++ src/paimon/core/table/source/split.cpp | 282 +++++++++++++---- test/test_data/compatibility/split-v1-chain | Bin 0 -> 1359 bytes .../compatibility/split-v1-fallback-data | Bin 0 -> 897 bytes test/test_data/compatibility/split-v1-indexed | Bin 0 -> 961 bytes 14 files changed, 601 insertions(+), 359 deletions(-) delete mode 100644 src/paimon/core/io/chain_data_file_path_factory.cpp create mode 100644 src/paimon/core/io/chain_split_file_path_factory.cpp rename src/paimon/core/io/{chain_data_file_path_factory.h => chain_split_file_path_factory.h} (61%) delete mode 100644 src/paimon/core/table/source/chain_data_split_test.cpp rename src/paimon/core/table/source/{chain_data_split_impl.h => chain_split_impl.h} (84%) create mode 100644 src/paimon/core/table/source/chain_split_test.cpp create mode 100644 test/test_data/compatibility/split-v1-chain create mode 100644 test/test_data/compatibility/split-v1-fallback-data create mode 100644 test/test_data/compatibility/split-v1-indexed diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index e12908c71..cf42f3080 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -220,7 +220,7 @@ set(PAIMON_CORE_SRCS core/io/data_file_meta_first_row_id_legacy_serializer.cpp core/io/data_file_meta.cpp core/io/data_file_meta_serializer.cpp - core/io/chain_data_file_path_factory.cpp + core/io/chain_split_file_path_factory.cpp core/io/data_file_path_factory.cpp core/io/append_data_file_writer_factory.cpp core/io/blob_data_file_writer_factory.cpp @@ -736,7 +736,7 @@ if(PAIMON_BUILD_TESTS) core/table/source/table_read_test.cpp core/table/source/append_count_reader_test.cpp core/table/source/pk_count_reader_test.cpp - core/table/source/chain_data_split_test.cpp + core/table/source/chain_split_test.cpp core/table/source/data_split_test.cpp core/table/source/deletion_file_test.cpp core/table/source/split_generator_test.cpp diff --git a/src/paimon/core/io/chain_data_file_path_factory.cpp b/src/paimon/core/io/chain_data_file_path_factory.cpp deleted file mode 100644 index c5419899f..000000000 --- a/src/paimon/core/io/chain_data_file_path_factory.cpp +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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/io/chain_data_file_path_factory.h" - -#include - -#include "paimon/common/utils/path_util.h" -#include "paimon/core/io/data_file_meta.h" - -namespace paimon { - -ChainDataFilePathFactory::ChainDataFilePathFactory( - std::shared_ptr fallback, - std::unordered_map file_bucket_path_mapping) - : fallback_(std::move(fallback)), - file_bucket_path_mapping_(std::move(file_bucket_path_mapping)) {} - -std::string ChainDataFilePathFactory::ToPath(const std::shared_ptr& file_meta) const { - if (file_meta->external_path) { - return file_meta->external_path.value(); - } - - auto it = file_bucket_path_mapping_.find(file_meta->file_name); - if (it != file_bucket_path_mapping_.end()) { - return PathUtil::JoinPath(it->second, file_meta->file_name); - } - - return fallback_->ToPath(file_meta); -} - -} // namespace paimon diff --git a/src/paimon/core/io/chain_split_file_path_factory.cpp b/src/paimon/core/io/chain_split_file_path_factory.cpp new file mode 100644 index 000000000..08c94dc1f --- /dev/null +++ b/src/paimon/core/io/chain_split_file_path_factory.cpp @@ -0,0 +1,66 @@ +/* + * 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/io/chain_split_file_path_factory.h" + +#include + +#include "fmt/format.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/status.h" + +namespace paimon { + +Result> ChainSplitFilePathFactory::Create( + const std::vector>& data_files, + std::unordered_map file_bucket_path_mapping) { + for (const auto& file : data_files) { + if (file->external_path) { + continue; + } + if (file_bucket_path_mapping.find(file->file_name) == file_bucket_path_mapping.end()) { + return Status::Invalid( + fmt::format("bucket path is missing for ChainSplit file {}", file->file_name)); + } + } + return std::make_shared(std::move(file_bucket_path_mapping)); +} + +ChainSplitFilePathFactory::ChainSplitFilePathFactory( + std::unordered_map file_bucket_path_mapping) + : file_bucket_path_mapping_(std::move(file_bucket_path_mapping)) {} + +std::string ChainSplitFilePathFactory::ToPath( + const std::shared_ptr& file_meta) const { + if (file_meta->external_path) { + return file_meta->external_path.value(); + } + + return PathUtil::JoinPath(file_bucket_path_mapping_.at(file_meta->file_name), + file_meta->file_name); +} + +std::string ChainSplitFilePathFactory::ToAlignedPath( + const std::string& file_name, const std::shared_ptr& aligned) const { + auto external_path = aligned->ExternalPathDir(); + if (external_path) { + return PathUtil::JoinPath(external_path.value(), file_name); + } + return PathUtil::JoinPath(file_bucket_path_mapping_.at(aligned->file_name), file_name); +} + +} // namespace paimon diff --git a/src/paimon/core/io/chain_data_file_path_factory.h b/src/paimon/core/io/chain_split_file_path_factory.h similarity index 61% rename from src/paimon/core/io/chain_data_file_path_factory.h rename to src/paimon/core/io/chain_split_file_path_factory.h index 30270a30f..70e2cd137 100644 --- a/src/paimon/core/io/chain_data_file_path_factory.h +++ b/src/paimon/core/io/chain_split_file_path_factory.h @@ -19,20 +19,27 @@ #include #include #include +#include #include "paimon/core/io/data_file_path_factory.h" +#include "paimon/result.h" namespace paimon { -class ChainDataFilePathFactory : public DataFilePathFactory { +class ChainSplitFilePathFactory : public DataFilePathFactory { public: - ChainDataFilePathFactory(std::shared_ptr fallback, - std::unordered_map file_bucket_path_mapping); + static Result> Create( + const std::vector>& data_files, + std::unordered_map file_bucket_path_mapping); + + explicit ChainSplitFilePathFactory( + std::unordered_map file_bucket_path_mapping); std::string ToPath(const std::shared_ptr& file_meta) const override; + std::string ToAlignedPath(const std::string& file_name, + const std::shared_ptr& aligned) const override; private: - std::shared_ptr fallback_; std::unordered_map file_bucket_path_mapping_; }; diff --git a/src/paimon/core/io/data_file_path_factory.h b/src/paimon/core/io/data_file_path_factory.h index e071c42c4..4935e3bea 100644 --- a/src/paimon/core/io/data_file_path_factory.h +++ b/src/paimon/core/io/data_file_path_factory.h @@ -77,8 +77,8 @@ class DataFilePathFactory : public PathFactory { } std::string ToFileIndexPath(const std::string& file_path) const; - std::string ToAlignedPath(const std::string& file_name, - const std::shared_ptr& aligned) const; + virtual std::string ToAlignedPath(const std::string& file_name, + const std::shared_ptr& aligned) const; std::vector CollectFiles(const std::shared_ptr& file_meta) const; bool IsExternalPath() const { diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp index 46d5cc86d..cd767ecca 100644 --- a/src/paimon/core/operation/abstract_split_read.cpp +++ b/src/paimon/core/operation/abstract_split_read.cpp @@ -33,7 +33,7 @@ #include "paimon/common/table/special_fields.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/object_utils.h" -#include "paimon/core/io/chain_data_file_path_factory.h" +#include "paimon/core/io/chain_split_file_path_factory.h" #include "paimon/core/io/complete_row_tracking_fields_reader.h" #include "paimon/core/io/data_file_meta.h" #include "paimon/core/io/data_file_path_factory.h" @@ -41,7 +41,7 @@ #include "paimon/core/operation/internal_read_context.h" #include "paimon/core/partition/partition_info.h" #include "paimon/core/schema/table_schema.h" -#include "paimon/core/table/source/chain_data_split_impl.h" +#include "paimon/core/table/source/chain_split_impl.h" #include "paimon/core/table/source/data_split_impl.h" #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/nested_projection_utils.h" @@ -120,17 +120,16 @@ Result> AbstractSplitRead::ApplyPredicateFilterIfNe Result> AbstractSplitRead::CreateDataFilePathFactory( const std::shared_ptr& data_split) const { + auto chain_split = std::dynamic_pointer_cast(data_split); + if (chain_split) { + return ChainSplitFilePathFactory::Create(chain_split->DataFiles(), + chain_split->FileBucketPathMapping()); + } + PAIMON_ASSIGN_OR_RAISE( std::shared_ptr base_factory, path_factory_->CreateDataFilePathFactory(data_split->Partition(), data_split->Bucket())); - - auto chain_split = std::dynamic_pointer_cast(data_split); - if (!chain_split) { - return base_factory; - } - - return std::make_shared(base_factory, - chain_split->FileBucketPathMapping()); + return base_factory; } Result> AbstractSplitRead::PrepareReaderBuilder( diff --git a/src/paimon/core/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp index 47f81b2d7..1b1c6128e 100644 --- a/src/paimon/core/operation/data_evolution_split_read.cpp +++ b/src/paimon/core/operation/data_evolution_split_read.cpp @@ -196,8 +196,8 @@ Result> DataEvolutionSplitRead::WrapWithBlobViewRes 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) { + auto split_impl = std::dynamic_pointer_cast(data_split); + if (!split_impl) { return Status::Invalid("unexpected error, split cast to impl failed"); } assert(raw_read_schema_->num_fields() > 0); @@ -214,9 +214,8 @@ Result> DataEvolutionSplitRead::CreateBlobViewReade } 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())); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_file_path_factory, + CreateDataFilePathFactory(split_impl)); // skip blob files: they only contain blob payloads, not the blob-view columns. std::vector> data_files; diff --git a/src/paimon/core/table/source/chain_data_split_test.cpp b/src/paimon/core/table/source/chain_data_split_test.cpp deleted file mode 100644 index 82f27d451..000000000 --- a/src/paimon/core/table/source/chain_data_split_test.cpp +++ /dev/null @@ -1,223 +0,0 @@ -/* - * 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 -#include -#include -#include -#include -#include - -#include "gtest/gtest.h" -#include "paimon/common/data/binary_row.h" -#include "paimon/common/io/memory_segment_output_stream.h" -#include "paimon/common/memory/memory_segment_utils.h" -#include "paimon/common/utils/serialization_utils.h" -#include "paimon/core/io/chain_data_file_path_factory.h" -#include "paimon/core/io/data_file_meta.h" -#include "paimon/core/io/data_file_meta_serializer.h" -#include "paimon/core/io/data_file_path_factory.h" -#include "paimon/core/manifest/file_source.h" -#include "paimon/core/stats/simple_stats.h" -#include "paimon/core/table/source/chain_data_split_impl.h" -#include "paimon/core/table/source/data_split_impl.h" -#include "paimon/core/table/source/deletion_file.h" -#include "paimon/data/timestamp.h" -#include "paimon/memory/bytes.h" -#include "paimon/memory/memory_pool.h" -#include "paimon/table/source/data_split.h" -#include "paimon/testing/utils/testharness.h" - -namespace paimon::test { -namespace { -constexpr const char* kFileName = "data-b446f78a-2cfb-4b3b-add8-31295d24a277-0.parquet"; -constexpr const char* kBucketPath = "data/parquet/append_09.db/append_09/f1=20/bucket-0"; -constexpr const char* kBranch = "delta"; -constexpr const char* kFallbackBucketPath = "data/parquet/append_09.db/append_09/f1=10/bucket-1"; -constexpr const char* kSingleSplitBucketPath = - "data/parquet/append_table_with_append_pt_branch.db/append_table_with_append_pt_branch/" - "pt=2/bucket-0"; -constexpr const char* kSingleSplitFileName0 = "data-39204ff8-55b2-497b-8e87-a1c736799eab-0.parquet"; -constexpr const char* kSingleSplitFileName1 = "data-625b3277-84d3-4320-80b9-89a5075bf5fd-0.parquet"; - -std::shared_ptr CreateDataFileMeta( - const std::string& file_name, std::optional external_path = std::nullopt) { - return std::make_shared( - file_name, /*file_size=*/1024, /*row_count=*/7, BinaryRow::EmptyRow(), - BinaryRow::EmptyRow(), SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), - /*min_sequence_number=*/0, /*max_sequence_number=*/6, /*schema_id=*/0, - DataFileMeta::DUMMY_LEVEL, std::vector>(), Timestamp(0, 0), - /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(), - /*value_stats_cols=*/std::nullopt, external_path, - /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt); -} - -void WriteStringMap(MemorySegmentOutputStream* out, - const std::unordered_map& values) { - out->WriteValue(values.size()); - for (const auto& [key, value] : values) { - out->WriteString(key); - out->WriteString(value); - } -} - -std::string ToString(const MemorySegmentOutputStream& out, - const std::shared_ptr& pool) { - PAIMON_UNIQUE_PTR bytes = - MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); - return std::string(bytes->data(), bytes->size()); -} - -std::string SerializeVersion7DataSplit(const std::shared_ptr& split, - const std::shared_ptr& pool) { - MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); - out.WriteValue(DataSplitImpl::MAGIC); - out.WriteValue(7); - out.WriteValue(split->SnapshotId()); - EXPECT_OK(SerializationUtils::SerializeBinaryRow(split->Partition(), &out)); - out.WriteValue(split->Bucket()); - out.WriteString(split->BucketPath()); - - out.WriteValue(split->TotalBuckets().has_value()); - if (split->TotalBuckets().has_value()) { - out.WriteValue(split->TotalBuckets().value()); - } - - DataFileMetaSerializer serializer(pool); - EXPECT_OK(serializer.SerializeList(split->BeforeFiles(), &out)); - DeletionFile::SerializeList(split->BeforeDeletionFiles(), &out); - EXPECT_OK(serializer.SerializeList(split->DataFiles(), &out)); - DeletionFile::SerializeList(split->DeletionFiles(), &out); - out.WriteValue(split->IsStreaming()); - out.WriteValue(split->RawConvertible()); - return ToString(out, pool); -} - -std::shared_ptr CreateBaseSplit(const std::shared_ptr& pool) { - DataSplitImpl::Builder builder(BinaryRow::EmptyRow(), /*bucket=*/118, - ChainDataSplitImpl::VIRTUAL_BUCKET_PATH, - {CreateDataFileMeta(kFileName)}); - return builder.WithSnapshot(42) - .WithTotalBuckets(256) - .IsStreaming(false) - .RawConvertible(true) - .Build() - .value(); -} - -std::string AppendChainDataSplitTail( - const std::string& base_bytes, const std::shared_ptr& pool, - const std::unordered_map& bucket_paths, - const std::unordered_map& branches) { - MemorySegmentOutputStream tail(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); - tail.WriteValue(true); - EXPECT_OK(SerializationUtils::SerializeBinaryRow(BinaryRow::EmptyRow(), &tail)); - WriteStringMap(&tail, bucket_paths); - WriteStringMap(&tail, branches); - return base_bytes + ToString(tail, pool); -} - -std::string AppendMalformedChainDataSplitTail(const std::string& base_bytes, - const std::shared_ptr& pool) { - MemorySegmentOutputStream tail(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); - tail.WriteValue(true); - return base_bytes + ToString(tail, pool); -} - -} // namespace - -TEST(ChainDataSplitTest, DeserializeChainDataSplitTail) { - auto pool = GetDefaultPool(); - auto base_split = CreateBaseSplit(pool); - ASSERT_OK_AND_ASSIGN(std::string base_bytes, Split::Serialize(base_split, pool)); - - std::string chain_bytes = AppendChainDataSplitTail(base_bytes, pool, {{kFileName, kBucketPath}}, - {{kFileName, kBranch}}); - - ASSERT_OK_AND_ASSIGN(std::shared_ptr split, - Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool)); - auto chain_split = std::dynamic_pointer_cast(split); - ASSERT_TRUE(chain_split); - EXPECT_TRUE(chain_split->AllSnapshotSplit()); - EXPECT_EQ(chain_split->BucketPath(), ChainDataSplitImpl::VIRTUAL_BUCKET_PATH); - EXPECT_EQ(chain_split->FileBucketPathMapping().at(kFileName), kBucketPath); - EXPECT_EQ(chain_split->FileBranchMapping().at(kFileName), kBranch); -} - -TEST(ChainDataSplitTest, DeserializeVersion7SingleSplitWithOriginalBucketPath) { - auto pool = GetDefaultPool(); - DataSplitImpl::Builder builder( - BinaryRow::EmptyRow(), /*bucket=*/3, kSingleSplitBucketPath, - {CreateDataFileMeta(kSingleSplitFileName0), CreateDataFileMeta(kSingleSplitFileName1)}); - ASSERT_OK_AND_ASSIGN(auto base_split, builder.WithSnapshot(42) - .WithTotalBuckets(256) - .IsStreaming(false) - .RawConvertible(true) - .Build()); - std::string base_bytes = SerializeVersion7DataSplit(base_split, pool); - std::string split_bytes = AppendChainDataSplitTail( - base_bytes, pool, - {{kSingleSplitFileName0, kSingleSplitBucketPath}, - {kSingleSplitFileName1, kSingleSplitBucketPath}}, - {{kSingleSplitFileName0, "snapshot"}, {kSingleSplitFileName1, "snapshot"}}); - - ASSERT_OK_AND_ASSIGN(std::shared_ptr split, - Split::Deserialize(split_bytes.data(), split_bytes.size(), pool)); - auto chain_split = std::dynamic_pointer_cast(split); - ASSERT_TRUE(chain_split); - EXPECT_TRUE(chain_split->AllSnapshotSplit()); - EXPECT_EQ(chain_split->BucketPath(), kSingleSplitBucketPath); - EXPECT_EQ(chain_split->FileBucketPathMapping().at(kSingleSplitFileName0), - kSingleSplitBucketPath); - EXPECT_EQ(chain_split->FileBucketPathMapping().at(kSingleSplitFileName1), - kSingleSplitBucketPath); - EXPECT_EQ(chain_split->FileBranchMapping().at(kSingleSplitFileName0), "snapshot"); - EXPECT_EQ(chain_split->FileBranchMapping().at(kSingleSplitFileName1), "snapshot"); -} - -TEST(ChainDataSplitTest, ChainDataFilePathFactoryUsesPerFileBucketPath) { - auto fallback = std::make_shared(); - ASSERT_OK(fallback->Init(kFallbackBucketPath, "parquet", "data-", - /*external_path_provider=*/nullptr)); - ChainDataFilePathFactory factory(fallback, {{kFileName, kBucketPath}}); - - auto file_meta = CreateDataFileMeta(kFileName); - EXPECT_EQ(factory.ToPath(file_meta), std::string(kBucketPath) + "/" + kFileName); -} - -TEST(ChainDataSplitTest, ChainDataFilePathFactoryPreservesExternalPath) { - auto fallback = std::make_shared(); - ASSERT_OK(fallback->Init(kFallbackBucketPath, "parquet", "data-", - /*external_path_provider=*/nullptr)); - ChainDataFilePathFactory factory(fallback, {{kFileName, kBucketPath}}); - - const std::string external_path = "hdfs://external/path/" + std::string(kFileName); - auto file_meta = CreateDataFileMeta(kFileName, external_path); - EXPECT_EQ(factory.ToPath(file_meta), external_path); -} - -TEST(ChainDataSplitTest, MalformedChainDataSplitTailReturnsContextualError) { - auto pool = GetDefaultPool(); - auto base_split = CreateBaseSplit(pool); - ASSERT_OK_AND_ASSIGN(std::string base_bytes, Split::Serialize(base_split, pool)); - - std::string chain_bytes = AppendMalformedChainDataSplitTail(base_bytes, pool); - - ASSERT_NOK_WITH_MSG(Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool), - "ChainDataSplit"); -} - -} // namespace paimon::test diff --git a/src/paimon/core/table/source/chain_data_split_impl.h b/src/paimon/core/table/source/chain_split_impl.h similarity index 84% rename from src/paimon/core/table/source/chain_data_split_impl.h rename to src/paimon/core/table/source/chain_split_impl.h index 7588a3bfc..0305f305a 100644 --- a/src/paimon/core/table/source/chain_data_split_impl.h +++ b/src/paimon/core/table/source/chain_split_impl.h @@ -27,14 +27,14 @@ namespace paimon { -class ChainDataSplitImpl : public DataSplitImpl { +class ChainSplitImpl : public DataSplitImpl { public: static constexpr const char* VIRTUAL_BUCKET_PATH = "placeholder::virtual-bucket-path"; - ChainDataSplitImpl(const std::shared_ptr& base_split, bool all_snapshot_split, - const BinaryRow& read_partition, - std::unordered_map&& file_bucket_path_mapping, - std::unordered_map&& file_branch_mapping) + ChainSplitImpl(const std::shared_ptr& base_split, bool all_snapshot_split, + const BinaryRow& read_partition, + std::unordered_map&& file_bucket_path_mapping, + std::unordered_map&& file_branch_mapping) : DataSplitImpl(base_split->Partition(), base_split->Bucket(), base_split->BucketPath(), std::vector>(base_split->DataFiles())), all_snapshot_split_(all_snapshot_split), diff --git a/src/paimon/core/table/source/chain_split_test.cpp b/src/paimon/core/table/source/chain_split_test.cpp new file mode 100644 index 000000000..33d599d6c --- /dev/null +++ b/src/paimon/core/table/source/chain_split_test.cpp @@ -0,0 +1,283 @@ +/* + * 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 +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/memory/memory_segment_utils.h" +#include "paimon/common/utils/serialization_utils.h" +#include "paimon/core/global_index/indexed_split_impl.h" +#include "paimon/core/io/chain_split_file_path_factory.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/io/data_file_meta_serializer.h" +#include "paimon/core/manifest/file_source.h" +#include "paimon/core/stats/simple_stats.h" +#include "paimon/core/table/source/chain_split_impl.h" +#include "paimon/core/table/source/data_split_impl.h" +#include "paimon/core/table/source/fallback_data_split.h" +#include "paimon/data/timestamp.h" +#include "paimon/fs/local/local_file_system.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/table/source/data_split.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::test { +namespace { +constexpr const char* kFileName = "data-b446f78a-2cfb-4b3b-add8-31295d24a277-0.parquet"; +constexpr const char* kBucketPath = "data/parquet/append_09.db/append_09/f1=20/bucket-0"; +constexpr const char* kBranch = "delta"; +constexpr const char* kSingleSplitBucketPath = + "data/parquet/append_table_with_append_pt_branch.db/append_table_with_append_pt_branch/" + "pt=2/bucket-0"; +constexpr int64_t kSplitSerializerMagic = 0x53504C49545F5631L; // "SPLIT_V1" +constexpr int32_t kSplitSerializerVersion = 1; +constexpr int32_t kSplitSerializerChainSplit = 4; + +Result ReadCompatibilityBytes(const std::string& file_name) { + auto file_system = std::make_unique(); + PAIMON_ASSIGN_OR_RAISE(auto input_stream, + file_system->Open(GetDataDir() + "/compatibility/" + file_name)); + std::string bytes(input_stream->Length().value_or(0), '\0'); + PAIMON_RETURN_NOT_OK(input_stream->Read(bytes.data(), bytes.size())); + PAIMON_RETURN_NOT_OK(input_stream->Close()); + return bytes; +} + +std::shared_ptr CreateDataFileMeta( + const std::string& file_name, std::optional external_path = std::nullopt) { + return std::make_shared( + file_name, /*file_size=*/1024, /*row_count=*/7, BinaryRow::EmptyRow(), + BinaryRow::EmptyRow(), SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), + /*min_sequence_number=*/0, /*max_sequence_number=*/6, /*schema_id=*/0, + DataFileMeta::DUMMY_LEVEL, std::vector>(), Timestamp(0, 0), + /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(), + /*value_stats_cols=*/std::nullopt, external_path, + /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt); +} + +void WriteStringMap(MemorySegmentOutputStream* out, + const std::unordered_map& values) { + out->WriteValue(true); + out->WriteValue(values.size()); + for (const auto& [key, value] : values) { + out->WriteValue(static_cast(key.size())); + out->Write(key.data(), static_cast(key.size())); + out->WriteValue(static_cast(value.size())); + out->Write(value.data(), static_cast(value.size())); + } +} + +std::string ToString(const MemorySegmentOutputStream& out, + const std::shared_ptr& pool) { + PAIMON_UNIQUE_PTR bytes = + MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); + return std::string(bytes->data(), bytes->size()); +} + +std::string SerializeChainSplit(const BinaryRow& logical_partition, + const std::vector>& data_files, + const std::unordered_map& bucket_paths, + const std::unordered_map& branches, + const std::shared_ptr& pool) { + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + out.WriteValue(kSplitSerializerMagic); + out.WriteValue(kSplitSerializerVersion); + out.WriteValue(kSplitSerializerChainSplit); + EXPECT_OK(SerializationUtils::SerializeBinaryRow(logical_partition, &out)); + DataFileMetaSerializer serializer(pool); + EXPECT_OK(serializer.SerializeList(data_files, &out)); + WriteStringMap(&out, bucket_paths); + WriteStringMap(&out, branches); + return ToString(out, pool); +} + +} // namespace + +TEST(ChainSplitTest, DeserializeChainSplit) { + auto pool = GetDefaultPool(); + + std::string chain_bytes = + SerializeChainSplit(BinaryRow::EmptyRow(), {CreateDataFileMeta(kFileName)}, + {{kFileName, kBucketPath}}, {{kFileName, kBranch}}, pool); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool)); + auto chain_split = std::dynamic_pointer_cast(split); + ASSERT_TRUE(chain_split); + EXPECT_FALSE(chain_split->AllSnapshotSplit()); + EXPECT_EQ(chain_split->BucketPath(), ChainSplitImpl::VIRTUAL_BUCKET_PATH); + ASSERT_EQ(chain_split->DataFiles().size(), 1U); + EXPECT_EQ(chain_split->DataFiles()[0]->file_name, kFileName); + EXPECT_EQ(chain_split->FileBucketPathMapping().at(kFileName), kBucketPath); + EXPECT_EQ(chain_split->FileBranchMapping().at(kFileName), kBranch); +} + +TEST(ChainSplitTest, DeserializeJavaGoldenChainSplit) { + auto pool = GetDefaultPool(); + ASSERT_OK_AND_ASSIGN(std::string chain_bytes, ReadCompatibilityBytes("split-v1-chain")); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool)); + auto chain_split = std::dynamic_pointer_cast(split); + ASSERT_TRUE(chain_split); + EXPECT_EQ(chain_split->FileBucketPathMapping().at("file-a"), "dt=20260706/bucket-3"); + EXPECT_EQ(chain_split->FileBucketPathMapping().at("file-b"), "dt=20260706/bucket-3"); + EXPECT_EQ(chain_split->FileBucketPathMapping().at("chain-file"), "dt=20260707/bucket-5"); + EXPECT_EQ(chain_split->FileBranchMapping().at("file-a"), "snapshot"); + EXPECT_EQ(chain_split->FileBranchMapping().at("file-b"), "snapshot"); + EXPECT_EQ(chain_split->FileBranchMapping().at("chain-file"), "delta"); +} + +TEST(ChainSplitTest, DeserializeJavaGoldenIndexedSplit) { + auto pool = GetDefaultPool(); + ASSERT_OK_AND_ASSIGN(std::string indexed_bytes, ReadCompatibilityBytes("split-v1-indexed")); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(indexed_bytes.data(), indexed_bytes.size(), pool)); + auto indexed_split = std::dynamic_pointer_cast(split); + ASSERT_TRUE(indexed_split); + ASSERT_EQ(indexed_split->RowRanges().size(), 2U); + EXPECT_EQ(indexed_split->RowRanges()[0], Range(1, 4)); + EXPECT_EQ(indexed_split->RowRanges()[1], Range(11, 13)); + ASSERT_EQ(indexed_split->Scores().size(), 3U); + EXPECT_FLOAT_EQ(indexed_split->Scores()[0], 0.5f); + EXPECT_FLOAT_EQ(indexed_split->Scores()[1], 0.25f); + EXPECT_FLOAT_EQ(indexed_split->Scores()[2], 0.125f); +} + +TEST(ChainSplitTest, DeserializeJavaGoldenFallbackDataSplit) { + auto pool = GetDefaultPool(); + ASSERT_OK_AND_ASSIGN(std::string fallback_bytes, + ReadCompatibilityBytes("split-v1-fallback-data")); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(fallback_bytes.data(), fallback_bytes.size(), pool)); + auto fallback_split = std::dynamic_pointer_cast(split); + ASSERT_TRUE(fallback_split); + EXPECT_TRUE(fallback_split->IsFallback()); + ASSERT_TRUE(std::dynamic_pointer_cast(fallback_split->GetSplit())); + + ASSERT_OK_AND_ASSIGN(std::string serialized, Split::Serialize(fallback_split, pool)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr round_trip, + Split::Deserialize(serialized.data(), serialized.size(), pool)); + auto round_trip_fallback = std::dynamic_pointer_cast(round_trip); + ASSERT_TRUE(round_trip_fallback); + EXPECT_TRUE(round_trip_fallback->IsFallback()); +} + +TEST(ChainSplitTest, DeserializeChainSplitWithMultipleFiles) { + auto pool = GetDefaultPool(); + const std::string file_name0 = "data-39204ff8-55b2-497b-8e87-a1c736799eab-0.parquet"; + const std::string file_name1 = "data-625b3277-84d3-4320-80b9-89a5075bf5fd-0.parquet"; + + std::string split_bytes = SerializeChainSplit( + BinaryRow::EmptyRow(), {CreateDataFileMeta(file_name0), CreateDataFileMeta(file_name1)}, + {{file_name0, kSingleSplitBucketPath}, {file_name1, kSingleSplitBucketPath}}, + {{file_name0, "snapshot"}, {file_name1, "snapshot"}}, pool); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(split_bytes.data(), split_bytes.size(), pool)); + auto chain_split = std::dynamic_pointer_cast(split); + ASSERT_TRUE(chain_split); + EXPECT_FALSE(chain_split->AllSnapshotSplit()); + EXPECT_EQ(chain_split->BucketPath(), ChainSplitImpl::VIRTUAL_BUCKET_PATH); + ASSERT_EQ(chain_split->DataFiles().size(), 2U); + EXPECT_EQ(chain_split->FileBucketPathMapping().at(file_name0), kSingleSplitBucketPath); + EXPECT_EQ(chain_split->FileBucketPathMapping().at(file_name1), kSingleSplitBucketPath); + EXPECT_EQ(chain_split->FileBranchMapping().at(file_name0), "snapshot"); + EXPECT_EQ(chain_split->FileBranchMapping().at(file_name1), "snapshot"); +} + +TEST(ChainSplitTest, ChainSplitFilePathFactoryUsesPerFileBucketPath) { + ChainSplitFilePathFactory factory( + std::unordered_map{{kFileName, kBucketPath}}); + + auto file_meta = CreateDataFileMeta(kFileName); + EXPECT_EQ(factory.ToPath(file_meta), std::string(kBucketPath) + "/" + kFileName); +} + +TEST(ChainSplitTest, ChainSplitFilePathFactoryUsesPerFileBucketPathForAlignedFile) { + ChainSplitFilePathFactory factory( + std::unordered_map{{kFileName, kBucketPath}}); + + auto file_meta = CreateDataFileMeta(kFileName); + EXPECT_EQ(factory.ToAlignedPath("index-0.index", file_meta), + std::string(kBucketPath) + "/index-0.index"); +} + +TEST(ChainSplitTest, ChainSplitFilePathFactoryPreservesExternalPath) { + ChainSplitFilePathFactory factory( + std::unordered_map{{kFileName, kBucketPath}}); + + const std::string external_path = "hdfs://external/path/" + std::string(kFileName); + auto file_meta = CreateDataFileMeta(kFileName, external_path); + EXPECT_EQ(factory.ToPath(file_meta), external_path); +} + +TEST(ChainSplitTest, ChainSplitFilePathFactoryPreservesExternalAlignedPath) { + ChainSplitFilePathFactory factory( + std::unordered_map{{kFileName, kBucketPath}}); + + const std::string external_path = "hdfs://external/path/" + std::string(kFileName); + auto file_meta = CreateDataFileMeta(kFileName, external_path); + EXPECT_EQ(factory.ToAlignedPath("index-0.index", file_meta), + std::string("hdfs://external/path/index-0.index")); +} + +TEST(ChainSplitTest, ChainSplitFilePathFactoryRejectsMissingBucketPathMapping) { + ASSERT_NOK_WITH_MSG(ChainSplitFilePathFactory::Create({CreateDataFileMeta(kFileName)}, {}), + "bucket path is missing for ChainSplit file"); +} + +TEST(ChainSplitTest, SerializeChainSplitPreservesMappings) { + auto pool = GetDefaultPool(); + + std::string chain_bytes = + SerializeChainSplit(BinaryRow::EmptyRow(), {CreateDataFileMeta(kFileName)}, + {{kFileName, kBucketPath}}, {{kFileName, kBranch}}, pool); + ASSERT_OK_AND_ASSIGN(std::shared_ptr split, + Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool)); + ASSERT_OK_AND_ASSIGN(std::string serialized, Split::Serialize(split, pool)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr round_trip, + Split::Deserialize(serialized.data(), serialized.size(), pool)); + + auto chain_split = std::dynamic_pointer_cast(round_trip); + ASSERT_TRUE(chain_split); + EXPECT_EQ(chain_split->FileBucketPathMapping().at(kFileName), kBucketPath); + EXPECT_EQ(chain_split->FileBranchMapping().at(kFileName), kBranch); +} + +TEST(ChainSplitTest, MalformedChainSplitReturnsError) { + auto pool = GetDefaultPool(); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + out.WriteValue(kSplitSerializerMagic); + out.WriteValue(kSplitSerializerVersion); + out.WriteValue(kSplitSerializerChainSplit); + EXPECT_OK(SerializationUtils::SerializeBinaryRow(BinaryRow::EmptyRow(), &out)); + std::string chain_bytes = ToString(out, pool); + + ASSERT_NOK(Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool)); +} + +} // namespace paimon::test diff --git a/src/paimon/core/table/source/split.cpp b/src/paimon/core/table/source/split.cpp index 462059b5a..fd7d1f14e 100644 --- a/src/paimon/core/table/source/split.cpp +++ b/src/paimon/core/table/source/split.cpp @@ -14,8 +14,12 @@ * limitations under the License. */ +#include +#include +#include #include #include +#include #include "fmt/format.h" #include "paimon/common/data/binary_row.h" @@ -25,7 +29,7 @@ #include "paimon/core/global_index/indexed_split_impl.h" #include "paimon/core/io/data_file_meta_first_row_id_legacy_serializer.h" #include "paimon/core/io/data_file_meta_serializer.h" -#include "paimon/core/table/source/chain_data_split_impl.h" +#include "paimon/core/table/source/chain_split_impl.h" #include "paimon/core/table/source/data_split_impl.h" #include "paimon/core/table/source/deletion_file.h" #include "paimon/core/table/source/fallback_data_split.h" @@ -40,6 +44,13 @@ namespace paimon { struct DataFileMeta; namespace { +constexpr int64_t kSplitSerializerMagic = 0x53504C49545F5631L; // "SPLIT_V1" +constexpr int32_t kSplitSerializerVersion = 1; +constexpr int32_t kSplitSerializerDataSplit = 1; +constexpr int32_t kSplitSerializerIndexedSplit = 3; +constexpr int32_t kSplitSerializerChainSplit = 4; +constexpr int32_t kSplitSerializerFallbackDataSplit = 6; + Result>> ReadVersion7DataFileMetaList( DataInputStream* in, const std::shared_ptr& pool) { PAIMON_ASSIGN_OR_RAISE(int32_t size, in->ReadValue()); @@ -193,42 +204,192 @@ Result> ReadDataSplitWithoutMagicNumber( return builder.Build(); } -Result> ReadStringMap(DataInputStream* in) { +Status ValidateFullyRead(const char* split_type, DataInputStream* in) { + PAIMON_ASSIGN_OR_RAISE(int64_t pos, in->GetPos()); + PAIMON_ASSIGN_OR_RAISE(int64_t stream_length, in->Length()); + if (pos != stream_length) { + return Status::Invalid( + fmt::format("invalid {} byte stream, remaining {} bytes after deserializing", + split_type, stream_length - pos)); + } + return Status::OK(); +} + +Result> ReadSplitSerializerString(DataInputStream* in) { + PAIMON_ASSIGN_OR_RAISE(int32_t length, in->ReadValue()); + if (length < 0) { + return std::optional(); + } + PAIMON_ASSIGN_OR_RAISE(int64_t pos, in->GetPos()); + PAIMON_ASSIGN_OR_RAISE(int64_t stream_length, in->Length()); + if (length > stream_length - pos) { + return Status::Invalid( + fmt::format("split serializer string length {} exceeds remaining stream bytes {}", + length, stream_length - pos)); + } + + std::string value(length, '\0'); + if (length > 0) { + PAIMON_RETURN_NOT_OK(in->Read(value.data(), length)); + } + return std::optional(std::move(value)); +} + +Result> ReadSplitSerializerStringMap( + DataInputStream* in) { + PAIMON_ASSIGN_OR_RAISE(bool exists, in->ReadValue()); + if (!exists) { + return std::unordered_map(); + } + PAIMON_ASSIGN_OR_RAISE(int32_t size, in->ReadValue()); if (size < 0) { - return Status::Invalid(fmt::format("invalid string map size: {}", size)); + return Status::Invalid(fmt::format("invalid split serializer string map size: {}", size)); } std::unordered_map result; result.reserve(size); for (int32_t i = 0; i < size; ++i) { - PAIMON_ASSIGN_OR_RAISE(std::string key, in->ReadString()); - PAIMON_ASSIGN_OR_RAISE(std::string value, in->ReadString()); - result.emplace(std::move(key), std::move(value)); + PAIMON_ASSIGN_OR_RAISE(std::optional key, ReadSplitSerializerString(in)); + PAIMON_ASSIGN_OR_RAISE(std::optional value, ReadSplitSerializerString(in)); + if (!key || !value) { + return Status::Invalid("split serializer string map does not support null entries"); + } + result.insert_or_assign(std::move(key.value()), std::move(value.value())); } return result; } -Result> ReadChainDataSplitTail( - const std::shared_ptr& base_split, const std::shared_ptr& pool, - DataInputStream* in) { - PAIMON_ASSIGN_OR_RAISE(bool all_snapshot_split, in->ReadValue()); - PAIMON_ASSIGN_OR_RAISE(BinaryRow read_partition, +void WriteSplitSerializerString(MemorySegmentOutputStream* out, const std::string& value) { + out->WriteValue(static_cast(value.size())); + out->Write(value.data(), static_cast(value.size())); +} + +void WriteSplitSerializerStringMap(MemorySegmentOutputStream* out, + const std::unordered_map& values) { + out->WriteValue(true); + out->WriteValue(static_cast(values.size())); + for (const auto& [key, value] : + std::map(values.begin(), values.end())) { + WriteSplitSerializerString(out, key); + WriteSplitSerializerString(out, value); + } +} + +Status WriteChainSplitPayload(const std::shared_ptr& chain_split, + MemorySegmentOutputStream* out, + const std::shared_ptr& pool) { + PAIMON_RETURN_NOT_OK(SerializationUtils::SerializeBinaryRow(chain_split->ReadPartition(), out)); + DataFileMetaSerializer serializer(pool); + PAIMON_RETURN_NOT_OK(serializer.SerializeList(chain_split->DataFiles(), out)); + WriteSplitSerializerStringMap(out, chain_split->FileBucketPathMapping()); + WriteSplitSerializerStringMap(out, chain_split->FileBranchMapping()); + return Status::OK(); +} + +Result> ReadChainSplitPayload( + DataInputStream* in, const std::shared_ptr& pool) { + PAIMON_ASSIGN_OR_RAISE(BinaryRow logical_partition, SerializationUtils::DeserializeBinaryRow(in, pool.get())); - PAIMON_ASSIGN_OR_RAISE(auto file_bucket_path_mapping, ReadStringMap(in)); - PAIMON_ASSIGN_OR_RAISE(auto file_branch_mapping, ReadStringMap(in)); - PAIMON_ASSIGN_OR_RAISE(int64_t pos, in->GetPos()); - PAIMON_ASSIGN_OR_RAISE(int64_t stream_length, in->Length()); - if (pos != stream_length) { - return Status::Invalid(fmt::format( - "invalid ChainDataSplit byte stream, remaining {} bytes after deserializing", - stream_length - pos)); + DataFileMetaSerializer serializer(pool); + PAIMON_ASSIGN_OR_RAISE(std::vector> data_files, + serializer.DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(auto file_bucket_path_mapping, ReadSplitSerializerStringMap(in)); + PAIMON_ASSIGN_OR_RAISE(auto file_branch_mapping, ReadSplitSerializerStringMap(in)); + PAIMON_RETURN_NOT_OK(ValidateFullyRead("ChainSplit", in)); + + DataSplitImpl::Builder builder(logical_partition, /*bucket=*/0, + ChainSplitImpl::VIRTUAL_BUCKET_PATH, std::move(data_files)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr base_split, builder.Build()); + return std::make_shared(base_split, /*all_snapshot_split=*/false, + logical_partition, std::move(file_bucket_path_mapping), + std::move(file_branch_mapping)); +} + +Result> ReadIndexedSplitPayload( + int64_t magic, DataInputStream* in, const std::shared_ptr& pool) { + if (magic != IndexedSplitImpl::MAGIC) { + return Status::Invalid(fmt::format("Corrupted IndexedSplit: wrong magic number {}", magic)); + } + + PAIMON_ASSIGN_OR_RAISE(int32_t version, in->ReadValue()); + if (version != IndexedSplitImpl::VERSION) { + return Status::Invalid(fmt::format("Unsupported IndexedSplit version: {}", version)); + } + PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in->ReadValue()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_split, + ReadDataSplitWithoutMagicNumber(data_split_magic, in, pool)); + PAIMON_ASSIGN_OR_RAISE(int32_t range_size, in->ReadValue()); + if (range_size < 0) { + return Status::Invalid(fmt::format("invalid IndexedSplit range size: {}", range_size)); + } + std::vector row_ranges; + row_ranges.reserve(range_size); + for (int32_t i = 0; i < range_size; ++i) { + PAIMON_ASSIGN_OR_RAISE(int64_t range_from, in->ReadValue()); + PAIMON_ASSIGN_OR_RAISE(int64_t range_to, in->ReadValue()); + row_ranges.emplace_back(range_from, range_to); + } + + std::vector scores; + PAIMON_ASSIGN_OR_RAISE(bool has_scores, in->ReadValue()); + if (has_scores) { + PAIMON_ASSIGN_OR_RAISE(int32_t scores_length, in->ReadValue()); + if (scores_length < 0) { + return Status::Invalid( + fmt::format("invalid IndexedSplit scores length: {}", scores_length)); + } + scores.resize(scores_length); + for (int32_t i = 0; i < scores_length; ++i) { + PAIMON_ASSIGN_OR_RAISE(float score, in->ReadValue()); + scores[i] = score; + } } - return std::make_shared(base_split, all_snapshot_split, read_partition, - std::move(file_bucket_path_mapping), - std::move(file_branch_mapping)); + return std::make_shared(data_split, row_ranges, scores); +} + +Result> ReadSplitSerializerPayload(DataInputStream* in, + const std::shared_ptr& pool) { + PAIMON_ASSIGN_OR_RAISE(int32_t version, in->ReadValue()); + if (version != kSplitSerializerVersion) { + return Status::Invalid(fmt::format("unsupported split serializer version: {}", version)); + } + + PAIMON_ASSIGN_OR_RAISE(int32_t type, in->ReadValue()); + switch (type) { + case kSplitSerializerDataSplit: { + PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in->ReadValue()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_split, + ReadDataSplitWithoutMagicNumber(data_split_magic, in, pool)); + PAIMON_RETURN_NOT_OK(ValidateFullyRead("DataSplit", in)); + return std::static_pointer_cast(data_split); + } + case kSplitSerializerIndexedSplit: { + PAIMON_ASSIGN_OR_RAISE(int64_t indexed_split_magic, in->ReadValue()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr indexed_split, + ReadIndexedSplitPayload(indexed_split_magic, in, pool)); + PAIMON_RETURN_NOT_OK(ValidateFullyRead("IndexedSplit", in)); + return std::static_pointer_cast(indexed_split); + } + case kSplitSerializerChainSplit: { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr chain_split, + ReadChainSplitPayload(in, pool)); + return std::static_pointer_cast(chain_split); + } + case kSplitSerializerFallbackDataSplit: { + PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in->ReadValue()); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_split, + ReadDataSplitWithoutMagicNumber(data_split_magic, in, pool)); + PAIMON_ASSIGN_OR_RAISE(bool is_fallback, in->ReadValue()); + PAIMON_RETURN_NOT_OK(ValidateFullyRead("FallbackDataSplit", in)); + return std::static_pointer_cast( + std::make_shared(data_split, is_fallback)); + } + default: + return Status::Invalid(fmt::format("unsupported split serializer type: {}", type)); + } } } // namespace @@ -236,11 +397,30 @@ Result> ReadChainDataSplitTail( Result Split::Serialize(const std::shared_ptr& split, const std::shared_ptr& pool) { MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); - if (auto data_split_impl = std::dynamic_pointer_cast(split)) { + if (auto fallback_data_split = std::dynamic_pointer_cast(split)) { + if (std::dynamic_pointer_cast(fallback_data_split->GetSplit())) { + return Status::Invalid("FallbackDataSplit cannot serialize a wrapped ChainSplit"); + } + auto data_split_impl = + std::dynamic_pointer_cast(fallback_data_split->GetSplit()); + if (!data_split_impl) { + return Status::Invalid("inner split in FallbackDataSplit is supposed to be DataSplit"); + } + PAIMON_RETURN_NOT_OK(WriteDataSplit(data_split_impl, &out, pool)); + out.WriteValue(fallback_data_split->IsFallback()); + } else if (auto chain_split_impl = std::dynamic_pointer_cast(split)) { + out.WriteValue(kSplitSerializerMagic); + out.WriteValue(kSplitSerializerVersion); + out.WriteValue(kSplitSerializerChainSplit); + PAIMON_RETURN_NOT_OK(WriteChainSplitPayload(chain_split_impl, &out, pool)); + } else if (auto data_split_impl = std::dynamic_pointer_cast(split)) { PAIMON_RETURN_NOT_OK(WriteDataSplit(data_split_impl, &out, pool)); } else if (auto indexed_split_impl = std::dynamic_pointer_cast(split)) { out.WriteValue(IndexedSplitImpl::MAGIC); out.WriteValue(IndexedSplitImpl::VERSION); + if (std::dynamic_pointer_cast(indexed_split_impl->GetDataSplit())) { + return Status::Invalid("IndexedSplit cannot serialize a wrapped ChainSplit"); + } auto inner_split_impl = std::dynamic_pointer_cast(indexed_split_impl->GetDataSplit()); if (!inner_split_impl) { @@ -265,7 +445,9 @@ Result Split::Serialize(const std::shared_ptr& split, out.WriteValue(false); } } else { - return Status::Invalid("invalid split, cannot cast to DataSplit or IndexedSplit"); + return Status::Invalid( + "invalid split, cannot cast to FallbackDataSplit, ChainSplit, DataSplit or " + "IndexedSplit"); } PAIMON_UNIQUE_PTR bytes = MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); @@ -280,37 +462,16 @@ Result> Split::Deserialize(const char* buffer, size_t len int64_t magic = -1; PAIMON_ASSIGN_OR_RAISE(magic, in.ReadValue()); - if (magic == IndexedSplitImpl::MAGIC) { - PAIMON_ASSIGN_OR_RAISE(int32_t version, in.ReadValue()); - if (version != IndexedSplitImpl::VERSION) { - return Status::Invalid(fmt::format("Unsupported IndexedSplit version: {}", version)); - } - PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in.ReadValue()); - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_split, - ReadDataSplitWithoutMagicNumber(data_split_magic, &in, pool)); - PAIMON_ASSIGN_OR_RAISE(int32_t range_size, in.ReadValue()); - std::vector row_ranges; - row_ranges.reserve(range_size); - for (int32_t i = 0; i < range_size; ++i) { - PAIMON_ASSIGN_OR_RAISE(int64_t range_from, in.ReadValue()); - PAIMON_ASSIGN_OR_RAISE(int64_t range_to, in.ReadValue()); - row_ranges.emplace_back(range_from, range_to); - } - std::vector scores; - PAIMON_ASSIGN_OR_RAISE(bool has_scores, in.ReadValue()); - if (has_scores) { - PAIMON_ASSIGN_OR_RAISE(int32_t scores_length, in.ReadValue()); - scores.resize(scores_length); - for (int32_t i = 0; i < scores_length; ++i) { - PAIMON_ASSIGN_OR_RAISE(float score, in.ReadValue()); - scores[i] = score; - } - } + if (magic == kSplitSerializerMagic) { + return ReadSplitSerializerPayload(&in, pool); + } else if (magic == IndexedSplitImpl::MAGIC) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr indexed_split, + ReadIndexedSplitPayload(magic, &in, pool)); // TODO(lisizhuo.lsz): support fallback split in IndexedSplit PAIMON_ASSIGN_OR_RAISE(int64_t pos, in.GetPos()); PAIMON_ASSIGN_OR_RAISE(int64_t stream_length, in.Length()); if (pos == stream_length) { - return std::make_shared(data_split, row_ranges, scores); + return indexed_split; } else if (pos == stream_length - 1) { return Status::Invalid( "invalid IndexedSplit, do not support FallbackSplit in IndexedSplit"); @@ -322,23 +483,18 @@ Result> Split::Deserialize(const char* buffer, size_t len } else if (magic == DataSplitImpl::MAGIC) { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_split, ReadDataSplitWithoutMagicNumber(magic, &in, pool)); + auto fully_read = ValidateFullyRead("DataSplit", &in); + if (fully_read.ok()) { + return data_split; + } PAIMON_ASSIGN_OR_RAISE(int64_t pos, in.GetPos()); PAIMON_ASSIGN_OR_RAISE(int64_t stream_length, in.Length()); - if (pos == stream_length) { - return data_split; - } else if (data_split->BucketPath() == ChainDataSplitImpl::VIRTUAL_BUCKET_PATH) { - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr chain_split, - ReadChainDataSplitTail(data_split, pool, &in)); - return std::static_pointer_cast(chain_split); - } else if (pos == stream_length - 1) { + if (pos == stream_length - 1) { PAIMON_ASSIGN_OR_RAISE(bool is_fallback, in.ReadValue()); return std::make_shared(data_split, is_fallback); - } else { - PAIMON_ASSIGN_OR_RAISE(std::shared_ptr chain_split, - ReadChainDataSplitTail(data_split, pool, &in)); - return std::static_pointer_cast(chain_split); } + return fully_read; } - return Status::Invalid("invalid split, must be DataSplit or IndexedSplit"); + return Status::Invalid("invalid split, must be SplitSerializer, DataSplit or IndexedSplit"); } } // namespace paimon diff --git a/test/test_data/compatibility/split-v1-chain b/test/test_data/compatibility/split-v1-chain new file mode 100644 index 0000000000000000000000000000000000000000..d9f12c976540a7d55a8aeb60d07d3a912b1aad71 GIT binary patch literal 1359 zcmWFz@bL_Z4>M$7U|!z*?$1tOS%N(vYl9QZ+;w9K4T-9&~qE)WL@ zU^Gbc3Lpm22Y}cEh;Kk?1&|F;kOE_%(Hua222dR?c{JttxNzIx42Wtt7v^SULwJB3 z1t12|20#pQj{^{s3aDicMwlm&ALb{}JOK*`T7)?_s$DRv!07^23KG@|K*2#2<_)lP z4oV{{fEbomU^FOw?*L*DeF2C)fcOQJHh}wXV1dKHh2lYxloDGb10yp7a|3hzq|)T<)Dm4& zpeiM$7U|6hdjRnbD6If84hm9W3^W>KTLw@aE_pQN__%P};0%arI2Yz-WJ7p>90ec-(FQ;a za*qQLlM1M14n~+KksszK&^!SP2wH?WH>zDQtH9|3RSFW;3P8cZ66PSsG6Ay_XG)nq TIGra!T@1@Xu!O_Rz{m&yE!-qE literal 0 HcmV?d00001 diff --git a/test/test_data/compatibility/split-v1-indexed b/test/test_data/compatibility/split-v1-indexed new file mode 100644 index 0000000000000000000000000000000000000000..0d20df101234947fef1436014942786e1cb21a11 GIT binary patch literal 961 zcmWFz@bL_Z4>M$7U| z4iLa-5g>a75QFFgK@u!K?zO3sfmcSStVp2TPcP x9LogETAV3m`rve)1a&bi2f-2!GdTajL}7kpfvScD0WU~`*&ayPH2^V?1^@wfDRck; literal 0 HcmV?d00001 From 06eb293cfd2a52af7fbb369a1ad26ceaebf5ae21 Mon Sep 17 00:00:00 2001 From: "xuweixin.rex" Date: Fri, 10 Jul 2026 15:22:15 +0800 Subject: [PATCH 6/6] Resolve chain split schemas by file branch --- .../core/io/chain_split_file_path_factory.cpp | 28 +++- .../core/io/chain_split_file_path_factory.h | 8 +- .../core/operation/abstract_split_read.cpp | 58 ++++++-- .../core/operation/abstract_split_read.h | 8 ++ .../operation/abstract_split_read_test.cpp | 130 ++++++++++++++++++ .../core/table/source/chain_split_test.cpp | 19 ++- 6 files changed, 238 insertions(+), 13 deletions(-) diff --git a/src/paimon/core/io/chain_split_file_path_factory.cpp b/src/paimon/core/io/chain_split_file_path_factory.cpp index 08c94dc1f..061e8fe04 100644 --- a/src/paimon/core/io/chain_split_file_path_factory.cpp +++ b/src/paimon/core/io/chain_split_file_path_factory.cpp @@ -27,8 +27,13 @@ namespace paimon { Result> ChainSplitFilePathFactory::Create( const std::vector>& data_files, - std::unordered_map file_bucket_path_mapping) { + std::unordered_map file_bucket_path_mapping, + std::unordered_map file_branch_mapping) { for (const auto& file : data_files) { + if (file_branch_mapping.find(file->file_name) == file_branch_mapping.end()) { + return Status::Invalid( + fmt::format("branch is missing for ChainSplit file {}", file->file_name)); + } if (file->external_path) { continue; } @@ -37,12 +42,20 @@ Result> ChainSplitFilePathFactory::Cr fmt::format("bucket path is missing for ChainSplit file {}", file->file_name)); } } - return std::make_shared(std::move(file_bucket_path_mapping)); + return std::make_shared(std::move(file_bucket_path_mapping), + std::move(file_branch_mapping)); } ChainSplitFilePathFactory::ChainSplitFilePathFactory( std::unordered_map file_bucket_path_mapping) - : file_bucket_path_mapping_(std::move(file_bucket_path_mapping)) {} + : ChainSplitFilePathFactory(std::move(file_bucket_path_mapping), + std::unordered_map()) {} + +ChainSplitFilePathFactory::ChainSplitFilePathFactory( + std::unordered_map file_bucket_path_mapping, + std::unordered_map file_branch_mapping) + : file_bucket_path_mapping_(std::move(file_bucket_path_mapping)), + file_branch_mapping_(std::move(file_branch_mapping)) {} std::string ChainSplitFilePathFactory::ToPath( const std::shared_ptr& file_meta) const { @@ -63,4 +76,13 @@ std::string ChainSplitFilePathFactory::ToAlignedPath( return PathUtil::JoinPath(file_bucket_path_mapping_.at(aligned->file_name), file_name); } +std::optional ChainSplitFilePathFactory::BranchForFile( + const std::string& file_name) const { + auto it = file_branch_mapping_.find(file_name); + if (it == file_branch_mapping_.end()) { + return std::nullopt; + } + return it->second; +} + } // namespace paimon diff --git a/src/paimon/core/io/chain_split_file_path_factory.h b/src/paimon/core/io/chain_split_file_path_factory.h index 70e2cd137..1a977c5eb 100644 --- a/src/paimon/core/io/chain_split_file_path_factory.h +++ b/src/paimon/core/io/chain_split_file_path_factory.h @@ -17,6 +17,7 @@ #pragma once #include +#include #include #include #include @@ -30,17 +31,22 @@ class ChainSplitFilePathFactory : public DataFilePathFactory { public: static Result> Create( const std::vector>& data_files, - std::unordered_map file_bucket_path_mapping); + std::unordered_map file_bucket_path_mapping, + std::unordered_map file_branch_mapping); explicit ChainSplitFilePathFactory( std::unordered_map file_bucket_path_mapping); + ChainSplitFilePathFactory(std::unordered_map file_bucket_path_mapping, + std::unordered_map file_branch_mapping); std::string ToPath(const std::shared_ptr& file_meta) const override; std::string ToAlignedPath(const std::string& file_name, const std::shared_ptr& aligned) const override; + std::optional BranchForFile(const std::string& file_name) const; private: std::unordered_map file_bucket_path_mapping_; + std::unordered_map file_branch_mapping_; }; } // namespace paimon diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp index cd767ecca..ada3cdab3 100644 --- a/src/paimon/core/operation/abstract_split_read.cpp +++ b/src/paimon/core/operation/abstract_split_read.cpp @@ -33,6 +33,7 @@ #include "paimon/common/table/special_fields.h" #include "paimon/common/types/data_field.h" #include "paimon/common/utils/object_utils.h" +#include "paimon/common/utils/string_utils.h" #include "paimon/core/io/chain_split_file_path_factory.h" #include "paimon/core/io/complete_row_tracking_fields_reader.h" #include "paimon/core/io/data_file_meta.h" @@ -43,6 +44,7 @@ #include "paimon/core/schema/table_schema.h" #include "paimon/core/table/source/chain_split_impl.h" #include "paimon/core/table/source/data_split_impl.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/nested_projection_utils.h" #include "paimon/format/file_format.h" @@ -123,7 +125,8 @@ Result> AbstractSplitRead::CreateDataFilePa auto chain_split = std::dynamic_pointer_cast(data_split); if (chain_split) { return ChainSplitFilePathFactory::Create(chain_split->DataFiles(), - chain_split->FileBucketPathMapping()); + chain_split->FileBucketPathMapping(), + chain_split->FileBranchMapping()); } PAIMON_ASSIGN_OR_RAISE( @@ -132,6 +135,50 @@ Result> AbstractSplitRead::CreateDataFilePa return base_factory; } +Result AbstractSplitRead::GetSchemaManagerForBranch( + const std::string& branch) const { + std::string normalized_branch = BranchManager::NormalizeBranch(branch); + std::string current_branch = BranchManager::NormalizeBranch(options_.GetBranch()); + if (StringUtils::ToLowerCase(normalized_branch) == StringUtils::ToLowerCase(current_branch)) { + return schema_manager_.get(); + } + + auto it = branch_schema_managers_.find(normalized_branch); + if (it != branch_schema_managers_.end()) { + return it->second.get(); + } + + auto schema_manager = std::make_unique(options_.GetFileSystem(), + context_->GetPath(), normalized_branch); + auto [inserted_it, _] = + branch_schema_managers_.emplace(normalized_branch, std::move(schema_manager)); + return inserted_it->second.get(); +} + +Result> AbstractSplitRead::ReadDataSchema( + const std::shared_ptr& file_meta, + const std::shared_ptr& data_file_path_factory) const { + const SchemaManager* schema_manager = schema_manager_.get(); + bool current_branch = true; + + auto chain_path_factory = + std::dynamic_pointer_cast(data_file_path_factory); + if (chain_path_factory) { + std::optional branch = chain_path_factory->BranchForFile(file_meta->file_name); + if (!branch) { + return Status::Invalid( + fmt::format("branch is missing for ChainSplit file {}", file_meta->file_name)); + } + PAIMON_ASSIGN_OR_RAISE(schema_manager, GetSchemaManagerForBranch(branch.value())); + current_branch = schema_manager == schema_manager_.get(); + } + + if (current_branch && file_meta->schema_id == context_->GetTableSchema()->Id()) { + return context_->GetTableSchema(); + } + return schema_manager->ReadSchema(file_meta->schema_id); +} + Result> AbstractSplitRead::PrepareReaderBuilder( const std::string& format_identifier) const { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr file_format, @@ -175,13 +222,8 @@ Result> AbstractSplitRead::CreateFieldMappingRe const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory, const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory) const { - std::shared_ptr data_schema; - if (file_meta->schema_id == context_->GetTableSchema()->Id()) { - data_schema = context_->GetTableSchema(); - } else { - // load schema to get data schema - PAIMON_ASSIGN_OR_RAISE(data_schema, schema_manager_->ReadSchema(file_meta->schema_id)); - } + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr data_schema, + ReadDataSchema(file_meta, data_file_path_factory)); PAIMON_ASSIGN_OR_RAISE(CoreOptions data_options, CoreOptions::FromMap(data_schema->Options(), options_.GetFileSystem())); auto blob_inline_fields = data_options.GetBlobInlineFields(); diff --git a/src/paimon/core/operation/abstract_split_read.h b/src/paimon/core/operation/abstract_split_read.h index 22a481041..cd223e285 100644 --- a/src/paimon/core/operation/abstract_split_read.h +++ b/src/paimon/core/operation/abstract_split_read.h @@ -16,6 +16,7 @@ #pragma once +#include #include #include #include @@ -112,6 +113,12 @@ class AbstractSplitRead : public SplitRead { const std::optional>& row_ranges, const std::shared_ptr& data_file_path_factory) const; + Result> ReadDataSchema( + const std::shared_ptr& file_meta, + const std::shared_ptr& data_file_path_factory) const; + + Result GetSchemaManagerForBranch(const std::string& branch) const; + Result, std::set>> ApplySharedShreddingReaderIfNeeded(std::unique_ptr&& file_reader, const std::shared_ptr& read_schema) const; @@ -128,6 +135,7 @@ class AbstractSplitRead : public SplitRead { std::shared_ptr raw_read_schema_; std::shared_ptr context_; std::unique_ptr schema_manager_; + mutable std::map> branch_schema_managers_; }; } // namespace paimon diff --git a/src/paimon/core/operation/abstract_split_read_test.cpp b/src/paimon/core/operation/abstract_split_read_test.cpp index f151d432d..b1f0f9bea 100644 --- a/src/paimon/core/operation/abstract_split_read_test.cpp +++ b/src/paimon/core/operation/abstract_split_read_test.cpp @@ -18,14 +18,90 @@ #include #include +#include +#include +#include +#include +#include +#include #include "arrow/type_fwd.h" #include "gtest/gtest.h" +#include "paimon/common/data/binary_row.h" #include "paimon/common/table/special_fields.h" #include "paimon/common/types/data_field.h" +#include "paimon/core/io/chain_split_file_path_factory.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/manifest/file_source.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/stats/simple_stats.h" +#include "paimon/data/timestamp.h" +#include "paimon/defs.h" +#include "paimon/fs/file_system.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/read_context.h" #include "paimon/testing/utils/testharness.h" namespace paimon::test { +namespace { + +Result> CreateTableForReadSchemaTest( + const std::shared_ptr& fs, const std::string& table_root, const std::string& branch, + const std::string& field_name) { + SchemaManager schema_manager(fs, table_root, branch); + auto schema = arrow::schema({arrow::field(field_name, arrow::int32())}); + std::map options = {{Options::FILE_FORMAT, "parquet"}}; + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr table_schema, + schema_manager.CreateTable(schema, /*partition_keys=*/{}, /*primary_keys=*/{}, options)); + return std::shared_ptr(std::move(table_schema)); +} + +std::shared_ptr CreateDataFileMetaForReadSchemaTest(const std::string& file_name, + int64_t schema_id) { + return std::make_shared( + file_name, /*file_size=*/1024, /*row_count=*/7, BinaryRow::EmptyRow(), + BinaryRow::EmptyRow(), SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), + /*min_sequence_number=*/0, /*max_sequence_number=*/6, schema_id, DataFileMeta::DUMMY_LEVEL, + std::vector>(), Timestamp(0, 0), + /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(), + /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt, + /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt); +} + +class AbstractSplitReadForTest : public AbstractSplitRead { + public: + AbstractSplitReadForTest(const std::shared_ptr& context, + std::unique_ptr&& schema_manager) + : AbstractSplitRead(/*path_factory=*/nullptr, context, std::move(schema_manager), + GetDefaultPool(), /*executor=*/nullptr) {} + + Result> CreateReader( + const std::shared_ptr& split) override { + (void)split; + return Status::NotImplemented("not used"); + } + + Result Match(const std::shared_ptr& split, bool force_keep_delete) const override { + (void)split; + (void)force_keep_delete; + return false; + } + + private: + Result> ApplyIndexAndDvReaderIfNeeded( + std::unique_ptr&& file_reader, const std::shared_ptr& file, + const std::shared_ptr& data_schema, + const std::shared_ptr& read_schema, + const std::shared_ptr& predicate, DeletionVector::Factory dv_factory, + const std::optional>& row_ranges, + const std::shared_ptr& data_file_path_factory) const override { + return std::move(file_reader); + } +}; + +} // namespace + TEST(AbstractSplitReadTest, TestNeedCompleteRowTrackingFields) { std::vector data_fields = {DataField(0, arrow::field("name", arrow::utf8())), DataField(1, arrow::field("sex", arrow::utf8())), @@ -156,4 +232,58 @@ TEST(AbstractSplitReadTest, TestProjectFieldsForRowTrackingAndDataEvolution) { } } +TEST(AbstractSplitReadTest, ReadDataSchemaUsesChainSplitFileBranchMapping) { + auto dir = UniqueTestDirectory::Create(); + auto fs = dir->GetFileSystem(); + const std::string table_root = dir->Str(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr main_schema, + CreateTableForReadSchemaTest(fs, table_root, "main", "main_field")); + ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::shared_ptr delta_schema, + CreateTableForReadSchemaTest(fs, table_root, "delta", "delta_field")); + + ReadContextBuilder context_builder(table_root); + context_builder.WithFileSystem(fs); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_context, context_builder.Finish()); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr internal_context, + InternalReadContext::Create(std::move(read_context), main_schema, main_schema->Options())); + AbstractSplitReadForTest split_read(internal_context, + std::make_unique(fs, table_root, "main")); + + const std::string file_name = "data-branch-aware.parquet"; + auto file_meta = CreateDataFileMetaForReadSchemaTest(file_name, /*schema_id=*/0); + auto chain_path_factory = std::make_shared( + std::unordered_map{{file_name, "bucket-0"}}, + std::unordered_map{{file_name, "delta"}}); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr data_schema, + split_read.ReadDataSchema(file_meta, chain_path_factory)); + EXPECT_EQ(data_schema->FieldNames(), std::vector({"delta_field"})); +} + +TEST(AbstractSplitReadTest, ReadDataSchemaRejectsChainSplitMissingBranchMapping) { + auto dir = UniqueTestDirectory::Create(); + auto fs = dir->GetFileSystem(); + const std::string table_root = dir->Str(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr main_schema, + CreateTableForReadSchemaTest(fs, table_root, "main", "main_field")); + + ReadContextBuilder context_builder(table_root); + context_builder.WithFileSystem(fs); + ASSERT_OK_AND_ASSIGN(std::unique_ptr read_context, context_builder.Finish()); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr internal_context, + InternalReadContext::Create(std::move(read_context), main_schema, main_schema->Options())); + AbstractSplitReadForTest split_read(internal_context, + std::make_unique(fs, table_root, "main")); + + const std::string file_name = "data-missing-branch.parquet"; + auto file_meta = CreateDataFileMetaForReadSchemaTest(file_name, /*schema_id=*/0); + auto chain_path_factory = std::make_shared( + std::unordered_map{{file_name, "bucket-0"}}); + + ASSERT_NOK_WITH_MSG(split_read.ReadDataSchema(file_meta, chain_path_factory), + "branch is missing for ChainSplit file"); +} + } // namespace paimon::test diff --git a/src/paimon/core/table/source/chain_split_test.cpp b/src/paimon/core/table/source/chain_split_test.cpp index 33d599d6c..82e8805c9 100644 --- a/src/paimon/core/table/source/chain_split_test.cpp +++ b/src/paimon/core/table/source/chain_split_test.cpp @@ -246,10 +246,27 @@ TEST(ChainSplitTest, ChainSplitFilePathFactoryPreservesExternalAlignedPath) { } TEST(ChainSplitTest, ChainSplitFilePathFactoryRejectsMissingBucketPathMapping) { - ASSERT_NOK_WITH_MSG(ChainSplitFilePathFactory::Create({CreateDataFileMeta(kFileName)}, {}), + ASSERT_NOK_WITH_MSG(ChainSplitFilePathFactory::Create({CreateDataFileMeta(kFileName)}, {}, + {{kFileName, kBranch}}), "bucket path is missing for ChainSplit file"); } +TEST(ChainSplitTest, ChainSplitFilePathFactoryRejectsMissingBranchMapping) { + ASSERT_NOK_WITH_MSG(ChainSplitFilePathFactory::Create({CreateDataFileMeta(kFileName)}, + {{kFileName, kBucketPath}}, {}), + "branch is missing for ChainSplit file"); +} + +TEST(ChainSplitTest, ChainSplitFilePathFactoryPreservesFileBranchMapping) { + ChainSplitFilePathFactory factory( + std::unordered_map{{kFileName, kBucketPath}}, + std::unordered_map{{kFileName, kBranch}}); + + ASSERT_TRUE(factory.BranchForFile(kFileName)); + EXPECT_EQ(factory.BranchForFile(kFileName).value(), kBranch); + EXPECT_EQ(factory.BranchForFile("missing"), std::nullopt); +} + TEST(ChainSplitTest, SerializeChainSplitPreservesMappings) { auto pool = GetDefaultPool();