Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/paimon/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
88 changes: 88 additions & 0 deletions src/paimon/core/io/chain_split_file_path_factory.cpp
Original file line number Diff line number Diff line change
@@ -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
52 changes: 52 additions & 0 deletions src/paimon/core/io/chain_split_file_path_factory.h
Original file line number Diff line number Diff line change
@@ -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
6 changes: 3 additions & 3 deletions src/paimon/core/io/data_file_path_factory.h
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
72 changes: 65 additions & 7 deletions src/paimon/core/operation/abstract_split_read.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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 {

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think there is still a semantic gap with the Java ChainSplit read path.

The PR now deserializes and preserves fileBranchMapping, and uses fileBucketPathMapping to resolve each file path. However, Java also consumes fileBranchMapping when reading each file: MergeFileSplitRead#createChainReader builds a ChainReadContext, and ChainKeyValueFileReaderFactory#getDataSchema uses chainReadContext.fileBranchMapping().get(fileMeta.fileName()) to choose the branch-specific SchemaManager before loading fileMeta.schemaId().

In the C++ path, AbstractSplitRead::CreateFieldMappingReader still loads schemas from the current table/current branch only via context_->GetTableSchema() or schema_manager_->ReadSchema(file_meta->schema_id). For a ChainSplit containing files from both snapshot and delta branches, the same schema_id may refer to different schemas across branches, or may not exist in the current branch at all. In that case C++ can read with the wrong schema or fail, while Java reads with the schema from the file’s own branch.

Could we either make ChainSplit reads branch-aware when resolving DataFileMeta::schema_id, using ChainSplitImpl::FileBranchMapping(), or explicitly fail fast for ChainSplits whose files belong to a non-current branch until branch-aware schema selection is supported?

PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileFormat> file_format,
Expand Down Expand Up @@ -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();
Expand Down
11 changes: 11 additions & 0 deletions src/paimon/core/operation/abstract_split_read.h
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

#pragma once

#include <map>
#include <memory>
#include <optional>
#include <set>
Expand Down Expand Up @@ -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(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: CreateDataFilePathFactory is currently in the public section, but it seems to only be called by subclasses internally. Would it be possible to move it into the protected section just below?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in the latest revision. CreateDataFilePathFactory is now protected since it is only used by AbstractSplitRead subclasses.

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(
Expand Down Expand Up @@ -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;
Expand All @@ -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
Loading
Loading