[rest] Add table commit leases with unified catalog locks

docs/docs/concepts/concurrency-control.md

@@ -101,6 +101,58 @@ which is stopped when the scope ends, so blocked renewals do not occupy other sc
 threads. Filesystem locks still depend on the storage's atomic publication behavior; they do not
 provide a storage fencing token.
 
+## REST Catalog Commit Leases
+
+Catalog-managed Paimon tables in the REST catalog expose table-bound commit leases through
+`Catalog.createLock` and the unified
+`Lock.acquire` / `Lock.runWithLock` contract. Set the persisted table property
+`rest.commit.lock-enabled = true` to enable server-side lease enforcement. Independently, set
+`rest.commit.acquire-lock = true` in a writer's dynamic options to acquire a lease for each
+publication attempt. Both options default to `false`. Enabling the table capability alone does
+not make clients acquire leases. Requesting acquisition without the table capability fails.
+External tables keep filesystem publication.
+Callers supply their exact `commitUser` and table UUID.
+Leases require the target branch's catalog-resolved UUID. Reload a view switched to another branch
+from the catalog before using leases.
+Lock contention waits until `commit.timeout` when configured, independently of publication retries.
+Scopes share a timer and a cached worker pool. The timer dispatches RPCs to workers so a blocked
+renewal cannot block other scopes' scheduling. Each scope permits only one in-flight renewal and
+cancels its own scheduled and in-flight renewal tasks when it ends, without stopping the shared
+executors. Timer and worker threads exit after one minute of inactivity and restart when needed.
+The timer stays alive while renewal tasks remain scheduled. The client validates the lease's
+monotonic deadline before publishing. A failed operation stops renewal and its lease expires
+automatically. The server keeps the lease duration fixed for each scope; duration changes take effect
+on the next acquisition.
+
+The server must validate the current table incarnation, branch, authenticated caller and exact
+owner, and atomically publish the snapshot and clear the lease. All snapshot writers must honor
+active leases, including writers that do not request one. Each successful snapshot publication
+ends its lease; a later publication must acquire another scope. Snapshot comparison and file
+conflict validation remain necessary.
+
+For a table with lease enforcement disabled, the server can skip lease storage access entirely.
+For a table with enforcement enabled, even writers with `rest.commit.acquire-lock = false` must
+respect another writer's active lease. The server uses its own persisted table property for this
+decision, rather than client options. Capability changes must be serialized with publication and
+lease acquisition; disabling enforcement must not bypass an active lease.
+
+Automatic conflict rollback always supplies the runtime branch and writer's `commitUser`, together
+with the expected head's `fromSnapshot` and `fromSnapshotUuid`. Snapshot IDs can be reused after a
+rollback, so the server must check both fields against the current head inside the mutation
+transaction, before changing partitions or deleting files. Automatic rollback skips legacy snapshots
+without a UUID. A known table or branch UUID is supplied independently of writer identity.
+
+`commitUser` identifies the writer; it does not require a lease to exist. If a live lease exists, the
+writer and authenticated caller must match its owner. Without a live lease, conditional optimistic
+rollback is allowed. The server must validate admission and the expected head, roll back and clear
+the lease atomically. Manual rollback can omit writer identity, but cannot change a leased head.
+The table UUID is omitted only when the target table or branch identity is unknown.
+
+This is a commit admission lease, not a generation or fencing token. Requests with the same
+commit user and authenticated caller share ownership. Use a unique commit user per logical writer
+and preserve it during recovery. A server without the lease protocol rejects acquisition rather
+than silently proceeding without protection.
+
 ## Files conflict
 
 A writer validates file-level changes as well as the snapshot ID. For example, if two compactors

docs/generated/core_configuration.html

@@ -1470,6 +1470,18 @@
             <td>String</td>
             <td>Time field for record level expire. It supports the following types: `timestamps in seconds with INT`,`timestamps in seconds with BIGINT`, `timestamps in milliseconds with BIGINT` or `timestamp`.</td>
         </tr>
