refactor(table): extract file store factories

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

src/paimon/core/table/file_store.cpp

@@ -0,0 +1,132 @@
+/*
+ * 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 <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#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/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"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+FileStore::FileStore(std::string root_path, CoreOptions options)
+    : root_path_(std::move(root_path)), options_(std::move(options)) {}
+
+Result<std::shared_ptr<FileStorePathFactory>> FileStore::NewPathFactory(
+    const std::shared_ptr<arrow::Schema>& arrow_schema,
+    const std::vector<std::string>& partition_keys, const std::string& file_format_identifier,
+    const std::shared_ptr<MemoryPool>& memory_pool) const {
+    PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> external_paths, options_.CreateExternalPaths());
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> 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<std::shared_ptr<SnapshotManager>> FileStore::NewSnapshotManager(
+    const std::shared_ptr<Catalog>& catalog,
+    const std::optional<Identifier>& catalog_identifier) const {
+    auto snapshot_manager = std::make_shared<SnapshotManager>(
+        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");
+        }
+        Identifier identifier = catalog_identifier.value();
+        snapshot_manager->SetSnapshotLoader(
+            [catalog, versioned, identifier]() -> Result<std::optional<Snapshot>> {
+                return versioned->LoadSnapshot(identifier);
+            });
+    }
+    return snapshot_manager;
+}
+
+Result<std::shared_ptr<ManifestList>> FileStore::NewManifestList(
+    const std::shared_ptr<FileStorePathFactory>& path_factory,
+    const std::shared_ptr<MemoryPool>& memory_pool) const {
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<ManifestList> manifest_list,
+        ManifestList::Create(options_.GetFileSystem(), options_.GetManifestFormat(),
+                             options_.GetManifestCompression(), path_factory, options_.GetCache(),
+                             memory_pool));
+    return manifest_list;
+}
+
+Result<std::shared_ptr<ManifestFile>> FileStore::NewManifestFile(
+    const std::shared_ptr<FileStorePathFactory>& path_factory,
+    const std::shared_ptr<arrow::Schema>& partition_schema,
+    const std::shared_ptr<MemoryPool>& memory_pool) const {
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<ManifestFile> manifest_file,
+        ManifestFile::Create(options_.GetFileSystem(), options_.GetManifestFormat(),
+                             options_.GetManifestCompression(), path_factory,
+                             options_.GetManifestTargetFileSize(), memory_pool, options_,
+                             partition_schema));
+    return manifest_file;
+}
+
+Result<std::unique_ptr<IndexManifestFile>> FileStore::NewIndexManifestFile(
+    BucketMode bucket_mode, const std::shared_ptr<FileStorePathFactory>& path_factory,
+    const std::shared_ptr<MemoryPool>& memory_pool) const {
+    return IndexManifestFile::Create(options_.GetFileSystem(), options_.GetManifestFormat(),
+                                     options_.GetManifestCompression(), path_factory, bucket_mode,
+                                     memory_pool, options_);
+}
+
+Result<std::unique_ptr<IndexFileHandler>> FileStore::NewIndexFileHandler(
+    BucketMode bucket_mode, const std::shared_ptr<FileStorePathFactory>& path_factory,
+    const std::shared_ptr<MemoryPool>& memory_pool) const {
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IndexManifestFile> index_manifest_file,
+                           NewIndexManifestFile(bucket_mode, path_factory, memory_pool));
+    return std::make_unique<IndexFileHandler>(
+        options_.GetFileSystem(), std::move(index_manifest_file),
+        std::make_shared<IndexFilePathFactories>(path_factory), options_.DeletionVectorsBitmap64(),
+        memory_pool);
+}
+
+Result<std::unique_ptr<BinaryRowPartitionComputer>> FileStore::NewPartitionComputer(
+    const std::shared_ptr<arrow::Schema>& arrow_schema,
+    const std::vector<std::string>& partition_keys,
+    const std::shared_ptr<MemoryPool>& memory_pool) const {
+    return BinaryRowPartitionComputer::Create(partition_keys, arrow_schema,
+                                              options_.GetPartitionDefaultName(),
+                                              options_.LegacyPartitionNameEnabled(), memory_pool);
+}
+
+}  // namespace paimon

src/paimon/core/table/file_store.h

@@ -0,0 +1,81 @@
+/*
+ * 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 <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#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, CoreOptions options);
+
+    Result<std::shared_ptr<FileStorePathFactory>> NewPathFactory(
+        const std::shared_ptr<arrow::Schema>& arrow_schema,
+        const std::vector<std::string>& partition_keys, const std::string& file_format_identifier,
+        const std::shared_ptr<MemoryPool>& memory_pool) const;
+    Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager(
+        const std::shared_ptr<Catalog>& catalog,
+        const std::optional<Identifier>& catalog_identifier) const;
+    Result<std::shared_ptr<ManifestList>> NewManifestList(
+        const std::shared_ptr<FileStorePathFactory>& path_factory,
+        const std::shared_ptr<MemoryPool>& memory_pool) const;
+    Result<std::shared_ptr<ManifestFile>> NewManifestFile(
+        const std::shared_ptr<FileStorePathFactory>& path_factory,
+        const std::shared_ptr<arrow::Schema>& partition_schema,
+        const std::shared_ptr<MemoryPool>& memory_pool) const;
+    Result<std::unique_ptr<IndexManifestFile>> NewIndexManifestFile(
+        BucketMode bucket_mode, const std::shared_ptr<FileStorePathFactory>& path_factory,
+        const std::shared_ptr<MemoryPool>& memory_pool) const;
+    Result<std::unique_ptr<IndexFileHandler>> NewIndexFileHandler(
+        BucketMode bucket_mode, const std::shared_ptr<FileStorePathFactory>& path_factory,
+        const std::shared_ptr<MemoryPool>& memory_pool) const;
+    Result<std::unique_ptr<BinaryRowPartitionComputer>> NewPartitionComputer(
+        const std::shared_ptr<arrow::Schema>& arrow_schema,
+        const std::vector<std::string>& partition_keys,
+        const std::shared_ptr<MemoryPool>& memory_pool) const;
+
+ private:
+    std::string root_path_;
+    CoreOptions options_;
+};
+}  // namespace paimon