(paimon) branch master updated: [flink] Support creating database with properties in flink (#3491)

[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/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new cf9de27e1 [flink] Support creating database with properties in flink 
(#3491)
cf9de27e1 is described below

commit cf9de27e13a606a750b1ba614dc7c0eeec17420f
Author: Fang Yong <zjur...@gmail.com>
AuthorDate: Mon Jun 10 22:24:06 2024 +0800

    [flink] Support creating database with properties in flink (#3491)
---
 .../java/org/apache/paimon/flink/FlinkCatalog.java   | 11 ++++-------
 .../org/apache/paimon/flink/FlinkCatalogTest.java    | 20 ++++++++++++++++----
 2 files changed, 20 insertions(+), 11 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 7896ec923..fc74b73a4 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
@@ -194,13 +194,7 @@ public class FlinkCatalog extends AbstractCatalog {
     @Override
     public void createDatabase(String name, CatalogDatabase database, boolean 
ignoreIfExists)
             throws DatabaseAlreadyExistException, CatalogException {
-        // todo: flink hive catalog support create db with props
         if (database != null) {
-            if (database.getProperties().size() > 0) {
-                throw new UnsupportedOperationException(
-                        "Create database with properties is unsupported.");
-            }
-
             if (database.getDescription().isPresent()
                     && !database.getDescription().get().equals("")) {
                 throw new UnsupportedOperationException(
@@ -209,7 +203,10 @@ public class FlinkCatalog extends AbstractCatalog {
         }
 
         try {
-            catalog.createDatabase(name, ignoreIfExists);
+            catalog.createDatabase(
+                    name,
+                    ignoreIfExists,
+                    database == null ? Collections.emptyMap() : 
database.getProperties());
         } catch (Catalog.DatabaseAlreadyExistException e) {
             throw new DatabaseAlreadyExistException(getName(), e.database());
         }
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 2ab2a0f53..587069ada 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
@@ -428,12 +428,24 @@ public class FlinkCatalogTest {
     }
 
     @Test
-    public void testCreateDb_DatabaseWithPropertiesException() {
+    public void testCreateDb_DatabaseWithProperties() throws Exception {
         CatalogDatabaseImpl database =
                 new CatalogDatabaseImpl(Collections.singletonMap("haa", 
"ccc"), null);
-        assertThatThrownBy(() -> 
catalog.createDatabase(path1.getDatabaseName(), database, false))
-                .isInstanceOf(UnsupportedOperationException.class)
-                .hasMessage("Create database with properties is unsupported.");
+        catalog.createDatabase(path1.getDatabaseName(), database, false);
+        assertThat(catalog.databaseExists(path1.getDatabaseName())).isTrue();
+        // TODO filesystem catalog will ignore all properties
+        
assertThat(catalog.getDatabase(path1.getDatabaseName()).getProperties().isEmpty()).isTrue();
+
+        // File system catalog doesn't support path for database.
+        CatalogDatabaseImpl databaseWithPath =
+                new CatalogDatabaseImpl(Collections.singletonMap("location", 
"/tmp"), null);
+        assertThatThrownBy(
+                        () ->
+                                catalog.createDatabase(
+                                        "test-database-with-location", 
databaseWithPath, false))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessage(
+                        "Cannot specify location for a database when using 
fileSystem catalog.");
     }
 
     @Test