[core] Unify catalog lock interfaces and snapshot commit scopes

docs/docs/concepts/concurrency-control.md

@@ -72,6 +72,35 @@ The commit mechanism depends on the catalog and storage:
 All writers of the same table must use a compatible commit mechanism and shared locking
 configuration. See [Catalog](./catalog) when choosing the metadata backend.
 
+## Commit Locks
+
+Configured catalog locks protect every snapshot publication attempt, including the first attempt.
+The writer acquires the lock before reading the latest snapshot, validating changes and preparing
+manifests. It checks that the lock is still valid immediately before publishing. Each snapshot has
+its own scope: APPEND and COMPACT snapshots produced by the same commit acquire separate locks.
+Post-commit callbacks run after the scope ends. Metadata-only manifest replacement validates the
+caller's prepared layout against the current head while holding the lock.
+
+`Catalog.createLock` supplies a table-bound `Lock` with one `runWithLock` contract. `CatalogLockFactory`
+binds the table, UUID and writer identity when creating a `Lock`.
+`SnapshotCommit` owns this lock together with its publisher resources. Its `commit` method always
+uses the bound lock, including direct publication. Writers use `SnapshotCommit.runWithLock` to
+cover head reads and preparation; publication reuses that scope and revalidates the lease.
+Each catalog interprets its own lock configuration. Hive and JDBC define the catalog options
+`lock.enabled`, `lock.type`, `lock-check-max-sleep` and `lock-acquire-timeout` in their own options
+classes. For these catalogs, locking is enabled on object stores and disabled on other filesystems
+unless explicitly set. The filesystem catalog does not interpret these options. Catalogs without
+a locking policy use a no-op lock.
+Hive locks send heartbeats and validate ownership before publication. The client's
+`hive.txn.timeout` must match the metastore timeout. JDBC locks renew their
+leases and use an internal owner ID for renewal and release. Existing JDBC lock tables receive a
+nullable `lock_owner` column when the catalog initializes; this requires permission to alter that
+table. Older clients can still read the table, but all writers should upgrade before relying on
+owner-checked release after lease expiry. Each active Hive or JDBC scope owns a renewal worker
+which is stopped when the scope ends, so blocked renewals do not occupy other scopes' scheduling
+threads. Filesystem locks still depend on the storage's atomic publication behavior; they do not
+provide a storage fencing token.
+
 ## Files conflict
 
 A writer validates file-level changes as well as the snapshot ID. For example, if two compactors

docs/generated/catalog_configuration.html

@@ -152,30 +152,6 @@
             <td>String</td>
             <td>Comma-separated list of file types to cache. Supported values: meta, global-index, bucket-index, data, file-index, or * for all of them.</td>
         </tr>
-        <tr>
-            <td><h5>lock-acquire-timeout</h5></td>
-            <td style="word-wrap: break-word;">8 min</td>
-            <td>Duration</td>
-            <td>The maximum time to wait for acquiring the lock.</td>
-        </tr>
-        <tr>
-            <td><h5>lock-check-max-sleep</h5></td>
-            <td style="word-wrap: break-word;">8 s</td>
-            <td>Duration</td>
-            <td>The maximum sleep time when retrying to check the lock.</td>
-        </tr>
-        <tr>
-            <td><h5>lock.enabled</h5></td>
-            <td style="word-wrap: break-word;">(none)</td>
-            <td>Boolean</td>
-            <td>Enable Catalog Lock.</td>
-        </tr>
-        <tr>
-            <td><h5>lock.type</h5></td>
-            <td style="word-wrap: break-word;">(none)</td>
-            <td>String</td>
-            <td>The Lock Type for Catalog, such as 'hive', 'zookeeper'.</td>
-        </tr>
         <tr>
             <td><h5>metastore</h5></td>
             <td style="word-wrap: break-word;">"filesystem"</td>

docs/generated/hive_catalog_configuration.html

@@ -77,6 +77,30 @@
 you can set this option to true.
 </td>
         </tr>
+        <tr>
+            <td><h5>lock-acquire-timeout</h5></td>
+            <td style="word-wrap: break-word;">8 min</td>
+            <td>Duration</td>
+            <td>The maximum time to wait for acquiring the lock.</td>
+        </tr>
+        <tr>
+            <td><h5>lock-check-max-sleep</h5></td>
+            <td style="word-wrap: break-word;">8 s</td>
+            <td>Duration</td>
+            <td>The maximum sleep time when retrying to check the lock.</td>
+        </tr>
+        <tr>
+            <td><h5>lock.enabled</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>Boolean</td>
+            <td>Enable catalog locking. Defaults to enabled on object stores.</td>
+        </tr>
+        <tr>
+            <td><h5>lock.type</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The catalog lock factory identifier. Defaults to the catalog's built-in lock factory.</td>
+        </tr>
         <tr>
             <td><h5>metastore.client.class</h5></td>
             <td style="word-wrap: break-word;">"org.apache.hadoop.hive.metastore.HiveMetaStoreClient"</td>

