[flink][spark] Check Exists before create database to avoid Permissio…

…n issue
apache · Dec 11, 2023 · 38aa07b · 38aa07b
1 parent ac328f1
commit 38aa07b
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 6 deletions.
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -148,9 +148,11 @@ public FlinkCatalog(
         this.logStoreAutoRegister = options.get(LOG_SYSTEM_AUTO_REGISTER);
         this.logStoreAutoRegisterTimeout = options.get(REGISTER_TIMEOUT);
         this.disableCreateTableInDefaultDatabase = options.get(DISABLE_CREATE_TABLE_IN_DEFAULT_DB);
-        try {
-            this.catalog.createDatabase(defaultDatabase, true);
-        } catch (Catalog.DatabaseAlreadyExistException ignore) {
+        if (!catalog.databaseExists(defaultDatabase)) {
+            try {
+                catalog.createDatabase(defaultDatabase, true);
+            } catch (Catalog.DatabaseAlreadyExistException ignore) {
+            }
         }
     }
 

diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -74,9 +74,11 @@ public void initialize(String name, CaseInsensitiveStringMap options) {
                         Options.fromMap(options),
                         SparkSession.active().sessionState().newHadoopConf());
         this.catalog = CatalogFactory.createCatalog(catalogContext);
-        try {
-            createNamespace(defaultNamespace(), new HashMap<>());
-        } catch (NamespaceAlreadyExistsException ignored) {
+        if (!catalog.databaseExists(defaultNamespace()[0])) {
+            try {
+                createNamespace(defaultNamespace(), new HashMap<>());
+            } catch (NamespaceAlreadyExistsException ignored) {
+            }
         }
     }