[#10737] fix(core): Avoid blocking dropCatalog on imported schemas

core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java

@@ -73,6 +73,7 @@
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.Schema;
 import org.apache.gravitino.StringIdentifier;
 import org.apache.gravitino.connector.BaseCatalog;
 import org.apache.gravitino.connector.CatalogOperations;
@@ -87,6 +88,7 @@
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NoSuchSchemaException;
 import org.apache.gravitino.exceptions.NonEmptyCatalogException;
 import org.apache.gravitino.exceptions.NonEmptyEntityException;
 import org.apache.gravitino.file.FilesetCatalog;
@@ -877,9 +879,37 @@ private boolean containsUserCreatedSchemas(
     Set<String> availableSchemaNames =
         Arrays.stream(allSchemas).map(NameIdentifier::name).collect(Collectors.toSet());
 
-    // some schemas are dropped externally, but still exist in the entity store, those schemas are
-    // invalid
-    return schemaEntities.stream().map(SchemaEntity::name).anyMatch(availableSchemaNames::contains);
+    // Some schemas are dropped externally but still exist in the entity store — those are invalid.
+    // Among schemas that exist in the underlying catalog, only those created via Gravitino carry a
+    // StringIdentifier in their external properties; imported schemas do not.
+    for (SchemaEntity schemaEntity : schemaEntities) {
+      if (!availableSchemaNames.contains(schemaEntity.name())) {
+        continue;
+      }
+
+      try {
+        Schema schema =
+            catalogWrapper.doWithSchemaOps(ops -> ops.loadSchema(schemaEntity.nameIdentifier()));
+        Map<String, String> props = schema.properties();
+        // If the backend cannot store a StringIdentifier (null or empty properties, e.g. MySQL
+        // which does not support schema comments), we cannot tell whether the schema was created
+        // by Gravitino or imported. Be conservative and treat it as user-created to avoid
+        // accidental data loss.
+        // Only skip a schema when properties are non-null, non-empty, and contain no
+        // StringIdentifier — the reliable signal that the schema was imported from an external
+        // catalog on a backend that does support identifier storage.
+        if (props == null || props.isEmpty() || StringIdentifier.fromProperties(props) != null) {
+          return true;
+        }
+      } catch (NoSuchSchemaException ex) {
+        // A race between listSchemas and loadSchema is expected; treat as non-user-created.
+        LOG.debug(
+            "Schema {} no longer exists while checking whether it is user-created",
+            schemaEntity.nameIdentifier());
+      }
+    }
+
+    return false;
   }
 
   /**

core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java

@@ -46,12 +46,14 @@
 import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
+import org.apache.gravitino.Schema;
 import org.apache.gravitino.connector.capability.Capability;
 import org.apache.gravitino.connector.capability.CapabilityResult;
 import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
 import org.apache.gravitino.exceptions.CatalogInUseException;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NoSuchSchemaException;
 import org.apache.gravitino.lock.LockManager;
 import org.apache.gravitino.meta.AuditInfo;
 import org.apache.gravitino.meta.BaseMetalake;
@@ -589,6 +591,167 @@ public void testDropCatalog() throws Exception {
     Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(ident));
   }
 
+  @Test
+  public void testDropCatalogSkipsImportedSchemas() throws Exception {
+    NameIdentifier ident = NameIdentifier.of("metalake", "test41");
+    Map<String, String> props =
+        ImmutableMap.of(
+            "provider",
+            "test",
+            PROPERTY_KEY1,
+            "value1",
+            PROPERTY_KEY2,
+            "value2",
+            PROPERTY_KEY5_PREFIX + "1",
+            "value3");
+    String comment = "comment";
+
+    Catalog catalog =
+        catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider, comment, props);
+    Mockito.doCallRealMethod().when(catalogManager).loadCatalogAndWrap(ident);
+    Assertions.assertDoesNotThrow(() -> catalogManager.disableCatalog(ident));
+    CatalogEntity catalogEntity = entityStore.get(ident, EntityType.CATALOG, CatalogEntity.class);
+    FieldUtils.writeField(catalog, "entity", catalogEntity, true);
+
+    SchemaEntity importedSchemaEntity =
+        SchemaEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("imported_schema")
+            .withNamespace(Namespace.of("metalake", "test41"))
+            .withAuditInfo(
+                AuditInfo.builder()
+                    .withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+                    .withCreateTime(Instant.now())
+                    .build())
+            .build();
+    entityStore.put(importedSchemaEntity);
+
+    Schema importedSchema = Mockito.mock(Schema.class);
+    // Non-empty properties without StringIdentifier simulate an imported schema on a backend
+    // that supports property storage (e.g., Hive, Iceberg) but did not create this schema
+    // via Gravitino.
+    Mockito.doReturn(ImmutableMap.of("owner", "external")).when(importedSchema).properties();
+    CatalogManager.CatalogWrapper wrapper = Mockito.mock(CatalogManager.CatalogWrapper.class);
+    Capability capability = Mockito.mock(Capability.class);
+    CapabilityResult unsupportedResult = CapabilityResult.unsupported("Not managed");
+    Mockito.doReturn(wrapper).when(catalogManager).loadCatalogAndWrap(ident);
+    Mockito.doReturn(catalog).when(wrapper).catalog();
+    Mockito.doReturn(capability).when(wrapper).capabilities();
+    Mockito.doReturn(unsupportedResult).when(capability).managedStorage(any());
+    Mockito.doReturn(
+            new NameIdentifier[] {NameIdentifier.of("metalake", "test41", "imported_schema")})
+        .doReturn(importedSchema)
+        .when(wrapper)
+        .doWithSchemaOps(any());
+
+    // Imported schema (no StringIdentifier in external catalog properties) should not block drop.
+    Assertions.assertTrue(catalogManager.dropCatalog(ident));
+  }
+
+  @Test
+  public void testDropCatalogIgnoresMissingSchema() throws Exception {
+    NameIdentifier ident = NameIdentifier.of("metalake", "test41");
+    Map<String, String> props =
+        ImmutableMap.of(
+            "provider",
+            "test",
+            PROPERTY_KEY1,
+            "value1",
+            PROPERTY_KEY2,
+            "value2",
+            PROPERTY_KEY5_PREFIX + "1",
+            "value3");
+    String comment = "comment";
+
+    Catalog catalog =
+        catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider, comment, props);
+    Mockito.doCallRealMethod().when(catalogManager).loadCatalogAndWrap(ident);
+    Assertions.assertDoesNotThrow(() -> catalogManager.disableCatalog(ident));
+    CatalogEntity catalogEntity = entityStore.get(ident, EntityType.CATALOG, CatalogEntity.class);
+    FieldUtils.writeField(catalog, "entity", catalogEntity, true);
+
+    SchemaEntity schemaEntity =
+        SchemaEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("default")
+            .withNamespace(Namespace.of("metalake", "test41"))
+            .withAuditInfo(
+                AuditInfo.builder()
+                    .withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+                    .withCreateTime(Instant.now())
+                    .build())
+            .build();
+    entityStore.put(schemaEntity);
+
+    CatalogManager.CatalogWrapper wrapper = Mockito.mock(CatalogManager.CatalogWrapper.class);
+    Capability capability = Mockito.mock(Capability.class);
+    CapabilityResult unsupportedResult = CapabilityResult.unsupported("Not managed");
+    Mockito.doReturn(wrapper).when(catalogManager).loadCatalogAndWrap(ident);
+    Mockito.doReturn(catalog).when(wrapper).catalog();
+    Mockito.doReturn(capability).when(wrapper).capabilities();
+    Mockito.doReturn(unsupportedResult).when(capability).managedStorage(any());
+    Mockito.doReturn(new NameIdentifier[] {NameIdentifier.of("metalake", "test41", "default")})
+        .doThrow(new NoSuchSchemaException("Schema not found"))
+        .when(wrapper)
+        .doWithSchemaOps(any());
+
+    // Schema disappearing between listSchemas and loadSchema should not block drop.
+    Assertions.assertTrue(catalogManager.dropCatalog(ident));
+  }
+
+  @Test
+  public void testDropCatalogFailsOnSchemaClassificationError() throws Exception {
+    NameIdentifier ident = NameIdentifier.of("metalake", "test41");
+    Map<String, String> props =
+        ImmutableMap.of(
+            "provider",
+            "test",
+            PROPERTY_KEY1,
+            "value1",
+            PROPERTY_KEY2,
+            "value2",
+            PROPERTY_KEY5_PREFIX + "1",
+            "value3");
+    String comment = "comment";
+
+    Catalog catalog =
+        catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider, comment, props);
+    Mockito.doCallRealMethod().when(catalogManager).loadCatalogAndWrap(ident);
+    Assertions.assertDoesNotThrow(() -> catalogManager.disableCatalog(ident));
+    CatalogEntity catalogEntity = entityStore.get(ident, EntityType.CATALOG, CatalogEntity.class);
+    FieldUtils.writeField(catalog, "entity", catalogEntity, true);
+
+    SchemaEntity schemaEntity =
+        SchemaEntity.builder()
+            .withId(RandomIdGenerator.INSTANCE.nextId())
+            .withName("test_schema1")
+            .withNamespace(Namespace.of("metalake", "test41"))
+            .withAuditInfo(
+                AuditInfo.builder()
+                    .withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+                    .withCreateTime(Instant.now())
+                    .build())
+            .build();
+    entityStore.put(schemaEntity);
+
+    CatalogManager.CatalogWrapper wrapper = Mockito.mock(CatalogManager.CatalogWrapper.class);
+    Capability capability = Mockito.mock(Capability.class);
+    CapabilityResult unsupportedResult = CapabilityResult.unsupported("Not managed");
+    Mockito.doReturn(wrapper).when(catalogManager).loadCatalogAndWrap(ident);
+    Mockito.doReturn(catalog).when(wrapper).catalog();
+    Mockito.doReturn(capability).when(wrapper).capabilities();
+    Mockito.doReturn(unsupportedResult).when(capability).managedStorage(any());
+    Mockito.doReturn(new NameIdentifier[] {NameIdentifier.of("metalake", "test41", "test_schema1")})
+        .doThrow(new RuntimeException("Failed connect"))
+        .when(wrapper)
+        .doWithSchemaOps(any());
+
+    // Unexpected errors during schema classification should propagate (fail-closed).
+    RuntimeException ex =
+        Assertions.assertThrows(RuntimeException.class, () -> catalogManager.dropCatalog(ident));
+    Assertions.assertTrue(ex.getCause().getMessage().contains("Failed connect"));
+  }
+
   @Test
   public void testForceDropCatalog() throws Exception {
     NameIdentifier ident = NameIdentifier.of("metalake", "test41");