feat: support to read chain data split

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_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
@@ -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_split_test.cpp
                     core/table/source/data_split_test.cpp
                     core/table/source/deletion_file_test.cpp
                     core/table/source/split_generator_test.cpp

src/paimon/core/io/chain_split_file_path_factory.cpp

@@ -0,0 +1,88 @@
+/*
+ * 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 <utility>
+
+#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<std::shared_ptr<ChainSplitFilePathFactory>> ChainSplitFilePathFactory::Create(
+    const std::vector<std::shared_ptr<DataFileMeta>>& data_files,
+    std::unordered_map<std::string, std::string> file_bucket_path_mapping,
+    std::unordered_map<std::string, std::string> 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;
+        }
+        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<ChainSplitFilePathFactory>(std::move(file_bucket_path_mapping),
+                                                       std::move(file_branch_mapping));
+}
+
+ChainSplitFilePathFactory::ChainSplitFilePathFactory(
+    std::unordered_map<std::string, std::string> file_bucket_path_mapping)
+    : ChainSplitFilePathFactory(std::move(file_bucket_path_mapping),
+                                std::unordered_map<std::string, std::string>()) {}
+
+ChainSplitFilePathFactory::ChainSplitFilePathFactory(
+    std::unordered_map<std::string, std::string> file_bucket_path_mapping,
+    std::unordered_map<std::string, std::string> 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<DataFileMeta>& 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<DataFileMeta>& 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);
+}
+
+std::optional<std::string> 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

src/paimon/core/io/chain_split_file_path_factory.h

@@ -0,0 +1,52 @@
+/*
+ * 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 <memory>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+class ChainSplitFilePathFactory : public DataFilePathFactory {
+ public:
+    static Result<std::shared_ptr<ChainSplitFilePathFactory>> Create(
+        const std::vector<std::shared_ptr<DataFileMeta>>& data_files,
+        std::unordered_map<std::string, std::string> file_bucket_path_mapping,
+        std::unordered_map<std::string, std::string> file_branch_mapping);
+
+    explicit ChainSplitFilePathFactory(
+        std::unordered_map<std::string, std::string> file_bucket_path_mapping);
+    ChainSplitFilePathFactory(std::unordered_map<std::string, std::string> file_bucket_path_mapping,
+                              std::unordered_map<std::string, std::string> file_branch_mapping);
+
+    std::string ToPath(const std::shared_ptr<DataFileMeta>& file_meta) const override;
+    std::string ToAlignedPath(const std::string& file_name,
+                              const std::shared_ptr<DataFileMeta>& aligned) const override;
+    std::optional<std::string> BranchForFile(const std::string& file_name) const;
+
+ private:
+    std::unordered_map<std::string, std::string> file_bucket_path_mapping_;
+    std::unordered_map<std::string, std::string> file_branch_mapping_;
+};
+
+}  // namespace paimon

src/paimon/core/io/data_file_path_factory.h

@@ -70,15 +70,15 @@ class DataFilePathFactory : public PathFactory {
     }
 
     std::string ToPath(const std::string& file_name) const override;
-    std::string ToPath(const std::shared_ptr<DataFileMeta>& file_meta) const;
+    virtual std::string ToPath(const std::shared_ptr<DataFileMeta>& file_meta) const;
 
     const std::string& GetUUID() const {
         return uuid_;
     }
 
     std::string ToFileIndexPath(const std::string& file_path) const;
-    std::string ToAlignedPath(const std::string& file_name,
-                              const std::shared_ptr<DataFileMeta>& aligned) const;
+    virtual std::string ToAlignedPath(const std::string& file_name,
+                                      const std::shared_ptr<DataFileMeta>& aligned) const;
 
     std::vector<std::string> CollectFiles(const std::shared_ptr<DataFileMeta>& file_meta) const;
     bool IsExternalPath() const {

src/paimon/core/operation/abstract_split_read.cpp

@@ -33,14 +33,18 @@
 #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"
 #include "paimon/core/io/data_file_path_factory.h"
 #include "paimon/core/io/field_mapping_reader.h"
 #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_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"
@@ -116,6 +120,65 @@ Result<std::unique_ptr<BatchReader>> AbstractSplitRead::ApplyPredicateFilterIfNe
     return PredicateBatchReader::Create(std::move(reader), predicate, pool_);
 }
 
+Result<std::shared_ptr<DataFilePathFactory>> AbstractSplitRead::CreateDataFilePathFactory(
+    const std::shared_ptr<DataSplitImpl>& data_split) const {
+    auto chain_split = std::dynamic_pointer_cast<ChainSplitImpl>(data_split);
+    if (chain_split) {
+        return ChainSplitFilePathFactory::Create(chain_split->DataFiles(),
+                                                 chain_split->FileBucketPathMapping(),
+                                                 chain_split->FileBranchMapping());
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<DataFilePathFactory> base_factory,
+        path_factory_->CreateDataFilePathFactory(data_split->Partition(), data_split->Bucket()));
+    return base_factory;
+}
+
+Result<const SchemaManager*> 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<SchemaManager>(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<std::shared_ptr<TableSchema>> AbstractSplitRead::ReadDataSchema(
+    const std::shared_ptr<DataFileMeta>& file_meta,
+    const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
+    const SchemaManager* schema_manager = schema_manager_.get();
+    bool current_branch = true;
+
+    auto chain_path_factory =
+        std::dynamic_pointer_cast<ChainSplitFilePathFactory>(data_file_path_factory);
+    if (chain_path_factory) {
+        std::optional<std::string> 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<std::unique_ptr<ReaderBuilder>> AbstractSplitRead::PrepareReaderBuilder(
     const std::string& format_identifier) const {
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> file_format,
@@ -159,13 +222,8 @@ Result<std::unique_ptr<FileBatchReader>> AbstractSplitRead::CreateFieldMappingRe
     const FieldMappingBuilder* field_mapping_builder, DeletionVector::Factory dv_factory,
     const std::optional<std::vector<Range>>& row_ranges,
     const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const {
-    std::shared_ptr<TableSchema> 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<TableSchema> 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();

src/paimon/core/operation/abstract_split_read.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include <map>
 #include <memory>
 #include <optional>
 #include <set>
@@ -76,6 +77,9 @@ class AbstractSplitRead : public SplitRead {
     Result<std::unique_ptr<BatchReader>> ApplyPredicateFilterIfNeeded(
         std::unique_ptr<BatchReader>&& reader, const std::shared_ptr<Predicate>& predicate) const;
 
+    Result<std::shared_ptr<DataFilePathFactory>> CreateDataFilePathFactory(
+        const std::shared_ptr<DataSplitImpl>& data_split) const;
+
  protected:
     // return nullptr if file is skipped by index or dv
     virtual Result<std::unique_ptr<FileBatchReader>> ApplyIndexAndDvReaderIfNeeded(
@@ -109,6 +113,12 @@ class AbstractSplitRead : public SplitRead {
         const std::optional<std::vector<Range>>& row_ranges,
         const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const;
 
+    Result<std::shared_ptr<TableSchema>> ReadDataSchema(
+        const std::shared_ptr<DataFileMeta>& file_meta,
+        const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const;
+
+    Result<const SchemaManager*> GetSchemaManagerForBranch(const std::string& branch) const;
+
     Result<std::pair<std::unique_ptr<FileBatchReader>, std::set<int32_t>>>
     ApplySharedShreddingReaderIfNeeded(std::unique_ptr<FileBatchReader>&& file_reader,
                                        const std::shared_ptr<arrow::Schema>& read_schema) const;
@@ -125,6 +135,7 @@ class AbstractSplitRead : public SplitRead {
     std::shared_ptr<arrow::Schema> raw_read_schema_;
     std::shared_ptr<InternalReadContext> context_;
     std::unique_ptr<SchemaManager> schema_manager_;
+    mutable std::map<std::string, std::unique_ptr<SchemaManager>> branch_schema_managers_;
 };
 
 }  // namespace paimon