[HUDI-3826] Make truncate partition use delete_partition operation

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/HoodieSqlCommonUtils.scala

@@ -29,13 +29,13 @@ import org.apache.hudi.common.util.PartitionPathEncodeUtils
 import org.apache.hudi.{AvroConversionUtils, SparkAdapterSupport}
 import org.apache.spark.api.java.JavaSparkContext
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.{Resolver, UnresolvedRelation}
-import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTableType, HoodieCatalogTable}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
+import org.apache.spark.sql.catalyst.catalog.{CatalogTable, HoodieCatalogTable}
 import org.apache.spark.sql.catalyst.expressions.{And, Attribute, Cast, Expression, Literal}
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
-import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
-import org.apache.spark.sql.types.{DataType, NullType, StringType, StructField, StructType}
+import org.apache.spark.sql.types._
 import org.apache.spark.sql.{AnalysisException, Column, DataFrame, SparkSession}
 
 import java.net.URI
@@ -128,6 +128,7 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
 
   /**
    * Add the hoodie meta fields to the schema.
+   *
    * @param schema
    * @return
    */
@@ -143,6 +144,7 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
 
   /**
    * Remove the meta fields from the schema.
+   *
    * @param schema
    * @return
    */
@@ -171,6 +173,7 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
 
   /**
    * Get the table location.
+   *
    * @param tableId
    * @param spark
    * @return
@@ -233,6 +236,7 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
 
   /**
    * Split the expression to a sub expression seq by the AND operation.
+   *
    * @param expression
    * @return
    */
@@ -262,15 +266,15 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
    * Currently we support three kinds of instant time format for time travel query:
    * 1、yyyy-MM-dd HH:mm:ss
    * 2、yyyy-MM-dd
