From fd2e7981156f6a95056e6714d81e33bfb6e7716c Mon Sep 17 00:00:00 2001
From: sunxiaojian <sunxiaojian926@163.com>
Date: Wed, 27 Mar 2024 23:46:20 +0800
Subject: [PATCH] Support common jdbc catalog lock for filesystem catalog

---
 docs/content/flink/sql-ddl.md                 | 100 ++++++++-------
 docs/content/spark/sql-ddl.md                 |   8 +-
 .../generated/catalog_configuration.html      |   8 +-
 .../apache/paimon/options/CatalogOptions.java |   9 +-
 .../paimon/catalog/AbstractCatalog.java       |  27 ++++
 .../org/apache/paimon/catalog/Catalog.java    |   5 +
 .../catalog/CatalogLockContextFactory.java    |  29 +++++
 .../paimon/catalog/FileSystemCatalog.java     |  13 ++
 .../catalog/FileSystemCatalogFactory.java     |   6 +
 .../org/apache/paimon/jdbc/JdbcCatalog.java   |   2 +-
 .../paimon/jdbc/JdbcCatalogLockContext.java   | 116 ++++++++++++++++--
 .../jdbc/JdbcCatalogLockContextFactory.java   |  37 ++++++
 .../org/apache/paimon/jdbc/JdbcUtils.java     |  13 ++
 .../org.apache.paimon.factories.Factory       |   1 +
 .../paimon/catalog/FileSystemCatalogTest.java |  72 +++++++++++
 .../paimon/flink/FileSystemCatalogITCase.java |  18 ++-
 .../org.apache.paimon.factories.Factory       |   6 +-
 .../org/apache/paimon/hive/HiveCatalog.java   |  15 ++-
 .../paimon/hive/HiveCatalogLockContext.java   |  10 ++
 .../hive/HiveCatalogLockContextFactory.java   |  36 ++++++
 .../org.apache.paimon.factories.Factory       |   1 +
 21 files changed, 466 insertions(+), 66 deletions(-)
 create mode 100644 paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockContextFactory.java
 create mode 100644 paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContextFactory.java
 create mode 100644 paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
 create mode 100644 paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContextFactory.java

diff --git a/docs/content/flink/sql-ddl.md b/docs/content/flink/sql-ddl.md
index 3e6af04a8ea5..84f374d5526c 100644
--- a/docs/content/flink/sql-ddl.md
+++ b/docs/content/flink/sql-ddl.md
@@ -3,7 +3,7 @@ title: "SQL DDL"
 weight: 2
 type: docs
 aliases:
-- /flink/sql-ddl.html
+  - /flink/sql-ddl.html
 ---
 <!--
 Licensed to the Apache Software Foundation (ASF) under one
@@ -51,6 +51,14 @@ USE CATALOG my_catalog;
 
 You can define any default table options with the prefix `table-default.` for tables created in the catalog.
 
+The FileSystem catalog supports jdbc lock and can take effect through the following configuration:
+
+> ```shell
+> 'lock.uri' = 'jdbc:mysql://<host>:<port>/<databaseName>'
+> 'lock.jdbc.user' = '...',
+> 'lock.jdbc.password' = '...',
+> ```
+
 ### Creating Hive Catalog
 
 By using Paimon Hive catalog, changes to the catalog will directly affect the corresponding Hive metastore. Tables created in such catalog can also be accessed directly from Hive.
@@ -68,7 +76,7 @@ The following Flink SQL registers and uses a Paimon Hive catalog named `my_hive`
 
 If your Hive requires security authentication such as Kerberos, LDAP, Ranger or you want the paimon table to be managed
 by Apache Atlas(Setting 'hive.metastore.event.listeners' in hive-site.xml). You can specify the hive-conf-dir and
