fix

Author: lucasfang

docs/source/user_guide/catalog.rst

@@ -162,10 +162,12 @@ already advanced the latest snapshot, the requested snapshot must be published
 under the table path. Recreating the writer restores offsets from the catalog's
 current snapshot.
 
-Both builders use the catalog's file system for manifests and data, preserving
-its object-store credentials. ``WithFileSystem`` overrides it; writers also
-allow ``WithFileSystemSchemeToIdentifierMap`` to override file-system selection.
-For format tables, use ``WriteContextBuilder(FormatTable)`` instead.
+Both builders use the file system of the table being written or committed for
+manifests and data, so a catalog that issues temporary credentials per table
+serves them through ``Catalog::GetTableFileSystem``. ``WithFileSystem`` overrides
+it; writers also allow ``WithFileSystemSchemeToIdentifierMap`` to override
+file-system selection. For format tables, use ``WriteContextBuilder(FormatTable)``
+instead.
 
 The snapshot being committed carries a uuid generated on the client, and the
 commit names the snapshot it is based on by that snapshot's uuid, so the server

include/paimon/write_context.h

@@ -263,11 +263,10 @@ class PAIMON_EXPORT WriteContextBuilder {
     /// snapshot. Use the same catalog and identifier as `CommitContextBuilder` to restore and
     /// refresh real-time progress.
     ///
-    /// A branch is addressed by the identifier, as `tbl$branch_dev`, so that the catalog answers
-    /// for that branch. Its schema is read from the branch directory rather than from the catalog,
-    /// just as a read of that branch reads it.
-    /// Data, manifests and historical metadata remain on the file system. The catalog supplies
-    /// that file system unless overridden by `WithFileSystem()` or
+    /// Only the main branch is supported. Data, manifests and historical metadata remain on the
+    /// file system. The catalog supplies that table's file system - including the per-table
+    /// temporary credentials a catalog that issues them hands out through
+    /// `Catalog::GetTableFileSystem` - unless overridden by `WithFileSystem()` or
     /// `WithFileSystemSchemeToIdentifierMap()`.
     /// @param catalog Non-null catalog, kept alive while the writer uses its snapshot loader.
     /// @param identifier The native table to write to.

src/paimon/core/operation/file_store_commit.cpp

@@ -195,7 +195,13 @@ Result<std::unique_ptr<FileStoreCommit>> FileStoreCommit::Create(
     // Use catalog credentials unless the caller supplied a file system.
     std::shared_ptr<FileSystem> specific_fs = ctx->GetSpecificFileSystem();
     if (specific_fs == nullptr && ctx->GetCatalog() != nullptr) {
-        specific_fs = ctx->GetCatalog()->GetFileSystem();
+        if (!ctx->GetIdentifier()) {
+            return Status::Invalid("a catalog commit requires a table identifier");
+        }
+        // Use the credentials of this table, which a catalog issuing per-table temporary
+        // ones only hands out through GetTableFileSystem.
+        PAIMON_ASSIGN_OR_RAISE(specific_fs,
+                               ctx->GetCatalog()->GetTableFileSystem(ctx->GetIdentifier().value()));
     }
     PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options,
                            CoreOptions::FromMap(ctx->GetOptions(), specific_fs));

src/paimon/core/operation/file_store_commit_test.cpp

@@ -39,13 +39,15 @@
 #include "paimon/core/io/compact_increment.h"
 #include "paimon/core/io/data_increment.h"
 #include "paimon/core/operation/file_store_commit_impl.h"
+#include "paimon/core/schema/table_schema.h"
 #include "paimon/core/table/sink/commit_message_impl.h"
 #include "paimon/core/utils/snapshot_manager.h"
 #include "paimon/defs.h"
 #include "paimon/executor.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/result.h"
+#include "paimon/testing/mock/mock_catalog.h"
 #include "paimon/testing/utils/binary_row_generator.h"
 #include "paimon/testing/utils/testharness.h"
 
@@ -103,6 +105,35 @@ TEST(FileStoreCommitTest, TestCreateWithCatalogRequiresIdentifier) {
                         "a catalog commit requires a table identifier");
 }
 
