HIVE-29528: Cleanup managed DB dir after async drop (apache#6397)

Author: VenuReddy2103

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CleanupRequest.java

@@ -30,6 +30,8 @@ public class CleanupRequest {
   private final String location;
   private final List<Path> obsoleteDirs;
   private final boolean purge;
+  private final boolean softDelete;
+  private final boolean sourceOfReplication;
   private final String runAs;
   private final String dbName;
   private final String fullPartitionName;
@@ -38,6 +40,8 @@ public CleanupRequest(CleanupRequestBuilder builder) {
     this.location = builder.location;
     this.obsoleteDirs = builder.obsoleteDirs;
     this.purge = builder.purge;
+    this.softDelete = builder.softDelete;
+    this.sourceOfReplication = builder.sourceOfReplication;
     this.runAs = builder.runAs;
     this.dbName = builder.dbName;
     this.fullPartitionName = builder.fullPartitionName;
@@ -55,6 +59,14 @@ public boolean isPurge() {
     return purge;
   }
 
+  public boolean isSoftDelete() {
+    return softDelete;
+  }
+
+  public boolean isSourceOfReplication() {
+    return sourceOfReplication;
+  }
+
   public String runAs() {
     return runAs;
   }
@@ -74,6 +86,8 @@ public static class CleanupRequestBuilder {
     private String location;
     private List<Path> obsoleteDirs;
     private boolean purge;
+    private boolean softDelete;
+    private boolean sourceOfReplication;
     private String runAs;
     private String dbName;
     private String fullPartitionName;
@@ -93,6 +107,16 @@ public CleanupRequestBuilder setPurge(boolean purge) {
       return this;
     }
 
+    public CleanupRequestBuilder setSoftDelete(boolean softDelete) {
+      this.softDelete = softDelete;
+      return this;
+    }
+
+    public CleanupRequestBuilder setSourceOfReplication(boolean sourceOfReplication) {
+      this.sourceOfReplication = sourceOfReplication;
+      return this;
+    }
+
     public CleanupRequestBuilder setDbName(String dbName) {
       this.dbName = dbName;
       return this;

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/FSRemover.java

@@ -99,17 +99,18 @@ private List<Path> removeFiles(CleanupRequest cr)
     LOG.info("About to remove {} obsolete directories from {}. {}", cr.getObsoleteDirs().size(),
             cr.getLocation(), CompactorUtil.getDebugInfo(cr.getObsoleteDirs()));
     boolean needCmRecycle;
+    Database db = null;
     try {
-      Database db = metadataCache.computeIfAbsent(cr.getDbName(),
+      db = metadataCache.computeIfAbsent(cr.getDbName(),
               () -> CompactorUtil.resolveDatabase(conf, cr.getDbName()));
       needCmRecycle = ReplChangeManager.isSourceOfReplication(db);
     } catch (NoSuchObjectException ex) {
       // can not drop a database which is a source of replication
-      needCmRecycle = false;
+      needCmRecycle = cr.isSourceOfReplication();
     } catch (RuntimeException ex) {
       if (ex.getCause() instanceof NoSuchObjectException) {
         // can not drop a database which is a source of replication
-        needCmRecycle = false;
+        needCmRecycle = cr.isSourceOfReplication();
       } else {
         throw ex;
       }
@@ -126,6 +127,21 @@ private List<Path> removeFiles(CleanupRequest cr)
         deleted.add(dead);
       }
     }
+    removeDbDirIfOrphaned(db, cr, fs);
     return deleted;
   }
+
+  private void removeDbDirIfOrphaned(Database db, CleanupRequest cr, FileSystem fs) throws IOException {
+    if (db != null || !cr.isSoftDelete()) {
+      return;
+    }
+    Path databasePath = new Path(cr.getLocation()).getParent();
+    if (!FileUtils.isDirEmpty(fs, databasePath)) {
+      return;
+    }
+    if (cr.isSourceOfReplication()) {
+      replChangeManager.recycle(databasePath, ReplChangeManager.RecycleType.MOVE, cr.isPurge());
+    }
+    FileUtils.deleteDir(fs, databasePath, cr.isPurge(), conf);
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/CompactionCleaner.java

@@ -198,8 +198,7 @@ private void cleanUsingLocation(CompactionInfo ci, String path, boolean requires
       deleted = fsRemover.clean(getCleaningRequestBasedOnLocation(ci, path));
     }
     if (!deleted.isEmpty()) {
-      ci.setSoftDelete(true);
-      txnHandler.markCleaned(ci);
+      txnHandler.markCleaned(ci.asSoftDeleted());
     } else {
       txnHandler.clearCleanerStart(ci);
     }
@@ -287,6 +286,8 @@ private CleanupRequest getCleaningRequestBasedOnLocation(CompactionInfo ci, Stri
         .setFullPartitionName(ci.getFullPartitionName())
         .setRunAs(ci.runAs)
         .setPurge(ifPurge)
+        .setSoftDelete(true)
+        .setSourceOfReplication(ci.isSourceOfReplication())
         .setObsoleteDirs(Collections.singletonList(obsoletePath))
         .build();
   }

ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hive.ql;
 
 import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -1827,11 +1828,9 @@ private void dropDatabaseCascadeNonBlocking() throws Exception {
 
     runCleaner(hiveConf);
 
-    stat = fs.listStatus(new Path(getWarehouseDir(), database + ".db"),
-      t -> t.getName().matches("(mv_)?" + tableName + "2" + SOFT_DELETE_TABLE_PATTERN));
-    if (stat.length != 0) {
-      Assert.fail("Table data was not removed from FS");
-    }
+    Assert.assertThrows(database + ".db directory should not exist", FileNotFoundException.class, () -> {
+      fs.listStatus(new Path(getWarehouseDir(), database + ".db"));
+    });
   }
 
   @Test

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/AcidEventListener.java

@@ -19,6 +19,7 @@
 package org.apache.hadoop.hive.metastore;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.common.repl.ReplConst;
 import org.apache.hadoop.hive.metastore.api.CompactionRequest;
 import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.Database;
@@ -91,6 +92,10 @@ public void onDropTable(DropTableEvent tableEvent) throws MetaException {
             rqst.setRunas(TxnUtils.findUserToRunAs(table.getSd().getLocation(), table, conf));
             rqst.putToProperties("location", table.getSd().getLocation());
             rqst.putToProperties("ifPurge", Boolean.toString(isMustPurge(tableEvent.getEnvironmentContext(), table)));
+            if (Boolean.parseBoolean(tableEvent.getEnvironmentContext().getProperties()
+                .get(ReplConst.SOURCE_OF_REPLICATION))) {
+              rqst.putToProperties(ReplConst.SOURCE_OF_REPLICATION, Boolean.TRUE.toString());
+            }
             txnHandler.submitForCleanup(rqst, table.getWriteId(), currentTxn);
           } catch (InterruptedException | IOException e) {
             throwMetaException(e);

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/DropDatabaseHandler.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.repl.ReplConst;
 import org.apache.hadoop.hive.metastore.Batchable;
 import org.apache.hadoop.hive.metastore.HMSHandler;
 import org.apache.hadoop.hive.metastore.IHMSHandler;
@@ -136,6 +137,9 @@ public DropDatabaseResult execute() throws TException, IOException {
         if (isSoftDelete) {
           context = new EnvironmentContext();
           context.putToProperties(hive_metastoreConstants.TXN_ID, String.valueOf(request.getTxnId()));
+          if (ReplChangeManager.isSourceOfReplication(db)) {
+            context.putToProperties(ReplConst.SOURCE_OF_REPLICATION, Boolean.TRUE.toString());
+          }
           request.setDeleteManagedDir(false);
         }
         DropTableRequest dropRequest = new DropTableRequest(name, table.getTableName());

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/entities/CompactionInfo.java

@@ -19,6 +19,7 @@
 
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.hadoop.hive.common.ValidCompactorWriteIdList;
+import org.apache.hadoop.hive.common.repl.ReplConst;
 import org.apache.hadoop.hive.metastore.api.CompactionInfoStruct;
 import org.apache.hadoop.hive.metastore.api.CompactionType;
 import org.apache.hadoop.hive.metastore.api.MetaException;
@@ -371,11 +372,16 @@ public boolean isAbortedTxnCleanup() {
     return type == CompactionType.ABORT_TXN_CLEANUP;
   }
 
-  public void setSoftDelete(boolean softDelete) {
-    this.softDelete = softDelete;
+  public CompactionInfo asSoftDeleted() {
+    this.softDelete = true;
+    return this;
   }
 
   public boolean isSoftDelete() {
     return softDelete;
   }
+
+  public boolean isSourceOfReplication() {
+    return Boolean.parseBoolean(getProperty(ReplConst.SOURCE_OF_REPLICATION));
+  }
 }