+        <tr>
+            <td><h5>rest.commit.acquire-lock</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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 server-side table.</td>
+        </tr>
+        <tr>
+            <td><h5>rest.commit.lock-enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Enable the REST catalog's server-side table commit lease capability. Writers acquire a lease only when rest.commit.acquire-lock is enabled.</td>
+        </tr>
         <tr>
             <td><h5>row-tracking.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

docs/static/rest-catalog-open-api.yaml

@@ -595,6 +595,12 @@ paths:
           required: true
           schema:
             type: string
+      description: >-
+        Validate the table incarnation, branch and expected head atomically with publication.
+        Every writer must honor an active commit lease, including writers that do not request
+        one. The authenticated caller and snapshot.commitUser must match the active owner.
+        Clear the lease only when a new snapshot is successfully published; rejected requests
+        and idempotent replays of an existing snapshot must not clear a newer lease.
       requestBody:
         content:
           application/json:
@@ -615,6 +621,112 @@ paths:
           $ref: '#/components/responses/TableNotExistErrorResponse'
         "500":
           $ref: '#/components/responses/ServerErrorResponse'
+  /v1/{prefix}/databases/{database}/tables/{table}/commit-lock:
+    post:
+      tags:
+        - table
+      summary: Acquire a table commit lease
+      operationId: acquireCommitLock
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: database
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: table
+          in: path
+          required: true
+          schema:
+            type: string
+      description: >-
+        Requires the server-side table property rest.commit.lock-enabled=true and the same
+        permission as committing snapshots. Clients opt into acquisition separately using
+        rest.commit.acquire-lock=true; that client option does not enable the table capability.
+        Ownership is the exact commitUser together with the authenticated caller. Contention
+        returns HTTP 200 with acquired=false. A grant includes the authoritative snapshot head.
+        Renewal never revives an expired lease. Publication retains snapshot CAS validation
+        and atomically releases the lease on a new successful commit. A rejected publication
+        does not release the lease. A replay of an already-published snapshot must not release
+        a newer lease. The lease duration is fixed for each scope. There is no unlock RPC.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CommitLockRequest'
+      responses:
+        "200":
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CommitLockResponse'
+        "400":
+          $ref: '#/components/responses/BadRequestErrorResponse'
+        "401":
+          $ref: '#/components/responses/UnauthorizedErrorResponse'
+        "403":
+          $ref: '#/components/responses/ForbiddenErrorResponse'
+        "404":
+          $ref: '#/components/responses/TableNotExistErrorResponse'
+        "500":
+          $ref: '#/components/responses/ServerErrorResponse'
+  /v1/{prefix}/databases/{database}/tables/{table}/commit-lock/renew:
+    post:
+      tags:
+        - table
+      summary: Renew a table commit lease
+      operationId: renewCommitLock
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: database
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: table
+          in: path
+          required: true
+          schema:
+            type: string
+      description: >-
+        Requires the same permission as committing snapshots. Validate the table incarnation,
+        branch, authenticated caller and exact commitUser against the unexpired lease.
+        Return acquired=false when ownership is lost or the lease has expired; renewal must
+        never acquire a new lease or revive an expired one. Keep leaseMillis fixed within
+        the scope. Each successful new snapshot publication ends its lease.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CommitLockRequest'
+      responses:
+        "200":
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CommitLockResponse'
+        "400":
+          $ref: '#/components/responses/BadRequestErrorResponse'
+        "401":
+          $ref: '#/components/responses/UnauthorizedErrorResponse'
+        "403":
+          $ref: '#/components/responses/ForbiddenErrorResponse'
+        "404":
+          $ref: '#/components/responses/TableNotExistErrorResponse'
+        "500":
+          $ref: '#/components/responses/ServerErrorResponse'
   /v1/{prefix}/databases/{database}/tables/{table}/rollback:
     post:
       tags:
@@ -3499,6 +3611,39 @@ components:
           $ref: '#/components/schemas/Identifier'
         destination:
           $ref: '#/components/schemas/Identifier'
