[SPARK-30312][SQL] Preserve path permission and acl when truncate table

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.command
 import java.net.{URI, URISyntaxException}
 
 import scala.collection.mutable.ArrayBuffer
-import scala.util.Try
+import scala.util.{Failure, Try}
 import scala.util.control.NonFatal
 
 import org.apache.hadoop.fs.{FileContext, FsConstants, Path}
@@ -499,8 +499,35 @@ case class TruncateTableCommand(
         val path = new Path(location.get)
         try {
           val fs = path.getFileSystem(hadoopConf)
+          val fileStatus = fs.getFileStatus(path)
+          // Not all fs impl. support these APIs.
+          val optPermission = Try(fileStatus.getPermission())
+          val optAcls = Try(fs.getAclStatus(path).getEntries)
+
           fs.delete(path, true)
+
+          // We should keep original permission/acl of the path.
+          // For owner/group, only super-user can set it, for example on HDFS. Because
+          // current user can delete the path, we assume the user/group is correct or not an issue.
           fs.mkdirs(path)
+          optPermission.foreach { permission =>
+            Try(fs.setPermission(path, permission)) match {
+              case Failure(e) =>
+                throw new AnalysisException(
+                  s"Failed to set original permission $permission back to " +
+                    s"the created path: $path. Exception: ${e.getMessage}")
+              case _ =>
+            }
+          }
+          optAcls.foreach { acls =>
+            Try(fs.setAcl(path, acls)) match {
+              case Failure(e) =>
+                throw new AnalysisException(
+                  s"Failed to set original ACL $acls back to " +
+                    s"the created path: $path. Exception: ${e.getMessage}")
+              case _ =>
+            }
+          }
         } catch {
           case NonFatal(e) =>
             throw new AnalysisException(

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala

@@ -21,7 +21,8 @@ import java.io.{File, PrintWriter}
 import java.net.URI
 import java.util.Locale
 
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{Path, RawLocalFileSystem}
+import org.apache.hadoop.fs.permission.{AclEntry, AclEntryScope, AclEntryType, AclStatus, FsAction, FsPermission}
 
 import org.apache.spark.internal.config
 import org.apache.spark.internal.config.RDD_PARALLEL_LISTING_THRESHOLD
@@ -1981,6 +1982,48 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     }
   }
 
+  test("truncate table - keep acl/permission") {
+    import testImplicits._
+
+    withSQLConf(
+      "fs.file.impl" -> classOf[FakeLocalFsFileSystem].getName,
+      "fs.file.impl.disable.cache" -> "true") {
+      withTable("tab1") {
+        sql("CREATE TABLE tab1 (col INT) USING parquet")
+        sql("INSERT INTO tab1 SELECT 1")
+        checkAnswer(spark.table("tab1"), Row(1))
+
+        val tablePath = new Path(spark.sessionState.catalog
+          .getTableMetadata(TableIdentifier("tab1")).storage.locationUri.get)
+
+        val hadoopConf = spark.sessionState.newHadoopConf()
+        val fs = tablePath.getFileSystem(hadoopConf)
+        val fileStatus = fs.getFileStatus(tablePath);
+
+        fs.setPermission(tablePath, new FsPermission("777"))
+        assert(fileStatus.getPermission().toString() == "rwxrwxrwx")
+
+        // Set ACL to table path.
+        val customAcl = new java.util.ArrayList[AclEntry]()
+        customAcl.add(new AclEntry.Builder()
+          .setType(AclEntryType.USER)
+          .setScope(AclEntryScope.ACCESS)
+          .setPermission(FsAction.READ).build())
+        fs.setAcl(tablePath, customAcl)
+        assert(fs.getAclStatus(tablePath).getEntries().get(0) == customAcl.get(0))
+
+        sql("TRUNCATE TABLE tab1")
+        assert(spark.table("tab1").collect().isEmpty)
+
+        val fileStatus2 = fs.getFileStatus(tablePath)
+        assert(fileStatus2.getPermission().toString() == "rwxrwxrwx")
+        val aclEntries = fs.getAclStatus(tablePath).getEntries()
+        assert(aclEntries.size() == 1)
+        assert(aclEntries.get(0) == customAcl.get(0))
+      }
+    }
+  }
+
   test("create temporary view with mismatched schema") {
     withTable("tab1") {
       spark.range(10).write.saveAsTable("tab1")
@@ -2879,3 +2922,25 @@ abstract class DDLSuite extends QueryTest with SQLTestUtils {
     }
   }
 }
+
+object FakeLocalFsFileSystem {
+  var aclStatus = new AclStatus.Builder().build()
+}
+
+// A fake test local filesystem used to test ACL. It keeps a ACL status. If deletes
+// a path of this filesystem, it will clean up the ACL status. Note that for test purpose,
+// it has only one ACL status for all paths.
+class FakeLocalFsFileSystem extends RawLocalFileSystem {
+  import FakeLocalFsFileSystem._
+
+  override def delete(f: Path, recursive: Boolean): Boolean = {
+    aclStatus = new AclStatus.Builder().build()
+    super.delete(f, recursive)
+  }
+
+  override def getAclStatus(path: Path): AclStatus = aclStatus
+
+  override def setAcl(path: Path, aclSpec: java.util.List[AclEntry]): Unit = {
+    aclStatus = new AclStatus.Builder().addEntries(aclSpec).build()
+  }
+}