diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 0c1716260f71..05f10032ffe4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -119,7 +119,7 @@ public void dropPartition(Identifier identifier, Map<String, String> partitionSp
                 Collections.singletonList(partitionSpec), BatchWriteBuilder.COMMIT_IDENTIFIER);
     }
 
-    protected abstract void createDatabaseImpl(String name);
+    protected abstract void createDatabaseImpl(String name) throws DatabaseAlreadyExistException;
 
     @Override
     public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade)
@@ -134,14 +134,14 @@ public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade
             throw new DatabaseNotExistException(name);
         }
 
-        if (!cascade && listTables(name).size() > 0) {
+        if (!cascade && !listTables(name).isEmpty()) {
             throw new DatabaseNotEmptyException(name);
         }
 
         dropDatabaseImpl(name);
     }
 
-    protected abstract void dropDatabaseImpl(String name);
+    protected abstract void dropDatabaseImpl(String name) throws DatabaseNotExistException;
 
     @Override
     public List<String> listTables(String databaseName) throws DatabaseNotExistException {
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
index 03523da580ee..c400826b73f5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/FileSystemCatalog.java
@@ -72,12 +72,12 @@ protected boolean databaseExistsImpl(String databaseName) {
     }
 
     @Override
-    protected void createDatabaseImpl(String name) {
+    public void createDatabaseImpl(String name) {
         uncheck(() -> fileIO.mkdirs(newDatabasePath(name)));
     }
 
     @Override
-    protected void dropDatabaseImpl(String name) {
+    public void dropDatabaseImpl(String name) {
         uncheck(() -> fileIO.delete(newDatabasePath(name), true));
     }
 
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java
index d85de5e7629e..517160204526 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogTestBase.java
@@ -54,7 +54,7 @@ public abstract class CatalogTestBase {
     @TempDir java.nio.file.Path tempFile;
     protected String warehouse;
     protected FileIO fileIO;
-    protected Catalog catalog;
+    protected AbstractCatalog catalog;
     protected static final Schema DEFAULT_TABLE_SCHEMA =
             new Schema(
                     Lists.newArrayList(
@@ -140,6 +140,21 @@ public void testCreateDatabase() throws Exception {
                 .doesNotThrowAnyException();
     }
 
+    @Test
+    public void testCreateDatabaseImpl() throws Exception {
+        // Create database creates a new database when it does not exist
+        catalog.createDatabaseImpl("new_db");
+        boolean exists = catalog.databaseExists("new_db");
+        assertThat(exists).isTrue();
+
+        catalog.createDatabaseImpl("existing_db");
+
+        // Create database throws DatabaseAlreadyExistException when database already exists
+        assertThatExceptionOfType(Catalog.DatabaseAlreadyExistException.class)
+                .isThrownBy(() -> catalog.createDatabaseImpl("existing_db"))
+                .withMessage("Database existing_db already exists.");
+    }
+
     @Test
     public void testDropDatabase() throws Exception {
         // Drop database deletes the database when it exists and there are no tables
@@ -173,6 +188,29 @@ public void testDropDatabase() throws Exception {
                 .withMessage("Database db_with_tables is not empty.");
     }
 
+    @Test
+    public void testDropDatabaseImpl() throws Exception {
+        // Drop database deletes the database when it exists and there are no tables
+        catalog.createDatabase("db_to_drop", false);
+        catalog.dropDatabase("db_to_drop", false, false);
+        boolean exists = catalog.databaseExists("db_to_drop");
+        assertThat(exists).isFalse();
+
+        // Drop database does throw exception when database does not exist
+        assertThatExceptionOfType(Catalog.DatabaseNotExistException.class)
+                .isThrownBy(() -> catalog.dropDatabaseImpl("non_existing_db"))
+                .withMessage("Database non_existing_db does not exist.");
+
+        // Drop database deletes all tables in the database
+        catalog.createDatabase("db_to_drop", false);
+        catalog.createTable(Identifier.create("db_to_drop", "table1"), DEFAULT_TABLE_SCHEMA, false);
+        catalog.createTable(Identifier.create("db_to_drop", "table2"), DEFAULT_TABLE_SCHEMA, false);
+
+        catalog.dropDatabaseImpl("db_to_drop");
+        exists = catalog.databaseExists("db_to_drop");
+        assertThat(exists).isFalse();
+    }
+
     @Test
     public void testListTables() throws Exception {
         // List tables returns an empty list when there are no tables in the 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 f7cfc76148fd..42cc831e653d 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
@@ -208,10 +208,7 @@ public List<String> listDatabases() {
     @Override
     protected boolean databaseExistsImpl(String databaseName) {
         try {
-            client.getDatabase(databaseName);
-            return true;
-        } catch (NoSuchObjectException e) {
-            return false;
+            return getDatabase(databaseName) != null;
         } catch (TException e) {
             throw new RuntimeException(
                     "Failed to determine if database " + databaseName + " exists", e);
@@ -219,32 +216,52 @@ protected boolean databaseExistsImpl(String databaseName) {
     }
 
     @Override
-    protected void createDatabaseImpl(String name) {
+    protected void createDatabaseImpl(String name) throws DatabaseAlreadyExistException {
         try {
-            Path databasePath = newDatabasePath(name);
-            locationHelper.createPathIfRequired(databasePath, fileIO);
-
-            Database database = new Database();
-            database.setName(name);
-            locationHelper.specifyDatabaseLocation(databasePath, database);
-            client.createDatabase(database);
+            Database database = getDatabase(name);
+            if (database == null) {
+                Path databasePath = newDatabasePath(name);
+                locationHelper.createPathIfRequired(databasePath, fileIO);
+
+                database = new Database();
+                database.setName(name);
+                locationHelper.specifyDatabaseLocation(databasePath, database);
+                client.createDatabase(database);
+            } else {
+                throw new DatabaseAlreadyExistException(name);
+            }
         } catch (TException | IOException e) {
             throw new RuntimeException("Failed to create database " + name, e);
         }
     }
 
     @Override
-    protected void dropDatabaseImpl(String name) {
+    protected void dropDatabaseImpl(String name) throws DatabaseNotExistException {
         try {
-            Database database = client.getDatabase(name);
-            String location = locationHelper.getDatabaseLocation(database);
-            locationHelper.dropPathIfRequired(new Path(location), fileIO);
-            client.dropDatabase(name, true, false, true);
+            Database database = getDatabase(name);
+            if (database != null) {
+                String location = locationHelper.getDatabaseLocation(database);
+                locationHelper.dropPathIfRequired(new Path(location), fileIO);
+                client.dropDatabase(name, true, false, true);
+            } else {
+                throw new DatabaseNotExistException(name);
+            }
         } catch (TException | IOException e) {
             throw new RuntimeException("Failed to drop database " + name, e);
         }
     }
 
+    private Database getDatabase(String databaseName) throws TException {
+        try {
+            return client.getDatabase(databaseName);
+        } catch (NoSuchObjectException e) {
+            return null;
+        } catch (TException e) {
+            throw new RuntimeException(
+                    "Failed to determine if database " + databaseName + " exists", e);
+        }
+    }
+
     @Override
     protected List<String> listTablesImpl(String databaseName) {
         try {