+    CommitLockRequest:
+      type: object
+      required: [tableId, commitUser]
+      properties:
+        tableId:
+          type: string
+          description: Stable resolved table or branch UUID; a recreated table cannot inherit a lease.
+        commitUser:
+          type: string
+          minLength: 1
+          maxLength: 255
+          description: Exact commitUser that will appear in the submitted snapshot.
+    CommitLockResponse:
+      type: object
+      required: [acquired, expiresAtMillis, leaseMillis]
+      properties:
+        acquired:
+          type: boolean
+        commitUser:
+          type: [string, "null"]
+        expiresAtMillis:
+          type: integer
+          format: int64
+          description: Server database clock expiration; zero when acquired is false.
+        leaseMillis:
+          type: integer
+          format: int64
+          description: Lease duration in milliseconds, fixed within a scope; zero when acquired is false.
+        snapshot:
+          anyOf:
+            - $ref: '#/components/schemas/Snapshot'
+            - type: "null"
+          description: Authoritative head at grant time; null for an empty table.
     CommitTableRequest:
       type: object
       properties:
@@ -3516,21 +3661,32 @@ components:
       type: object
       description: >-
         tableId identifies the expected table incarnation independently of lock ownership.
-        For rollback within a writer's commit lock, also supply commitUser and fromSnapshot so
-        the server can validate the expected head and owner. Omit commitUser when no writer
-        lock is used, and omit tableId only when its UUID is unavailable.
+        Writer rollback supplies commitUser, fromSnapshot and fromSnapshotUuid regardless of
+        whether the writer acquired a lease. commitUser identifies the writer, not a requirement
+        for a lease to exist. A live lease requires the exact owner and authenticated caller;
+        without a live lease, conditional optimistic rollback is allowed. Snapshot IDs can be
+        reused after rollback, so both expected head fields must be checked against the current
+        head inside the mutation transaction before partition changes or file cleanup. The server
+        must validate table incarnation, admission and expected head, then roll back and clear
+        the lease atomically. Manual rollback may omit writer identity, but cannot change a
+        leased head. Omit tableId only when its UUID is unavailable.
       properties:
         instant:
           $ref: '#/components/schemas/Instant'
         fromSnapshot:
           type: [ integer, "null" ]
           format: int64
+          description: Expected latest snapshot ID, required for writer rollback.
+        fromSnapshotUuid:
+          type: [ string, "null" ]
+          minLength: 1
+          description: Expected latest snapshot UUID; requires fromSnapshot and is required for writer rollback.
         tableId:
           type: [ string, "null" ]
           description: Expected table or branch UUID, independently of commitUser.
         commitUser:
           type: [ string, "null" ]
-          description: Exact owner of the writer's lock.
+          description: Writer identity for admission against any active lease; requires both expected head fields.
     RollbackSchemaRequest:
       type: object
       required:

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -973,6 +973,23 @@ public InlineElement getDescription() {
                             "Specify how to initialize the next sequence number for primary key "
                                     + "table writers.");
 
+    public static final ConfigOption<Boolean> REST_COMMIT_LOCK_ENABLED =
+            key("rest.commit.lock-enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Enable the REST catalog's server-side table commit lease capability. "
+                                    + "Writers acquire a lease only when rest.commit.acquire-lock is enabled.");
+
+    public static final ConfigOption<Boolean> REST_COMMIT_ACQUIRE_LOCK =
+            key("rest.commit.acquire-lock")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "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 server-side table.");
+
     public static final ConfigOption<Duration> COMMIT_TIMEOUT =
             key("commit.timeout")
                     .durationType()
@@ -3900,6 +3917,14 @@ public SequenceNumberInitMode writeSequenceNumberInitMode() {
         return options.get(WRITE_SEQUENCE_NUMBER_INIT_MODE);
     }
 
