feat(rest): support table commit leases

docs/source/user_guide/catalog.rst

@@ -197,6 +197,58 @@ can tell a commit that raced another. A commit which lost such a race is
 rebased on the new latest snapshot and retried, bounded by ``commit.max-retries``
 and ``commit.timeout`` as a file-system commit is.
 
+A REST catalog can also give a writer a commit lease, so concurrent writers take
+turns instead of racing. The table property ``rest.commit.lock-enabled = true``
+turns on lease enforcement on the server, and a writer that sets
+``rest.commit.acquire-lock = true``, usually as a commit context option, acquires
+a lease for each publication attempt. Both default to ``false``; enabling the
+table property alone does not make writers acquire, and asking to acquire without
+it fails when the commit is created. A writer with an active lease keeps the
+others out, including writers that do not acquire one. The lease is bound to the
+table or branch, to the table UUID, which has to be passed to ``WithTableId``,
+and to the exact commit user, so give each logical writer its own commit user and
+keep it across recovery.
+
+.. code-block:: cpp
+
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::Table> table,
+                          shared_catalog->GetTable(paimon::Identifier("db", "tbl")));
+   std::optional<std::string> table_uuid = table->CatalogUuid();
+   if (!table_uuid) {
+       return paimon::Status::Invalid("a commit lease needs the table UUID");
+   }
+   paimon::CommitContextBuilder builder(table_path, "writer-1");
+   builder.WithCatalog(shared_catalog, paimon::Identifier("db", "tbl"))
+       .WithTableId(table_uuid.value())
+       .AddOption(paimon::Options::REST_COMMIT_ACQUIRE_LOCK, "true");
+
+Each attempt acquires its own lease before reading the latest snapshot, waits
+while another writer holds it until ``commit.timeout`` without using up
+``commit.max-retries``, renews it in the background and checks it is still
+valid right before publishing. An attempt that fails to acquire the lease, or
+loses it before publishing, publishes nothing and is retried within
+``commit.max-retries`` and ``commit.timeout``, as after a conflict; an attempt
+that lost its lease also removes the manifests it wrote. When the retries run
+out, the error names the lease failure rather than a conflict.
+``RollbackToAsLatest`` holds a lease the same way but is not retried: failing to
+acquire or losing the lease fails it.
+
+``commit.timeout`` is not a strict deadline for waiting on the lease. Each
+acquisition may wait the full timeout, however much of it earlier attempts of
+the same commit used. The timeout is also checked only after each acquisition
+request returns, without interrupting a request in flight, so the wait can run
+past it by the duration of one request, which only the HTTP client's own
+timeouts bound.
+
+There is no unlock request. The server ends a lease when a snapshot is published
+under it. Ending an attempt that published nothing only stops renewing the
+lease, without waiting for a renewal in flight, and the lease stays held until
+it expires.
+
+Leases need ``WithCatalog``: a commit that only builds the commit table request
+for the caller to send cannot hold one, so asking to acquire is rejected there.
+The snapshot UUID checks above still apply.
+
 The HTTP client neither retries nor follows redirects for a commit request.
 Transport errors and non-success responses, including 429, 503 and redirects,
 leave the outcome uncertain: the server may already have accepted the snapshot.

include/paimon/defs.h

