From 988369e6ab8444e3582216d20e0702d8f1ffaa9a Mon Sep 17 00:00:00 2001
From: "wenchao.wu" <wenchao.wu@shopee.com>
Date: Wed, 17 Apr 2024 14:32:49 +0800
Subject: [PATCH] [PAIMON-3219][Flink] support create database with properties
 when inner catalog supports.

---
 .../org/apache/paimon/catalog/Catalog.java    |  8 +++++
 .../org/apache/paimon/jdbc/JdbcCatalog.java   |  5 ++++
 .../org/apache/paimon/flink/FlinkCatalog.java | 29 ++++++++++---------
 .../org/apache/paimon/hive/HiveCatalog.java   |  5 ++++
 4 files changed, 33 insertions(+), 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
index 99b71e8deb1e..e179088be61b 100644
--- 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
index 7e7718b5bee9..670b4cec8c31 100644
--- 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
index 98ca840f0e65..cb0e66467812 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
@@ -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
index 32d25e7db199..e7eed57ef94d 100644
--- 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);