+    public boolean restCommitLockEnabled() {
+        return options.get(REST_COMMIT_LOCK_ENABLED);
+    }
+
+    public boolean restCommitAcquireLock() {
+        return options.get(REST_COMMIT_ACQUIRE_LOCK);
+    }
+
     public long commitTimeout() {
         return options.get(COMMIT_TIMEOUT) == null
                 ? Long.MAX_VALUE

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -45,6 +45,7 @@
 import org.apache.paimon.rest.requests.AlterTableRequest;
 import org.apache.paimon.rest.requests.AlterViewRequest;
 import org.apache.paimon.rest.requests.AuthTableQueryRequest;
+import org.apache.paimon.rest.requests.CommitLockRequest;
 import org.apache.paimon.rest.requests.CommitTableRequest;
 import org.apache.paimon.rest.requests.CreateBranchRequest;
 import org.apache.paimon.rest.requests.CreateDatabaseRequest;
@@ -72,6 +73,7 @@
 import org.apache.paimon.rest.requests.UpsertSemanticViewRequest;
 import org.apache.paimon.rest.responses.AlterDatabaseResponse;
 import org.apache.paimon.rest.responses.AuthTableQueryResponse;
+import org.apache.paimon.rest.responses.CommitLockResponse;
 import org.apache.paimon.rest.responses.CommitTableResponse;
 import org.apache.paimon.rest.responses.ConfigResponse;
 import org.apache.paimon.rest.responses.CreatePartitionsResponse;
@@ -734,6 +736,25 @@ public boolean commitSnapshot(
         return response.isSuccess();
     }
 
+    public CommitLockResponse acquireCommitLock(
+            Identifier identifier, String tableUuid, String commitUser) {
+        return client.post(
+                resourcePaths.commitLock(identifier.getDatabaseName(), identifier.getObjectName()),
+                new CommitLockRequest(tableUuid, commitUser),
+                CommitLockResponse.class,
+                restAuthFunction);
+    }
+
+    public CommitLockResponse renewCommitLock(
+            Identifier identifier, String tableUuid, String commitUser) {
+        return client.post(
+                resourcePaths.commitLock(identifier.getDatabaseName(), identifier.getObjectName())
+                        + "/renew",
+                new CommitLockRequest(tableUuid, commitUser),
+                CommitLockResponse.class,
+                restAuthFunction);
+    }
+
     /**
      * Rollback instant for table.
      *
@@ -742,7 +763,8 @@ public boolean commitSnapshot(
      * @param instant instant to rollback
      * @param fromSnapshot snapshot from, success only occurs when the latest snapshot is this
      *     snapshot.
-     * @param commitUser owner of the writer lock, or null without a writer lock
+     * @param fromSnapshotUuid expected snapshot UUID, checked together with fromSnapshot
+     * @param commitUser writer identity for active-lock admission, or null without a writer
      * @throws NoSuchResourceException Exception thrown on HTTP 404 means the table or the snapshot
      *     or the tag not exists
      * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for
@@ -753,9 +775,11 @@ public void rollbackTo(
             @Nullable String tableUuid,
             Instant instant,
             @Nullable Long fromSnapshot,
+            @Nullable String fromSnapshotUuid,
             @Nullable String commitUser) {
         RollbackTableRequest request =
-                new RollbackTableRequest(instant, fromSnapshot, tableUuid, commitUser);
+                new RollbackTableRequest(
+                        instant, fromSnapshot, fromSnapshotUuid, tableUuid, commitUser);
         client.post(
                 resourcePaths.rollbackTable(
                         identifier.getDatabaseName(), identifier.getObjectName()),

paimon-api/src/main/java/org/apache/paimon/rest/ResourcePaths.java

@@ -196,6 +196,10 @@ public String commitTable(String databaseName, String objectName) {
                 "commit");
     }
 
+    public String commitLock(String databaseName, String objectName) {
+        return table(databaseName, objectName) + "/commit-lock";
+    }
+
     public String rollbackTable(String databaseName, String objectName) {
         return SLASH.join(
                 V1,