From 1c7736ad1d6b5f4c56057ef253496662a307fa07 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Thu, 5 Nov 2026 15:58:20 +0800 Subject: [PATCH 1/2] refactor(table): extract file store factories Centralize path, snapshot, manifest, index and partition component construction in an operation-local FileStore. Preserve existing option parsing, validation, dispatch and commit behavior. --- src/paimon/CMakeLists.txt | 1 + src/paimon/core/table/file_store.cpp | 138 +++++++++++++++++++++ src/paimon/core/table/file_store.h | 82 +++++++++++++ src/paimon/core/table/table.cpp | 177 +++++++-------------------- 4 files changed, 267 insertions(+), 131 deletions(-) create mode 100644 src/paimon/core/table/file_store.cpp create mode 100644 src/paimon/core/table/file_store.h diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 4db62cb79..2a1e8a803 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -448,6 +448,7 @@ set(PAIMON_CORE_SRCS core/stats/simple_stats_converter.cpp core/stats/simple_stats.cpp core/stats/simple_stats_evolution.cpp + core/table/file_store.cpp core/table/table.cpp core/table/bucket_mode.cpp core/table/format/format_file_listing.cpp diff --git a/src/paimon/core/table/file_store.cpp b/src/paimon/core/table/file_store.cpp new file mode 100644 index 000000000..4e6fb0afc --- /dev/null +++ b/src/paimon/core/table/file_store.cpp @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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/table/file_store.h" + +#include +#include +#include +#include +#include + +#include "paimon/catalog/catalog.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/common/utils/binary_row_partition_computer.h" +#include "paimon/core/catalog/version_managed_catalog.h" +#include "paimon/core/core_options.h" +#include "paimon/core/index/index_file_handler.h" +#include "paimon/core/manifest/index_manifest_file.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/utils/file_store_path_factory.h" +#include "paimon/core/utils/index_file_path_factories.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +FileStore::FileStore(std::string root_path, std::string branch, CoreOptions options) + : root_path_(std::move(root_path)), branch_(std::move(branch)), options_(std::move(options)) {} + +Result> FileStore::NewPathFactory( + const std::shared_ptr& arrow_schema, + const std::vector& partition_keys, const std::string& file_format_identifier, + const std::shared_ptr& memory_pool) const { + const CoreOptions& options = options_; + PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, options.CreateExternalPaths()); + PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, + options.CreateGlobalIndexExternalPath()); + return FileStorePathFactory::Create( + root_path_, arrow_schema, partition_keys, options.GetPartitionDefaultName(), + file_format_identifier, options.DataFilePrefix(), options.LegacyPartitionNameEnabled(), + external_paths, global_index_external_path, options.IndexFileInDataFileDir(), memory_pool); +} + +Result> FileStore::NewSnapshotManager( + const std::shared_ptr& catalog, + const std::optional& catalog_identifier) const { + const CoreOptions& options = options_; + auto snapshot_manager = + std::make_shared(options.GetFileSystem(), root_path_, branch_); + if (VersionManagedCatalog* versioned = AsVersionManaged(catalog)) { + if (!catalog_identifier) { + return Status::Invalid("a version-managed catalog requires a table identifier"); + } + Identifier identifier = catalog_identifier.value(); + snapshot_manager->SetSnapshotLoader( + [catalog, versioned, identifier]() -> Result> { + return versioned->LoadSnapshot(identifier); + }); + } + return snapshot_manager; +} + +Result> FileStore::NewManifestList( + const std::shared_ptr& path_factory, + const std::shared_ptr& memory_pool) const { + const CoreOptions& options = options_; + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr manifest_list, + ManifestList::Create(options.GetFileSystem(), options.GetManifestFormat(), + options.GetManifestCompression(), path_factory, options.GetCache(), + memory_pool)); + return manifest_list; +} + +Result> FileStore::NewManifestFile( + const std::shared_ptr& path_factory, + const std::shared_ptr& partition_schema, + const std::shared_ptr& memory_pool) const { + const CoreOptions& options = options_; + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr manifest_file, + ManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), + options.GetManifestCompression(), path_factory, + options.GetManifestTargetFileSize(), memory_pool, options, + partition_schema)); + return manifest_file; +} + +Result> FileStore::NewIndexManifestFile( + BucketMode bucket_mode, const std::shared_ptr& path_factory, + const std::shared_ptr& memory_pool) const { + const CoreOptions& options = options_; + return IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), + options.GetManifestCompression(), path_factory, bucket_mode, + memory_pool, options); +} + +Result> FileStore::NewIndexFileHandler( + BucketMode bucket_mode, const std::shared_ptr& path_factory, + const std::shared_ptr& memory_pool) const { + const CoreOptions& options = options_; + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_manifest_file, + NewIndexManifestFile(bucket_mode, path_factory, memory_pool)); + return std::make_unique( + options.GetFileSystem(), std::move(index_manifest_file), + std::make_shared(path_factory), options.DeletionVectorsBitmap64(), + memory_pool); +} + +Result> FileStore::NewPartitionComputer( + const std::shared_ptr& arrow_schema, + const std::vector& partition_keys, + const std::shared_ptr& memory_pool) const { + const CoreOptions& options = options_; + return BinaryRowPartitionComputer::Create(partition_keys, arrow_schema, + options.GetPartitionDefaultName(), + options.LegacyPartitionNameEnabled(), memory_pool); +} + +} // namespace paimon diff --git a/src/paimon/core/table/file_store.h b/src/paimon/core/table/file_store.h new file mode 100644 index 000000000..a03420390 --- /dev/null +++ b/src/paimon/core/table/file_store.h @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include +#include + +#include "paimon/catalog/identifier.h" +#include "paimon/core/core_options.h" +#include "paimon/core/table/bucket_mode.h" +#include "paimon/result.h" +#include "paimon/type_fwd.h" + +namespace arrow { +class Schema; +} // namespace arrow + +namespace paimon { +class BinaryRowPartitionComputer; +class Catalog; +class FileStorePathFactory; +class IndexFileHandler; +class IndexManifestFile; +class ManifestFile; +class ManifestList; +class SnapshotManager; + +/// Component factories for one managed-table operation using already resolved options. +class FileStore { + public: + FileStore(std::string root_path, std::string branch, CoreOptions options); + + Result> NewPathFactory( + const std::shared_ptr& arrow_schema, + const std::vector& partition_keys, const std::string& file_format_identifier, + const std::shared_ptr& memory_pool) const; + Result> NewSnapshotManager( + const std::shared_ptr& catalog, + const std::optional& catalog_identifier) const; + Result> NewManifestList( + const std::shared_ptr& path_factory, + const std::shared_ptr& memory_pool) const; + Result> NewManifestFile( + const std::shared_ptr& path_factory, + const std::shared_ptr& partition_schema, + const std::shared_ptr& memory_pool) const; + Result> NewIndexManifestFile( + BucketMode bucket_mode, const std::shared_ptr& path_factory, + const std::shared_ptr& memory_pool) const; + Result> NewIndexFileHandler( + BucketMode bucket_mode, const std::shared_ptr& path_factory, + const std::shared_ptr& memory_pool) const; + Result> NewPartitionComputer( + const std::shared_ptr& arrow_schema, + const std::vector& partition_keys, + const std::shared_ptr& memory_pool) const; + + private: + std::string root_path_; + std::string branch_; + CoreOptions options_; +}; +} // namespace paimon diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp index 7370377fb..b81010cd5 100644 --- a/src/paimon/core/table/table.cpp +++ b/src/paimon/core/table/table.cpp @@ -74,6 +74,7 @@ #include "paimon/core/schema/schema_validation.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/table/bucket_mode.h" +#include "paimon/core/table/file_store.h" #include "paimon/core/table/format/format_table_file_store_commit.h" #include "paimon/core/table/format/format_table_file_store_write.h" #include "paimon/core/table/format/format_table_loader.h" @@ -98,7 +99,6 @@ #include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/file_store_path_factory.h" -#include "paimon/core/utils/index_file_path_factories.h" #include "paimon/core/utils/primary_key_table_utils.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/defs.h" @@ -331,29 +331,14 @@ Result> FileStoreWrite::Create(std::unique_ptr partition_schema, FieldMapping::GetPartitionSchema(arrow_schema, schema->PartitionKeys())); - PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, options.CreateExternalPaths()); - PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, - options.CreateGlobalIndexExternalPath()); - + FileStore file_store(ctx->GetRootPath(), branch, options); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr file_store_path_factory, - FileStorePathFactory::Create(ctx->GetRootPath(), arrow_schema, schema->PartitionKeys(), - options.GetPartitionDefaultName(), - options.GetWriteFileFormat(/*level=*/0)->Identifier(), - options.DataFilePrefix(), options.LegacyPartitionNameEnabled(), - external_paths, global_index_external_path, - options.IndexFileInDataFileDir(), ctx->GetMemoryPool())); - auto snapshot_manager = - std::make_shared(options.GetFileSystem(), ctx->GetRootPath(), branch); - if (VersionManagedCatalog* versioned = AsVersionManaged(ctx->GetCatalog())) { - std::shared_ptr catalog = ctx->GetCatalog(); - Identifier identifier = catalog_identifier.value(); - // Keep the catalog alive while the writer uses its snapshot loader. - snapshot_manager->SetSnapshotLoader( - [catalog, versioned, identifier]() -> Result> { - return versioned->LoadSnapshot(identifier); - }); - } + file_store.NewPathFactory(arrow_schema, schema->PartitionKeys(), + options.GetWriteFileFormat(/*level=*/0)->Identifier(), + ctx->GetMemoryPool())); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot_manager, + file_store.NewSnapshotManager(ctx->GetCatalog(), catalog_identifier)); std::shared_ptr io_manager; const auto& io_temp_dir = ctx->GetTempDirectory(); if (!io_temp_dir.empty()) { @@ -424,15 +409,9 @@ Result> FileStoreWrite::Create(std::unique_ptr dv_maintainer_factory; if (need_dv_maintainer_factory) { PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr index_manifest_file, - IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), file_store_path_factory, - ResolveBucketMode(options.GetBucket(), schema), - ctx->GetMemoryPool(), options)); - auto index_file_handler = std::make_shared( - options.GetFileSystem(), std::move(index_manifest_file), - std::make_shared(file_store_path_factory), - options.DeletionVectorsBitmap64(), ctx->GetMemoryPool()); + std::shared_ptr index_file_handler, + file_store.NewIndexFileHandler(ResolveBucketMode(options.GetBucket(), schema), + file_store_path_factory, ctx->GetMemoryPool())); dv_maintainer_factory = std::make_shared(index_file_handler); } @@ -526,15 +505,9 @@ Result> FileStoreWrite::Create(std::unique_ptr index_file_handler; if (options.DeletionVectorsEnabled() || has_btree_index || has_existing_index_manifest) { PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr index_manifest_file, - IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), file_store_path_factory, - ResolveBucketMode(options.GetBucket(), schema), - ctx->GetMemoryPool(), options)); - index_file_handler = std::make_shared( - options.GetFileSystem(), std::move(index_manifest_file), - std::make_shared(file_store_path_factory), - options.DeletionVectorsBitmap64(), ctx->GetMemoryPool()); + index_file_handler, + file_store.NewIndexFileHandler(ResolveBucketMode(options.GetBucket(), schema), + file_store_path_factory, ctx->GetMemoryPool())); } bool has_restored_btree_index = false; if (!has_btree_index && has_existing_index_manifest) { @@ -590,23 +563,20 @@ class TableScanImpl { const std::shared_ptr& executor, const std::shared_ptr& memory_pool, const ScanContext* context) { auto fs = core_options.GetFileSystem(); - auto manifest_file_format = core_options.GetManifestFormat(); std::string branch = BranchManager::NormalizeBranch(core_options.GetBranch()); - auto snapshot_manager = std::make_shared(fs, context->GetPath(), branch); + FileStore file_store(context->GetPath(), branch, core_options); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot_manager, + file_store.NewSnapshotManager(nullptr, std::nullopt)); // TODO(liancheng.lsz): support fallback branch in scan auto schema_manager = std::make_shared(fs, context->GetPath(), branch); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr manifest_list, - ManifestList::Create(fs, manifest_file_format, core_options.GetManifestCompression(), - path_factory, core_options.GetCache(), memory_pool)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr manifest_list, + file_store.NewManifestList(path_factory, memory_pool)); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr partition_schema, FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys())); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr manifest_file, - ManifestFile::Create(fs, manifest_file_format, core_options.GetManifestCompression(), - path_factory, core_options.GetManifestTargetFileSize(), - memory_pool, core_options, partition_schema)); + file_store.NewManifestFile(path_factory, partition_schema, memory_pool)); std::unique_ptr scan; if (table_schema->PrimaryKeys().empty()) { if (core_options.DataEvolutionEnabled()) { @@ -668,22 +638,6 @@ class TableScanImpl { key_comparator); } } - - static Result> CreateIndexFileHandler( - const std::shared_ptr& table_schema, const CoreOptions& core_options, - const std::shared_ptr& path_factory, - const std::shared_ptr& memory_pool) { - PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_manifest_file, - IndexManifestFile::Create( - core_options.GetFileSystem(), core_options.GetManifestFormat(), - core_options.GetManifestCompression(), path_factory, - ResolveBucketMode(core_options.GetBucket(), table_schema), - memory_pool, core_options)); - return std::make_unique( - core_options.GetFileSystem(), std::move(index_manifest_file), - std::make_shared(path_factory), - core_options.DeletionVectorsBitmap64(), memory_pool); - } }; /// @param loaded_schema The table's schema, when the caller already read it, or null when it has @@ -895,19 +849,12 @@ Result> NewDataTableScan( *arrow_schema, context->GetScanFilters()->GetPredicate(), /*validate_field_idx=*/false)); } - PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, - core_options.CreateExternalPaths()); - PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, - core_options.CreateGlobalIndexExternalPath()); - - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr path_factory, - FileStorePathFactory::Create( - context->GetPath(), arrow_schema, table_schema->PartitionKeys(), - core_options.GetPartitionDefaultName(), core_options.GetFileFormat()->Identifier(), - core_options.DataFilePrefix(), core_options.LegacyPartitionNameEnabled(), - external_paths, global_index_external_path, core_options.IndexFileInDataFileDir(), - context->GetMemoryPool())); + FileStore file_store(context->GetPath(), + BranchManager::NormalizeBranch(core_options.GetBranch()), core_options); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr path_factory, + file_store.NewPathFactory(arrow_schema, table_schema->PartitionKeys(), + core_options.GetFileFormat()->Identifier(), + context->GetMemoryPool())); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr file_store_scan, TableScanImpl::CreateFileStoreScan( @@ -916,9 +863,10 @@ Result> NewDataTableScan( PAIMON_ASSIGN_OR_RAISE( std::unique_ptr split_generator, TableScanImpl::CreateSplitGenerator(table_schema, core_options, context.get())); - PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_file_handler, - TableScanImpl::CreateIndexFileHandler( - table_schema, core_options, path_factory, context->GetMemoryPool())); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr index_file_handler, + file_store.NewIndexFileHandler(ResolveBucketMode(core_options.GetBucket(), table_schema), + path_factory, context->GetMemoryPool())); auto snapshot_reader = std::make_shared( file_store_scan, path_factory, std::move(split_generator), std::move(index_file_handler)); const bool pk_table = !table_schema->PrimaryKeys().empty(); @@ -1021,19 +969,12 @@ Result> CreateTableRead( } PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateRealtime(*table_schema, core_options)); auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); - PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, - core_options.CreateExternalPaths()); - PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, - core_options.CreateGlobalIndexExternalPath()); - + FileStore file_store(internal_context->GetPath(), + BranchManager::NormalizeBranch(core_options.GetBranch()), core_options); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr path_factory, - FileStorePathFactory::Create( - internal_context->GetPath(), arrow_schema, table_schema->PartitionKeys(), - core_options.GetPartitionDefaultName(), core_options.GetFileFormat()->Identifier(), - core_options.DataFilePrefix(), core_options.LegacyPartitionNameEnabled(), - external_paths, global_index_external_path, core_options.IndexFileInDataFileDir(), - memory_pool)); + file_store.NewPathFactory(arrow_schema, table_schema->PartitionKeys(), + core_options.GetFileFormat()->Identifier(), memory_pool)); if (internal_context->GetPrimaryKeys().empty()) { return std::make_unique(path_factory, internal_context, memory_pool, @@ -1383,55 +1324,29 @@ Result> FileStoreCommit::Create( return Status::NotImplemented( "commit operation does not support object store file system for now"); } + FileStore file_store(root_path, branch, options); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr partition_computer, - BinaryRowPartitionComputer::Create(schema->PartitionKeys(), arrow_schema, - options.GetPartitionDefaultName(), - options.LegacyPartitionNameEnabled(), - ctx->GetMemoryPool())); - PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, options.CreateExternalPaths()); - PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, - options.CreateGlobalIndexExternalPath()); - + file_store.NewPartitionComputer(arrow_schema, schema->PartitionKeys(), + ctx->GetMemoryPool())); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr path_factory, - FileStorePathFactory::Create( - root_path, arrow_schema, schema->PartitionKeys(), options.GetPartitionDefaultName(), - options.GetFileFormat()->Identifier(), options.DataFilePrefix(), - options.LegacyPartitionNameEnabled(), external_paths, global_index_external_path, - options.IndexFileInDataFileDir(), ctx->GetMemoryPool())); - - auto snapshot_manager = - std::make_shared(options.GetFileSystem(), root_path, branch); - if (IsCatalogCommit(*ctx)) { - std::shared_ptr catalog = ctx->GetCatalog(); - VersionManagedCatalog* versioned = AsVersionManaged(catalog); - Identifier identifier = catalog_identifier.value(); - // Capture the catalog to keep the versioned interface alive. - snapshot_manager->SetSnapshotLoader( - [catalog, versioned, identifier]() -> Result> { - return versioned->LoadSnapshot(identifier); - }); - } + file_store.NewPathFactory(arrow_schema, schema->PartitionKeys(), + options.GetFileFormat()->Identifier(), ctx->GetMemoryPool())); PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr manifest_list, - ManifestList::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), path_factory, options.GetCache(), - ctx->GetMemoryPool())); - + std::shared_ptr snapshot_manager, + file_store.NewSnapshotManager(IsCatalogCommit(*ctx) ? ctx->GetCatalog() : nullptr, + catalog_identifier)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr manifest_list, + file_store.NewManifestList(path_factory, ctx->GetMemoryPool())); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr partition_schema, FieldMapping::GetPartitionSchema(arrow_schema, schema->PartitionKeys())); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr manifest_file, - ManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), path_factory, - options.GetManifestTargetFileSize(), ctx->GetMemoryPool(), options, - partition_schema)); + file_store.NewManifestFile(path_factory, partition_schema, ctx->GetMemoryPool())); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr index_manifest_file, - IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), path_factory, - ResolveBucketMode(options.GetBucket(), schema), - ctx->GetMemoryPool(), options)); + file_store.NewIndexManifestFile(ResolveBucketMode(options.GetBucket(), schema), + path_factory, ctx->GetMemoryPool())); auto expire_snapshots = std::make_shared( snapshot_manager, path_factory, manifest_list, manifest_file, index_manifest_file, From 545ff44f456e8ca5ab1365be9b26abf33ec3f65f Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Fri, 6 Nov 2026 11:49:37 +0800 Subject: [PATCH 2/2] refactor(table): simplify file store option handling --- src/paimon/core/table/file_store.cpp | 50 ++++++++++++---------------- src/paimon/core/table/file_store.h | 3 +- src/paimon/core/table/table.cpp | 21 +++++------- 3 files changed, 32 insertions(+), 42 deletions(-) diff --git a/src/paimon/core/table/file_store.cpp b/src/paimon/core/table/file_store.cpp index 4e6fb0afc..8f1c6d7a8 100644 --- a/src/paimon/core/table/file_store.cpp +++ b/src/paimon/core/table/file_store.cpp @@ -34,6 +34,7 @@ #include "paimon/core/manifest/index_manifest_file.h" #include "paimon/core/manifest/manifest_file.h" #include "paimon/core/manifest/manifest_list.h" +#include "paimon/core/utils/branch_manager.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/index_file_path_factories.h" #include "paimon/core/utils/snapshot_manager.h" @@ -42,29 +43,27 @@ namespace paimon { -FileStore::FileStore(std::string root_path, std::string branch, CoreOptions options) - : root_path_(std::move(root_path)), branch_(std::move(branch)), options_(std::move(options)) {} +FileStore::FileStore(std::string root_path, CoreOptions options) + : root_path_(std::move(root_path)), options_(std::move(options)) {} Result> FileStore::NewPathFactory( const std::shared_ptr& arrow_schema, const std::vector& partition_keys, const std::string& file_format_identifier, const std::shared_ptr& memory_pool) const { - const CoreOptions& options = options_; - PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, options.CreateExternalPaths()); + PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, options_.CreateExternalPaths()); PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, - options.CreateGlobalIndexExternalPath()); + options_.CreateGlobalIndexExternalPath()); return FileStorePathFactory::Create( - root_path_, arrow_schema, partition_keys, options.GetPartitionDefaultName(), - file_format_identifier, options.DataFilePrefix(), options.LegacyPartitionNameEnabled(), - external_paths, global_index_external_path, options.IndexFileInDataFileDir(), memory_pool); + root_path_, arrow_schema, partition_keys, options_.GetPartitionDefaultName(), + file_format_identifier, options_.DataFilePrefix(), options_.LegacyPartitionNameEnabled(), + external_paths, global_index_external_path, options_.IndexFileInDataFileDir(), memory_pool); } Result> FileStore::NewSnapshotManager( const std::shared_ptr& catalog, const std::optional& catalog_identifier) const { - const CoreOptions& options = options_; - auto snapshot_manager = - std::make_shared(options.GetFileSystem(), root_path_, branch_); + auto snapshot_manager = std::make_shared( + options_.GetFileSystem(), root_path_, BranchManager::NormalizeBranch(options_.GetBranch())); if (VersionManagedCatalog* versioned = AsVersionManaged(catalog)) { if (!catalog_identifier) { return Status::Invalid("a version-managed catalog requires a table identifier"); @@ -81,11 +80,10 @@ Result> FileStore::NewSnapshotManager( Result> FileStore::NewManifestList( const std::shared_ptr& path_factory, const std::shared_ptr& memory_pool) const { - const CoreOptions& options = options_; PAIMON_ASSIGN_OR_RAISE( std::shared_ptr manifest_list, - ManifestList::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), path_factory, options.GetCache(), + ManifestList::Create(options_.GetFileSystem(), options_.GetManifestFormat(), + options_.GetManifestCompression(), path_factory, options_.GetCache(), memory_pool)); return manifest_list; } @@ -94,12 +92,11 @@ Result> FileStore::NewManifestFile( const std::shared_ptr& path_factory, const std::shared_ptr& partition_schema, const std::shared_ptr& memory_pool) const { - const CoreOptions& options = options_; PAIMON_ASSIGN_OR_RAISE( std::shared_ptr manifest_file, - ManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), path_factory, - options.GetManifestTargetFileSize(), memory_pool, options, + ManifestFile::Create(options_.GetFileSystem(), options_.GetManifestFormat(), + options_.GetManifestCompression(), path_factory, + options_.GetManifestTargetFileSize(), memory_pool, options_, partition_schema)); return manifest_file; } @@ -107,21 +104,19 @@ Result> FileStore::NewManifestFile( Result> FileStore::NewIndexManifestFile( BucketMode bucket_mode, const std::shared_ptr& path_factory, const std::shared_ptr& memory_pool) const { - const CoreOptions& options = options_; - return IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), - options.GetManifestCompression(), path_factory, bucket_mode, - memory_pool, options); + return IndexManifestFile::Create(options_.GetFileSystem(), options_.GetManifestFormat(), + options_.GetManifestCompression(), path_factory, bucket_mode, + memory_pool, options_); } Result> FileStore::NewIndexFileHandler( BucketMode bucket_mode, const std::shared_ptr& path_factory, const std::shared_ptr& memory_pool) const { - const CoreOptions& options = options_; PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_manifest_file, NewIndexManifestFile(bucket_mode, path_factory, memory_pool)); return std::make_unique( - options.GetFileSystem(), std::move(index_manifest_file), - std::make_shared(path_factory), options.DeletionVectorsBitmap64(), + options_.GetFileSystem(), std::move(index_manifest_file), + std::make_shared(path_factory), options_.DeletionVectorsBitmap64(), memory_pool); } @@ -129,10 +124,9 @@ Result> FileStore::NewPartitionCompu const std::shared_ptr& arrow_schema, const std::vector& partition_keys, const std::shared_ptr& memory_pool) const { - const CoreOptions& options = options_; return BinaryRowPartitionComputer::Create(partition_keys, arrow_schema, - options.GetPartitionDefaultName(), - options.LegacyPartitionNameEnabled(), memory_pool); + options_.GetPartitionDefaultName(), + options_.LegacyPartitionNameEnabled(), memory_pool); } } // namespace paimon diff --git a/src/paimon/core/table/file_store.h b/src/paimon/core/table/file_store.h index a03420390..521a7bc39 100644 --- a/src/paimon/core/table/file_store.h +++ b/src/paimon/core/table/file_store.h @@ -47,7 +47,7 @@ class SnapshotManager; /// Component factories for one managed-table operation using already resolved options. class FileStore { public: - FileStore(std::string root_path, std::string branch, CoreOptions options); + FileStore(std::string root_path, CoreOptions options); Result> NewPathFactory( const std::shared_ptr& arrow_schema, @@ -76,7 +76,6 @@ class FileStore { private: std::string root_path_; - std::string branch_; CoreOptions options_; }; } // namespace paimon diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp index b81010cd5..7bf9cdeb8 100644 --- a/src/paimon/core/table/table.cpp +++ b/src/paimon/core/table/table.cpp @@ -331,7 +331,7 @@ Result> FileStoreWrite::Create(std::unique_ptr partition_schema, FieldMapping::GetPartitionSchema(arrow_schema, schema->PartitionKeys())); - FileStore file_store(ctx->GetRootPath(), branch, options); + FileStore file_store(ctx->GetRootPath(), options); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr file_store_path_factory, file_store.NewPathFactory(arrow_schema, schema->PartitionKeys(), @@ -564,9 +564,10 @@ class TableScanImpl { const ScanContext* context) { auto fs = core_options.GetFileSystem(); std::string branch = BranchManager::NormalizeBranch(core_options.GetBranch()); - FileStore file_store(context->GetPath(), branch, core_options); + FileStore file_store(context->GetPath(), core_options); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot_manager, - file_store.NewSnapshotManager(nullptr, std::nullopt)); + file_store.NewSnapshotManager(/*catalog=*/nullptr, + /*catalog_identifier=*/std::nullopt)); // TODO(liancheng.lsz): support fallback branch in scan auto schema_manager = std::make_shared(fs, context->GetPath(), branch); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr manifest_list, @@ -849,8 +850,7 @@ Result> NewDataTableScan( *arrow_schema, context->GetScanFilters()->GetPredicate(), /*validate_field_idx=*/false)); } - FileStore file_store(context->GetPath(), - BranchManager::NormalizeBranch(core_options.GetBranch()), core_options); + FileStore file_store(context->GetPath(), core_options); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr path_factory, file_store.NewPathFactory(arrow_schema, table_schema->PartitionKeys(), core_options.GetFileFormat()->Identifier(), @@ -969,8 +969,7 @@ Result> CreateTableRead( } PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateRealtime(*table_schema, core_options)); auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); - FileStore file_store(internal_context->GetPath(), - BranchManager::NormalizeBranch(core_options.GetBranch()), core_options); + FileStore file_store(internal_context->GetPath(), core_options); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr path_factory, file_store.NewPathFactory(arrow_schema, table_schema->PartitionKeys(), @@ -1324,7 +1323,7 @@ Result> FileStoreCommit::Create( return Status::NotImplemented( "commit operation does not support object store file system for now"); } - FileStore file_store(root_path, branch, options); + FileStore file_store(root_path, options); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr partition_computer, file_store.NewPartitionComputer(arrow_schema, schema->PartitionKeys(), ctx->GetMemoryPool())); @@ -1332,10 +1331,8 @@ Result> FileStoreCommit::Create( std::shared_ptr path_factory, file_store.NewPathFactory(arrow_schema, schema->PartitionKeys(), options.GetFileFormat()->Identifier(), ctx->GetMemoryPool())); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr snapshot_manager, - file_store.NewSnapshotManager(IsCatalogCommit(*ctx) ? ctx->GetCatalog() : nullptr, - catalog_identifier)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr snapshot_manager, + file_store.NewSnapshotManager(ctx->GetCatalog(), catalog_identifier)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr manifest_list, file_store.NewManifestList(path_factory, ctx->GetMemoryPool())); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr partition_schema,