yuqi1129 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 28, 2026 · Sep 30, 2026
diff --git a/api/src/main/java/org/apache/gravitino/exceptions/NonEmptyCatalogException.java b/api/src/main/java/org/apache/gravitino/exceptions/NonEmptyCatalogException.java
@@ -34,4 +34,16 @@ public class NonEmptyCatalogException extends GravitinoRuntimeException {
   public NonEmptyCatalogException(@FormatString String message, Object... args) {
     super(message, args);
   }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public NonEmptyCatalogException(Throwable cause, @FormatString String message, Object... args) {
+    super(cause, message, args);
+  }
 }
diff --git a/core/src/main/java/org/apache/gravitino/SupportsConditionalCatalogDelete.java b/core/src/main/java/org/apache/gravitino/SupportsConditionalCatalogDelete.java
@@ -0,0 +1,43 @@
+/*
+ * 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.gravitino;
+
+import java.io.IOException;
+import java.util.Set;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+
+/**
+ * An optional capability for deleting a catalog after checking its remaining schemas atomically.
+ */
+public interface SupportsConditionalCatalogDelete {
+
+  /**
+   * Delete a catalog only if all remaining schemas have IDs in the allowlist. The check and
+   * deletion must run in one transaction and be serialized with schema creation.
+   *
+   * @param ident the catalog identifier
+   * @param allowedSchemaIds IDs of schemas that may be deleted with the catalog
+   * @return true if the catalog was deleted, or false only if it was already absent
+   * @throws NonEmptyEntityException if a remaining schema is not in {@code allowedSchemaIds}; an
+   *     implementation must not return false for this case
+   * @throws IOException if the store operation fails
+   */
+  boolean deleteCatalogWithAllowedSchemas(NameIdentifier ident, Set<Long> allowedSchemaIds)
+      throws IOException;
+}
diff --git a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -81,6 +81,7 @@
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.Schema;
 import org.apache.gravitino.StringIdentifier;
+import org.apache.gravitino.SupportsConditionalCatalogDelete;
 import org.apache.gravitino.connector.BaseCatalog;
 import org.apache.gravitino.connector.CatalogDropAware;
 import org.apache.gravitino.connector.CatalogOperations;
@@ -1414,7 +1415,40 @@ public boolean dropCatalog(NameIdentifier ident, boolean force)
             // the cache with stale data between invalidate and delete.
             Map<String, String> catalogProperties =
                 copyProperties(catalogWrapper.catalog().entity().getProperties());
-            boolean deleted = store.delete(ident, EntityType.CATALOG, true);
+            boolean deleted;
+            if (force) {
+              deleted = store.delete(ident, EntityType.CATALOG, true);
+            } else {
+              try {
+                if (schemaEntities.isEmpty()) {
+                  deleted = store.delete(ident, EntityType.CATALOG, false);
+                } else {
+                  // Managed-storage catalogs cannot reach this branch: any schema in one blocks
+                  // non-force drop before external schema deletion or secret cleanup starts.
+                  Set<Long> allowedSchemaIds =
+                      schemaEntities.stream().map(SchemaEntity::id).collect(Collectors.toSet());
+                  if (!(store instanceof SupportsConditionalCatalogDelete)) {
+                    // Fail closed: an unconditional cascade could delete a schema created after
+                    // the classification above.
+                    throw new UnsupportedOperationException(
+                        String.format(
+                            "Catalog %s still has built-in, imported, or externally removed "
+                                + "schemas, and entity store %s cannot delete it atomically with "
+                                + "them. Use the force option, or use an entity store that "
+                                + "implements %s",
+                            ident,
+                            store.getClass().getName(),
+                            SupportsConditionalCatalogDelete.class.getSimpleName()));
+                  }
+                  deleted =
+                      ((SupportsConditionalCatalogDelete) store)
+                          .deleteCatalogWithAllowedSchemas(ident, allowedSchemaIds);
+                }
+              } catch (NonEmptyEntityException e) {
+                throw new NonEmptyCatalogException(
+                    e, "Catalog %s has schemas, please drop them first or use force option", ident);
+              }
+            }
             if (deleted) {
               markLocalMutation(ident);
               try {
@@ -1448,7 +1482,9 @@ public boolean dropCatalog(NameIdentifier ident, boolean force)
             // reaches here: it maps a missing entity to false on its own.
             catalogCache.invalidate(ident);
             return false;
-          } catch (GravitinoRuntimeException e) {
+          } catch (GravitinoRuntimeException | UnsupportedOperationException e) {
+            // Keep UnsupportedOperationException unwrapped so REST reports an unsupported
+            // operation instead of an internal error.
             throw e;
           } catch (Exception e) {
             throw new RuntimeException(e);

diff --git a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
@@ -378,14 +378,15 @@ public boolean dropMetalake(NameIdentifier ident, boolean force)
                     "Metalake %s is in use, please disable it first or use force option", ident);
               }
 
