Core: Fix HadoopCatalog drop table behavior (apache#3097)

Dropping a table should return false if it does not exist.

Co-authored-by: adityasingh <[email protected]>

core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java

@@ -252,21 +252,19 @@ public boolean dropTable(TableIdentifier identifier, boolean purge) {
 
     Path tablePath = new Path(defaultWarehouseLocation(identifier));
     TableOperations ops = newTableOps(identifier);
-    TableMetadata lastMetadata;
-    if (purge && ops.current() != null) {
-      lastMetadata = ops.current();
-    } else {
-      lastMetadata = null;
-    }
-
+    TableMetadata lastMetadata = ops.current();
     try {
-      if (purge && lastMetadata != null) {
-        // Since the data files and the metadata files may store in different locations,
-        // so it has to call dropTableData to force delete the data file.
-        CatalogUtil.dropTableData(ops.io(), lastMetadata);
+      if (lastMetadata == null) {
+        LOG.debug("Not an iceberg table: {}", identifier);
+        return false;
+      } else {
+        if (purge) {
+          // Since the data files and the metadata files may store in different locations,
+          // so it has to call dropTableData to force delete the data file.
+          CatalogUtil.dropTableData(ops.io(), lastMetadata);
+        }
+        return fs.delete(tablePath, true /* recursive */);
       }
-      fs.delete(tablePath, true /* recursive */);
-      return true;
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to delete file: %s", tablePath);
     }

core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java

@@ -237,6 +237,24 @@ public void testDropTable() throws Exception {
     Assert.assertFalse(fs.isDirectory(new Path(metaLocation)));
   }
 
+  @Test
+  public void testDropNonIcebergTable() throws Exception {
+    Configuration conf = new Configuration();
+    String warehousePath = temp.newFolder().getAbsolutePath();
+    HadoopCatalog catalog = new HadoopCatalog(conf, warehousePath);
+    TableIdentifier testTable = TableIdentifier.of("db", "ns1", "ns2", "tbl");
+    String metaLocation = catalog.defaultWarehouseLocation(testTable);
+    //testing with non existent directory
+    Assert.assertFalse(catalog.dropTable(testTable));
+
+    FileSystem fs = Util.getFs(new Path(metaLocation), conf);
+    fs.mkdirs(new Path(metaLocation));
+    Assert.assertTrue(fs.isDirectory(new Path(metaLocation)));
+
+    Assert.assertFalse(catalog.dropTable(testTable));
+    Assert.assertTrue(fs.isDirectory(new Path(metaLocation)));
+  }
+
   @Test
   public void testRenameTable() throws Exception {
     Configuration conf = new Configuration();