Throw Exception If catalog default database is not existed.

Author: loserwang1024

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

@@ -108,7 +108,7 @@ public class FlinkCatalog extends AbstractCatalog {
     protected final ClassLoader classLoader;
 
     protected final String catalogName;
-    protected final @Nullable String defaultDatabase;
+    protected final String defaultDatabase;
     protected final String bootstrapServers;
     private final Map<String, String> securityConfigs;
     protected Connection connection;
@@ -117,7 +117,7 @@ public class FlinkCatalog extends AbstractCatalog {
 
     public FlinkCatalog(
             String name,
-            @Nullable String defaultDatabase,
+            String defaultDatabase,
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs) {
@@ -142,6 +142,10 @@ public void open() throws CatalogException {
 
         connection = ConnectionFactory.createConnection(Configuration.fromMap(flussConfigs));
         admin = connection.getAdmin();
+        if (!databaseExists(defaultDatabase)) {
+            throw new CatalogException(
+                    String.format("Database %s does not exist in fluss server.", defaultDatabase));
+        }
     }
 
     @Override

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java

@@ -33,6 +33,7 @@
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.CloseableIterator;
 import org.apache.flink.util.CollectionUtil;
@@ -657,6 +658,20 @@ void testAuthentication() throws Exception {
         }
     }
 
+    @Test
+    void createCatalogWithUnexistedDatabase() {
+        assertThatThrownBy(
+                        () ->
+                                tEnv.executeSql(
+                                        String.format(
+                                                "create catalog test_non_exist_database_catalog with ('type' = 'fluss', '%s' = '%s', 'default-database' = 'non-exist')",
+                                                BOOTSTRAP_SERVERS.key(),
+                                                FLUSS_CLUSTER_EXTENSION.getBootstrapServers())))
+                .rootCause()
+                .isExactlyInstanceOf(CatalogException.class)
+                .hasMessage("Database non-exist does not exist in fluss server.");
+    }
+
     private static void assertOptionsEqual(
             Map<String, String> actualOptions, Map<String, String> expectedOptions) {
         actualOptions.remove(ConfigOptions.BOOTSTRAP_SERVERS.key());

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java

@@ -84,7 +84,7 @@ class FlinkCatalogTest {
             FlussClusterExtension.builder().setNumOfTabletServers(1).build();
 
     private static final String CATALOG_NAME = "test-catalog";
-    private static final String DEFAULT_DB = "default";
+    private static final String DEFAULT_DB = FlinkCatalogOptions.DEFAULT_DATABASE.defaultValue();
     static Catalog catalog;
     private final ObjectPath tableInDefaultDb = new ObjectPath(DEFAULT_DB, "t1");
 
@@ -407,8 +407,7 @@ void testDatabase() throws Exception {
         CatalogTable expectedTable = addOptions(table, addedOptions);
         checkEqualsRespectSchema((CatalogTable) tableCreated, expectedTable);
         assertThat(catalog.listTables("db1")).isEqualTo(Collections.singletonList("t1"));
-        assertThat(catalog.listDatabases())
-                .isEqualTo(Arrays.asList(DEFAULT_DB, "db1", "db2", "fluss"));
+        assertThat(catalog.listDatabases()).isEqualTo(Arrays.asList("db1", "db2", DEFAULT_DB));
         // test drop db1;
         // should throw exception since db1 is not empty and we set cascade = false
         assertThatThrownBy(() -> catalog.dropDatabase("db1", false, false))
@@ -424,10 +423,10 @@ void testDatabase() throws Exception {
         // should be ok since we set ignoreIfNotExists = true
         catalog.dropDatabase("db1", true, true);
         // test list db
-        assertThat(catalog.listDatabases()).isEqualTo(Arrays.asList(DEFAULT_DB, "db2", "fluss"));
+        assertThat(catalog.listDatabases()).isEqualTo(Arrays.asList("db2", DEFAULT_DB));
         catalog.dropDatabase("db2", false, true);
         // should be empty
-        assertThat(catalog.listDatabases()).isEqualTo(Arrays.asList(DEFAULT_DB, "fluss"));
+        assertThat(catalog.listDatabases()).isEqualTo(Collections.singletonList(DEFAULT_DB));
         // should throw exception since the db is not exist and we set ignoreIfNotExists = false
         assertThatThrownBy(() -> catalog.listTables("unknown"))
                 .isInstanceOf(DatabaseNotExistException.class)
@@ -465,7 +464,7 @@ void testOperatePartitions() throws Exception {
         catalog.createTable(path1, table, false);
         assertThatThrownBy(() -> catalog.listPartitions(path1))
                 .isInstanceOf(TableNotPartitionedException.class)
-                .hasMessage("Table default.t1 in catalog test-catalog is not partitioned.");
+                .hasMessage("Table fluss.t1 in catalog test-catalog is not partitioned.");
 
         // create partition table and list partitions.
         ObjectPath path2 = new ObjectPath(DEFAULT_DB, "partitioned_t1");
@@ -501,7 +500,7 @@ void testOperatePartitions() throws Exception {
         assertThatThrownBy(() -> catalog.listPartitions(path2, invalidTestSpec))
                 .isInstanceOf(CatalogException.class)
                 .hasMessage(
-                        "Failed to list partitions of table default.partitioned_t1 in test-catalog, by partitionSpec CatalogPartitionSpec{{second=}}");
+                        "Failed to list partitions of table fluss.partitioned_t1 in test-catalog, by partitionSpec CatalogPartitionSpec{{second=}}");
 
         // NEW: Test dropPartition functionality
         CatalogPartitionSpec firstPartSpec = catalogPartitionSpecs.get(0);
@@ -520,7 +519,7 @@ void testOperatePartitions() throws Exception {
                 .isInstanceOf(
                         org.apache.flink.table.catalog.exceptions.PartitionNotExistException.class)
                 .hasMessage(
-                        "Partition CatalogPartitionSpec{{first=999}} of table default.partitioned_t1 in catalog test-catalog does not exist.");
+                        "Partition CatalogPartitionSpec{{first=999}} of table fluss.partitioned_t1 in catalog test-catalog does not exist.");
 
         // Should not throw with ignoreIfNotExists = true
         catalog.dropPartition(path2, nonExistentSpec, true);