[SPARK-32480] Support insert overwrite to move data to trash

Author: Udbhav30

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -50,7 +50,7 @@ import com.google.common.net.InetAddresses
 import org.apache.commons.codec.binary.Hex
 import org.apache.commons.lang3.SystemUtils
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
+import org.apache.hadoop.fs.{FileSystem, FileUtil, Path, Trash}
 import org.apache.hadoop.io.compress.{CompressionCodecFactory, SplittableCompressionCodec}
 import org.apache.hadoop.security.UserGroupInformation
 import org.apache.hadoop.yarn.conf.YarnConfiguration
@@ -269,6 +269,20 @@ private[spark] object Utils extends Logging {
     file.setExecutable(true, true)
   }
 
+  def moveToTrashIfEnabled(
+      fs: FileSystem,
+      partitionPath: Path,
+      trashInterval: Int,
+      hadoopConf: Configuration): Unit = {
+    if (trashInterval < 0) {
+      fs.delete(partitionPath, true)
+    } else {
+      logDebug(s"will move data ${partitionPath.toString} to trash")
+      hadoopConf.setInt("fs.trash.interval", trashInterval)
+      Trash.moveToAppropriateTrash(fs, partitionPath, hadoopConf)
+    }
+  }
+
   /**
    * Create a directory given the abstract pathname
    * @return true, if the directory is successfully created; otherwise, return false.

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2701,6 +2701,13 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val INSERT_OVERWRITE_TRASH_INTERVAL =
+    buildConf("spark.insertOverwrite.trash.interval")
+      .doc("This Configuration will decide whether move files to trash on insert overwrite" +
+        "If -1 files will be deleted without moving to trash")
+      .intConf
+      .createWithDefault(-1)
+
   /**
    * Holds information about keys that have been deprecated.
    *
@@ -3311,6 +3318,8 @@ class SQLConf extends Serializable with Logging {
   def optimizeNullAwareAntiJoin: Boolean =
     getConf(SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN)
 
+  def insertOverwriteTrashInterval: Int = getConf(SQLConf.INSERT_OVERWRITE_TRASH_INTERVAL)
+
   /** ********************** SQLConf functionality methods ************ */
 
   /** Set Spark SQL configuration properties. */

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveDirCommand.scala

@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.hive.client.HiveClientImpl
 import org.apache.spark.sql.util.SchemaUtils
+import org.apache.spark.util.Utils
 
 /**
  * Command for writing the results of `query` to file system.
@@ -108,8 +109,11 @@ case class InsertIntoHiveDirCommand(
         outputLocation = tmpPath.toString)
 
       if (overwrite && fs.exists(writeToPath)) {
+        val trashInterval = sparkSession.sessionState.conf.insertOverwriteTrashInterval
         fs.listStatus(writeToPath).foreach { existFile =>
-          if (Option(existFile.getPath) != createdTempDir) fs.delete(existFile.getPath, true)
+          if (Option(existFile.getPath) != createdTempDir) {
+            Utils.moveToTrashIfEnabled(fs, existFile.getPath, trashInterval, hadoopConf)
+          }
         }
       }
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.hive.execution
 import java.util.Locale
 
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{Path, Trash}
 import org.apache.hadoop.hive.ql.ErrorMsg
 import org.apache.hadoop.hive.ql.plan.TableDesc
 
@@ -309,9 +309,16 @@ case class InsertIntoHiveTable(
             partitionPath.foreach { path =>
               val fs = path.getFileSystem(hadoopConf)
               if (fs.exists(path)) {
-                if (!fs.delete(path, true)) {
-                  throw new RuntimeException(
-                    s"Cannot remove partition directory '$path'")
+                val trashInterval = sparkSession.sessionState.conf.insertOverwriteTrashInterval
+                if (trashInterval < 0 ) {
+                  if (!fs.delete(path, true)) {
+                    throw new RuntimeException(
+                      s"Cannot remove partition directory '$path'")
+                  }
+                } else {
+                  logDebug(s"will move data ${path.toString} to trash")
+                  hadoopConf.setInt("fs.trash.interval", trashInterval)
+                  Trash.moveToAppropriateTrash(fs, path, hadoopConf)
                 }
                 // Don't let Hive do overwrite operation since it is slower.
                 doHiveOverwrite = false