+TEST(FileStoreCommitTest, TestCatalogCommitUsesPerTableFileSystem) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    const std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar");
+    const Identifier identifier("foo", "bar");
+    const auto logical_schema = arrow::schema(
+        {arrow::field("id", arrow::int64(), false), arrow::field("value", arrow::utf8())});
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> schema,
+        TableSchema::Create(0, logical_schema, /*partition_keys=*/{},
+                            /*primary_keys=*/{}, {{Options::FILE_FORMAT, "parquet"}}));
+    auto catalog = std::make_shared<MockVersionManagedCatalog>();
+    catalog->SetTableSchema(schema);
+    // The catalog-wide file system is a different instance, so a commit that reached for it
+    // instead of the table's own would not be served the credentials the table needs.
+    catalog->SetFileSystem(std::make_shared<LocalFileSystem>());
+    catalog->SetTableFileSystem(dir->GetFileSystem());
+
+    CommitContextBuilder builder(table_path, "commit_user");
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitContext> ctx,
+                         builder.WithCatalog(catalog, identifier).Finish());
+    ASSERT_OK_AND_ASSIGN(auto commit, FileStoreCommit::Create(std::move(ctx)));
+
+    // The file system was resolved per table, not from the catalog-wide one.
+    const std::vector<Identifier>& requests = catalog->TableFileSystemRequests();
+    ASSERT_FALSE(requests.empty());
+    ASSERT_EQ(requests.back().GetFullName(), identifier.GetFullName());
+}
+
 TEST(FileStoreCommitTest, TestAppendDvIndexShouldUseOverwriteCommitKind) {
     auto string_field = arrow::field("f0", arrow::utf8());
     auto int_field = arrow::field("f1", arrow::int32());

src/paimon/core/operation/file_store_write.cpp

@@ -157,7 +157,13 @@ Result<std::unique_ptr<FileStoreWrite>> FileStoreWrite::Create(std::unique_ptr<W
     std::shared_ptr<FileSystem> specific_fs = ctx->GetSpecificFileSystem();
     if (specific_fs == nullptr && ctx->GetFileSystemSchemeToIdentifierMap().empty() &&
         ctx->GetCatalog() != nullptr) {
-        specific_fs = ctx->GetCatalog()->GetFileSystem();
+        if (!ctx->GetIdentifier()) {
+            return Status::Invalid("a catalog write requires a table identifier");
+        }
+        // A catalog issuing per-table temporary credentials only hands them out through
+        // GetTableFileSystem, so the write uses the credentials of this table.
+        PAIMON_ASSIGN_OR_RAISE(specific_fs,
+                               ctx->GetCatalog()->GetTableFileSystem(ctx->GetIdentifier().value()));
     }
     PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options,
                            CoreOptions::FromMap(ctx->GetOptions(), specific_fs,

src/paimon/core/operation/file_store_write_test.cpp

@@ -222,6 +222,37 @@ TEST(FileStoreWriteTest, TestCatalogWriteAsksTheCatalogOnlyForTheMainBranchSchem
     ASSERT_EQ(catalog->LoadTableSchemaIdentifiers().size(), 1u);
 }
 
+TEST(FileStoreWriteTest, TestCatalogWriteUsesPerTableFileSystem) {
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    const std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar");
+    const Identifier identifier("foo", "bar");
+    const auto logical_schema = arrow::schema(
+        {arrow::field("id", arrow::int64(), false), arrow::field("value", arrow::utf8())});
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> schema,
+        TableSchema::Create(0, logical_schema, /*partition_keys=*/{},
+                            /*primary_keys=*/{}, {{Options::FILE_FORMAT, "parquet"}}));
+    auto catalog = std::make_shared<MockVersionManagedCatalog>();
+    catalog->SetTableSchema(schema);
+    // The catalog-wide file system is a different instance, so a write that reached for it
+    // instead of the table's own would not be served the credentials the table needs.
+    catalog->SetFileSystem(std::make_shared<LocalFileSystem>());
+    catalog->SetTableFileSystem(dir->GetFileSystem());
+
+    WriteContextBuilder builder(table_path, "writer");
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> ctx,
+                         builder.WithCatalog(catalog, identifier).Finish());
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
+                         FileStoreWrite::Create(std::move(ctx)));
+    ASSERT_OK(writer->Close());
+
+    // The file system was resolved per table, not from the catalog-wide one.
+    const std::vector<Identifier>& requests = catalog->TableFileSystemRequests();
+    ASSERT_FALSE(requests.empty());
+    ASSERT_EQ(requests.back().GetFullName(), identifier.GetFullName());
+}
+
 TEST(FileStoreWriteTest, TestCreateWriterForLoadedMapBlobTable) {
     auto dir = UniqueTestDirectory::Create();
     std::string table_path = PathUtil::JoinPath(dir->Str(), "foo.db/bar");

src/paimon/testing/mock/mock_catalog.h

@@ -227,6 +227,27 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog {
     std::shared_ptr<FileSystem> GetFileSystem() const override {
         return file_system_;
     }
+
+    /// Records what was asked and serves the per-table file system when one was set, so a
+    /// test can tell it apart from the catalog-wide one; falls back to `GetFileSystem()`,
+    /// matching the base default, when none was set.
+    Result<std::shared_ptr<FileSystem>> GetTableFileSystem(
+        const Identifier& identifier) const override {
+        table_file_system_requests_.push_back(identifier);
+        if (table_file_system_ != nullptr) {
+            return table_file_system_;
+        }
+        return GetFileSystem();
+    }
+
+    void SetTableFileSystem(const std::shared_ptr<FileSystem>& file_system) {
+        table_file_system_ = file_system;
+    }
+
+    const std::vector<Identifier>& TableFileSystemRequests() const {
+        return table_file_system_requests_;
+    }
+
     const std::map<std::string, std::string>& GetOptions() const override {
         return options_;
     }
@@ -244,6 +265,8 @@ class MockVersionManagedCatalog : public Catalog, public VersionManagedCatalog {
     std::string table_uuid_;
     std::shared_ptr<Schema> table_schema_;
     std::shared_ptr<FileSystem> file_system_;
+    std::shared_ptr<FileSystem> table_file_system_;
+    mutable std::vector<Identifier> table_file_system_requests_;
     bool supports_version_management_ = true;
     std::function<void()> on_commit_;
     bool check_table_uuid_ = false;