From 23645d9b85172c6040fb47830028bf8a75e05310 Mon Sep 17 00:00:00 2001
From: pradeep <pradeep@apache.org>
Date: Tue, 11 May 2021 11:46:39 +0530
Subject: [PATCH] [I3SI4C] Delete empty staging directory and subdirectories
 through a single api call with recursive option

---
 .../SemiTransactionalHiveMetastore.java       | 24 ++++++++++++++++++-
 1 file changed, 23 insertions(+), 1 deletion(-)

diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/SemiTransactionalHiveMetastore.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
index 90a130307..b3e974fd2 100644
--- a/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
+++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
@@ -1667,7 +1667,7 @@ public class SemiTransactionalHiveMetastore
 
             if (pathExists(hdfsContext, hdfsEnvironment, currentPath)) {
                 if (!targetPath.equals(currentPath)) {
-                    renameNewPartitionDirectory(
+                    renamePartitionDirectory(
                             hdfsContext,
                             hdfsEnvironment,
                             currentPath,
@@ -2264,6 +2264,22 @@ public class SemiTransactionalHiveMetastore
         }
     }
 
+    private static void renamePartitionDirectory(HdfsContext context,
+            HdfsEnvironment hdfsEnvironment,
+            Path source,
+            Path target,
+            List<DirectoryCleanUpTask> cleanUpTasksForAbort)
+    {
+        try {
+            if (hdfsEnvironment.getFileSystem(context, source).rename(source, target)) {
+                cleanUpTasksForAbort.add(new DirectoryCleanUpTask(context, target, true));
+            }
+        }
+        catch (IOException e) {
+            renameNewPartitionDirectory(context, hdfsEnvironment, source, target, cleanUpTasksForAbort);
+        }
+    }
+
     private static void renameDirectory(HdfsEnvironment.HdfsContext context, HdfsEnvironment hdfsEnvironment, Path source, Path target, Runnable runWhenRenameSuccess)
     {
         if (pathExists(context, hdfsEnvironment, target)) {
@@ -2312,6 +2328,12 @@ public class SemiTransactionalHiveMetastore
             return new RecursiveDeleteResult(false, ImmutableList.of());
         }
 
+        if (directory.getName().startsWith(".staging")) {
+            if (deleteIfExists(fileSystem, directory, true)) {
+                return new RecursiveDeleteResult(true, ImmutableList.of());
+            }
+        }
+
         FileStatus[] allFiles;
         try {
             allFiles = fileSystem.listStatus(directory);