@@ -314,6 +314,16 @@ struct PAIMON_EXPORT Options {
     /// Default value is "false".
     static const char COMMIT_DISCARD_DUPLICATE_FILES[];
 
+    /// "rest.commit.lock-enabled" - Enable the REST catalog's server-side table commit lease
+    /// capability. Writers acquire a lease only when "rest.commit.acquire-lock" is enabled.
+    /// Default value is "false".
+    static const char REST_COMMIT_LOCK_ENABLED[];
+
+    /// "rest.commit.acquire-lock" - Whether this writer acquires a REST commit lease for each
+    /// publication attempt. Can be set dynamically for a writer and requires
+    /// "rest.commit.lock-enabled" on the table. Default value is "false".
+    static const char REST_COMMIT_ACQUIRE_LOCK[];
+
     /// "compaction.max-size-amplification-percent" - The size amplification is defined as the
     /// amount (in percentage) of additional storage needed to store a single byte of data in the
     /// merge tree for changelog mode table. Default value is 200.

src/paimon/CMakeLists.txt

@@ -514,6 +514,7 @@ if(PAIMON_ENABLE_REST)
          rest/rest_auth.cpp
          rest/dlf_auth.cpp
          rest/rest_catalog.cpp
+         rest/rest_commit_lock.cpp
          rest/rest_messages.cpp
          rest/rest_token_file_system.cpp
          rest/rest_credential_provider.cpp
@@ -1100,6 +1101,7 @@ if(PAIMON_BUILD_TESTS)
                         rest/resource_paths_test.cpp
                         rest/dlf_auth_test.cpp
                         rest/rest_catalog_test.cpp
+                        rest/rest_commit_lock_test.cpp
                         rest/rest_messages_test.cpp
                         rest/rest_token_file_system_test.cpp
                         rest/rest_credential_provider_test.cpp

src/paimon/common/defs.cpp

@@ -89,6 +89,8 @@ const char Options::COMMIT_MAX_RETRIES[] = "commit.max-retries";
 const char Options::COMMIT_MIN_RETRY_WAIT[] = "commit.min-retry-wait";
 const char Options::COMMIT_MAX_RETRY_WAIT[] = "commit.max-retry-wait";
 const char Options::COMMIT_DISCARD_DUPLICATE_FILES[] = "commit.discard-duplicate-files";
+const char Options::REST_COMMIT_LOCK_ENABLED[] = "rest.commit.lock-enabled";
+const char Options::REST_COMMIT_ACQUIRE_LOCK[] = "rest.commit.acquire-lock";
 const char Options::SEQUENCE_FIELD[] = "sequence.field";
 const char Options::SEQUENCE_FIELD_SORT_ORDER[] = "sequence.field.sort-order";
 const char Options::MERGE_ENGINE[] = "merge-engine";

src/paimon/core/catalog/catalog_snapshot_commit.h

@@ -27,6 +27,7 @@
 #include "paimon/catalog/catalog.h"
 #include "paimon/catalog/identifier.h"
 #include "paimon/core/catalog/catalog_utils.h"
+#include "paimon/core/catalog/commit_lock.h"
 #include "paimon/core/catalog/commit_table_request.h"
 #include "paimon/core/catalog/snapshot_commit.h"
 #include "paimon/core/catalog/version_managed_catalog.h"
@@ -45,17 +46,27 @@ class CatalogSnapshotCommit : public SnapshotCommit {
     /// @param catalog Catalog implementing version management.
     /// @param identifier Identifier of the table to commit to.
     /// @param table_id Catalog UUID captured before preparing changes; null if unavailable.
+    /// @param lock Lock taken around each publication attempt; null to publish without one.
     CatalogSnapshotCommit(const std::shared_ptr<Catalog>& catalog, Identifier identifier,
-                          const std::optional<std::string>& table_id)
+                          const std::optional<std::string>& table_id,
+                          std::unique_ptr<CommitLock> lock = nullptr)
         : catalog_(catalog),
           version_managed_catalog_(AsVersionManaged(catalog)),
           identifier_(std::move(identifier)),
-          table_id_(table_id) {}
+          table_id_(table_id),
+          lock_(std::move(lock)) {}
 
     /// @param table_id Catalog UUID to include in the request; null if unavailable.
     explicit CatalogSnapshotCommit(const std::optional<std::string>& table_id = std::nullopt)
         : identifier_("", ""), table_id_(table_id) {}
 
+    Result<std::unique_ptr<CommitLockScope>> AcquireLock() override {
+        if (lock_ == nullptr) {
+            return std::unique_ptr<CommitLockScope>();
+        }
+        return lock_->Acquire();
+    }
+
     Result<bool> Commit(const std::optional<std::string>& base_snapshot_uuid,
                         const Snapshot& snapshot, const std::string& branch,
                         const std::vector<PartitionStatistics>& statistics) override {
@@ -110,6 +121,7 @@ class CatalogSnapshotCommit : public SnapshotCommit {
     VersionManagedCatalog* version_managed_catalog_ = nullptr;
     Identifier identifier_;
     std::optional<std::string> table_id_;
+    std::unique_ptr<CommitLock> lock_;
     std::optional<CommitTableRequest> commit_table_request_;
 };
 

src/paimon/core/catalog/catalog_snapshot_commit_test.cpp

@@ -26,6 +26,7 @@
 
 #include "gtest/gtest.h"
 #include "paimon/catalog/identifier.h"
+#include "paimon/core/catalog/commit_lock.h"
 #include "paimon/core/catalog/commit_table_request.h"
 #include "paimon/core/partition/partition_statistics.h"
 #include "paimon/result.h"
@@ -191,4 +192,47 @@ TEST(CatalogSnapshotCommitTest, TestBuildRequestWithoutCatalog) {
     ASSERT_EQ(bare_request.GetBaseSnapshotUuid(), std::nullopt);
 }
 
+TEST(CatalogSnapshotCommitTest, TestAcquireLockUsesTheLockItWasGiven) {
+    class CountingLock : public CommitLock {
+     public:
+        explicit CountingLock(int32_t* acquisitions) : acquisitions_(acquisitions) {}
+
+        Result<std::unique_ptr<CommitLockScope>> Acquire() override {
+            ++*acquisitions_;
+            if (*acquisitions_ > 1) {
+                return Status::IOError("lock service unavailable");
+            }
+            return std::unique_ptr<CommitLockScope>(std::make_unique<Scope>());
+        }
+
+     private:
+        class Scope : public CommitLockScope {
+         public:
+            Status EnsureValid() override {
+                return Status::OK();
+            }
+        };
+
+        int32_t* acquisitions_;
+    };
+
+    auto catalog = std::make_shared<MockVersionManagedCatalog>(true);
+    CatalogSnapshotCommit unlocked(catalog, Identifier("db", "tbl"), "table-uuid");
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CommitLockScope> scope, unlocked.AcquireLock());
+    ASSERT_FALSE(scope);
+    CatalogSnapshotCommit request_only("table-uuid");
+    ASSERT_OK_AND_ASSIGN(scope, request_only.AcquireLock());
+    ASSERT_FALSE(scope);
+
+    int32_t acquisitions = 0;
+    CatalogSnapshotCommit locked(catalog, Identifier("db", "tbl"), "table-uuid",
+                                 std::make_unique<CountingLock>(&acquisitions));
+    ASSERT_OK_AND_ASSIGN(scope, locked.AcquireLock());
+    ASSERT_TRUE(scope);
+    ASSERT_OK(scope->EnsureValid());
+    ASSERT_NOK_WITH_MSG(locked.AcquireLock(), "lock service unavailable");
+    ASSERT_EQ(2, acquisitions);
+    ASSERT_TRUE(catalog->CommitCalls().empty());
+}
+
 }  // namespace paimon::test