-hadoop-conf-dir parameter to the hive-site.xml file path. 
+hadoop-conf-dir parameter to the hive-site.xml file path.
 
 ```sql
 CREATE CATALOG my_hive WITH (
@@ -105,8 +113,8 @@ If you want to see a partitioned table in Hive and also synchronize newly create
 
 #### Adding Parameters to a Hive Table
 
-Using the table option facilitates the convenient definition of Hive table parameters. 
-Parameters prefixed with `hive.` will be automatically defined in the `TBLPROPERTIES` of the Hive table. 
+Using the table option facilitates the convenient definition of Hive table parameters.
+Parameters prefixed with `hive.` will be automatically defined in the `TBLPROPERTIES` of the Hive table.
 For instance, using the option `hive.table.owner=Jon` will automatically add the parameter `table.owner=Jon` to the table properties during the creation process.
 
 #### Setting Location in Properties
@@ -155,17 +163,17 @@ You can define any default table options with the prefix `table-default.` for ta
 After use Paimon catalog, you can create and drop tables. Tables created in Paimon Catalogs are managed by the catalog.
 When the table is dropped from catalog, its table files will also be deleted.
 
-The following SQL assumes that you have registered and are using a Paimon catalog. It creates a managed table named 
+The following SQL assumes that you have registered and are using a Paimon catalog. It creates a managed table named
 `my_table` with five columns in the catalog's `default` database, where `dt`, `hh` and `user_id` are the primary keys.
 
 ```sql
 CREATE TABLE my_table (
-    user_id BIGINT,
-    item_id BIGINT,
-    behavior STRING,
-    dt STRING,
-    hh STRING,
-    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
+                        user_id BIGINT,
+                        item_id BIGINT,
+                        behavior STRING,
+                        dt STRING,
+                        hh STRING,
+                        PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
 );
 ```
 
@@ -173,12 +181,12 @@ You can create partitioned table:
 
 ```sql
 CREATE TABLE my_table (
-    user_id BIGINT,
-    item_id BIGINT,
-    behavior STRING,
-    dt STRING,
-    hh STRING,
-    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
+                        user_id BIGINT,
+                        item_id BIGINT,
+                        behavior STRING,
+                        dt STRING,
+                        hh STRING,
+                        PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
 ) PARTITIONED BY (dt, hh);
 ```
 
@@ -220,48 +228,48 @@ We can specify the primary key or partition when use `CREATE TABLE AS SELECT`, f
 /* For streaming mode, you need to enable the checkpoint. */
 
 CREATE TABLE my_table (
-    user_id BIGINT,
-    item_id BIGINT
+                        user_id BIGINT,
+                        item_id BIGINT
 );
 CREATE TABLE my_table_as AS SELECT * FROM my_table;
 
 /* partitioned table */
 CREATE TABLE my_table_partition (
-     user_id BIGINT,
-     item_id BIGINT,
-     behavior STRING,
-     dt STRING,
-     hh STRING
+                                  user_id BIGINT,
+                                  item_id BIGINT,
+                                  behavior STRING,
+                                  dt STRING,
+                                  hh STRING
 ) PARTITIONED BY (dt, hh);
 CREATE TABLE my_table_partition_as WITH ('partition' = 'dt') AS SELECT * FROM my_table_partition;
-    
+
 /* change options */
 CREATE TABLE my_table_options (
-       user_id BIGINT,
-       item_id BIGINT
+                                user_id BIGINT,
+                                item_id BIGINT
 ) WITH ('file.format' = 'orc');
 CREATE TABLE my_table_options_as WITH ('file.format' = 'parquet') AS SELECT * FROM my_table_options;
 
 /* primary key */
 CREATE TABLE my_table_pk (
-      user_id BIGINT,
-      item_id BIGINT,
-      behavior STRING,
-      dt STRING,
-      hh STRING,
-      PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
+                           user_id BIGINT,
+                           item_id BIGINT,
+                           behavior STRING,
+                           dt STRING,
+                           hh STRING,
+                           PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
 );
 CREATE TABLE my_table_pk_as WITH ('primary-key' = 'dt,hh') AS SELECT * FROM my_table_pk;
 
 
 /* primary key + partition */
 CREATE TABLE my_table_all (
-      user_id BIGINT,
-      item_id BIGINT,
-      behavior STRING,
-      dt STRING,
-      hh STRING,
-      PRIMARY KEY (dt, hh, user_id) NOT ENFORCED 
+                            user_id BIGINT,
+                            item_id BIGINT,
+                            behavior STRING,
+                            dt STRING,
+                            hh STRING,
+                            PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
 ) PARTITIONED BY (dt, hh);
 CREATE TABLE my_table_all_as WITH ('primary-key' = 'dt,hh', 'partition' = 'dt') AS SELECT * FROM my_table_all;
 ```