docs/generated/jdbc_catalog_configuration.html

@@ -32,11 +32,35 @@
             <td>String</td>
             <td>Custom jdbc catalog store key.</td>
         </tr>
+        <tr>
+            <td><h5>lock-acquire-timeout</h5></td>
+            <td style="word-wrap: break-word;">8 min</td>
+            <td>Duration</td>
+            <td>The maximum time to wait for acquiring the lock.</td>
+        </tr>
+        <tr>
+            <td><h5>lock-check-max-sleep</h5></td>
+            <td style="word-wrap: break-word;">8 s</td>
+            <td>Duration</td>
+            <td>The maximum sleep time when retrying to check the lock.</td>
+        </tr>
         <tr>
             <td><h5>lock-key-max-length</h5></td>
             <td style="word-wrap: break-word;">255</td>
             <td>Integer</td>
             <td>Set the maximum length of the lock key. The 'lock-key' is composed of concatenating three fields : 'catalog-key', 'database', and 'table'.</td>
         </tr>
+        <tr>
+            <td><h5>lock.enabled</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>Boolean</td>
+            <td>Enable catalog locking. Defaults to enabled on object stores.</td>
+        </tr>
+        <tr>
+            <td><h5>lock.type</h5></td>
+            <td style="word-wrap: break-word;">(none)</td>
+            <td>String</td>
+            <td>The catalog lock factory identifier. Defaults to the catalog's built-in lock factory.</td>
+        </tr>
     </tbody>
 </table>

paimon-api/src/main/java/org/apache/paimon/options/CatalogOptions.java

