diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/DefaultErrorHandler.java b/paimon-core/src/main/java/org/apache/paimon/rest/DefaultErrorHandler.java
index 1a8618c1c603..ce2cbb56ae24 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/DefaultErrorHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/DefaultErrorHandler.java
@@ -18,8 +18,10 @@
 
 package org.apache.paimon.rest;
 
+import org.apache.paimon.rest.exceptions.AlreadyExistsException;
 import org.apache.paimon.rest.exceptions.BadRequestException;
 import org.apache.paimon.rest.exceptions.ForbiddenException;
+import org.apache.paimon.rest.exceptions.NoSuchResourceException;
 import org.apache.paimon.rest.exceptions.NotAuthorizedException;
 import org.apache.paimon.rest.exceptions.RESTException;
 import org.apache.paimon.rest.exceptions.ServiceFailureException;
@@ -28,6 +30,7 @@
 
 /** Default error handler. */
 public class DefaultErrorHandler extends ErrorHandler {
+
     private static final ErrorHandler INSTANCE = new DefaultErrorHandler();
 
     public static ErrorHandler getInstance() {
@@ -36,26 +39,32 @@ public static ErrorHandler getInstance() {
 
     @Override
     public void accept(ErrorResponse error) {
-        int code = error.code();
+        int code = error.getCode();
+        String message = error.getMessage();
         switch (code) {
             case 400:
-                throw new BadRequestException(
-                        String.format("Malformed request: %s", error.message()));
+                throw new BadRequestException(String.format("Malformed request: %s", message));
             case 401:
-                throw new NotAuthorizedException("Not authorized: %s", error.message());
+                throw new NotAuthorizedException("Not authorized: %s", message);
             case 403:
-                throw new ForbiddenException("Forbidden: %s", error.message());
+                throw new ForbiddenException("Forbidden: %s", message);
+            case 404:
+                throw new NoSuchResourceException("%s", message);
             case 405:
             case 406:
                 break;
+            case 409:
+                throw new AlreadyExistsException("%s", message);
             case 500:
-                throw new ServiceFailureException("Server error: %s", error.message());
+                throw new ServiceFailureException("Server error: %s", message);
             case 501:
-                throw new UnsupportedOperationException(error.message());
+                throw new UnsupportedOperationException(message);
             case 503:
-                throw new ServiceUnavailableException("Service unavailable: %s", error.message());
+                throw new ServiceUnavailableException("Service unavailable: %s", message);
+            default:
+                break;
         }
 
-        throw new RESTException("Unable to process: %s", error.message());
+        throw new RESTException("Unable to process: %s", message);
     }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/HttpClient.java b/paimon-core/src/main/java/org/apache/paimon/rest/HttpClient.java
index e092711e5f97..97696aef09ed 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/HttpClient.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/HttpClient.java
@@ -95,6 +95,23 @@ public <T extends RESTResponse> T post(
         }
     }
 
+    @Override
+    public <T extends RESTResponse> T delete(
+            String path, RESTRequest body, Map<String, String> headers) {
+        try {
+            RequestBody requestBody = buildRequestBody(body);
+            Request request =
+                    new Request.Builder()
+                            .url(uri + path)
+                            .delete(requestBody)
+                            .headers(Headers.of(headers))
+                            .build();
+            return exec(request, null);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     @Override
     public void close() throws IOException {
         okHttpClient.dispatcher().cancelAll();
@@ -111,10 +128,13 @@ private <T extends RESTResponse> T exec(Request request, Class<T> responseType)
                                 response.code());
                 errorHandler.accept(error);
             }
-            if (responseBodyStr == null) {
+            if (responseType != null && responseBodyStr != null) {
+                return mapper.readValue(responseBodyStr, responseType);
+            } else if (responseType == null) {
+                return null;
+            } else {
                 throw new RESTException("response body is null.");
             }
-            return mapper.readValue(responseBodyStr, responseType);
         } catch (Exception e) {
             throw new RESTException(e, "rest exception");
         }
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index f3007bf4bf02..3c2538df0ca2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -29,12 +29,21 @@
 import org.apache.paimon.rest.auth.AuthSession;
 import org.apache.paimon.rest.auth.CredentialsProvider;
 import org.apache.paimon.rest.auth.CredentialsProviderFactory;
+import org.apache.paimon.rest.exceptions.AlreadyExistsException;
+import org.apache.paimon.rest.exceptions.NoSuchResourceException;
+import org.apache.paimon.rest.requests.CreateDatabaseRequest;
+import org.apache.paimon.rest.requests.DropDatabaseRequest;
 import org.apache.paimon.rest.responses.ConfigResponse;
+import org.apache.paimon.rest.responses.CreateDatabaseResponse;
+import org.apache.paimon.rest.responses.DatabaseName;
+import org.apache.paimon.rest.responses.GetDatabaseResponse;
+import org.apache.paimon.rest.responses.ListDatabasesResponse;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.table.Table;
 
 import org.apache.paimon.shade.guava30.com.google.common.annotations.VisibleForTesting;
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 
 import java.time.Duration;
@@ -42,6 +51,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
 
 import static org.apache.paimon.utils.ThreadPoolUtils.createScheduledThreadPool;
 
@@ -113,24 +123,49 @@ public FileIO fileIO() {
 
     @Override
     public List<String> listDatabases() {
-        throw new UnsupportedOperationException();
+        ListDatabasesResponse response =
+                client.get(resourcePaths.databases(), ListDatabasesResponse.class, headers());
+        if (response.getDatabases() != null) {
+            return response.getDatabases().stream()
+                    .map(DatabaseName::getName)
+                    .collect(Collectors.toList());
+        }
+        return ImmutableList.of();
     }
 
     @Override
     public void createDatabase(String name, boolean ignoreIfExists, Map<String, String> properties)
             throws DatabaseAlreadyExistException {
-        throw new UnsupportedOperationException();
+        CreateDatabaseRequest request = new CreateDatabaseRequest(name, ignoreIfExists, properties);
+        try {
+            client.post(
+                    resourcePaths.databases(), request, CreateDatabaseResponse.class, headers());
+        } catch (AlreadyExistsException e) {
+            throw new DatabaseAlreadyExistException(name);
+        }
     }
 
     @Override
     public Database getDatabase(String name) throws DatabaseNotExistException {
-        throw new UnsupportedOperationException();
+        try {
+            GetDatabaseResponse response =
+                    client.get(resourcePaths.database(name), GetDatabaseResponse.class, headers());
+            return new Database.DatabaseImpl(
+                    name, response.options(), response.comment().orElseGet(() -> null));
+        } catch (NoSuchResourceException e) {
+            throw new DatabaseNotExistException(name);
+        }
     }
 
     @Override
     public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade)
             throws DatabaseNotExistException, DatabaseNotEmptyException {
-        throw new UnsupportedOperationException();
+        DropDatabaseRequest request = new DropDatabaseRequest(ignoreIfNotExists, cascade);
+        try {
+            client.delete(resourcePaths.database(name), request, headers());
+        } catch (NoSuchResourceException e) {
+            throw new DatabaseNotExistException(name);
+        }
     }
 
     @Override
@@ -208,7 +243,7 @@ public void close() throws Exception {
     Map<String, String> fetchOptionsFromServer(
             Map<String, String> headers, Map<String, String> clientProperties) {
         ConfigResponse response =
-                client.get(ResourcePaths.V1_CONFIG, ConfigResponse.class, headers());
+                client.get(ResourcePaths.V1_CONFIG, ConfigResponse.class, headers);
         return response.merge(clientProperties);
     }
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogInternalOptions.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogInternalOptions.java
index 62a8bf134ae5..722010923c46 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogInternalOptions.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalogInternalOptions.java
@@ -33,4 +33,9 @@ public class RESTCatalogInternalOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription("REST Catalog auth credentials provider.");
+    public static final ConfigOption<String> DATABASE_COMMENT =
+            ConfigOptions.key("comment")
+                    .stringType()
+                    .defaultValue(null)
+                    .withDescription("REST Catalog database comment.");
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTClient.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTClient.java
index feeed06a417a..d0244f309ef4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTClient.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTClient.java
@@ -28,4 +28,6 @@ public interface RESTClient extends Closeable {
 
     <T extends RESTResponse> T post(
             String path, RESTRequest body, Class<T> responseType, Map<String, String> headers);
+
+    <T extends RESTResponse> T delete(String path, RESTRequest body, Map<String, String> headers);
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTMessage.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTMessage.java
index 6cb0b6fa6573..31d46df7ef0f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTMessage.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTMessage.java
@@ -18,5 +18,8 @@
 
 package org.apache.paimon.rest;
 
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+
 /** Interface to mark both REST requests and responses. */
+@JsonIgnoreProperties(ignoreUnknown = true)
 public interface RESTMessage {}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/ResourcePaths.java b/paimon-core/src/main/java/org/apache/paimon/rest/ResourcePaths.java
index aaca6193802d..a6d0000a225b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/ResourcePaths.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/ResourcePaths.java
@@ -18,10 +18,13 @@
 
 package org.apache.paimon.rest;
 
+import java.util.StringJoiner;
+
 /** Resource paths for REST catalog. */
 public class ResourcePaths {
 
     public static final String V1_CONFIG = "/api/v1/config";
+    private static final StringJoiner SLASH = new StringJoiner("/");
 
     public static ResourcePaths forCatalogProperties(String prefix) {
         return new ResourcePaths(prefix);
@@ -32,4 +35,12 @@ public static ResourcePaths forCatalogProperties(String prefix) {
     public ResourcePaths(String prefix) {
         this.prefix = prefix;
     }
+
+    public String databases() {
+        return SLASH.add("api").add("v1").add(prefix).add("databases").toString();
+    }
+
+    public String database(String databaseName) {
+        return SLASH.add("api").add("v1").add(prefix).add("databases").add(databaseName).toString();
+    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/auth/AuthSession.java b/paimon-core/src/main/java/org/apache/paimon/rest/auth/AuthSession.java
index 74efb8508a06..3ca7590e5f96 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/auth/AuthSession.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/auth/AuthSession.java
@@ -33,9 +33,10 @@
 public class AuthSession {
 
     static final int TOKEN_REFRESH_NUM_RETRIES = 5;
+    static final long MIN_REFRESH_WAIT_MILLIS = 10;
+    static final long MAX_REFRESH_WINDOW_MILLIS = 300_000; // 5 minutes
+
     private static final Logger log = LoggerFactory.getLogger(AuthSession.class);
-    private static final long MAX_REFRESH_WINDOW_MILLIS = 300_000; // 5 minutes
-    private static final long MIN_REFRESH_WAIT_MILLIS = 10;
     private final CredentialsProvider credentialsProvider;
     private volatile Map<String, String> headers;
 
@@ -76,12 +77,38 @@ public Map<String, String> getHeaders() {
         return headers;
     }
 
+    public Boolean refresh() {
+        if (this.credentialsProvider.supportRefresh()
+                && this.credentialsProvider.keepRefreshed()
+                && this.credentialsProvider.expiresInMills().isPresent()) {
+            boolean isSuccessful = this.credentialsProvider.refresh();
+            if (isSuccessful) {
+                Map<String, String> currentHeaders = this.headers;
+                this.headers =
+                        RESTUtil.merge(currentHeaders, this.credentialsProvider.authHeader());
+            }
+            return isSuccessful;
+        }
+
+        return false;
+    }
+
     @VisibleForTesting
     static void scheduleTokenRefresh(
             ScheduledExecutorService executor, AuthSession session, long expiresAtMillis) {
         scheduleTokenRefresh(executor, session, expiresAtMillis, 0);
     }
 
+    @VisibleForTesting
+    static long getTimeToWaitByExpiresInMills(long expiresInMillis) {
+        // how much ahead of time to start the refresh to allow it to complete
+        long refreshWindowMillis = Math.min(expiresInMillis, MAX_REFRESH_WINDOW_MILLIS);
+        // how much time to wait before expiration
+        long waitIntervalMillis = expiresInMillis - refreshWindowMillis;
+        // how much time to actually wait
+        return Math.max(waitIntervalMillis, MIN_REFRESH_WAIT_MILLIS);
+    }
+
     private static void scheduleTokenRefresh(
             ScheduledExecutorService executor,
             AuthSession session,
@@ -89,12 +116,7 @@ private static void scheduleTokenRefresh(
             int retryTimes) {
         if (retryTimes < TOKEN_REFRESH_NUM_RETRIES) {
             long expiresInMillis = expiresAtMillis - System.currentTimeMillis();
-            // how much ahead of time to start the refresh to allow it to complete
-            long refreshWindowMillis = Math.min(expiresInMillis, MAX_REFRESH_WINDOW_MILLIS);
-            // how much time to wait before expiration
-            long waitIntervalMillis = expiresInMillis - refreshWindowMillis;
-            // how much time to actually wait
-            long timeToWait = Math.max(waitIntervalMillis, MIN_REFRESH_WAIT_MILLIS);
+            long timeToWait = getTimeToWaitByExpiresInMills(expiresInMillis);
 
             executor.schedule(
                     () -> {
@@ -118,20 +140,4 @@ private static void scheduleTokenRefresh(
             log.warn("Failed to refresh token after {} retries.", TOKEN_REFRESH_NUM_RETRIES);
         }
     }
-
-    public Boolean refresh() {
-        if (this.credentialsProvider.supportRefresh()
-                && this.credentialsProvider.keepRefreshed()
-                && this.credentialsProvider.expiresInMills().isPresent()) {
-            boolean isSuccessful = this.credentialsProvider.refresh();
-            if (isSuccessful) {
-                Map<String, String> currentHeaders = this.headers;
-                this.headers =
-                        RESTUtil.merge(currentHeaders, this.credentialsProvider.authHeader());
-            }
-            return isSuccessful;
-        }
-
-        return false;
-    }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/exceptions/AlreadyExistsException.java b/paimon-core/src/main/java/org/apache/paimon/rest/exceptions/AlreadyExistsException.java
new file mode 100644
index 000000000000..8e30c8375bf9
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/exceptions/AlreadyExistsException.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.exceptions;
+
+/** Exception thrown on HTTP 409 means a resource already exists. */
+public class AlreadyExistsException extends RESTException {
+
+    public AlreadyExistsException(String message, Object... args) {
+        super(message, args);
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/exceptions/NoSuchResourceException.java b/paimon-core/src/main/java/org/apache/paimon/rest/exceptions/NoSuchResourceException.java
new file mode 100644
index 000000000000..cc4c7881f465
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/exceptions/NoSuchResourceException.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.exceptions;
+
+/** Exception thrown on HTTP 404 means a resource not exists. */
+public class NoSuchResourceException extends RESTException {
+
+    public NoSuchResourceException(String message, Object... args) {
+        super(message, args);
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/requests/CreateDatabaseRequest.java b/paimon-core/src/main/java/org/apache/paimon/rest/requests/CreateDatabaseRequest.java
new file mode 100644
index 000000000000..6067bf544b87
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/requests/CreateDatabaseRequest.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.requests;
+
+import org.apache.paimon.rest.RESTRequest;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Map;
+
+/** Request for creating database. */
+public class CreateDatabaseRequest implements RESTRequest {
+
+    private static final String FIELD_NAME = "name";
+    private static final String FIELD_IGNORE_IF_EXISTS = "ignoreIfExists";
+    private static final String FIELD_OPTIONS = "options";
+
+    @JsonProperty(FIELD_NAME)
+    private String name;
+
+    @JsonProperty(FIELD_IGNORE_IF_EXISTS)
+    private boolean ignoreIfExists;
+
+    @JsonProperty(FIELD_OPTIONS)
+    private Map<String, String> options;
+
+    @JsonCreator
+    public CreateDatabaseRequest(
+            @JsonProperty(FIELD_NAME) String name,
+            @JsonProperty(FIELD_IGNORE_IF_EXISTS) boolean ignoreIfExists,
+            @JsonProperty(FIELD_OPTIONS) Map<String, String> options) {
+        this.name = name;
+        this.ignoreIfExists = ignoreIfExists;
+        this.options = options;
+    }
+
+    @JsonGetter(FIELD_NAME)
+    public String getName() {
+        return name;
+    }
+
+    @JsonGetter(FIELD_IGNORE_IF_EXISTS)
+    public boolean getIgnoreIfExists() {
+        return ignoreIfExists;
+    }
+
+    @JsonGetter(FIELD_OPTIONS)
+    public Map<String, String> getOptions() {
+        return options;
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/requests/DropDatabaseRequest.java b/paimon-core/src/main/java/org/apache/paimon/rest/requests/DropDatabaseRequest.java
new file mode 100644
index 000000000000..d97f211c1caa
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/requests/DropDatabaseRequest.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.requests;
+
+import org.apache.paimon.rest.RESTRequest;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** Request for DropDatabase. */
+public class DropDatabaseRequest implements RESTRequest {
+
+    private static final String FIELD_IGNORE_IF_EXISTS = "ignoreIfExists";
+    private static final String FIELD_CASCADE = "cascade";
+
+    @JsonProperty(FIELD_IGNORE_IF_EXISTS)
+    private final boolean ignoreIfNotExists;
+
+    @JsonProperty(FIELD_CASCADE)
+    private final boolean cascade;
+
+    @JsonCreator
+    public DropDatabaseRequest(
+            @JsonProperty(FIELD_IGNORE_IF_EXISTS) boolean ignoreIfNotExists,
+            @JsonProperty(FIELD_CASCADE) boolean cascade) {
+        this.ignoreIfNotExists = ignoreIfNotExists;
+        this.cascade = cascade;
+    }
+
+    @JsonGetter(FIELD_IGNORE_IF_EXISTS)
+    public boolean getIgnoreIfNotExists() {
+        return ignoreIfNotExists;
+    }
+
+    @JsonGetter(FIELD_CASCADE)
+    public boolean getCascade() {
+        return cascade;
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/responses/ConfigResponse.java b/paimon-core/src/main/java/org/apache/paimon/rest/responses/ConfigResponse.java
index 903cfc84b46d..e8fff88b09c2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/responses/ConfigResponse.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/responses/ConfigResponse.java
@@ -23,17 +23,16 @@
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
 import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
-import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.beans.ConstructorProperties;
 import java.util.Map;
 import java.util.Objects;
 
 /** Response for getting config. */
-@JsonIgnoreProperties(ignoreUnknown = true)
 public class ConfigResponse implements RESTResponse {
+
     private static final String FIELD_DEFAULTS = "defaults";
     private static final String FIELD_OVERRIDES = "overrides";
 
@@ -43,8 +42,10 @@ public class ConfigResponse implements RESTResponse {
     @JsonProperty(FIELD_OVERRIDES)
     private Map<String, String> overrides;
 
-    @ConstructorProperties({FIELD_DEFAULTS, FIELD_OVERRIDES})
-    public ConfigResponse(Map<String, String> defaults, Map<String, String> overrides) {
+    @JsonCreator
+    public ConfigResponse(
+            @JsonProperty(FIELD_DEFAULTS) Map<String, String> defaults,
+            @JsonProperty(FIELD_OVERRIDES) Map<String, String> overrides) {
         this.defaults = defaults;
         this.overrides = overrides;
     }
@@ -65,12 +66,12 @@ public Map<String, String> merge(Map<String, String> clientProperties) {
     }
 
     @JsonGetter(FIELD_DEFAULTS)
-    public Map<String, String> defaults() {
+    public Map<String, String> getDefaults() {
         return defaults;
     }
 
     @JsonGetter(FIELD_OVERRIDES)
-    public Map<String, String> overrides() {
+    public Map<String, String> getOverrides() {
         return overrides;
     }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/responses/CreateDatabaseResponse.java b/paimon-core/src/main/java/org/apache/paimon/rest/responses/CreateDatabaseResponse.java
new file mode 100644
index 000000000000..43c99254f399
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/responses/CreateDatabaseResponse.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.responses;
+
+import org.apache.paimon.rest.RESTResponse;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Map;
+
+/** Response for creating database. */
+public class CreateDatabaseResponse implements RESTResponse {
+
+    private static final String FIELD_NAME = "name";
+    private static final String FIELD_OPTIONS = "options";
+
+    @JsonProperty(FIELD_NAME)
+    private String name;
+
+    @JsonProperty(FIELD_OPTIONS)
+    private Map<String, String> options;
+
+    @JsonCreator
+    public CreateDatabaseResponse(
+            @JsonProperty(FIELD_NAME) String name,
+            @JsonProperty(FIELD_OPTIONS) Map<String, String> options) {
+        this.name = name;
+        this.options = options;
+    }
+
+    @JsonGetter(FIELD_NAME)
+    public String getName() {
+        return name;
+    }
+
+    @JsonGetter(FIELD_OPTIONS)
+    public Map<String, String> getOptions() {
+        return options;
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/responses/DatabaseName.java b/paimon-core/src/main/java/org/apache/paimon/rest/responses/DatabaseName.java
new file mode 100644
index 000000000000..9a93b2fd1e3d
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/responses/DatabaseName.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.responses;
+
+import org.apache.paimon.rest.RESTMessage;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** Class for Database entity. */
+public class DatabaseName implements RESTMessage {
+
+    private static final String FIELD_NAME = "name";
+
+    @JsonProperty(FIELD_NAME)
+    private String name;
+
+    @JsonCreator
+    public DatabaseName(@JsonProperty(FIELD_NAME) String name) {
+        this.name = name;
+    }
+
+    @JsonGetter(FIELD_NAME)
+    public String getName() {
+        return this.name;
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java b/paimon-core/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
index 685fe53071b6..d24c8f0f9936 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/responses/ErrorResponse.java
@@ -18,10 +18,12 @@
 
 package org.apache.paimon.rest.responses;
 
+import org.apache.paimon.rest.RESTResponse;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
-import java.beans.ConstructorProperties;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.ArrayList;
@@ -29,7 +31,8 @@
 import java.util.List;
 
 /** Response for error. */
-public class ErrorResponse {
+public class ErrorResponse implements RESTResponse {
+
     private static final String FIELD_MESSAGE = "message";
     private static final String FIELD_CODE = "code";
     private static final String FIELD_STACK = "stack";
@@ -49,8 +52,11 @@ public ErrorResponse(String message, Integer code) {
         this.stack = new ArrayList<String>();
     }
 
-    @ConstructorProperties({FIELD_MESSAGE, FIELD_CODE, FIELD_STACK})
-    public ErrorResponse(String message, int code, List<String> stack) {
+    @JsonCreator
+    public ErrorResponse(
+            @JsonProperty(FIELD_MESSAGE) String message,
+            @JsonProperty(FIELD_CODE) int code,
+            @JsonProperty(FIELD_STACK) List<String> stack) {
         this.message = message;
         this.code = code;
         this.stack = stack;
@@ -63,17 +69,17 @@ public ErrorResponse(String message, int code, Throwable throwable) {
     }
 
     @JsonGetter(FIELD_MESSAGE)
-    public String message() {
+    public String getMessage() {
         return message;
     }
 
     @JsonGetter(FIELD_CODE)
-    public Integer code() {
+    public Integer getCode() {
         return code;
     }
 
     @JsonGetter(FIELD_STACK)
-    public List<String> stack() {
+    public List<String> getStack() {
         return stack;
     }
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/responses/GetDatabaseResponse.java b/paimon-core/src/main/java/org/apache/paimon/rest/responses/GetDatabaseResponse.java
new file mode 100644
index 000000000000..f8f7c8794b7b
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/responses/GetDatabaseResponse.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.responses;
+
+import org.apache.paimon.catalog.Database;
+import org.apache.paimon.rest.RESTResponse;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.paimon.rest.RESTCatalogInternalOptions.DATABASE_COMMENT;
+
+/** Response for getting database. */
+public class GetDatabaseResponse implements RESTResponse, Database {
+
+    private static final String FIELD_NAME = "name";
+    private static final String FIELD_OPTIONS = "options";
+
+    @JsonProperty(FIELD_NAME)
+    private final String name;
+
+    @JsonProperty(FIELD_OPTIONS)
+    private final Map<String, String> options;
+
+    @JsonCreator
+    public GetDatabaseResponse(
+            @JsonProperty(FIELD_NAME) String name,
+            @JsonProperty(FIELD_OPTIONS) Map<String, String> options) {
+        this.name = name;
+        this.options = options;
+    }
+
+    @JsonGetter(FIELD_NAME)
+    public String getName() {
+        return name;
+    }
+
+    @JsonGetter(FIELD_OPTIONS)
+    public Map<String, String> getOptions() {
+        return options;
+    }
+
+    @Override
+    public String name() {
+        return this.getName();
+    }
+
+    @Override
+    public Map<String, String> options() {
+        return this.getOptions();
+    }
+
+    @Override
+    public Optional<String> comment() {
+        return Optional.ofNullable(
+                this.options.getOrDefault(DATABASE_COMMENT.key(), DATABASE_COMMENT.defaultValue()));
+    }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/responses/ListDatabasesResponse.java b/paimon-core/src/main/java/org/apache/paimon/rest/responses/ListDatabasesResponse.java
new file mode 100644
index 000000000000..38773f354b77
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/responses/ListDatabasesResponse.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest.responses;
+
+import org.apache.paimon.rest.RESTResponse;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.List;
+
+/** Response for listing databases. */
+public class ListDatabasesResponse implements RESTResponse {
+    private static final String FIELD_DATABASES = "databases";
+
+    @JsonProperty(FIELD_DATABASES)
+    private List<DatabaseName> databases;
+
+    @JsonCreator
+    public ListDatabasesResponse(@JsonProperty(FIELD_DATABASES) List<DatabaseName> databases) {
+        this.databases = databases;
+    }
+
+    @JsonGetter(FIELD_DATABASES)
+    public List<DatabaseName> getDatabases() {
+        return this.databases;
+    }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/DefaultErrorHandlerTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/DefaultErrorHandlerTest.java
index 1f1b9c01aace..340e38f6a7f8 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/DefaultErrorHandlerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/DefaultErrorHandlerTest.java
@@ -18,8 +18,10 @@
 
 package org.apache.paimon.rest;
 
+import org.apache.paimon.rest.exceptions.AlreadyExistsException;
 import org.apache.paimon.rest.exceptions.BadRequestException;
 import org.apache.paimon.rest.exceptions.ForbiddenException;
+import org.apache.paimon.rest.exceptions.NoSuchResourceException;
 import org.apache.paimon.rest.exceptions.NotAuthorizedException;
 import org.apache.paimon.rest.exceptions.RESTException;
 import org.apache.paimon.rest.exceptions.ServiceFailureException;
@@ -54,10 +56,16 @@ public void testHandleErrorResponse() {
         assertThrows(
                 ForbiddenException.class,
                 () -> defaultErrorHandler.accept(generateErrorResponse(403)));
+        assertThrows(
+                NoSuchResourceException.class,
+                () -> defaultErrorHandler.accept(generateErrorResponse(404)));
         assertThrows(
                 RESTException.class, () -> defaultErrorHandler.accept(generateErrorResponse(405)));
         assertThrows(
                 RESTException.class, () -> defaultErrorHandler.accept(generateErrorResponse(406)));
+        assertThrows(
+                AlreadyExistsException.class,
+                () -> defaultErrorHandler.accept(generateErrorResponse(409)));
         assertThrows(
                 ServiceFailureException.class,
                 () -> defaultErrorHandler.accept(generateErrorResponse(500)));
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java
index 17c13b932fd2..f12af12a9d35 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/HttpClientTest.java
@@ -110,6 +110,20 @@ public void testPostFail() {
         verify(errorHandler, times(1)).accept(any());
     }
 
+    @Test
+    public void testDeleteSuccess() {
+        mockHttpCallWithCode(mockResponseDataStr, 200);
+        MockRESTData response = httpClient.delete(MOCK_PATH, mockResponseData, headers);
+        verify(errorHandler, times(0)).accept(any());
+    }
+
+    @Test
+    public void testDeleteFail() {
+        mockHttpCallWithCode(mockResponseDataStr, 400);
+        httpClient.delete(MOCK_PATH, mockResponseData, headers);
+        verify(errorHandler, times(1)).accept(any());
+    }
+
     private Map<String, String> headers(String token) {
         Map<String, String> header = new HashMap<>();
         header.put("Authorization", "Bearer " + token);
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java
new file mode 100644
index 000000000000..f111c41f6ada
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTMessage.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.rest;
+
+import org.apache.paimon.rest.requests.CreateDatabaseRequest;
+import org.apache.paimon.rest.requests.DropDatabaseRequest;
+import org.apache.paimon.rest.responses.CreateDatabaseResponse;
+import org.apache.paimon.rest.responses.DatabaseName;
+import org.apache.paimon.rest.responses.GetDatabaseResponse;
+import org.apache.paimon.rest.responses.ListDatabasesResponse;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.rest.RESTCatalogInternalOptions.DATABASE_COMMENT;
+
+/** Mock REST message. */
+public class MockRESTMessage {
+
+    public static String databaseName() {
+        return "database";
+    }
+
+    public static CreateDatabaseRequest createDatabaseRequest(String name) {
+        boolean ignoreIfExists = true;
+        Map<String, String> options = new HashMap<>();
+        options.put("a", "b");
+        return new CreateDatabaseRequest(name, ignoreIfExists, options);
+    }
+
+    public static DropDatabaseRequest dropDatabaseRequest() {
+        boolean ignoreIfNotExists = true;
+        boolean cascade = true;
+        return new DropDatabaseRequest(ignoreIfNotExists, cascade);
+    }
+
+    public static CreateDatabaseResponse createDatabaseResponse(String name) {
+        Map<String, String> options = new HashMap<>();
+        options.put("a", "b");
+        return new CreateDatabaseResponse(name, options);
+    }
+
+    public static GetDatabaseResponse getDatabaseResponse(String name) {
+        Map<String, String> options = new HashMap<>();
+        options.put("a", "b");
+        options.put(DATABASE_COMMENT.key(), "comment");
+        return new GetDatabaseResponse(name, options);
+    }
+
+    public static ListDatabasesResponse listDatabasesResponse(String name) {
+        DatabaseName databaseName = new DatabaseName(name);
+        List<DatabaseName> databaseNameList = new ArrayList<>();
+        databaseNameList.add(databaseName);
+        return new ListDatabasesResponse(databaseNameList);
+    }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index f3f56e97215f..cffac6046623 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -18,8 +18,15 @@
 
 package org.apache.paimon.rest;
 
+import org.apache.paimon.catalog.Database;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.rest.responses.CreateDatabaseResponse;
+import org.apache.paimon.rest.responses.GetDatabaseResponse;
+import org.apache.paimon.rest.responses.ListDatabasesResponse;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
@@ -29,14 +36,17 @@
 
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 
 /** Test for REST Catalog. */
 public class RESTCatalogTest {
 
+    private ObjectMapper mapper = RESTObjectMapper.create();
     private MockWebServer mockWebServer;
     private RESTCatalog restCatalog;
 
@@ -50,7 +60,11 @@ public void setUp() throws IOException {
         String initToken = "init_token";
         options.set(RESTCatalogOptions.TOKEN, initToken);
         options.set(RESTCatalogOptions.THREAD_POOL_SIZE, 1);
-        mockOptions(RESTCatalogInternalOptions.PREFIX.key(), "prefix");
+        String mockResponse =
+                String.format(
+                        "{\"defaults\": {\"%s\": \"%s\"}}",
+                        RESTCatalogInternalOptions.PREFIX.key(), "prefix");
+        mockResponse(mockResponse);
         restCatalog = new RESTCatalog(options);
     }
 
@@ -70,14 +84,50 @@ public void testInitFailWhenDefineWarehouse() {
     public void testGetConfig() {
         String key = "a";
         String value = "b";
-        mockOptions(key, value);
+        String mockResponse = String.format("{\"defaults\": {\"%s\": \"%s\"}}", key, value);
+        mockResponse(mockResponse);
         Map<String, String> header = new HashMap<>();
         Map<String, String> response = restCatalog.fetchOptionsFromServer(header, new HashMap<>());
         assertEquals(value, response.get(key));
     }
 
-    private void mockOptions(String key, String value) {
-        String mockResponse = String.format("{\"defaults\": {\"%s\": \"%s\"}}", key, value);
+    @Test
+    public void testListDatabases() throws JsonProcessingException {
+        String name = MockRESTMessage.databaseName();
+        ListDatabasesResponse response = MockRESTMessage.listDatabasesResponse(name);
+        mockResponse(mapper.writeValueAsString(response));
+        List<String> result = restCatalog.listDatabases();
+        assertEquals(response.getDatabases().size(), result.size());
+        assertEquals(name, result.get(0));
+    }
+
+    @Test
+    public void testCreateDatabase() throws Exception {
+        String name = MockRESTMessage.databaseName();
+        CreateDatabaseResponse response = MockRESTMessage.createDatabaseResponse(name);
+        mockResponse(mapper.writeValueAsString(response));
+        assertDoesNotThrow(() -> restCatalog.createDatabase(name, false, response.getOptions()));
+    }
+
+    @Test
+    public void testGetDatabase() throws Exception {
+        String name = MockRESTMessage.databaseName();
+        GetDatabaseResponse response = MockRESTMessage.getDatabaseResponse(name);
+        mockResponse(mapper.writeValueAsString(response));
+        Database result = restCatalog.getDatabase(name);
+        assertEquals(name, result.name());
+        assertEquals(response.getOptions().size(), result.options().size());
+        assertEquals(response.comment().get(), result.comment().get());
+    }
+
+    @Test
+    public void testDropDatabase() {
+        String name = "name";
+        mockResponse("");
+        assertDoesNotThrow(() -> restCatalog.dropDatabase(name, false, false));
+    }
+
+    private void mockResponse(String mockResponse) {
         MockResponse mockResponseObj =
                 new MockResponse()
                         .setBody(mockResponse)
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTObjectMapperTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTObjectMapperTest.java
index 83a8805d29a0..622a98993692 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTObjectMapperTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTObjectMapperTest.java
@@ -18,8 +18,13 @@
 
 package org.apache.paimon.rest;
 
+import org.apache.paimon.rest.requests.CreateDatabaseRequest;
+import org.apache.paimon.rest.requests.DropDatabaseRequest;
 import org.apache.paimon.rest.responses.ConfigResponse;
+import org.apache.paimon.rest.responses.CreateDatabaseResponse;
 import org.apache.paimon.rest.responses.ErrorResponse;
+import org.apache.paimon.rest.responses.GetDatabaseResponse;
+import org.apache.paimon.rest.responses.ListDatabasesResponse;
 
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -43,7 +48,7 @@ public void configResponseParseTest() throws Exception {
         ConfigResponse response = new ConfigResponse(conf, conf);
         String responseStr = mapper.writeValueAsString(response);
         ConfigResponse parseData = mapper.readValue(responseStr, ConfigResponse.class);
-        assertEquals(conf.get(confKey), parseData.defaults().get(confKey));
+        assertEquals(conf.get(confKey), parseData.getDefaults().get(confKey));
     }
 
     @Test
@@ -53,7 +58,60 @@ public void errorResponseParseTest() throws Exception {
         ErrorResponse response = new ErrorResponse(message, code, new ArrayList<String>());
         String responseStr = mapper.writeValueAsString(response);
         ErrorResponse parseData = mapper.readValue(responseStr, ErrorResponse.class);
-        assertEquals(message, parseData.message());
-        assertEquals(code, parseData.code());
+        assertEquals(message, parseData.getMessage());
+        assertEquals(code, parseData.getCode());
+    }
+
+    @Test
+    public void createDatabaseRequestParseTest() throws Exception {
+        String name = MockRESTMessage.databaseName();
+        CreateDatabaseRequest request = MockRESTMessage.createDatabaseRequest(name);
+        String requestStr = mapper.writeValueAsString(request);
+        CreateDatabaseRequest parseData = mapper.readValue(requestStr, CreateDatabaseRequest.class);
+        assertEquals(request.getName(), parseData.getName());
+        assertEquals(request.getIgnoreIfExists(), parseData.getIgnoreIfExists());
+        assertEquals(request.getOptions().size(), parseData.getOptions().size());
+    }
+
+    @Test
+    public void dropDatabaseRequestParseTest() throws Exception {
+        DropDatabaseRequest request = MockRESTMessage.dropDatabaseRequest();
+        String requestStr = mapper.writeValueAsString(request);
+        DropDatabaseRequest parseData = mapper.readValue(requestStr, DropDatabaseRequest.class);
+        assertEquals(request.getIgnoreIfNotExists(), parseData.getIgnoreIfNotExists());
+        assertEquals(request.getCascade(), parseData.getCascade());
+    }
+
+    @Test
+    public void createDatabaseResponseParseTest() throws Exception {
+        String name = MockRESTMessage.databaseName();
+        CreateDatabaseResponse response = MockRESTMessage.createDatabaseResponse(name);
+        String responseStr = mapper.writeValueAsString(response);
+        CreateDatabaseResponse parseData =
+                mapper.readValue(responseStr, CreateDatabaseResponse.class);
+        assertEquals(name, parseData.getName());
+        assertEquals(response.getOptions().size(), parseData.getOptions().size());
+    }
+
+    @Test
+    public void getDatabaseResponseParseTest() throws Exception {
+        String name = MockRESTMessage.databaseName();
+        GetDatabaseResponse response = MockRESTMessage.getDatabaseResponse(name);
+        String responseStr = mapper.writeValueAsString(response);
+        GetDatabaseResponse parseData = mapper.readValue(responseStr, GetDatabaseResponse.class);
+        assertEquals(name, parseData.getName());
+        assertEquals(response.getOptions().size(), parseData.getOptions().size());
+        assertEquals(response.comment().get(), parseData.comment().get());
+    }
+
+    @Test
+    public void listDatabaseResponseParseTest() throws Exception {
+        String name = MockRESTMessage.databaseName();
+        ListDatabasesResponse response = MockRESTMessage.listDatabasesResponse(name);
+        String responseStr = mapper.writeValueAsString(response);
+        ListDatabasesResponse parseData =
+                mapper.readValue(responseStr, ListDatabasesResponse.class);
+        assertEquals(response.getDatabases().size(), parseData.getDatabases().size());
+        assertEquals(name, parseData.getDatabases().get(0).getName());
     }
 }
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/auth/AuthSessionTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/auth/AuthSessionTest.java
index 81b3ea57b703..1f4a48fd5e8c 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/auth/AuthSessionTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/auth/AuthSessionTest.java
@@ -35,6 +35,8 @@
 import java.util.UUID;
 import java.util.concurrent.ScheduledExecutorService;
 
+import static org.apache.paimon.rest.auth.AuthSession.MAX_REFRESH_WINDOW_MILLIS;
+import static org.apache.paimon.rest.auth.AuthSession.MIN_REFRESH_WAIT_MILLIS;
 import static org.apache.paimon.rest.auth.AuthSession.TOKEN_REFRESH_NUM_RETRIES;
 import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.verify;
@@ -121,6 +123,22 @@ public void testRetryWhenRefreshFail() throws Exception {
         verify(credentialsProvider, Mockito.times(TOKEN_REFRESH_NUM_RETRIES + 1)).refresh();
     }
 
+    @Test
+    public void testGetTimeToWaitByExpiresInMills() {
+        long expiresInMillis = -100L;
+        long timeToWait = AuthSession.getTimeToWaitByExpiresInMills(expiresInMillis);
+        assertEquals(MIN_REFRESH_WAIT_MILLIS, timeToWait);
+        expiresInMillis = (long) (MAX_REFRESH_WINDOW_MILLIS * 0.5);
+        timeToWait = AuthSession.getTimeToWaitByExpiresInMills(expiresInMillis);
+        assertEquals(MIN_REFRESH_WAIT_MILLIS, timeToWait);
+        expiresInMillis = MAX_REFRESH_WINDOW_MILLIS;
+        timeToWait = AuthSession.getTimeToWaitByExpiresInMills(expiresInMillis);
+        assertEquals(timeToWait, MIN_REFRESH_WAIT_MILLIS);
+        expiresInMillis = MAX_REFRESH_WINDOW_MILLIS * 2L;
+        timeToWait = AuthSession.getTimeToWaitByExpiresInMills(expiresInMillis);
+        assertEquals(timeToWait, MAX_REFRESH_WINDOW_MILLIS);
+    }
+
     private Pair<File, String> generateTokenAndWriteToFile(String fileName) throws IOException {
         File tokenFile = folder.newFile(fileName);
         String token = UUID.randomUUID().toString();
diff --git a/paimon-open-api/generate.sh b/paimon-open-api/generate.sh
index b63aa538abc4..619b642ab760 100755
--- a/paimon-open-api/generate.sh
+++ b/paimon-open-api/generate.sh
@@ -17,6 +17,7 @@
 
 # Start the application
 cd ..
+mvn spotless:apply
 mvn clean install -DskipTests
 cd ./paimon-open-api
 mvn spring-boot:run &
diff --git a/paimon-open-api/rest-catalog-open-api.yaml b/paimon-open-api/rest-catalog-open-api.yaml
index 432ee123b8d4..2a5d1dc58418 100644
--- a/paimon-open-api/rest-catalog-open-api.yaml
+++ b/paimon-open-api/rest-catalog-open-api.yaml
@@ -28,6 +28,120 @@ servers:
   - url: http://localhost:8080
     description: Server URL in Development environment
 paths:
+  /api/v1/{prefix}/databases:
+    get:
+      tags:
+        - database
+      summary: List Databases
+      operationId: listDatabases
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ListDatabasesResponse'
+        "500":
+          description: Internal Server Error
+    post:
+      tags:
+        - database
+      summary: Create Databases
+      operationId: createDatabases
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CreateDatabaseRequest'
+      responses:
+        "500":
+          description: Internal Server Error
+        "409":
+          description: Resource has exist
+          content:
+            '*/*':
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "200":
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CreateDatabaseResponse'
+  /api/v1/{prefix}/databases/{database}:
+    get:
+      tags:
+        - database
+      summary: Get Database
+      operationId: getDatabases
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: database
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: OK
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/GetDatabaseResponse'
+        "404":
+          description: Resource not found
+          content:
+            '*/*':
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "500":
+          description: Internal Server Error
+    delete:
+      tags:
+        - database
+      summary: Drop Database
+      operationId: dropDatabases
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: database
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/DropDatabaseRequest'
+      responses:
+        "404":
+          description: Resource not found
+          content:
+            '*/*':
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
+        "500":
+          description: Internal Server Error
   /api/v1/config:
     get:
       tags:
@@ -37,14 +151,67 @@ paths:
       responses:
         "500":
           description: Internal Server Error
-        "201":
-          description: Created
+        "200":
+          description: OK
           content:
             application/json:
               schema:
                 $ref: '#/components/schemas/ConfigResponse'
 components:
   schemas:
+    CreateDatabaseRequest:
+      type: object
+      properties:
+        name:
+          type: string
+        ignoreIfExists:
+          type: boolean
+        options:
+          type: object
+          additionalProperties:
+            type: string
+    ErrorResponse:
+      type: object
+      properties:
+        message:
+          type: string
+        code:
+          type: integer
+          format: int32
+        stack:
+          type: array
+          items:
+            type: string
+    CreateDatabaseResponse:
+      type: object
+      properties:
+        name:
+          type: string
+        options:
+          type: object
+          additionalProperties:
+            type: string
+    DatabaseName:
+      type: object
+      properties:
+        name:
+          type: string
+    ListDatabasesResponse:
+      type: object
+      properties:
+        databases:
+          type: array
+          items:
+            $ref: '#/components/schemas/DatabaseName'
+    GetDatabaseResponse:
+      type: object
+      properties:
+        name:
+          type: string
+        options:
+          type: object
+          additionalProperties:
+            type: string
     ConfigResponse:
       type: object
       properties:
@@ -52,9 +219,14 @@ components:
           type: object
           additionalProperties:
             type: string
-          writeOnly: true
         overrides:
           type: object
           additionalProperties:
             type: string
-          writeOnly: true
+    DropDatabaseRequest:
+      type: object
+      properties:
+        ignoreIfNotExists:
+          type: boolean
+        cascade:
+          type: boolean
diff --git a/paimon-open-api/src/main/java/org/apache/paimon/open/api/RESTCatalogController.java b/paimon-open-api/src/main/java/org/apache/paimon/open/api/RESTCatalogController.java
index b47554057105..364cc5adbb2c 100644
--- a/paimon-open-api/src/main/java/org/apache/paimon/open/api/RESTCatalogController.java
+++ b/paimon-open-api/src/main/java/org/apache/paimon/open/api/RESTCatalogController.java
@@ -19,17 +19,28 @@
 package org.apache.paimon.open.api;
 
 import org.apache.paimon.rest.ResourcePaths;
+import org.apache.paimon.rest.requests.CreateDatabaseRequest;
+import org.apache.paimon.rest.requests.DropDatabaseRequest;
 import org.apache.paimon.rest.responses.ConfigResponse;
+import org.apache.paimon.rest.responses.CreateDatabaseResponse;
+import org.apache.paimon.rest.responses.DatabaseName;
+import org.apache.paimon.rest.responses.ErrorResponse;
+import org.apache.paimon.rest.responses.GetDatabaseResponse;
+import org.apache.paimon.rest.responses.ListDatabasesResponse;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
 
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.media.Content;
 import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.parameters.RequestBody;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import io.swagger.v3.oas.annotations.responses.ApiResponses;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.CrossOrigin;
+import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RestController;
 
 import java.util.HashMap;
@@ -45,7 +56,7 @@ public class RESTCatalogController {
             tags = {"config"})
     @ApiResponses({
         @ApiResponse(
-                responseCode = "201",
+                responseCode = "200",
                 content = {
                     @Content(
                             schema = @Schema(implementation = ConfigResponse.class),
@@ -56,14 +67,99 @@ public class RESTCatalogController {
                 content = {@Content(schema = @Schema())})
     })
     @GetMapping(ResourcePaths.V1_CONFIG)
-    public ResponseEntity<ConfigResponse> getConfig() {
-        try {
-            Map<String, String> defaults = new HashMap<>();
-            Map<String, String> overrides = new HashMap<>();
-            ConfigResponse response = new ConfigResponse(defaults, overrides);
-            return new ResponseEntity<>(response, HttpStatus.CREATED);
-        } catch (Exception e) {
-            return new ResponseEntity<>(null, HttpStatus.INTERNAL_SERVER_ERROR);
-        }
+    public ConfigResponse getConfig() {
+        Map<String, String> defaults = new HashMap<>();
+        Map<String, String> overrides = new HashMap<>();
+        return new ConfigResponse(defaults, overrides);
+    }
+
+    @Operation(
+            summary = "List Databases",
+            tags = {"database"})
+    @ApiResponses({
+        @ApiResponse(
+                responseCode = "200",
+                content = {
+                    @Content(
+                            schema = @Schema(implementation = ListDatabasesResponse.class),
+                            mediaType = "application/json")
+                }),
+        @ApiResponse(
+                responseCode = "500",
+                content = {@Content(schema = @Schema())})
+    })
+    @GetMapping("/api/v1/{prefix}/databases")
+    public ListDatabasesResponse listDatabases(@PathVariable String prefix) {
+        return new ListDatabasesResponse(ImmutableList.of(new DatabaseName("account")));
+    }
+
+    @Operation(
+            summary = "Create Databases",
+            tags = {"database"})
+    @ApiResponses({
+        @ApiResponse(
+                responseCode = "200",
+                content = {
+                    @Content(
+                            schema = @Schema(implementation = CreateDatabaseResponse.class),
+                            mediaType = "application/json")
+                }),
+        @ApiResponse(
+                responseCode = "409",
+                description = "Resource has exist",
+                content = {@Content(schema = @Schema(implementation = ErrorResponse.class))}),
+        @ApiResponse(
+                responseCode = "500",
+                content = {@Content(schema = @Schema())})
+    })
+    @PostMapping("/api/v1/{prefix}/databases")
+    public CreateDatabaseResponse createDatabases(
+            @PathVariable String prefix, @RequestBody CreateDatabaseRequest request) {
+        Map<String, String> properties = new HashMap<>();
+        return new CreateDatabaseResponse("name", properties);
     }
+
+    @Operation(
+            summary = "Get Database",
+            tags = {"database"})
+    @ApiResponses({
+        @ApiResponse(
+                responseCode = "200",
+                content = {
+                    @Content(
+                            schema = @Schema(implementation = GetDatabaseResponse.class),
+                            mediaType = "application/json")
+                }),
+        @ApiResponse(
+                responseCode = "404",
+                description = "Resource not found",
+                content = {@Content(schema = @Schema(implementation = ErrorResponse.class))}),
+        @ApiResponse(
+                responseCode = "500",
+                content = {@Content(schema = @Schema())})
+    })
+    @GetMapping("/api/v1/{prefix}/databases/{database}")
+    public GetDatabaseResponse getDatabases(
+            @PathVariable String prefix, @PathVariable String database) {
+        Map<String, String> options = new HashMap<>();
+        return new GetDatabaseResponse("name", options);
+    }
+
+    @Operation(
+            summary = "Drop Database",
+            tags = {"database"})
+    @ApiResponses({
+        @ApiResponse(
+                responseCode = "404",
+                description = "Resource not found",
+                content = {@Content(schema = @Schema(implementation = ErrorResponse.class))}),
+        @ApiResponse(
+                responseCode = "500",
+                content = {@Content(schema = @Schema())})
+    })
+    @DeleteMapping("/api/v1/{prefix}/databases/{database}")
+    public void dropDatabases(
+            @PathVariable String prefix,
+            @PathVariable String database,
+            @RequestBody DropDatabaseRequest request) {}
 }
diff --git a/paimon-open-api/src/main/java/org/apache/paimon/open/api/config/OpenAPIConfig.java b/paimon-open-api/src/main/java/org/apache/paimon/open/api/config/OpenAPIConfig.java
index 01234c41bbff..0e28cd95f9d2 100644
--- a/paimon-open-api/src/main/java/org/apache/paimon/open/api/config/OpenAPIConfig.java
+++ b/paimon-open-api/src/main/java/org/apache/paimon/open/api/config/OpenAPIConfig.java
@@ -32,7 +32,6 @@
 /** Config for OpenAPI. */
 @Configuration
 public class OpenAPIConfig {
-
     @Value("${openapi.url}")
     private String devUrl;