@@ -272,12 +280,12 @@ To create a table with the same schema, partition, and table properties as anoth
 
 ```sql
 CREATE TABLE my_table (
-    user_id BIGINT,
-    item_id BIGINT,
-    behavior STRING,
-    dt STRING,
-    hh STRING,
-    PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
+                        user_id BIGINT,
+                        item_id BIGINT,
+                        behavior STRING,
+                        dt STRING,
+                        hh STRING,
+                        PRIMARY KEY (dt, hh, user_id) NOT ENFORCED
 );
 
 CREATE TABLE my_table_like LIKE my_table;
@@ -315,4 +323,4 @@ CREATE TEMPORARY TABLE temp_table (
 );
 
 SELECT my_table.k, my_table.v, temp_table.v FROM my_table JOIN temp_table ON my_table.k = temp_table.k;
-```
+```
\ No newline at end of file
diff --git a/docs/content/spark/sql-ddl.md b/docs/content/spark/sql-ddl.md
index 182d94ad0ddb..17cfc59de734 100644
--- a/docs/content/spark/sql-ddl.md
+++ b/docs/content/spark/sql-ddl.md
@@ -49,9 +49,15 @@ spark-sql ... \
 ```
 
 You can define any default table options with the prefix `spark.sql.catalog.paimon.table-default.` for tables created in the catalog.
+The FileSystem catalog supports jdbc lock and can take effect through the following configuration:
 
-After `spark-sql` is started, you can switch to the `default` database of the `paimon` catalog with the following SQL.
+> ```shell
+> 'lock.uri' = 'jdbc:mysql://<host>:<port>/<databaseName>'
+> 'lock.jdbc.user' = '...',
+> 'lock.jdbc.password' = '...',
+> ```
 
+After `spark-sql` is started, you can switch to the `default` database of the `paimon` catalog with the following SQL.
 ```sql
 USE paimon.default;
 ```
diff --git a/docs/layouts/shortcodes/generated/catalog_configuration.html b/docs/layouts/shortcodes/generated/catalog_configuration.html
index cab6e731e851..17929ec2e369 100644
--- a/docs/layouts/shortcodes/generated/catalog_configuration.html
+++ b/docs/layouts/shortcodes/generated/catalog_configuration.html
@@ -26,6 +26,12 @@
         </tr>
     </thead>
     <tbody>
+        <tr>
+            <td><h5>client-pool-cache-eviction-interval-ms</h5></td>
+            <td style="word-wrap: break-word;">300000</td>
+            <td>Long</td>
+            <td>Client pool cache eviction interval ms.</td>
+        </tr>
         <tr>
             <td><h5>client-pool-size</h5></td>
             <td style="word-wrap: break-word;">2</td>
@@ -66,7 +72,7 @@
             <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>
+            <td>The Lock Type for Catalog, such as 'hive', 'zookeeper', 'jdbc'.</td>
         </tr>
         <tr>
             <td><h5>metastore</h5></td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
