From 9156464b4018c4d2d184c2eafa4983c7a488e423 Mon Sep 17 00:00:00 2001
From: itachi-sharingan <guardianaditya@gmail.com>
Date: Mon, 13 Sep 2021 23:16:10 +0530
Subject: [PATCH] Core: Fix HadoopCatalog drop table behavior (#3097)

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

Co-authored-by: adityasingh <guardinaditya@gmail.com>
---
 .../apache/iceberg/hadoop/HadoopCatalog.java  | 24 +++++++++----------
 .../iceberg/hadoop/TestHadoopCatalog.java     | 18 ++++++++++++++
 2 files changed, 29 insertions(+), 13 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
index 9d06c08b0044..7aa43491e003 100644
--- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java
+++ b/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);
     }
diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java b/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
index bebf5b5b32c1..9d71ea469a0f 100644
--- a/core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java
+++ b/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();