[core] Fix drop table when table directory is missing on disk

paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java

@@ -364,32 +364,37 @@ public void dropTable(Identifier identifier, boolean ignoreIfNotExists)
         checkNotBranch(identifier, "dropTable");
         checkNotSystemTable(identifier, "dropTable");
 
+        if (!tableExists(identifier)) {
+            if (ignoreIfNotExists) {
+                return;
+            }
+            throw new TableNotExistException(identifier);
+        }
+
         Set<Path> externalPaths = new HashSet<>();
-        try {
+        if (tableExistsInFileSystem(getTableLocation(identifier), DEFAULT_MAIN_BRANCH)) {
             Table table = getTable(identifier);
             if (table instanceof FileStoreTable) {
                 FileStoreTable fileStoreTable = (FileStoreTable) table;
                 List<Path> schemaExternalPaths =
                         getSchemaExternalPaths(fileStoreTable.schemaManager().listAll());
                 externalPaths.addAll(schemaExternalPaths);
-                // get table branch external path
                 List<String> branches = fileStoreTable.branchManager().branches();
                 for (String branch : branches) {
                     SchemaManager schemaManager =
                             fileStoreTable.schemaManager().copyWithBranch(branch);
                     externalPaths.addAll(getSchemaExternalPaths(schemaManager.listAll()));
                 }
             }
-        } catch (TableNotExistException e) {
-            if (ignoreIfNotExists) {
-                return;
-            }
-            throw new TableNotExistException(identifier);
         }
 
         dropTableImpl(identifier, new ArrayList<>(externalPaths));
     }
 
+    protected boolean tableExists(Identifier identifier) {
+        return tableExistsInFileSystem(getTableLocation(identifier), DEFAULT_MAIN_BRANCH);
+    }
+
     private List<Path> getSchemaExternalPaths(List<TableSchema> schemas) {
         if (schemas == null) {
             return Collections.emptyList();

paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java

@@ -451,6 +451,16 @@ protected TableSchema loadTableSchema(Identifier identifier) throws TableNotExis
                         () -> new RuntimeException("There is no paimon table in " + tableLocation));
     }
 
+    @Override
+    protected boolean tableExists(Identifier identifier) {
+        return JdbcUtils.tableExists(
+                        connections,
+                        catalogKey,
+                        identifier.getDatabaseName(),
+                        identifier.getTableName())
+                || super.tableExists(identifier);
+    }
+
     @Override
     public boolean caseSensitive() {
         return false;

paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcCatalogTest.java

@@ -22,6 +22,7 @@
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.CatalogTestBase;
 import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.Schema;
@@ -86,6 +87,27 @@ private JdbcCatalog initCatalog(Map<String, String> props) {
     @Test
     public void testGetTable() throws Exception {}
 
+    @Test
+    public void testDropTableWhenTablePathMissing() throws Exception {
+        String databaseName = "test_db";
+        String tableName = "new_table";
+        catalog.createDatabase(databaseName, false);
+        Identifier identifier = Identifier.create(databaseName, tableName);
+        catalog.createTable(identifier, DEFAULT_TABLE_SCHEMA, false);
+
+        JdbcCatalog jdbcCatalog = (JdbcCatalog) catalog;
+        Path path = jdbcCatalog.getTableLocation(identifier);
+        jdbcCatalog.fileIO().deleteDirectoryQuietly(path);
+
+        assertThatThrownBy(() -> catalog.getTable(identifier))
+                .isInstanceOf(RuntimeException.class)
+                .hasMessage("There is no paimon table in " + path);
+        assertThat(jdbcCatalog.listTables(databaseName)).contains(tableName);
+
+        jdbcCatalog.dropTable(identifier, false);
+        assertThat(jdbcCatalog.listTables(databaseName)).doesNotContain(tableName);
+    }
+
     @Test
     public void testAcquireLockFail() throws SQLException, InterruptedException {
         String lockId = "jdbc.testDb.testTable";

paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java

@@ -303,6 +303,27 @@ public List<String> listDatabases() {
         }
     }
 
+    @Override
+    protected boolean tableExists(Identifier identifier) {
+        try {
+            boolean inHms =
+                    clients()
+                            .run(
+                                    client ->
+                                            client.tableExists(
+                                                    identifier.getDatabaseName(),
+                                                    identifier.getTableName()));
+            return inHms || super.tableExists(identifier);
+        } catch (TException e) {
+            throw new RuntimeException(
+                    "Cannot determine if table " + identifier.getFullName() + " exists.", e);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException(
+                    "Interrupted in call to tableExists " + identifier.getFullName(), e);
+        }
+    }
+
     @Override
     protected void createDatabaseImpl(String name, Map<String, String> properties) {
         try {

paimon-hive/paimon-hive-catalog/src/test/java/org/apache/paimon/hive/HiveCatalogTest.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.catalog.CatalogTestBase;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.client.ClientPool;
+import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
@@ -281,6 +282,37 @@ public void testAlterHiveTableParameters() {
         }
     }
 
+    @Test
+    public void testDropTableWhenTablePathMissing() throws Exception {
+        String databaseName = "test_db";
+        String tableName = "new_table";
+        catalog.createDatabase(databaseName, false);
+        Identifier identifier = Identifier.create(databaseName, tableName);
+
+        Schema schema =
+                new Schema(
+                        Lists.newArrayList(
+                                new DataField(0, "pk", DataTypes.INT()),
+                                new DataField(1, "col1", DataTypes.STRING()),
+                                new DataField(2, "col2", DataTypes.STRING())),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        new HashMap<>(),
+                        "");
+        catalog.createTable(identifier, schema, false);
+
+        HiveCatalog hiveCatalog = (HiveCatalog) catalog;
+        Path path = hiveCatalog.getTableLocation(identifier);
+        hiveCatalog.fileIO().deleteDirectoryQuietly(path);
+
+        assertThatThrownBy(() -> hiveCatalog.getTable(identifier))
+                .isInstanceOf(Catalog.TableNotExistException.class);
+        assertThat(hiveCatalog.listTables(databaseName)).contains(tableName);
+
+        hiveCatalog.dropTable(identifier, false);
+        assertThat(hiveCatalog.listTables(databaseName)).doesNotContain(tableName);
+    }
+
     @Test
     public void testListTablesLock() {
         try {