index f00a35a75094..858005c0fde4 100644
--- a/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
@@ -64,7 +64,8 @@ public class CatalogOptions {
             ConfigOptions.key("lock.type")
                     .stringType()
                     .noDefaultValue()
-                    .withDescription("The Lock Type for Catalog, such as 'hive', 'zookeeper'.");
+                    .withDescription(
+                            "The Lock Type for Catalog, such as 'hive', 'zookeeper', 'jdbc'.");
 
     public static final ConfigOption<Duration> LOCK_CHECK_MAX_SLEEP =
             key("lock-check-max-sleep")
@@ -110,4 +111,10 @@ public class CatalogOptions {
                                             TextElement.text(
                                                     "\"custom\": You can implement LineageMetaFactory and LineageMeta to store lineage information in customized storage."))
                                     .build());
+
+    public static final ConfigOption<Long> CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS =
+            ConfigOptions.key("client-pool-cache-eviction-interval-ms")
+                    .longType()
+                    .defaultValue(5 * 60 * 1000L)
+                    .withDescription("Client pool cache eviction interval ms.");
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 7d26a2197c10..a06dd50c22ce 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -53,6 +53,7 @@
 import static org.apache.paimon.options.CatalogOptions.LINEAGE_META;
 import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
 import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
+import static org.apache.paimon.options.CatalogOptions.METASTORE;
 import static org.apache.paimon.options.OptionsUtils.convertToPropertiesPrefixKey;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
