(incubator-paimon) branch master updated: [flink] Do not create default database if ddl in it is disable (#2500)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new c4f74c4c8 [flink] Do not create default database if ddl in it is 
disable (#2500)
c4f74c4c8 is described below

commit c4f74c4c83afdb05fc2619c1f5a2ae593f0d57e5
Author: Fang Yong <zjur...@gmail.com>
AuthorDate: Wed Dec 13 16:49:41 2023 +0800

    [flink] Do not create default database if ddl in it is disable (#2500)
---
 .../src/main/java/org/apache/paimon/flink/FlinkCatalog.java    | 10 ++++++----
 .../test/java/org/apache/paimon/flink/FlinkCatalogTest.java    |  3 +++
 2 files changed, 9 insertions(+), 4 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
index eeae7331e..c854be93a 100644
--- 
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,10 +148,12 @@ public class FlinkCatalog extends AbstractCatalog {
         this.logStoreAutoRegister = options.get(LOG_SYSTEM_AUTO_REGISTER);
         this.logStoreAutoRegisterTimeout = options.get(REGISTER_TIMEOUT);
         this.disableCreateTableInDefaultDatabase = 
options.get(DISABLE_CREATE_TABLE_IN_DEFAULT_DB);
-        if (!catalog.databaseExists(defaultDatabase)) {
-            try {
-                catalog.createDatabase(defaultDatabase, true);
-            } catch (Catalog.DatabaseAlreadyExistException ignore) {
+        if (!disableCreateTableInDefaultDatabase) {
+            if (!catalog.databaseExists(defaultDatabase)) {
+                try {
+                    catalog.createDatabase(defaultDatabase, true);
+                } catch (Catalog.DatabaseAlreadyExistException ignore) {
+                }
             }
         }
     }
diff --git 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
index cdca5e7ea..7f60a2b67 100644
--- 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
+++ 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
@@ -79,6 +79,7 @@ import static 
org.apache.paimon.flink.FlinkCatalogOptions.LOG_SYSTEM_AUTO_REGIST
 import static org.apache.paimon.flink.FlinkConnectorOptions.LOG_SYSTEM;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatCollection;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** Test for {@link FlinkCatalog}. */
@@ -536,10 +537,12 @@ public class FlinkCatalogTest {
                 .isInstanceOf(UnsupportedOperationException.class)
                 .hasMessage(
                         "Creating table in default database is disabled, 
please specify a database name.");
+        assertThatCollection(catalog.listDatabases()).isEmpty();
 
         catalog.createDatabase("db1", null, false);
         assertThatCode(() -> catalog.createTable(path1, this.createTable(new 
HashMap<>(0)), false))
                 .doesNotThrowAnyException();
+        assertThat(catalog.listDatabases()).containsExactlyInAnyOrder("db1");
 
         conf.set(FlinkCatalogOptions.DEFAULT_DATABASE, "default-db");
         Catalog catalog1 =