diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index b044badea..cf42f3080 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_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
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..061e8fe04
--- /dev/null
+++ b/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
diff --git a/src/paimon/core/io/chain_split_file_path_factory.h b/src/paimon/core/io/chain_split_file_path_factory.h
new file mode 100644
index 000000000..1a977c5eb
--- /dev/null
+++ b/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
diff --git a/src/paimon/core/io/data_file_path_factory.h b/src/paimon/core/io/data_file_path_factory.h
index 110a7035e..4935e3bea 100644
--- a/src/paimon/core/io/data_file_path_factory.h
+++ b/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 {
diff --git a/src/paimon/core/operation/abstract_split_read.cpp b/src/paimon/core/operation/abstract_split_read.cpp
index 1e70c0554..ada3cdab3 100644
--- a/src/paimon/core/operation/abstract_split_read.cpp
+++ b/src/paimon/core/operation/abstract_split_read.cpp
@@ -33,6 +33,8 @@
 #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"
@@ -40,7 +42,9 @@
 #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();
diff --git a/src/paimon/core/operation/abstract_split_read.h b/src/paimon/core/operation/abstract_split_read.h
index 4d49d78bc..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 <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
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 <cstdint>
 #include <limits>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
 
 #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<std::shared_ptr<TableSchema>> CreateTableForReadSchemaTest(
+    const std::shared_ptr<FileSystem>& 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<std::string, std::string> options = {{Options::FILE_FORMAT, "parquet"}};
+    PAIMON_ASSIGN_OR_RAISE(
+        std::unique_ptr<TableSchema> table_schema,
+        schema_manager.CreateTable(schema, /*partition_keys=*/{}, /*primary_keys=*/{}, options));
+    return std::shared_ptr<TableSchema>(std::move(table_schema));
+}
+
+std::shared_ptr<DataFileMeta> CreateDataFileMetaForReadSchemaTest(const std::string& file_name,
+                                                                  int64_t schema_id) {
+    return std::make_shared<DataFileMeta>(
+        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<std::optional<std::string>>(), 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<InternalReadContext>& context,
+                             std::unique_ptr<SchemaManager>&& schema_manager)
+        : AbstractSplitRead(/*path_factory=*/nullptr, context, std::move(schema_manager),
+                            GetDefaultPool(), /*executor=*/nullptr) {}
+
+    Result<std::unique_ptr<BatchReader>> CreateReader(
+        const std::shared_ptr<Split>& split) override {
+        (void)split;
+        return Status::NotImplemented("not used");
+    }
+
+    Result<bool> Match(const std::shared_ptr<Split>& split, bool force_keep_delete) const override {
+        (void)split;
+        (void)force_keep_delete;
+        return false;
+    }
+
+ private:
+    Result<std::unique_ptr<FileBatchReader>> ApplyIndexAndDvReaderIfNeeded(
+        std::unique_ptr<FileBatchReader>&& file_reader, const std::shared_ptr<DataFileMeta>& file,
+        const std::shared_ptr<arrow::Schema>& data_schema,
+        const std::shared_ptr<arrow::Schema>& read_schema,
+        const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory dv_factory,
+        const std::optional<std::vector<Range>>& row_ranges,
+        const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const override {
+        return std::move(file_reader);
+    }
+};
+
+}  // namespace
+
 TEST(AbstractSplitReadTest, TestNeedCompleteRowTrackingFields) {
     std::vector<DataField> 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<TableSchema> main_schema,
+                         CreateTableForReadSchemaTest(fs, table_root, "main", "main_field"));
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::shared_ptr<TableSchema> 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<ReadContext> read_context, context_builder.Finish());
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<InternalReadContext> internal_context,
+        InternalReadContext::Create(std::move(read_context), main_schema, main_schema->Options()));
+    AbstractSplitReadForTest split_read(internal_context,
+                                        std::make_unique<SchemaManager>(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<ChainSplitFilePathFactory>(
+        std::unordered_map<std::string, std::string>{{file_name, "bucket-0"}},
+        std::unordered_map<std::string, std::string>{{file_name, "delta"}});
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> data_schema,
+                         split_read.ReadDataSchema(file_meta, chain_path_factory));
+    EXPECT_EQ(data_schema->FieldNames(), std::vector<std::string>({"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<TableSchema> 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<ReadContext> read_context, context_builder.Finish());
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<InternalReadContext> internal_context,
+        InternalReadContext::Create(std::move(read_context), main_schema, main_schema->Options()));
+    AbstractSplitReadForTest split_read(internal_context,
+                                        std::make_unique<SchemaManager>(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<ChainSplitFilePathFactory>(
+        std::unordered_map<std::string, std::string>{{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/operation/data_evolution_split_read.cpp b/src/paimon/core/operation/data_evolution_split_read.cpp
index fab21ac01..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<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::WrapWithBlobViewRes
 Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateBlobViewReader(
     const std::shared_ptr<DataSplit>& data_split,
     const std::vector<std::string>& read_blob_view_fields) const {
-    auto split_impl = dynamic_cast<DataSplitImpl*>(data_split.get());
-    if (split_impl == nullptr) {
+    auto split_impl = std::dynamic_pointer_cast<DataSplitImpl>(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<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::CreateBlobViewReade
     }
     auto blob_view_schema = arrow::schema(std::move(blob_view_arrow_fields));
 
-    PAIMON_ASSIGN_OR_RAISE(
-        std::shared_ptr<DataFilePathFactory> data_file_path_factory,
-        path_factory_->CreateDataFilePathFactory(split_impl->Partition(), split_impl->Bucket()));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory> data_file_path_factory,
+                           CreateDataFilePathFactory(split_impl));
 
     // skip blob files: they only contain blob payloads, not the blob-view columns.
     std::vector<std::shared_ptr<DataFileMeta>> data_files;
@@ -289,14 +288,13 @@ Result<std::unordered_set<BlobViewStruct>> DataEvolutionSplitRead::ExtractBlobVi
 Result<std::unique_ptr<BatchReader>> DataEvolutionSplitRead::InnerCreateReader(
     const std::shared_ptr<DataSplit>& data_split,
     const std::optional<std::vector<Range>>& row_ranges) const {
-    auto split_impl = dynamic_cast<DataSplitImpl*>(data_split.get());
-    if (split_impl == nullptr) {
+    auto split_impl = std::dynamic_pointer_cast<DataSplitImpl>(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<DataFilePathFactory> data_file_path_factory,
-        path_factory_->CreateDataFilePathFactory(split_impl->Partition(), split_impl->Bucket()));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory> data_file_path_factory,
+                           CreateDataFilePathFactory(split_impl));
     auto metas = split_impl->DataFiles();
     PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>> 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<std::unique_ptr<BatchReader>> 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<DataFilePathFactory> data_file_path_factory,
-        path_factory_->CreateDataFilePathFactory(data_split->Partition(), data_split->Bucket()));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory> data_file_path_factory,
+                           CreateDataFilePathFactory(data_split));
     std::unique_ptr<BatchReader> 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..2b01caf43 100644
--- a/src/paimon/core/operation/raw_file_split_read.cpp
+++ b/src/paimon/core/operation/raw_file_split_read.cpp
@@ -66,17 +66,31 @@ Result<std::unique_ptr<BatchReader>> 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<DataFilePathFactory> 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<std::unique_ptr<BatchReader>> RawFileSplitRead::CreateReader(
     const BinaryRow& partition, int32_t bucket,
     const std::vector<std::shared_ptr<DataFileMeta>>& data_files,
     DeletionVector::Factory dv_factory) {
-    const auto& predicate = context_->GetPredicate();
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFilePathFactory> data_file_path_factory,
                            path_factory_->CreateDataFilePathFactory(partition, bucket));
+    return CreateReader(partition, bucket, data_files, dv_factory, data_file_path_factory);
+}
+
+Result<std::unique_ptr<BatchReader>> RawFileSplitRead::CreateReader(
+    const BinaryRow& partition, int32_t bucket,
+    const std::vector<std::shared_ptr<DataFileMeta>>& data_files,
+    DeletionVector::Factory dv_factory,
+    std::shared_ptr<DataFilePathFactory> data_file_path_factory) {
+    const auto& predicate = context_->GetPredicate();
 
     PAIMON_ASSIGN_OR_RAISE(
         std::vector<std::unique_ptr<FileBatchReader>> 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..04a71c2a2 100644
--- a/src/paimon/core/operation/raw_file_split_read.h
+++ b/src/paimon/core/operation/raw_file_split_read.h
@@ -82,6 +82,12 @@ class RawFileSplitRead : public AbstractSplitRead {
         const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory dv_factory,
         const std::optional<std::vector<Range>>& ranges,
         const std::shared_ptr<DataFilePathFactory>& data_file_path_factory) const override;
+
+ private:
+    Result<std::unique_ptr<BatchReader>> CreateReader(
+        const BinaryRow& partition, int32_t bucket,
+        const std::vector<std::shared_ptr<DataFileMeta>>& files, DeletionVector::Factory dv_factory,
+        std::shared_ptr<DataFilePathFactory> data_file_path_factory);
 };
 
 }  // namespace paimon
diff --git a/src/paimon/core/table/source/chain_split_impl.h b/src/paimon/core/table/source/chain_split_impl.h
new file mode 100644
index 000000000..0305f305a
--- /dev/null
+++ b/src/paimon/core/table/source/chain_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 <memory>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/table/source/data_split_impl.h"
+
+namespace paimon {
+
+class ChainSplitImpl : public DataSplitImpl {
+ public:
+    static constexpr const char* VIRTUAL_BUCKET_PATH = "placeholder::virtual-bucket-path";
+
+    ChainSplitImpl(const std::shared_ptr<DataSplitImpl>& base_split, bool all_snapshot_split,
+                   const BinaryRow& read_partition,
+                   std::unordered_map<std::string, std::string>&& file_bucket_path_mapping,
+                   std::unordered_map<std::string, std::string>&& file_branch_mapping)
+        : DataSplitImpl(base_split->Partition(), base_split->Bucket(), base_split->BucketPath(),
+                        std::vector<std::shared_ptr<DataFileMeta>>(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<std::string, std::string>& FileBucketPathMapping() const {
+        return file_bucket_path_mapping_;
+    }
+
+    const std::unordered_map<std::string, std::string>& FileBranchMapping() const {
+        return file_branch_mapping_;
+    }
+
+ private:
+    bool all_snapshot_split_;
+    BinaryRow read_partition_;
+    std::unordered_map<std::string, std::string> file_bucket_path_mapping_;
+    std::unordered_map<std::string, std::string> file_branch_mapping_;
+};
+
+}  // namespace paimon
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..82e8805c9
--- /dev/null
+++ b/src/paimon/core/table/source/chain_split_test.cpp
@@ -0,0 +1,300 @@
+/*
+ * 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 <memory>
+#include <optional>
+#include <string>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#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<std::string> ReadCompatibilityBytes(const std::string& file_name) {
+    auto file_system = std::make_unique<LocalFileSystem>();
+    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<DataFileMeta> CreateDataFileMeta(
+    const std::string& file_name, std::optional<std::string> external_path = std::nullopt) {
+    return std::make_shared<DataFileMeta>(
+        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<std::optional<std::string>>(), 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<std::string, std::string>& values) {
+    out->WriteValue<bool>(true);
+    out->WriteValue<int32_t>(values.size());
+    for (const auto& [key, value] : values) {
+        out->WriteValue<int32_t>(static_cast<int32_t>(key.size()));
+        out->Write(key.data(), static_cast<uint32_t>(key.size()));
+        out->WriteValue<int32_t>(static_cast<int32_t>(value.size()));
+        out->Write(value.data(), static_cast<uint32_t>(value.size()));
+    }
+}
+
+std::string ToString(const MemorySegmentOutputStream& out,
+                     const std::shared_ptr<MemoryPool>& pool) {
+    PAIMON_UNIQUE_PTR<Bytes> 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<std::shared_ptr<DataFileMeta>>& data_files,
+                                const std::unordered_map<std::string, std::string>& bucket_paths,
+                                const std::unordered_map<std::string, std::string>& branches,
+                                const std::shared_ptr<MemoryPool>& pool) {
+    MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
+    out.WriteValue<int64_t>(kSplitSerializerMagic);
+    out.WriteValue<int32_t>(kSplitSerializerVersion);
+    out.WriteValue<int32_t>(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,
+                         Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool));
+    auto chain_split = std::dynamic_pointer_cast<ChainSplitImpl>(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,
+                         Split::Deserialize(chain_bytes.data(), chain_bytes.size(), pool));
+    auto chain_split = std::dynamic_pointer_cast<ChainSplitImpl>(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,
+                         Split::Deserialize(indexed_bytes.data(), indexed_bytes.size(), pool));
+    auto indexed_split = std::dynamic_pointer_cast<IndexedSplitImpl>(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,
+                         Split::Deserialize(fallback_bytes.data(), fallback_bytes.size(), pool));
+    auto fallback_split = std::dynamic_pointer_cast<FallbackDataSplit>(split);
+    ASSERT_TRUE(fallback_split);
+    EXPECT_TRUE(fallback_split->IsFallback());
+    ASSERT_TRUE(std::dynamic_pointer_cast<DataSplitImpl>(fallback_split->GetSplit()));
+
+    ASSERT_OK_AND_ASSIGN(std::string serialized, Split::Serialize(fallback_split, pool));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Split> round_trip,
+                         Split::Deserialize(serialized.data(), serialized.size(), pool));
+    auto round_trip_fallback = std::dynamic_pointer_cast<FallbackDataSplit>(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,
+                         Split::Deserialize(split_bytes.data(), split_bytes.size(), pool));
+    auto chain_split = std::dynamic_pointer_cast<ChainSplitImpl>(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<std::string, std::string>{{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<std::string, std::string>{{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<std::string, std::string>{{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<std::string, std::string>{{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)}, {},
+                                                          {{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<std::string, std::string>{{kFileName, kBucketPath}},
+        std::unordered_map<std::string, std::string>{{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();
+
+    std::string chain_bytes =
+        SerializeChainSplit(BinaryRow::EmptyRow(), {CreateDataFileMeta(kFileName)},
+                            {{kFileName, kBucketPath}}, {{kFileName, kBranch}}, pool);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Split> 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<Split> round_trip,
+                         Split::Deserialize(serialized.data(), serialized.size(), pool));
+
+    auto chain_split = std::dynamic_pointer_cast<ChainSplitImpl>(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<int64_t>(kSplitSerializerMagic);
+    out.WriteValue<int32_t>(kSplitSerializerVersion);
+    out.WriteValue<int32_t>(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/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<std::shared_ptr<DataFileMeta>>&& 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<std::optional<int64_t>> RawMergedRowCount(DeletionVector::Factory dv_factory) const;
     bool DataEvolutionRowCountAvailable() const;
     Result<int64_t> 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..fd7d1f14e 100644
--- a/src/paimon/core/table/source/split.cpp
+++ b/src/paimon/core/table/source/split.cpp
@@ -14,7 +14,12 @@
  * limitations under the License.
  */
 
+#include <map>
+#include <optional>
+#include <string>
+#include <unordered_map>
 #include <utility>
+#include <vector>
 
 #include "fmt/format.h"
 #include "paimon/common/data/binary_row.h"
@@ -22,7 +27,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_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 +44,72 @@
 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<std::vector<std::shared_ptr<DataFileMeta>>> ReadVersion7DataFileMetaList(
+    DataInputStream* in, const std::shared_ptr<MemoryPool>& pool) {
+    PAIMON_ASSIGN_OR_RAISE(int32_t size, in->ReadValue<int32_t>());
+    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<std::shared_ptr<DataFileMeta>> result;
+    result.reserve(size);
+    for (int32_t i = 0; i < size; ++i) {
+        PAIMON_ASSIGN_OR_RAISE(int32_t row_size, in->ReadValue<int32_t>());
+        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 = 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<int64_t>(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<int32_t>(file_name_offset_and_size >> 32);
+
+        int32_t arity = 0;
+        const ObjectSerializer<std::shared_ptr<DataFileMeta>>* 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 = &current_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<DataFileMeta> meta, serializer->FromRow(row));
+        result.emplace_back(std::move(meta));
+    }
+    return result;
+}
+
+Result<std::vector<std::shared_ptr<DataFileMeta>>> ReadDataFileMetaList(
+    int32_t version, const ObjectSerializer<std::shared_ptr<DataFileMeta>>* data_file_serializer,
+    DataInputStream* in, const std::shared_ptr<MemoryPool>& pool) {
+    if (version == 7) {
+        return ReadVersion7DataFileMetaList(in, pool);
+    }
+    return data_file_serializer->DeserializeList(in);
+}
+
 Status WriteDataSplit(const std::shared_ptr<DataSplitImpl>& data_split_impl,
                       MemorySegmentOutputStream* out, const std::shared_ptr<MemoryPool>& pool) {
     out->WriteValue<int64_t>(DataSplitImpl::MAGIC);
@@ -98,13 +171,15 @@ Result<std::shared_ptr<DataSplitImpl>> ReadDataSplitWithoutMagicNumber(
         std::unique_ptr<ObjectSerializer<std::shared_ptr<DataFileMeta>>> data_file_serializer,
         DataSplitImpl::GetFileMetaSerializer(version, pool));
     std::vector<std::shared_ptr<DataFileMeta>> 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<std::optional<DeletionFile>> before_deletion_files;
     PAIMON_ASSIGN_OR_RAISE(before_deletion_files, DeletionFile::DeserializeList(in, version));
 
     std::vector<std::shared_ptr<DataFileMeta>> 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<std::optional<DeletionFile>> data_deletion_files;
     PAIMON_ASSIGN_OR_RAISE(data_deletion_files, DeletionFile::DeserializeList(in, version));
@@ -129,16 +204,223 @@ Result<std::shared_ptr<DataSplitImpl>> ReadDataSplitWithoutMagicNumber(
     return builder.Build();
 }
 
+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<std::optional<std::string>> ReadSplitSerializerString(DataInputStream* in) {
+    PAIMON_ASSIGN_OR_RAISE(int32_t length, in->ReadValue<int32_t>());
+    if (length < 0) {
+        return std::optional<std::string>();
+    }
+    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::string>(std::move(value));
+}
+
+Result<std::unordered_map<std::string, std::string>> ReadSplitSerializerStringMap(
+    DataInputStream* in) {
+    PAIMON_ASSIGN_OR_RAISE(bool exists, in->ReadValue<bool>());
+    if (!exists) {
+        return std::unordered_map<std::string, std::string>();
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(int32_t size, in->ReadValue<int32_t>());
+    if (size < 0) {
+        return Status::Invalid(fmt::format("invalid split serializer string map size: {}", size));
+    }
+
+    std::unordered_map<std::string, std::string> result;
+    result.reserve(size);
+    for (int32_t i = 0; i < size; ++i) {
+        PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> key, ReadSplitSerializerString(in));
+        PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> 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;
+}
+
+void WriteSplitSerializerString(MemorySegmentOutputStream* out, const std::string& value) {
+    out->WriteValue<int32_t>(static_cast<int32_t>(value.size()));
+    out->Write(value.data(), static_cast<uint32_t>(value.size()));
+}
+
+void WriteSplitSerializerStringMap(MemorySegmentOutputStream* out,
+                                   const std::unordered_map<std::string, std::string>& values) {
+    out->WriteValue<bool>(true);
+    out->WriteValue<int32_t>(static_cast<int32_t>(values.size()));
+    for (const auto& [key, value] :
+         std::map<std::string, std::string>(values.begin(), values.end())) {
+        WriteSplitSerializerString(out, key);
+        WriteSplitSerializerString(out, value);
+    }
+}
+
+Status WriteChainSplitPayload(const std::shared_ptr<ChainSplitImpl>& chain_split,
+                              MemorySegmentOutputStream* out,
+                              const std::shared_ptr<MemoryPool>& 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<std::shared_ptr<ChainSplitImpl>> ReadChainSplitPayload(
+    DataInputStream* in, const std::shared_ptr<MemoryPool>& pool) {
+    PAIMON_ASSIGN_OR_RAISE(BinaryRow logical_partition,
+                           SerializationUtils::DeserializeBinaryRow(in, pool.get()));
+
+    DataFileMetaSerializer serializer(pool);
+    PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> 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<DataSplitImpl> base_split, builder.Build());
+    return std::make_shared<ChainSplitImpl>(base_split, /*all_snapshot_split=*/false,
+                                            logical_partition, std::move(file_bucket_path_mapping),
+                                            std::move(file_branch_mapping));
+}
+
+Result<std::shared_ptr<IndexedSplitImpl>> ReadIndexedSplitPayload(
+    int64_t magic, DataInputStream* in, const std::shared_ptr<MemoryPool>& 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<int32_t>());
+    if (version != IndexedSplitImpl::VERSION) {
+        return Status::Invalid(fmt::format("Unsupported IndexedSplit version: {}", version));
+    }
+    PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in->ReadValue<int64_t>());
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataSplitImpl> data_split,
+                           ReadDataSplitWithoutMagicNumber(data_split_magic, in, pool));
+    PAIMON_ASSIGN_OR_RAISE(int32_t range_size, in->ReadValue<int32_t>());
+    if (range_size < 0) {
+        return Status::Invalid(fmt::format("invalid IndexedSplit range size: {}", range_size));
+    }
+    std::vector<Range> 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<int64_t>());
+        PAIMON_ASSIGN_OR_RAISE(int64_t range_to, in->ReadValue<int64_t>());
+        row_ranges.emplace_back(range_from, range_to);
+    }
+
+    std::vector<float> scores;
+    PAIMON_ASSIGN_OR_RAISE(bool has_scores, in->ReadValue<bool>());
+    if (has_scores) {
+        PAIMON_ASSIGN_OR_RAISE(int32_t scores_length, in->ReadValue<int32_t>());
+        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<float>());
+            scores[i] = score;
+        }
+    }
+
+    return std::make_shared<IndexedSplitImpl>(data_split, row_ranges, scores);
+}
+
+Result<std::shared_ptr<Split>> ReadSplitSerializerPayload(DataInputStream* in,
+                                                          const std::shared_ptr<MemoryPool>& pool) {
+    PAIMON_ASSIGN_OR_RAISE(int32_t version, in->ReadValue<int32_t>());
+    if (version != kSplitSerializerVersion) {
+        return Status::Invalid(fmt::format("unsupported split serializer version: {}", version));
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(int32_t type, in->ReadValue<int32_t>());
+    switch (type) {
+        case kSplitSerializerDataSplit: {
+            PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in->ReadValue<int64_t>());
+            PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataSplitImpl> data_split,
+                                   ReadDataSplitWithoutMagicNumber(data_split_magic, in, pool));
+            PAIMON_RETURN_NOT_OK(ValidateFullyRead("DataSplit", in));
+            return std::static_pointer_cast<Split>(data_split);
+        }
+        case kSplitSerializerIndexedSplit: {
+            PAIMON_ASSIGN_OR_RAISE(int64_t indexed_split_magic, in->ReadValue<int64_t>());
+            PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<IndexedSplitImpl> indexed_split,
+                                   ReadIndexedSplitPayload(indexed_split_magic, in, pool));
+            PAIMON_RETURN_NOT_OK(ValidateFullyRead("IndexedSplit", in));
+            return std::static_pointer_cast<Split>(indexed_split);
+        }
+        case kSplitSerializerChainSplit: {
+            PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<ChainSplitImpl> chain_split,
+                                   ReadChainSplitPayload(in, pool));
+            return std::static_pointer_cast<Split>(chain_split);
+        }
+        case kSplitSerializerFallbackDataSplit: {
+            PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in->ReadValue<int64_t>());
+            PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataSplitImpl> data_split,
+                                   ReadDataSplitWithoutMagicNumber(data_split_magic, in, pool));
+            PAIMON_ASSIGN_OR_RAISE(bool is_fallback, in->ReadValue<bool>());
+            PAIMON_RETURN_NOT_OK(ValidateFullyRead("FallbackDataSplit", in));
+            return std::static_pointer_cast<Split>(
+                std::make_shared<FallbackDataSplit>(data_split, is_fallback));
+        }
+        default:
+            return Status::Invalid(fmt::format("unsupported split serializer type: {}", type));
+    }
+}
+
 }  // namespace
 
 Result<std::string> Split::Serialize(const std::shared_ptr<Split>& split,
                                      const std::shared_ptr<MemoryPool>& pool) {
     MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
-    if (auto data_split_impl = std::dynamic_pointer_cast<DataSplitImpl>(split)) {
+    if (auto fallback_data_split = std::dynamic_pointer_cast<FallbackDataSplit>(split)) {
+        if (std::dynamic_pointer_cast<ChainSplitImpl>(fallback_data_split->GetSplit())) {
+            return Status::Invalid("FallbackDataSplit cannot serialize a wrapped ChainSplit");
+        }
+        auto data_split_impl =
+            std::dynamic_pointer_cast<DataSplitImpl>(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<bool>(fallback_data_split->IsFallback());
+    } else if (auto chain_split_impl = std::dynamic_pointer_cast<ChainSplitImpl>(split)) {
+        out.WriteValue<int64_t>(kSplitSerializerMagic);
+        out.WriteValue<int32_t>(kSplitSerializerVersion);
+        out.WriteValue<int32_t>(kSplitSerializerChainSplit);
+        PAIMON_RETURN_NOT_OK(WriteChainSplitPayload(chain_split_impl, &out, pool));
+    } else if (auto data_split_impl = std::dynamic_pointer_cast<DataSplitImpl>(split)) {
         PAIMON_RETURN_NOT_OK(WriteDataSplit(data_split_impl, &out, pool));
     } else if (auto indexed_split_impl = std::dynamic_pointer_cast<IndexedSplitImpl>(split)) {
         out.WriteValue<int64_t>(IndexedSplitImpl::MAGIC);
         out.WriteValue<int32_t>(IndexedSplitImpl::VERSION);
+        if (std::dynamic_pointer_cast<ChainSplitImpl>(indexed_split_impl->GetDataSplit())) {
+            return Status::Invalid("IndexedSplit cannot serialize a wrapped ChainSplit");
+        }
         auto inner_split_impl =
             std::dynamic_pointer_cast<DataSplitImpl>(indexed_split_impl->GetDataSplit());
         if (!inner_split_impl) {
@@ -163,7 +445,9 @@ Result<std::string> Split::Serialize(const std::shared_ptr<Split>& split,
             out.WriteValue<bool>(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> bytes =
         MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get());
@@ -178,37 +462,16 @@ Result<std::shared_ptr<Split>> Split::Deserialize(const char* buffer, size_t len
     int64_t magic = -1;
     PAIMON_ASSIGN_OR_RAISE(magic, in.ReadValue<int64_t>());
 
-    if (magic == IndexedSplitImpl::MAGIC) {
-        PAIMON_ASSIGN_OR_RAISE(int32_t version, in.ReadValue<int32_t>());
-        if (version != IndexedSplitImpl::VERSION) {
-            return Status::Invalid(fmt::format("Unsupported IndexedSplit version: {}", version));
-        }
-        PAIMON_ASSIGN_OR_RAISE(int64_t data_split_magic, in.ReadValue<int64_t>());
-        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataSplitImpl> data_split,
-                               ReadDataSplitWithoutMagicNumber(data_split_magic, &in, pool));
-        PAIMON_ASSIGN_OR_RAISE(int32_t range_size, in.ReadValue<int32_t>());
-        std::vector<Range> 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<int64_t>());
-            PAIMON_ASSIGN_OR_RAISE(int64_t range_to, in.ReadValue<int64_t>());
-            row_ranges.emplace_back(range_from, range_to);
-        }
-        std::vector<float> scores;
-        PAIMON_ASSIGN_OR_RAISE(bool has_scores, in.ReadValue<bool>());
-        if (has_scores) {
-            PAIMON_ASSIGN_OR_RAISE(int32_t scores_length, in.ReadValue<int32_t>());
-            scores.resize(scores_length);
-            for (int32_t i = 0; i < scores_length; ++i) {
-                PAIMON_ASSIGN_OR_RAISE(float score, in.ReadValue<float>());
-                scores[i] = score;
-            }
-        }
+    if (magic == kSplitSerializerMagic) {
+        return ReadSplitSerializerPayload(&in, pool);
+    } else if (magic == IndexedSplitImpl::MAGIC) {
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<IndexedSplitImpl> 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<IndexedSplitImpl>(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");
@@ -220,19 +483,18 @@ Result<std::shared_ptr<Split>> Split::Deserialize(const char* buffer, size_t len
     } else if (magic == DataSplitImpl::MAGIC) {
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataSplitImpl> 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 (pos == stream_length - 1) {
+        if (pos == stream_length - 1) {
             PAIMON_ASSIGN_OR_RAISE(bool is_fallback, in.ReadValue<bool>());
             return std::make_shared<FallbackDataSplit>(data_split, is_fallback);
-        } else {
-            return Status::Invalid(fmt::format(
-                "invalid data split byte stream, remaining {} bytes after deserializing",
-                stream_length - pos));
         }
+        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 000000000..d9f12c976
Binary files /dev/null and b/test/test_data/compatibility/split-v1-chain differ
diff --git a/test/test_data/compatibility/split-v1-fallback-data b/test/test_data/compatibility/split-v1-fallback-data
new file mode 100644
index 000000000..24be9e0ad
Binary files /dev/null and b/test/test_data/compatibility/split-v1-fallback-data differ
diff --git a/test/test_data/compatibility/split-v1-indexed b/test/test_data/compatibility/split-v1-indexed
new file mode 100644
index 000000000..0d20df101
Binary files /dev/null and b/test/test_data/compatibility/split-v1-indexed differ