-   *   This will convert to 'yyyyMMdd000000'.
+   * This will convert to 'yyyyMMdd000000'.
    * 3、yyyyMMddHHmmss
    */
   def formatQueryInstant(queryInstant: String): String = {
     val instantLength = queryInstant.length
     if (instantLength == 19 || instantLength == 23) { // for yyyy-MM-dd HH:mm:ss[.SSS]
       HoodieInstantTimeGenerator.getInstantForDateString(queryInstant)
     } else if (instantLength == HoodieInstantTimeGenerator.SECS_INSTANT_ID_LENGTH
-      || instantLength  == HoodieInstantTimeGenerator.MILLIS_INSTANT_ID_LENGTH) { // for yyyyMMddHHmmss[SSS]
+      || instantLength == HoodieInstantTimeGenerator.MILLIS_INSTANT_ID_LENGTH) { // for yyyyMMddHHmmss[SSS]
       HoodieActiveTimeline.parseDateFromInstantTime(queryInstant) // validate the format
       queryInstant
     } else if (instantLength == 10) { // for yyyy-MM-dd
@@ -300,7 +304,7 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
 
   // Find the origin column from schema by column name, throw an AnalysisException if the column
   // reference is invalid.
-  def findColumnByName(schema: StructType, name: String, resolver: Resolver):Option[StructField] = {
+  def findColumnByName(schema: StructType, name: String, resolver: Resolver): Option[StructField] = {
     schema.fields.collectFirst {
       case field if resolver(field.name, name) => field
     }
@@ -372,4 +376,32 @@ object HoodieSqlCommonUtils extends SparkAdapterSupport {
     }.mkString(",")
     partitionsToDrop
   }
+
+  def getPartitionPathToTruncate(hoodieCatalogTable: HoodieCatalogTable,
+                                 table: CatalogTable,
+                                 partitionSpec: Option[TablePartitionSpec],
+                                 resolver: Resolver): String = {
+    val partCols = table.partitionColumnNames
+    val normalizedSpec: Seq[Map[String, String]] = Seq(partitionSpec.map { spec =>
+      normalizePartitionSpec(
+        spec,
+        partCols,
+        table.identifier.quotedString,
+        resolver)
+    }.get)
+
+    val allPartitionPaths = hoodieCatalogTable.getPartitionPaths
+    val enableEncodeUrl = isUrlEncodeEnabled(allPartitionPaths, table)
+
+    val partitionsToTruncate = normalizedSpec.map { spec =>
+      hoodieCatalogTable.partitionFields.map { partitionColumn =>
+        if (enableEncodeUrl) {
+          partitionColumn + "=" + "\"" + spec(partitionColumn) + "\""
+        } else {
+          partitionColumn + "=" + "'" + spec(partitionColumn) + "'"
+        }
+      }.mkString(" and ")
+    }.mkString
+    partitionsToTruncate
+  }
 }

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/ProvidesHoodieConfig.scala

@@ -21,7 +21,7 @@ import org.apache.hudi.DataSourceWriteOptions._
 import org.apache.hudi.common.config.TypedProperties
 import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload
 import org.apache.hudi.common.table.HoodieTableConfig
-import org.apache.hudi.config.HoodieWriteConfig.TBL_NAME
+import org.apache.hudi.config.HoodieWriteConfig.{AVRO_SCHEMA_VALIDATE_ENABLE, TBL_NAME}
 import org.apache.hudi.config.{HoodieIndexConfig, HoodieWriteConfig}
 import org.apache.hudi.hive.ddl.HiveSyncMode
 import org.apache.hudi.hive.{HiveSyncConfig, MultiPartKeysValueExtractor}
@@ -40,7 +40,6 @@ import org.apache.spark.sql.types.StructType
 
 import java.util
 import java.util.Locale
-
 import scala.collection.JavaConverters._
 
 trait ProvidesHoodieConfig extends Logging {

hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/TruncateHoodieTableCommand.scala

@@ -17,17 +17,12 @@
 
 package org.apache.spark.sql.hudi.command
 
-import org.apache.hadoop.fs.Path
-import org.apache.hudi.common.fs.FSUtils
-import org.apache.hudi.common.table.HoodieTableMetaClient
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
-import org.apache.spark.sql.catalyst.catalog.{CatalogStatistics, CatalogTableType, HoodieCatalogTable}
-import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{getPartitionPathToDrop, normalizePartitionSpec}
+import org.apache.spark.sql.catalyst.catalog.{CatalogTableType, HoodieCatalogTable}
+import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.getPartitionPathToTruncate
 import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
 
-import scala.util.control.NonFatal
-
 /**
  * Command for truncate hudi table.
  */
@@ -36,97 +31,44 @@ case class TruncateHoodieTableCommand(
    partitionSpec: Option[TablePartitionSpec])
   extends HoodieLeafRunnableCommand {
 
-  override def run(spark: SparkSession): Seq[Row] = {
+  override def run(sparkSession: SparkSession): Seq[Row] = {
     val fullTableName = s"${tableIdentifier.database}.${tableIdentifier.table}"
     logInfo(s"start execute truncate table command for $fullTableName")
 
-    val hoodieCatalogTable = HoodieCatalogTable(spark, tableIdentifier)
-    val properties = hoodieCatalogTable.tableConfig.getProps
-
-    try {
-      // Delete all data in the table directory
-      val catalog = spark.sessionState.catalog
-      val table = catalog.getTableMetadata(tableIdentifier)
-      val tableIdentWithDB = table.identifier.quotedString
-
-      if (table.tableType == CatalogTableType.VIEW) {
-        throw new AnalysisException(
-          s"Operation not allowed: TRUNCATE TABLE on views: $tableIdentWithDB")
-      }
-
-      if (table.partitionColumnNames.isEmpty && partitionSpec.isDefined) {
-        throw new AnalysisException(
-          s"Operation not allowed: TRUNCATE TABLE ... PARTITION is not supported " +
-            s"for tables that are not partitioned: $tableIdentWithDB")
-      }
+    val hoodieCatalogTable = HoodieCatalogTable(sparkSession, tableIdentifier)
 
-      val basePath = hoodieCatalogTable.tableLocation
-      val partCols = table.partitionColumnNames
-      val locations = if (partitionSpec.isEmpty || partCols.isEmpty) {
-        Seq(basePath)
-      } else {
-        val normalizedSpec: Seq[Map[String, String]] = Seq(partitionSpec.map { spec =>
-          normalizePartitionSpec(
-            spec,
-            partCols,
-            table.identifier.quotedString,
-            spark.sessionState.conf.resolver)
-        }.get)
+    val catalog = sparkSession.sessionState.catalog
+    val table = catalog.getTableMetadata(tableIdentifier)
+    val tableIdentWithDB = table.identifier.quotedString
 
-        val fullPartitionPath = FSUtils.getPartitionPath(basePath, getPartitionPathToDrop(hoodieCatalogTable, normalizedSpec))
-
-        Seq(fullPartitionPath)
-      }
+    if (table.tableType == CatalogTableType.VIEW) {
+      throw new AnalysisException(
+        s"Operation not allowed: TRUNCATE TABLE on views: $tableIdentWithDB")
+    }
 
-      val hadoopConf = spark.sessionState.newHadoopConf()
-      locations.foreach { location =>
-        val path = new Path(location.toString)
-        try {
-          val fs = path.getFileSystem(hadoopConf)
-          fs.delete(path, true)
-          fs.mkdirs(path)
-        } catch {
-          case NonFatal(e) =>
-            throw new AnalysisException(
-              s"Failed to truncate table $tableIdentWithDB when removing data of the path: $path " +
-                s"because of ${e.toString}")
-        }
-      }
+    if (table.partitionColumnNames.isEmpty && partitionSpec.isDefined) {
+      throw new AnalysisException(
+        s"Operation not allowed: TRUNCATE TABLE ... PARTITION is not supported " +
+          s"for tables that are not partitioned: $tableIdentWithDB")
+    }
 
-      // Also try to drop the contents of the table from the columnar cache
-      try {
-        spark.sharedState.cacheManager.uncacheQuery(spark.table(table.identifier), cascade = true)
-      } catch {
-        case NonFatal(_) =>
-      }
+    val basePath = hoodieCatalogTable.tableLocation
 
-      if (table.stats.nonEmpty) {
-        // empty table after truncation
-        val newStats = CatalogStatistics(sizeInBytes = 0, rowCount = Some(0))
-        catalog.alterTableStats(tableIdentifier, Some(newStats))
-      }
-      Seq.empty[Row]
-    } catch {
-      // TruncateTableCommand will delete the related directories first, and then refresh the table.
-      // It will fail when refresh table, because the hudi meta directory(.hoodie) has been deleted at the first step.
-      // So here ignore this failure, and refresh table later.
-      case NonFatal(e) =>
-        throw new AnalysisException(s"Exception when attempting to truncate table ${tableIdentifier.quotedString}: " + e)
-    }
+    val df = sparkSession.sqlContext.read
+      .format("hudi")
+      .load(basePath)
 
-    // If we have not specified the partition, truncate will delete all the data in the table path
-    // include the hoodie.properties. In this case we should reInit the table.
     if (partitionSpec.isEmpty) {
-      val hadoopConf = spark.sessionState.newHadoopConf()
-      // ReInit hoodie.properties
-      HoodieTableMetaClient.withPropertyBuilder()
-        .fromProperties(properties)
-        .initTable(hadoopConf, hoodieCatalogTable.tableLocation)
+      df.sqlContext.sql(s"delete from ${hoodieCatalogTable.tableName}")
+    } else {
+      val resolver = sparkSession.sessionState.conf.resolver
+      val partitionsToTruncate: String = getPartitionPathToTruncate(hoodieCatalogTable, table, partitionSpec, resolver)
+      df.sqlContext.sql(s"""delete from ${hoodieCatalogTable.tableName} where $partitionsToTruncate""")
     }
 
     // After deleting the data, refresh the table to make sure we don't keep around a stale
     // file relation in the metastore cache and cached table data in the cache manager.
-    spark.catalog.refreshTable(hoodieCatalogTable.table.identifier.quotedString)
+    sparkSession.catalog.refreshTable(table.identifier.quotedString)
     Seq.empty[Row]
   }
 }

hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala

@@ -539,7 +539,7 @@ case class HoodiePostAnalysisRule(sparkSession: SparkSession) extends Rule[Logic
       // Rewrite TruncateTableCommand to TruncateHoodieTableCommand
       case TruncateTableCommand(tableName, partitionSpec)
         if sparkAdapter.isHoodieTable(tableName, sparkSession) =>
-        new TruncateHoodieTableCommand(tableName, partitionSpec)
+        TruncateHoodieTableCommand(tableName, partitionSpec)
       case _ => plan
     }
   }

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/TestTruncateTable.scala

@@ -69,7 +69,7 @@ class TestTruncateTable extends TestHoodieSqlBase {
 
         df.write.format("hudi")
           .option(HoodieWriteConfig.TBL_NAME.key, tableName)
-          .option(TABLE_TYPE.key, MOR_TABLE_TYPE_OPT_VAL)
+          .option(TABLE_TYPE.key, COW_TABLE_TYPE_OPT_VAL)
           .option(RECORDKEY_FIELD.key, "id")
           .option(PRECOMBINE_FIELD.key, "ts")
           .option(PARTITIONPATH_FIELD.key, "dt")