[flink] Validate database existence when resolving catalog functions in FlinkCatalog

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

@@ -802,18 +802,30 @@ public void alterPartition(
     @Override
     public List<String> listFunctions(String dbName)
             throws DatabaseNotExistException, CatalogException {
+        if (!databaseExists(dbName)) {
+            throw new DatabaseNotExistException(getName(), dbName);
+        }
         return new ArrayList<>(BUILTIN_BITMAP_FUNCTIONS.keySet());
     }
 
     @Override
     public boolean functionExists(ObjectPath objectPath) throws CatalogException {
+        if (!databaseExists(objectPath.getDatabaseName())) {
+            return false;
+        }
         return BUILTIN_BITMAP_FUNCTIONS.containsKey(
                 objectPath.getObjectName().toLowerCase(Locale.ROOT));
     }
 
     @Override
     public CatalogFunction getFunction(ObjectPath functionPath)
             throws FunctionNotExistException, CatalogException {
+        if (!databaseExists(functionPath.getDatabaseName())) {
+            throw new CatalogException(
+                    String.format(
+                            "Database %s does not exist in catalog %s.",
+                            functionPath.getDatabaseName(), getName()));
+        }
         String className =
                 BUILTIN_BITMAP_FUNCTIONS.get(functionPath.getObjectName().toLowerCase(Locale.ROOT));
         if (className == null) {

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

@@ -1004,6 +1004,17 @@ void testCreateCatalogWithUnexistedDatabase() {
                         "The configured default-database 'non-exist' does not exist in the Fluss cluster.");
     }
 
+    @Test
+    void testBitmapFunctionFailsForFullyQualifiedNonexistentDatabase() {
+        assertThatThrownBy(
+                        () ->
+                                tEnv.executeSql(
+                                        "SELECT "
+                                                + CATALOG_NAME
+                                                + ".nonexistent_db.rb_build(ARRAY[1,2])"))
+                .hasMessageContaining("nonexistent_db");
+    }
+
     @Test
     void testCreateCatalogWithLakeProperties() throws Exception {
         Map<String, String> properties = new HashMap<>();

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

@@ -1221,4 +1221,38 @@ void testBitmapFunctionsRegistered() throws Exception {
         // verify unknown still returns false
         assertThat(catalog.functionExists(new ObjectPath(DEFAULT_DB, "unknown_fn"))).isFalse();
     }
+
+    @Test
+    void testBuiltinFunctionsRequireExistingDatabase() throws Exception {
+        String nonexistentDb = "nonexistent_db_for_functions";
+        assertThat(catalog.databaseExists(nonexistentDb)).isFalse();
+
+        // listFunctions on a nonexistent database throws
+        assertThatThrownBy(() -> catalog.listFunctions(nonexistentDb))
+                .isInstanceOf(DatabaseNotExistException.class)
+                .hasMessage(
+                        "Database %s does not exist in Catalog %s.", nonexistentDb, CATALOG_NAME);
+
+        // functionExists on a nonexistent database returns false, not throw
+        ObjectPath qualifiedInNonexistentDb = new ObjectPath(nonexistentDb, "rb_build");
+        assertThat(catalog.functionExists(qualifiedInNonexistentDb)).isFalse();
+
+        // getFunction on a nonexistent database throws CatalogException with db-not-found message
+        assertThatThrownBy(() -> catalog.getFunction(qualifiedInNonexistentDb))
+                .isInstanceOf(CatalogException.class)
+                .hasMessageContaining("Database " + nonexistentDb + " does not exist");
+
+        // built-in functions still resolve from every existing database, not just DEFAULT_DB
+        String secondDb = "second_db_for_functions";
+        catalog.createDatabase(
+                secondDb, new CatalogDatabaseImpl(Collections.emptyMap(), null), true);
+        try {
+            assertThat(catalog.functionExists(new ObjectPath(secondDb, "rb_build"))).isTrue();
+            assertThat(catalog.getFunction(new ObjectPath(secondDb, "rb_build"))).isNotNull();
+            assertThat(catalog.listFunctions(secondDb))
+                    .contains("rb_build_agg", "rb_or_agg", "rb_and_agg", "rb_xor_agg");
+        } finally {
+            catalog.dropDatabase(secondDb, true, true);
+        }
+    }
 }