@@ -53,30 +53,6 @@ public class CatalogOptions {
                     .defaultValue(CatalogTableType.MANAGED)
                     .withDescription("Type of table.");
 
-    public static final ConfigOption<Boolean> LOCK_ENABLED =
-            ConfigOptions.key("lock.enabled")
-                    .booleanType()
-                    .noDefaultValue()
-                    .withDescription("Enable Catalog Lock.");
-
-    public static final ConfigOption<String> LOCK_TYPE =
-            ConfigOptions.key("lock.type")
-                    .stringType()
-                    .noDefaultValue()
-                    .withDescription("The Lock Type for Catalog, such as 'hive', 'zookeeper'.");
-
-    public static final ConfigOption<Duration> LOCK_CHECK_MAX_SLEEP =
-            key("lock-check-max-sleep")
-                    .durationType()
-                    .defaultValue(Duration.ofSeconds(8))
-                    .withDescription("The maximum sleep time when retrying to check the lock.");
-
-    public static final ConfigOption<Duration> LOCK_ACQUIRE_TIMEOUT =
-            key("lock-acquire-timeout")
-                    .durationType()
-                    .defaultValue(Duration.ofMinutes(8))
-                    .withDescription("The maximum time to wait for acquiring the lock.");
-
     public static final ConfigOption<Integer> CLIENT_POOL_SIZE =
             key("client-pool-size")
                     .intType()

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -20,7 +20,6 @@
 
 import org.apache.paimon.CoreOptions.ExternalPathStrategy;
 import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.catalog.RenamingSnapshotCommit;
 import org.apache.paimon.catalog.SnapshotCommit;
 import org.apache.paimon.catalog.TableRollback;
 import org.apache.paimon.data.InternalRow;
@@ -43,7 +42,6 @@
 import org.apache.paimon.operation.ChainTablePartitionExpire;
 import org.apache.paimon.operation.ChangelogDeletion;
 import org.apache.paimon.operation.FileStoreCommitImpl;
-import org.apache.paimon.operation.Lock;
 import org.apache.paimon.operation.ManifestsReader;
 import org.apache.paimon.operation.NormalPartitionExpire;
 import org.apache.paimon.operation.PartitionExpire;
@@ -290,10 +288,8 @@ public boolean mergeSchema(
     @Override
     public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
         SnapshotManager snapshotManager = snapshotManager();
-        SnapshotCommit snapshotCommit = catalogEnvironment.snapshotCommit(snapshotManager);
-        if (snapshotCommit == null) {
-            snapshotCommit = new RenamingSnapshotCommit(snapshotManager, Lock.empty());
-        }
+        SnapshotCommit snapshotCommit =
+                catalogEnvironment.snapshotCommit(snapshotManager, options, commitUser);
         ConflictDetection.Factory conflictDetectFactory =
                 scanner ->
                         ConflictDetection.create(

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -22,14 +22,14 @@
 import org.apache.paimon.PagedList;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.TableType;
-import org.apache.paimon.factories.FactoryUtil;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.FileStatus;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.cache.CachingFileIO;
 import org.apache.paimon.fs.cache.LocalCacheManager;
 import org.apache.paimon.function.Function;
 import org.apache.paimon.function.FunctionChange;
+import org.apache.paimon.operation.Lock;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
 import org.apache.paimon.partition.PartitionStatistics;
@@ -80,8 +80,6 @@
 import static org.apache.paimon.catalog.CatalogUtils.listPartitionsFromFileSystem;
 import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable;
 import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
-import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
-import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
 
 /** Common implementation of {@link Catalog}. */
 public abstract class AbstractCatalog implements Catalog {
@@ -122,19 +120,28 @@ protected FileIO fileIO(Path path) {
         return fileIO;
     }
 
+    @Override
+    public Lock createLock(
+            Identifier identifier,
+            @Nullable String tableUuid,
+            @Nullable String commitUser,
+            Options tableOptions) {
+        return lockFactory()
+                .map(
+                        factory ->
+                                factory.createLock(
+                                        lockContext().orElse(null),
+                                        identifier,
+                                        tableUuid,
+                                        commitUser,
+                                        tableOptions))
+                .map(Lock::reentrant)
+                .orElseGet(Lock::empty);
+    }
+
+    /** Lock factory selected by the catalog's own locking policy. */
     public Optional<CatalogLockFactory> lockFactory() {
-        if (!lockEnabled()) {
-            return Optional.empty();
-        }
-
-        String lock = context.options().get(LOCK_TYPE);
-        if (lock == null) {
-            return defaultLockFactory();
-        }
-
-        return Optional.of(
-                FactoryUtil.discoverFactory(
-                        AbstractCatalog.class.getClassLoader(), CatalogLockFactory.class, lock));
+        return Optional.empty();
     }
 
     public Optional<CatalogLockFactory> defaultLockFactory() {
@@ -146,7 +153,7 @@ public Optional<CatalogLockContext> lockContext() {
     }
 
     protected boolean lockEnabled() {
-        return context.options().getOptional(LOCK_ENABLED).orElse(fileIO.isObjectStore());
+        return false;
     }
 
     protected boolean allowCustomTablePath() {

paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java

@@ -25,6 +25,8 @@
 import org.apache.paimon.consumer.ConsumerInfo;
 import org.apache.paimon.function.Function;
 import org.apache.paimon.function.FunctionChange;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
 import org.apache.paimon.partition.PartitionStatistics;
 import org.apache.paimon.predicate.Predicate;
@@ -1321,6 +1323,28 @@ void alterFunction(
     TableQueryAuthResult authTableQuery(Identifier identifier, @Nullable List<String> select)
             throws TableNotExistException;
 
+    // ========================== Locking ==============================
+
+    /**
+     * Create a lock bound to the table incarnation and exact writer identity.
+     *
+     * <p>The caller owns the returned lock and must close it. A lease may end when its snapshot is
+     * published; each additional snapshot must use a new {@link Lock#runWithLock} scope. Catalogs
+     * without configured locking return a no-op lock.
+     *
+     * @param identifier table and branch to lock
+     * @param tableUuid table incarnation, required by catalogs that fence table recreation
+     * @param commitUser exact logical writer identity for commits; null for catalog operations
+     * @param tableOptions runtime table options controlling lock support
+     */
+    default Lock createLock(
+            Identifier identifier,
+            @Nullable String tableUuid,
+            @Nullable String commitUser,
+            Options tableOptions) {
+        return Lock.empty();
+    }
+
     // ==================== Catalog Information ==========================
 
     /** Catalog options for re-creating this catalog. */

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockFactory.java

@@ -19,11 +19,30 @@
 package org.apache.paimon.catalog;
 
 import org.apache.paimon.factories.Factory;
+import org.apache.paimon.factories.FactoryUtil;
+import org.apache.paimon.operation.Lock;
+import org.apache.paimon.options.Options;
+
+import javax.annotation.Nullable;
 
 import java.io.Serializable;
 
-/** Factory to create {@link CatalogLock}. */
+/** Factory to create table-bound {@link Lock} instances. */
 public interface CatalogLockFactory extends Factory, Serializable {
 
-    CatalogLock createLock(CatalogLockContext context);
+    /**
+     * Create a lock bound to the table and its branch. The UUID and writer identity may be null for
+     * catalog operations; snapshot commits supply them along with the runtime table options.
+     */
+    Lock createLock(
+            CatalogLockContext context,
+            Identifier identifier,
+            @Nullable String tableUuid,
+            @Nullable String commitUser,
+            Options tableOptions);
+
+    static CatalogLockFactory discover(String identifier) {
+        return FactoryUtil.discoverFactory(
+                CatalogLockFactory.class.getClassLoader(), CatalogLockFactory.class, identifier);
+    }
 }