feat(api): Support specifying file system for more APIs (alibaba#78)

Author: Eyizoha

include/paimon/catalog/catalog.h

@@ -46,9 +46,13 @@ class PAIMON_EXPORT Catalog {
     ///
     /// @param root_path Path to the root directory where the catalog is located.
     /// @param options Configuration options for catalog initialization.
+    /// @param file_system Specifies the file system for file operations.
+    ///                    If not set, use default file system (configured in
+    ///                    `Options::FILE_SYSTEM`)
     /// @return A result containing a unique pointer to a `Catalog` instance, or an error status.
     static Result<std::unique_ptr<Catalog>> Create(
-        const std::string& root_path, const std::map<std::string, std::string>& options);
+        const std::string& root_path, const std::map<std::string, std::string>& options,
+        const std::shared_ptr<FileSystem>& file_system = nullptr);
 
     virtual ~Catalog() = default;
 
@@ -124,6 +128,16 @@ class PAIMON_EXPORT Catalog {
     /// @return A string representing the expected location of the table.
     virtual std::string GetTableLocation(const Identifier& identifier) const = 0;
 
+    /// Returns the root path of the catalog.
+    ///
+    /// @return A string representing the root path of the catalog.
+    virtual std::string GetRootPath() const = 0;
+
+    /// Returns the file system used by the catalog.
+    ///
+    /// @return A shared pointer to the file system instance.
+    virtual std::shared_ptr<FileSystem> GetFileSystem() const = 0;
+
     /// Loads the latest schema of a specified table.
     ///
     /// @note System tables will not be supported.

include/paimon/commit_context.h

@@ -39,6 +39,7 @@ class PAIMON_EXPORT CommitContext {
                   bool ignore_empty_commit, bool use_rest_catalog_commit,
                   const std::shared_ptr<MemoryPool>& memory_pool,
                   const std::shared_ptr<Executor>& executor,
+                  const std::shared_ptr<FileSystem>& specific_file_system,
                   const std::map<std::string, std::string>& options);
     ~CommitContext();
 
@@ -66,6 +67,10 @@ class PAIMON_EXPORT CommitContext {
         return executor_;
     }
 
+    std::shared_ptr<FileSystem> GetSpecificFileSystem() const {
+        return specific_file_system_;
+    }
+
     const std::map<std::string, std::string>& GetOptions() const {
         return options_;
     }
@@ -77,6 +82,7 @@ class PAIMON_EXPORT CommitContext {
     bool use_rest_catalog_commit_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> options_;
 };
 
@@ -129,6 +135,15 @@ class PAIMON_EXPORT CommitContextBuilder {
     /// @return Reference to this builder for method chaining.
     CommitContextBuilder& WithExecutor(const std::shared_ptr<Executor>& executor);
 
+    /// Sets a custom file system instance to be used for all file operations in this commit
+    /// context.
+    /// This bypasses the global file system registry and uses the provided implementation directly.
+    ///
+    /// @param file_system The file system to use.
+    /// @return Reference to this builder for method chaining.
+    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
+    CommitContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
+
     /// Build and return a `CommitContext` instance with input validation.
     /// @return Result containing the constructed `CommitContext` or an error status.
     Result<std::unique_ptr<CommitContext>> Finish();

include/paimon/global_index/global_index_write_task.h

@@ -44,13 +44,16 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
     /// @param options      Index-specific configuration (e.g., false positive rate for bloom
     /// filters).
     /// @param pool         Memory pool for temporary allocations during index construction.
-    //                      If `nullptr`, the system's default memory pool will be used.
+    ///                     If `nullptr`, the system's default memory pool will be used.
+    /// @param file_system  Specifies the file system for file operations.
+    ///                     If `nullptr`, use default file system.
     /// @return A `Result` containing a shared pointer to the `CommitMessage` with index metadata,
     ///         or an error if indexing fails (e.g., unsupported type, I/O error).
     static Result<std::shared_ptr<CommitMessage>> WriteIndex(
         const std::string& table_path, const std::string& field_name, const std::string& index_type,
         const std::shared_ptr<IndexedSplit>& indexed_split,
-        const std::map<std::string, std::string>& options, const std::shared_ptr<MemoryPool>& pool);
+        const std::map<std::string, std::string>& options, const std::shared_ptr<MemoryPool>& pool,
+        const std::shared_ptr<FileSystem>& file_system = nullptr);
 };
 
 }  // namespace paimon

include/paimon/migrate/file_meta_utils.h