-              List<CatalogEntity> catalogEntities =
-                  store.list(Namespace.of(ident.name()), CatalogEntity.class, EntityType.CATALOG);
-              if (!catalogEntities.isEmpty() && !force) {
+              if (force) {
+                return store.delete(ident, EntityType.METALAKE, true);
+              }
+              try {
+                return store.delete(ident, EntityType.METALAKE, false);
+              } catch (NonEmptyEntityException e) {
                 throw new NonEmptyMetalakeException(
                     "Metalake %s has catalogs, please drop them first or use force option", ident);
               }
-
-              return store.delete(ident, EntityType.METALAKE, true);
             } catch (NoSuchMetalakeException | NoSuchEntityException e) {
               // Another server may have completed the drop after the initial existence check.
               // Dropping an already-removed metalake remains an idempotent false result.

diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
@@ -30,6 +30,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -46,6 +47,7 @@
 import org.apache.gravitino.RelationQuery;
 import org.apache.gravitino.RelationUpdate;
 import org.apache.gravitino.RelationalEntity;
+import org.apache.gravitino.SupportsConditionalCatalogDelete;
 import org.apache.gravitino.SupportsRelationOperations;
 import org.apache.gravitino.UnsupportedEntityTypeException;
 import org.apache.gravitino.cache.BaseEntityCache;
@@ -110,7 +112,10 @@
  * syntax, please implement the SQL statements and methods in MyBatis Mapper separately and switch
  * according to the {@link Configs#ENTITY_RELATIONAL_JDBC_BACKEND_URL_KEY} parameter.
  */
-public class JDBCBackend implements RelationalBackend, SupportsOrphanedRelationCleanup {
+public class JDBCBackend
+    implements RelationalBackend,
+        SupportsOrphanedRelationCleanup,
+        SupportsConditionalCatalogDelete {
 
   private static final Logger LOG = LoggerFactory.getLogger(JDBCBackend.class);
 
@@ -413,6 +418,32 @@ public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolea
     }
   }
 
+  @Override
+  public boolean deleteCatalogWithAllowedSchemas(NameIdentifier ident, Set<Long> allowedSchemaIds)
+      throws IOException {
+    boolean transactionOwner = !SessionUtils.isInTransaction();
+    if (transactionOwner) {
+      SessionUtils.beginTransaction();
+    }
+    boolean committed = false;
+    try {
+      boolean deleted =
+          CatalogMetaService.getInstance().deleteCatalogWithAllowedSchemas(ident, allowedSchemaIds);
+      if (deleted) {
+        insertEntityChange(ident, Entity.EntityType.CATALOG, OperateType.DROP);
+      }
+      if (transactionOwner) {
+        SessionUtils.commitTransaction();
+      }
+      committed = true;
+      return deleted;
+    } finally {
+      if (transactionOwner && !committed) {
+        SessionUtils.rollbackTransaction();
+      }
+    }
+  }
+
   @Override
   public <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
       NameIdentifier ident,

diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -28,6 +28,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
@@ -47,6 +48,7 @@
 import org.apache.gravitino.RelationQuery;
 import org.apache.gravitino.RelationUpdate;
 import org.apache.gravitino.RelationalEntity;
+import org.apache.gravitino.SupportsConditionalCatalogDelete;
 import org.apache.gravitino.SupportsRelationOperations;
 import org.apache.gravitino.cache.BaseEntityCache;
 import org.apache.gravitino.cache.CacheFactory;
@@ -69,7 +71,10 @@
  * RelationalBackend} interface. The default JDBC backend is {@link JDBCBackend}.
  */
 public class RelationalEntityStore
-    implements EntityStore, SupportsRelationOperations, SupportsEntityChangeLog {
+    implements EntityStore,
+        SupportsRelationOperations,
+        SupportsEntityChangeLog,
+        SupportsConditionalCatalogDelete {
   private static final Logger LOGGER = LoggerFactory.getLogger(RelationalEntityStore.class);
   public static final ImmutableMap<String, String> RELATIONAL_BACKENDS =
       ImmutableMap.of(
@@ -311,6 +316,28 @@ public boolean delete(NameIdentifier ident, Entity.EntityType entityType, boolea
     }
   }
 
+  @Override
+  public boolean deleteCatalogWithAllowedSchemas(NameIdentifier ident, Set<Long> allowedSchemaIds)
+      throws IOException {
+    if (!(backend instanceof SupportsConditionalCatalogDelete)) {
+      throw new UnsupportedOperationException(
+          String.format(
+              "Catalog %s cannot be deleted atomically with its allowed schemas by backend %s. "
+                  + "Use the force option or a backend that implements %s",
+              ident,
+              backend.getClass().getName(),
+              SupportsConditionalCatalogDelete.class.getSimpleName()));
+    }
+    try {
+      return ((SupportsConditionalCatalogDelete) backend)
+          .deleteCatalogWithAllowedSchemas(ident, allowedSchemaIds);
+    } catch (NoSuchEntityException e) {
+      return false;
+    } finally {
+      invalidateCache(ident, Entity.EntityType.CATALOG);
+    }
+  }
+
   @Override
   public <E extends Entity & HasIdentifier> Optional<E> deleteAndGet(
       NameIdentifier ident,