src/paimon/core/catalog/commit_lock.h

@@ -0,0 +1,49 @@
+/*
+ * 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 "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+/// One acquisition of a `CommitLock`, held for a single snapshot publication attempt. Destroying
+/// the scope ends it.
+class CommitLockScope {
+ public:
+    virtual ~CommitLockScope() = default;
+
+    /// Returns an error once the scope can no longer protect a publication, for example after its
+    /// lease expired. Called right before the snapshot is published.
+    virtual Status EnsureValid() = 0;
+};
+
+/// A lock taken around each snapshot publication attempt, mirroring Java
+/// `org.apache.paimon.operation.Lock`.
+class CommitLock {
+ public:
+    virtual ~CommitLock() = default;
+
+    /// Acquires a scope for one publication attempt, waiting while another writer holds the lock.
+    virtual Result<std::unique_ptr<CommitLockScope>> Acquire() = 0;
+};
+
+}  // namespace paimon

src/paimon/core/catalog/snapshot_commit.h

@@ -18,10 +18,12 @@
 
 #pragma once
 
+#include <memory>
 #include <optional>
 #include <string>
 #include <vector>
 
+#include "paimon/core/catalog/commit_lock.h"
 #include "paimon/core/partition/partition_statistics.h"
 #include "paimon/core/snapshot.h"
 
@@ -32,6 +34,14 @@ class SnapshotCommit {
  public:
     virtual ~SnapshotCommit() = default;
 
+    /// Acquires the lock scope of one publication attempt. The caller holds it from reading the
+    /// base snapshot until `Commit()` returns, and checks it right before committing.
+    ///
+    /// @return The scope, or null when publication takes no lock.
+    virtual Result<std::unique_ptr<CommitLockScope>> AcquireLock() {
+        return std::unique_ptr<CommitLockScope>();
+    }
+
     /// @param base_snapshot_uuid Base snapshot UUID; null for an absent or legacy snapshot.
     /// @param snapshot Snapshot to be committed.
     /// @param branch Branch to publish the snapshot on; an empty name is the main branch.

src/paimon/core/catalog/version_managed_catalog.h

@@ -25,12 +25,14 @@
 
 #include "paimon/catalog/catalog.h"
 #include "paimon/catalog/identifier.h"
+#include "paimon/core/catalog/commit_lock.h"
 #include "paimon/core/partition/partition_statistics.h"
 #include "paimon/core/snapshot.h"
 #include "paimon/result.h"
 #include "paimon/status.h"
 
 namespace paimon {
+class CoreOptions;
 
 /// Snapshot publication interface for catalogs returning true from `SupportsVersionManagement()`.
 ///
@@ -59,6 +61,20 @@ class VersionManagedCatalog {
                                         const std::optional<std::string>& base_snapshot_uuid,
                                         const Snapshot& snapshot,
                                         const std::vector<PartitionStatistics>& statistics) = 0;
+
+    /// Creates the lock taken around each snapshot publication to a table or branch, mirroring
+    /// Java `Catalog.createLock`.
+    ///
+    /// @param identifier Identifier of the table or branch the snapshots are published to.
+    /// @param table_uuid Catalog UUID of the table or branch; null if unavailable.
+    /// @param commit_user Commit user of the snapshots published under the lock.
+    /// @param options Table options merged with the writer's dynamic options.
+    /// @return The lock, or null when publication takes none.
+    virtual Result<std::unique_ptr<CommitLock>> CreateCommitLock(
+        const Identifier& identifier, const std::optional<std::string>& table_uuid,
+        const std::string& commit_user, const CoreOptions& options) {
+        return std::unique_ptr<CommitLock>();
+    }
 };
 
 /// Returns version management only when the catalog both advertises and implements it.

src/paimon/core/core_options.cpp

@@ -474,6 +474,8 @@ struct CoreOptions::Impl {
     bool global_index_enabled = true;
     bool commit_force_compact = false;
     bool commit_discard_duplicate_files = false;
+    bool rest_commit_lock_enabled = false;
+    bool rest_commit_acquire_lock = false;
     bool dynamic_partition_overwrite = true;
     bool overwrite_upgrade = true;
     bool compaction_force_rewrite_all_files = false;
@@ -704,6 +706,12 @@ struct CoreOptions::Impl {
         // Parse commit.discard-duplicate-files - whether to discard duplicate files on append
         PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::COMMIT_DISCARD_DUPLICATE_FILES,
                                                 &commit_discard_duplicate_files));
+        // Parse rest.commit.lock-enabled - server-side commit lease capability of the table
+        PAIMON_RETURN_NOT_OK(
+            parser.Parse<bool>(Options::REST_COMMIT_LOCK_ENABLED, &rest_commit_lock_enabled));
+        // Parse rest.commit.acquire-lock - whether this writer acquires a commit lease
+        PAIMON_RETURN_NOT_OK(
+            parser.Parse<bool>(Options::REST_COMMIT_ACQUIRE_LOCK, &rest_commit_acquire_lock));
         // Parse dynamic-partition-overwrite - whether overwrite only dynamic partitions
         // for partitioned table overwrite.
         PAIMON_RETURN_NOT_OK(
@@ -1343,6 +1351,14 @@ bool CoreOptions::CommitDiscardDuplicateFiles() const {
     return impl_->commit_discard_duplicate_files;
 }
 
+bool CoreOptions::RestCommitLockEnabled() const {
+    return impl_->rest_commit_lock_enabled;
+}
+
+bool CoreOptions::RestCommitAcquireLock() const {
+    return impl_->rest_commit_acquire_lock;
+}
+
 bool CoreOptions::DynamicPartitionOverwrite() const {
     return impl_->dynamic_partition_overwrite;
 }

src/paimon/core/core_options.h

@@ -142,6 +142,8 @@ class PAIMON_EXPORT CoreOptions {
     int64_t GetCommitMinRetryWait() const;
     int64_t GetCommitMaxRetryWait() const;
     bool CommitDiscardDuplicateFiles() const;
+    bool RestCommitLockEnabled() const;
+    bool RestCommitAcquireLock() const;
     bool DynamicPartitionOverwrite() const;
     bool OverwriteUpgrade() const;
     int32_t GetCompactionMinFileNum() const;

src/paimon/core/core_options_test.cpp

@@ -107,6 +107,8 @@ TEST(CoreOptionsTest, TestDefaultValue) {
     ASSERT_EQ(10, core_options.GetCommitMinRetryWait());
     ASSERT_EQ(10 * 1000, core_options.GetCommitMaxRetryWait());
     ASSERT_FALSE(core_options.CommitDiscardDuplicateFiles());
+    ASSERT_FALSE(core_options.RestCommitLockEnabled());
+    ASSERT_FALSE(core_options.RestCommitAcquireLock());
     ExpireConfig expire_config = core_options.GetExpireConfig();
     ASSERT_EQ(10, expire_config.GetSnapshotRetainMin());
     ASSERT_EQ(std::numeric_limits<int32_t>::max(), expire_config.GetSnapshotRetainMax());
@@ -244,6 +246,8 @@ TEST(CoreOptionsTest, TestFromMap) {
         {Options::COMMIT_MIN_RETRY_WAIT, "5ms"},
         {Options::COMMIT_MAX_RETRY_WAIT, "3s"},
         {Options::COMMIT_DISCARD_DUPLICATE_FILES, "true"},
+        {Options::REST_COMMIT_LOCK_ENABLED, "true"},
+        {Options::REST_COMMIT_ACQUIRE_LOCK, "true"},
         {Options::DYNAMIC_PARTITION_OVERWRITE, "false"},
         {Options::OVERWRITE_UPGRADE, "false"},
         {Options::SCAN_SNAPSHOT_ID, "5"},
@@ -398,6 +402,8 @@ TEST(CoreOptionsTest, TestFromMap) {
     ASSERT_EQ(5, core_options.GetCommitMinRetryWait());
     ASSERT_EQ(3 * 1000, core_options.GetCommitMaxRetryWait());
     ASSERT_TRUE(core_options.CommitDiscardDuplicateFiles());
+    ASSERT_TRUE(core_options.RestCommitLockEnabled());
+    ASSERT_TRUE(core_options.RestCommitAcquireLock());
     ASSERT_EQ(5, core_options.GetScanSnapshotId().value_or(-1));
     ASSERT_EQ(7, core_options.GetScanManifestEntryCacheMaxSnapshots());
     ASSERT_FALSE(core_options.ScanManifestEntryLazyDecodeEnabled());