@@ -56,12 +56,15 @@ class PAIMON_EXPORT FileMetaUtils {
     ///                         tables. Use empty map for non-partitioned tables.
     /// @param options Set a configuration options map to set some option entries which are not
     ///                defined in the table schema or whose values you want to overwrite.
+    /// @param file_system  Specifies the file system for file operations.
+    ///                     If `nullptr`, use default file system.
     /// @return Result containing a unique pointer to the generated `CommitMessage`,
     ///         or an error status if the migration cannot be performed.
     static Result<std::unique_ptr<CommitMessage>> GenerateCommitMessage(
         const std::vector<std::string>& src_data_files, const std::string& dst_table_path,
         const std::map<std::string, std::string>& partition_values,
-        const std::map<std::string, std::string>& options);
+        const std::map<std::string, std::string>& options,
+        const std::shared_ptr<FileSystem>& file_system = nullptr);
 };
 
 }  // namespace paimon

include/paimon/orphan_files_cleaner.h

@@ -41,6 +41,7 @@ class PAIMON_EXPORT CleanContext {
     CleanContext(const std::string& root_path, const std::map<std::string, std::string>& options,
                  int64_t older_than_ms, const std::shared_ptr<MemoryPool>& pool,
                  const std::shared_ptr<Executor>& executor,
+                 const std::shared_ptr<FileSystem>& specific_file_system,
                  std::function<bool(const std::string&)> should_be_retained);
     ~CleanContext();
 
@@ -64,6 +65,10 @@ class PAIMON_EXPORT CleanContext {
         return executor_;
     }
 
+    std::shared_ptr<FileSystem> GetSpecificFileSystem() const {
+        return specific_file_system_;
+    }
+
     std::function<bool(const std::string&)> GetFileRetainCondition() const {
         return should_be_retained_;
     }
@@ -74,6 +79,7 @@ class PAIMON_EXPORT CleanContext {
     int64_t older_than_ms_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<FileSystem> specific_file_system_;
     std::function<bool(const std::string&)> should_be_retained_;
 };
 
@@ -124,6 +130,14 @@ class PAIMON_EXPORT CleanContextBuilder {
     /// @return Reference to this builder for method chaining.
     CleanContextBuilder& WithExecutor(const std::shared_ptr<Executor>& executor);
 
+    /// Sets a custom file system instance to be used for all file operations in this clean context.
+    /// This bypasses the global file system registry and uses the provided implementation directly.
+    ///
+    /// @param file_system The file system to use.
+    /// @return Reference to this builder for method chaining.
+    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
+    CleanContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
+
     /// Build and return a `CleanContext` instance with input validation.
     /// @return Result containing the constructed `CleanContext` or an error status.
     Result<std::unique_ptr<CleanContext>> Finish();

include/paimon/read_context.h

@@ -334,6 +334,10 @@ class PAIMON_EXPORT ReadContextBuilder {
 
     /// Sets a custom file system instance to be used for all file operations in this read context.
     /// This bypasses the global file system registry and uses the provided implementation directly.
+    ///
+    /// @param file_system The file system to use.
+    /// @return Reference to this builder for method chaining.
+    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
     ReadContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
 
     /// Build and return a `ReadContext` instance with input validation.

include/paimon/scan_context.h

@@ -48,6 +48,7 @@ class PAIMON_EXPORT ScanContext {
                 const std::shared_ptr<GlobalIndexResult>& global_index_result,
                 const std::shared_ptr<MemoryPool>& memory_pool,
                 const std::shared_ptr<Executor>& executor,
+                const std::shared_ptr<FileSystem>& specific_file_system,
                 const std::map<std::string, std::string>& options);
 
     ~ScanContext();
@@ -82,6 +83,10 @@ class PAIMON_EXPORT ScanContext {
         return global_index_result_;
     }
 
+    std::shared_ptr<FileSystem> GetSpecificFileSystem() const {
+        return specific_file_system_;
+    }
+
  private:
     std::string path_;
     bool is_streaming_mode_;
@@ -90,6 +95,7 @@ class PAIMON_EXPORT ScanContext {
     std::shared_ptr<GlobalIndexResult> global_index_result_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> options_;
 };
 
@@ -174,6 +180,12 @@ class PAIMON_EXPORT ScanContextBuilder {
     /// @return Reference to this builder for method chaining.
     /// @note if not set, executor is default executor
     ScanContextBuilder& WithExecutor(const std::shared_ptr<Executor>& executor);
+    /// Sets a custom file system instance to be used for all file operations in this scan context.
+    /// This bypasses the global file system registry and uses the provided implementation directly.
+    /// @param file_system The file system to use.
+    /// @return Reference to this builder for method chaining.
+    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
+    ScanContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
 
     /// Build and return a `ScanContext` instance with input validation.
     /// @return Result containing the constructed `ScanContext` or an error status.

include/paimon/write_context.h

@@ -44,6 +44,7 @@ class PAIMON_EXPORT WriteContext {
                  const std::vector<std::string>& write_schema,
                  const std::shared_ptr<MemoryPool>& memory_pool,
                  const std::shared_ptr<Executor>& executor,
+                 const std::shared_ptr<FileSystem>& specific_file_system,
                  const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
                  const std::map<std::string, std::string>& options);
 
@@ -97,6 +98,10 @@ class PAIMON_EXPORT WriteContext {
         return executor_;
     }
 
+    std::shared_ptr<FileSystem> GetSpecificFileSystem() const {
+        return specific_file_system_;
+    }
+
  private:
     std::string root_path_;
     std::string commit_user_;
@@ -108,6 +113,7 @@ class PAIMON_EXPORT WriteContext {
     std::vector<std::string> write_schema_;
     std::shared_ptr<MemoryPool> memory_pool_;
     std::shared_ptr<Executor> executor_;
+    std::shared_ptr<FileSystem> specific_file_system_;
     std::map<std::string, std::string> fs_scheme_to_identifier_map_;
     std::map<std::string, std::string> options_;
 };
@@ -175,12 +181,30 @@ class PAIMON_EXPORT WriteContextBuilder {
     /// @return Reference to this builder for method chaining.
     WriteContextBuilder& WithWriteSchema(const std::vector<std::string>& write_schema);
 
-    /// Set the file system scheme to identifier mapping for custom file system configurations.
-    /// This allows using different file system implementations for different URI schemes.
+    /// Sets a custom file system instance to be used for all file operations in this write context.
+    /// This bypasses the global file system registry and uses the provided implementation directly.
+    ///
+    /// @param file_system The file system to use.
+    /// @return Reference to this builder for method chaining.
+    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`)
+    WriteContextBuilder& WithFileSystem(const std::shared_ptr<FileSystem>& file_system);
+
+    /// Sets a mapping from URI schemes (e.g., "file", "oss") to registered file system
+    /// identifiers. This allows selecting different pre-registered file system implementations
+    /// based on the URI scheme at runtime.
     ///
-    /// @param fs_scheme_to_identifier_map Map from URI scheme to file system identifier.
+    /// @param fs_scheme_to_identifier_map Map from URI scheme (like "oss") to the corresponding
+    /// file system identifier.
     /// @return Reference to this builder for method chaining.
-    /// @note If not set, use default file system (configured in `Options::FILE_SYSTEM`).
+    /// @note
+    ///   - This method is intended for environments where multiple file systems are pre-registered.
+    ///   - The specified identifiers must correspond to file systems that have been registered at
+    ///   compile time or initialization.
+    ///   - Cannot be used together with `WithFileSystem()`.
+    ///   - If not set, use default file system (configured in `Options::FILE_SYSTEM`).
+    /// Example:
+    ///   builder.WithFileSystemSchemeToIdentifierMap({{"oss", "jindo"}, {"file", "local"}});
+    ///
     WriteContextBuilder& WithFileSystemSchemeToIdentifierMap(
         const std::map<std::string, std::string>& fs_scheme_to_identifier_map);
 

src/paimon/core/catalog/catalog.cpp

@@ -28,9 +28,10 @@ const char Catalog::SYSTEM_TABLE_SPLITTER[] = "$";
 const char Catalog::DB_SUFFIX[] = ".db";
 const char Catalog::DB_LOCATION_PROP[] = "location";
 
-Result<std::unique_ptr<Catalog>> Catalog::Create(
-    const std::string& root_path, const std::map<std::string, std::string>& options) {
-    PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CoreOptions::FromMap(options));
+Result<std::unique_ptr<Catalog>> Catalog::Create(const std::string& root_path,
+                                                 const std::map<std::string, std::string>& options,
+                                                 const std::shared_ptr<FileSystem>& file_system) {
+    PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CoreOptions::FromMap(options, file_system));
     return std::make_unique<FileSystemCatalog>(core_options.GetFileSystem(), root_path);
 }
 

src/paimon/core/catalog/catalog_test.cpp

@@ -18,6 +18,7 @@
 
 #include "gtest/gtest.h"
 #include "paimon/defs.h"
+#include "paimon/testing/mock/mock_file_system.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
@@ -29,4 +30,13 @@ TEST(CatalogTest, Create) {
     ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create("path", options));
 }
 
+TEST(CatalogTest, CreateWithSpecificFileSystem) {
+    std::map<std::string, std::string> options;
+    const std::string path = "path";
+    const auto fs = std::make_shared<MockFileSystem>();
+    ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(path, options, fs));
+    ASSERT_EQ(path, catalog->GetRootPath());
+    ASSERT_EQ(fs, catalog->GetFileSystem());
+}
+
 }  // namespace paimon::test