@@ -62,6 +63,7 @@ public abstract class AbstractCatalog implements Catalog {
     public static final String DB_SUFFIX = ".db";
     protected static final String TABLE_DEFAULT_OPTION_PREFIX = "table-default.";
     protected static final String DB_LOCATION_PROP = "location";
+    protected static final String LOCK_PROP_PREFIX = "lock.";
 
     protected final FileIO fileIO;
     protected final Map<String, String> tableDefaultOptions;
@@ -115,6 +117,31 @@ public Optional<CatalogLockFactory> defaultLockFactory() {
         return Optional.empty();
     }
 
+    @Override
+    public Optional<CatalogLockContextFactory> lockContextFactory() {
+        String lock = catalogOptions.get(LOCK_TYPE);
+        if (lock == null) {
+            return Optional.empty();
+        }
+        return Optional.of(
+                FactoryUtil.discoverFactory(
+                        AbstractCatalog.class.getClassLoader(),
+                        CatalogLockContextFactory.class,
+                        lock));
+    }
+
+    public Options extractLockConfiguration(Map<String, String> properties) {
+        Map<String, String> result = new HashMap<>();
+        result.put(METASTORE.key(), properties.get(METASTORE.key()));
+        properties.forEach(
+                (key, value) -> {
+                    if (key.startsWith(LOCK_PROP_PREFIX)) {
+                        result.put(key.substring(LOCK_PROP_PREFIX.length()), value);
+                    }
+                });
+        return Options.fromMap(result);
+    }
+
     @Override
     public Optional<CatalogLockContext> lockContext() {
         return Optional.of(CatalogLockContext.fromOptions(catalogOptions));
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
index 4f63414171f5..4cd4cd19d939 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -60,6 +60,11 @@ public interface Catalog extends AutoCloseable {
      */
     Optional<CatalogLockFactory> lockFactory();
 
+    /** Get lock context factory for lock factory to create a lock. */
+    default Optional<CatalogLockContextFactory> lockContextFactory() {
+        return Optional.empty();
+    }
+
     /** Get lock context for lock factory to create a lock. */
     default Optional<CatalogLockContext> lockContext() {
         return Optional.empty();
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockContextFactory.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockContextFactory.java
new file mode 100644
index 000000000000..543dd478cd2e
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogLockContextFactory.java
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.factories.Factory;
+import org.apache.paimon.options.Options;
+
+import java.io.Serializable;
+
+/** Context for lock context factory to create lock. */
+public interface CatalogLockContextFactory extends Factory, Serializable {
+    CatalogLockContext createLockContext(Options lockOptions);
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
index 1e4e5b0ebaaa..a386dec342ac 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
@@ -35,6 +35,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 
 import static org.apache.paimon.catalog.FileSystemCatalogOptions.CASE_SENSITIVE;
@@ -199,4 +200,16 @@ public String warehouse() {
     public boolean caseSensitive() {
         return catalogOptions.get(CASE_SENSITIVE);
     }
+
+    @Override
+    public Optional<CatalogLockContext> lockContext() {
+        Optional<CatalogLockContextFactory> catalogLockContextFactory = lockContextFactory();
+        if (!catalogLockContextFactory.isPresent()) {
+            return super.lockContext();
+        }
+        return catalogLockContextFactory.map(
+                factory ->
+                        factory.createLockContext(
+                                extractLockConfiguration(catalogOptions.toMap())));
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogFactory.java b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogFactory.java
index 8a0b1643fb2b..89aaaf90a993 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogFactory.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalogFactory.java
@@ -25,6 +25,8 @@
 import org.apache.paimon.privilege.PrivilegedCatalog;
 import org.apache.paimon.table.TableType;
 
+import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
+import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
 import static org.apache.paimon.options.CatalogOptions.TABLE_TYPE;
 
 /** Factory to create {@link FileSystemCatalog}. */
@@ -44,6 +46,10 @@ public Catalog create(FileIO fileIO, Path warehouse, CatalogContext context) {
                     "Only managed table is supported in File system catalog.");
         }
 
+        if (context.options().get(LOCK_ENABLED) && context.options().get(LOCK_TYPE) == null) {
+            throw new IllegalArgumentException("Please configure the lock type correctly.");
+        }
+
         Catalog catalog = new FileSystemCatalog(fileIO, warehouse, context.options());
 
         PrivilegeManager privilegeManager =
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
index 7e7718b5bee9..55f9261edf03 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
@@ -350,7 +350,7 @@ public Optional<CatalogLockFactory> defaultLockFactory() {
 
     @Override
     public Optional<CatalogLockContext> lockContext() {
-        return Optional.of(new JdbcCatalogLockContext(connections, catalogKey, options));
+        return Optional.of(new JdbcCatalogLockContext(connections, options));
     }
 
     private Lock lock(Identifier identifier) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContext.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContext.java
index e56b3474ccb3..ed57c1a63d85 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContext.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContext.java
@@ -19,19 +19,96 @@
 package org.apache.paimon.jdbc;
 
 import org.apache.paimon.catalog.CatalogLockContext;
+import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Scheduler;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
+
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 
 /** Jdbc lock context. */
 public class JdbcCatalogLockContext implements CatalogLockContext {
 
-    private final JdbcClientPool connections;
+    private static final String CONF_KEY_PREFIX = "lockConfigKey:";
+
+    // Cache access connections for non Jdbc catalogs.
+    protected static Cache<String, JdbcClientPool> clientPoolCache;
+    private long evictionInterval;
+    private String key;
+
+    private JdbcClientPool connections;
     private final String catalogKey;
     private final Options options;
+    private final String metastore;
 
-    public JdbcCatalogLockContext(JdbcClientPool connections, String catalogKey, Options options) {
-        this.connections = connections;
-        this.catalogKey = catalogKey;
+    public JdbcCatalogLockContext(Options options) {
         this.options = options;
+        this.catalogKey = options.get(JdbcCatalogOptions.CATALOG_KEY);
+        this.metastore = options.get(CatalogOptions.METASTORE);
+        if (!metastore.equals(JdbcCatalogFactory.IDENTIFIER)) {
+            this.evictionInterval =
+                    options.get(CatalogOptions.CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS);
+            this.key = extractKey();
+            init();
+        }
+    }
+
+    /** Use all connection parameters of jdbc to form a unique connection. */
+    private String extractKey() {
+        Map<String, String> props =
+                JdbcUtils.extractJdbcConfigurationToMap(
+                        options.toMap(), JdbcCatalog.PROPERTY_PREFIX);
+        List<String> elements = Lists.newArrayList();
+        props.forEach(
+                (key, value) -> {
+                    elements.add(value);
+                });
+        return CONF_KEY_PREFIX
+                .concat(catalogKey)
+                .concat(":")
+                .concat(StringUtils.join(elements, "."));
+    }
+
+    JdbcCatalogLockContext(JdbcClientPool connections, Options options) {
+        this(options);
+        this.connections = connections;
+    }
+
+    protected synchronized void init() {
+        if (clientPoolCache == null) {
+            clientPoolCache =
+                    Caffeine.newBuilder()
+                            .expireAfterAccess(evictionInterval, TimeUnit.MILLISECONDS)
+                            .removalListener(
+                                    (ignored, value, cause) -> ((JdbcClientPool) value).close())
+                            .scheduler(
+                                    Scheduler.forScheduledExecutorService(
+                                            new ScheduledThreadPoolExecutor(
+                                                    1,
+                                                    new ThreadFactory() {
+                                                        final ThreadFactory defaultFactory =
+                                                                Executors.defaultThreadFactory();
+
+                                                        @Override
+                                                        public Thread newThread(Runnable r) {
+                                                            Thread thread =
+                                                                    defaultFactory.newThread(r);
+                                                            thread.setDaemon(true);
+                                                            return thread;
+                                                        }
+                                                    })))
+                            .build();
+        }
     }
 
     @Override
@@ -39,11 +116,34 @@ public Options options() {
         return options;
     }
 
-    public JdbcClientPool connections() {
-        return connections;
-    }
-
     public String catalogKey() {
         return catalogKey;
     }
+
+    public JdbcClientPool connections() {
+        // Cache connection information is required for non jdbc catalogs.
+        if (!metastore.equals(JdbcCatalogFactory.IDENTIFIER)) {
+            return clientPoolCache.get(
+                    key,
+                    k -> {
+                        JdbcClientPool connections =
+                                new JdbcClientPool(
+                                        options.get(CatalogOptions.CLIENT_POOL_SIZE),
+                                        options.get(CatalogOptions.URI.key()),
+                                        options.toMap());
+                        try {
+                            JdbcUtils.createDistributedLockTable(connections, options);
+                        } catch (SQLException e) {
+                            throw new RuntimeException(
+                                    "Cannot initialize JDBC distributed lock.", e);
+                        } catch (InterruptedException e) {
+                            throw new RuntimeException("Interrupted in call to initialize", e);
+                        }
+                        return connections;
+                    });
+        } else {
+            // The use of lock in the jdbc catalog is maintained by the jdbc catalog itself
+            return connections;
+        }
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContextFactory.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContextFactory.java
new file mode 100644
index 000000000000..ead2c4e9c353
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalogLockContextFactory.java
@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.jdbc;
+
+import org.apache.paimon.catalog.CatalogLockContext;
+import org.apache.paimon.catalog.CatalogLockContextFactory;
+import org.apache.paimon.options.Options;
+
+/** Factory for jdbc catalog lock context. */
+public class JdbcCatalogLockContextFactory implements CatalogLockContextFactory {
+
+    @Override
+    public String identifier() {
+        return JdbcCatalogLockFactory.IDENTIFIER;
+    }
+
+    @Override
+    public CatalogLockContext createLockContext(Options lockOptions) {
+        return new JdbcCatalogLockContext(lockOptions);
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java
index 4acb0f25aa91..682dd308aedd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java
@@ -30,6 +30,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.SQLIntegrityConstraintViolationException;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 import java.util.function.Consumer;
@@ -225,6 +226,18 @@ public class JdbcUtils {
     static final String ACQUIRED_AT = "acquired_at";
     static final String EXPIRE_TIME = "expire_time_seconds";
 
+    public static Map<String, String> extractJdbcConfigurationToMap(
+            Map<String, String> properties, String prefix) {
+        Map<String, String> result = new HashMap<>();
+        properties.forEach(
+                (key, value) -> {
+                    if (key.startsWith(prefix)) {
+                        result.put(key.substring(prefix.length()), value);
+                    }
+                });
+        return result;
+    }
+
     public static Properties extractJdbcConfiguration(
             Map<String, String> properties, String prefix) {
         Properties result = new Properties();
diff --git a/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 0f87c96b0d4e..bf7c9333bc22 100644
--- a/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-core/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -16,3 +16,4 @@
 org.apache.paimon.catalog.FileSystemCatalogFactory
 org.apache.paimon.jdbc.JdbcCatalogFactory
 org.apache.paimon.jdbc.JdbcCatalogLockFactory
+org.apache.paimon.jdbc.JdbcCatalogLockContextFactory
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
new file mode 100644
index 000000000000..ed597979a309
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.jdbc.JdbcCatalog;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+
+import org.junit.jupiter.api.BeforeEach;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FileSystemCatalog}. */
+public class FileSystemCatalogTest extends CatalogTestBase {
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        super.setUp();
+        catalog = initCatalog(Maps.newHashMap());
+    }
+
+    private FileSystemCatalog initCatalog(Map<String, String> props) {
+        Map<String, String> properties = Maps.newHashMap();
+        properties.put(
+                AbstractCatalog.LOCK_PROP_PREFIX + CatalogOptions.URI.key(),
+                "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
+
+        properties.put(
+                AbstractCatalog.LOCK_PROP_PREFIX + JdbcCatalog.PROPERTY_PREFIX + "username",
+                "user");
+        properties.put(
+                AbstractCatalog.LOCK_PROP_PREFIX + JdbcCatalog.PROPERTY_PREFIX + "password",
+                "password");
+        properties.put(CatalogOptions.WAREHOUSE.key(), warehouse);
+        properties.put(CatalogOptions.LOCK_ENABLED.key(), "true");
+        properties.put(CatalogOptions.LOCK_TYPE.key(), "jdbc");
+        properties.putAll(props);
+        FileSystemCatalog catalog =
+                new FileSystemCatalog(fileIO, new Path(warehouse), Options.fromMap(properties));
+        return catalog;
+    }
+
+    @Override
+    public void testListDatabasesWhenNoDatabases() {
+        List<String> databases = catalog.listDatabases();
+        assertThat(databases).isEqualTo(new ArrayList<>());
+    }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
index 451e4c78bcec..dc221b146096 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FileSystemCatalogITCase.java
@@ -22,10 +22,12 @@
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogLock;
 import org.apache.paimon.catalog.CatalogLockContext;
+import org.apache.paimon.catalog.CatalogLockContextFactory;
 import org.apache.paimon.catalog.CatalogLockFactory;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.flink.util.AbstractTestBase;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.utils.BlockingIterator;
 
 import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
@@ -191,7 +193,7 @@ private List<Row> collect(String sql) throws Exception {
     /** Lock factory for file system catalog. */
     public static class FileSystemCatalogDummyLockFactory implements CatalogLockFactory {
 
-        private static final String IDENTIFIER = "DUMMY";
+        public static final String IDENTIFIER = "DUMMY";
 
         @Override
         public String identifier() {
@@ -213,4 +215,18 @@ public void close() throws IOException {}
             };
         }
     }
+
+    /** Lock context factory for file system catalog. */
+    public static class DummyCatalogLockContextFactory implements CatalogLockContextFactory {
+
+        @Override
+        public String identifier() {
+            return FileSystemCatalogDummyLockFactory.IDENTIFIER;
+        }
+
+        @Override
+        public CatalogLockContext createLockContext(Options lockOptions) {
+            return CatalogLockContext.fromOptions(lockOptions);
+        }
+    }
 }
diff --git a/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
index fcb6fe982943..170577467c10 100644
--- a/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -19,4 +19,8 @@ org.apache.paimon.flink.FlinkCatalogTest$TestingLogSoreRegisterFactory
 org.apache.paimon.flink.FlinkLineageITCase$TestingMemoryLineageMetaFactory
 
 # Catalog lock factory
-org.apache.paimon.flink.FileSystemCatalogITCase$FileSystemCatalogDummyLockFactory
\ No newline at end of file
+org.apache.paimon.flink.FileSystemCatalogITCase$FileSystemCatalogDummyLockFactory
+
+
+# Catalog lock context factory
+org.apache.paimon.flink.FileSystemCatalogITCase$DummyCatalogLockContextFactory
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index 0d71e70a6479..6df57e2817c5 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -710,14 +710,17 @@ public static Catalog createHiveCatalog(CatalogContext context) {
     }
 
     public static HiveConf createHiveConf(CatalogContext context) {
-        String uri = context.options().get(CatalogOptions.URI);
-        String hiveConfDir = context.options().get(HIVE_CONF_DIR);
-        String hadoopConfDir = context.options().get(HADOOP_CONF_DIR);
-        HiveConf hiveConf =
-                HiveCatalog.createHiveConf(hiveConfDir, hadoopConfDir, context.hadoopConf());
+        return createHiveConf(context.options(), context.hadoopConf());
+    }
+
+    public static HiveConf createHiveConf(Options options, Configuration hadoopConf) {
+        String uri = options.get(CatalogOptions.URI);
+        String hiveConfDir = options.get(HIVE_CONF_DIR);
+        String hadoopConfDir = options.get(HADOOP_CONF_DIR);
+        HiveConf hiveConf = HiveCatalog.createHiveConf(hiveConfDir, hadoopConfDir, hadoopConf);
 
         // always using user-set parameters overwrite hive-site.xml parameters
-        context.options().toMap().forEach(hiveConf::set);
+        options.toMap().forEach(hiveConf::set);
         if (uri != null) {
             hiveConf.set(HiveConf.ConfVars.METASTOREURIS.varname, uri);
         }
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContext.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContext.java
index ecffd7f1e633..9d96edb247a3 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContext.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContext.java
@@ -20,6 +20,7 @@
 
 import org.apache.paimon.catalog.CatalogLockContext;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.HadoopUtils;
 
 /** Hive {@link CatalogLockContext}. */
 public class HiveCatalogLockContext implements CatalogLockContext {
@@ -28,6 +29,15 @@ public class HiveCatalogLockContext implements CatalogLockContext {
     private final String clientClassName;
     private final Options options;
 
+    public HiveCatalogLockContext(Options options) {
+        this.hiveConf =
+                new SerializableHiveConf(
+                        HiveCatalog.createHiveConf(
+                                options, HadoopUtils.getHadoopConfiguration(options)));
+        this.clientClassName = options.get(HiveCatalogFactory.METASTORE_CLIENT_CLASS);
+        this.options = options;
+    }
+
     public HiveCatalogLockContext(
             SerializableHiveConf hiveConf, String clientClassName, Options options) {
         this.hiveConf = hiveConf;
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContextFactory.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContextFactory.java
new file mode 100644
index 000000000000..b00aa4cb4e2c
--- /dev/null
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalogLockContextFactory.java
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.hive;
+
+import org.apache.paimon.catalog.CatalogLockContext;
+import org.apache.paimon.catalog.CatalogLockContextFactory;
+import org.apache.paimon.options.Options;
+
+/** Hive {@link CatalogLockContextFactory}. */
+public class HiveCatalogLockContextFactory implements CatalogLockContextFactory {
+    @Override
+    public String identifier() {
+        return HiveCatalogLock.LOCK_IDENTIFIER;
+    }
+
+    @Override
+    public CatalogLockContext createLockContext(Options lockOptions) {
+        return new HiveCatalogLockContext(lockOptions);
+    }
+}
diff --git a/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index baab92184129..e02ec0750d43 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-hive/paimon-hive-catalog/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -15,3 +15,4 @@
 
 org.apache.paimon.hive.HiveCatalogFactory
 org.apache.paimon.hive.HiveCatalogLockFactory
+org.apache.paimon.hive.HiveCatalogLockContextFactory