[PAIMON-3219][Flink] support create database with properties when inn…

…er catalog supports.
Stephen0421 · Apr 17, 2024 · 988369e · 988369e
1 parent c05cbca
commit 988369e
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 14 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -249,6 +249,14 @@ default boolean caseSensitive() {
         return true;
     }
 
+    /**
+     * Return a boolean that indicates whether this catalog supports create database with
+     * properties.
+     */
+    default boolean supportDatabaseProperties() {
+        return false;
+    }
+
     /** Exception for trying to drop on a database that is not empty. */
     class DatabaseNotEmptyException extends Exception {
         private static final String MSG = "Database %s is not empty.";

diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java
@@ -353,6 +353,11 @@ public Optional<CatalogLockContext> lockContext() {
         return Optional.of(new JdbcCatalogLockContext(connections, catalogKey, options));
     }
 
+    @Override
+    public boolean supportDatabaseProperties() {
+        return true;
+    }
+
     private Lock lock(Identifier identifier) {
         if (!lockEnabled()) {
             return new Lock.EmptyLock();

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
@@ -191,22 +191,23 @@ public CatalogDatabase getDatabase(String databaseName)
     @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.");
-            }
+        try {
+            if (database != null) {
+                if (database.getProperties().size() > 0 && !catalog.supportDatabaseProperties()) {
+                    throw new UnsupportedOperationException(
+                            "Create database with properties is unsupported.");
+                }
 
-            if (database.getDescription().isPresent()
-                    && !database.getDescription().get().equals("")) {
-                throw new UnsupportedOperationException(
-                        "Create database with description is unsupported.");
-            }
-        }
+                if (database.getDescription().isPresent()
+                        && !database.getDescription().get().equals("")) {
+                    throw new UnsupportedOperationException(
+                            "Create database with description is unsupported.");
+                }
 
-        try {
-            catalog.createDatabase(name, ignoreIfExists);
+                catalog.createDatabase(name, ignoreIfExists, database.getProperties());
+            } else {
+                catalog.createDatabase(name, ignoreIfExists);
+            }
         } catch (Catalog.DatabaseAlreadyExistException e) {
             throw new DatabaseAlreadyExistException(getName(), e.database());
         }

diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -236,6 +236,11 @@ protected void createDatabaseImpl(String name, Map<String, String> properties) {
         }
     }
 
+    @Override
+    public boolean supportDatabaseProperties() {
+        return true;
+    }
+
     private Database convertToHiveDatabase(String name, Map<String, String> properties) {
         Database database = new Database();
         database.setName(name);