This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 2f35b7c9cf7 [HUDI-8529] Add truncate table procedure (#12262)
2f35b7c9cf7 is described below
commit 2f35b7c9cf73757d47aacc8948d02d84febabd92
Author: fhan <[email protected]>
AuthorDate: Fri Nov 29 09:45:22 2024 +0800
[HUDI-8529] Add truncate table procedure (#12262)
---
.../hudi/command/procedures/HoodieProcedures.scala | 1 +
.../procedures/TruncateTableProcedure.scala | 129 +++++++++++++++++++++
.../procedure/TestTruncateTableProcedure.scala | 125 ++++++++++++++++++++
3 files changed, 255 insertions(+)
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
index 501bdbc2da0..0a6afd4e00a 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedures.scala
@@ -95,6 +95,7 @@ object HoodieProcedures {
,(ArchiveCommitsProcedure.NAME, ArchiveCommitsProcedure.builder)
,(RunTTLProcedure.NAME, RunTTLProcedure.builder)
,(DropPartitionProcedure.NAME, DropPartitionProcedure.builder)
+ ,(TruncateTableProcedure.NAME, TruncateTableProcedure.builder)
)
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/TruncateTableProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/TruncateTableProcedure.scala
new file mode 100644
index 00000000000..ad0b1d48448
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/TruncateTableProcedure.scala
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.command.procedures
+
+import org.apache.hudi.{HoodieCLIUtils, HoodieSparkSqlWriter}
+import org.apache.hudi.client.common.HoodieSparkEngineContext
+import org.apache.hudi.common.fs.FSUtils
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.exception.HoodieException
+import org.apache.hudi.hadoop.fs.HadoopFSUtils
+import org.apache.hudi.storage.{HoodieStorageUtils, StoragePath}
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.{AnalysisException, Row, SaveMode}
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.CatalogTableType
+import org.apache.spark.sql.hudi.ProvidesHoodieConfig
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField,
StructType}
+
+import java.util.function.Supplier
+
+class TruncateTableProcedure extends BaseProcedure
+ with ProcedureBuilder
+ with ProvidesHoodieConfig
+ with Logging {
+ def build: Procedure = new TruncateTableProcedure()
+
+ private val PARAMETERS = Array[ProcedureParameter](
+ ProcedureParameter.required(0, "table", DataTypes.StringType),
+ ProcedureParameter.optional(1, "partitions", DataTypes.StringType)
+ )
+
+ def parameters: Array[ProcedureParameter] = PARAMETERS
+
+ def outputType: StructType = StructType(Array[StructField](
+ StructField("result", DataTypes.StringType, nullable = true,
Metadata.empty),
+ StructField("time_cost", DataTypes.LongType, nullable = true,
Metadata.empty)
+ ))
+
+ override def call(args: ProcedureArgs): Seq[Row] = {
+ super.checkArgs(parameters, args)
+
+ val tableNameStr = getArgValueOrDefault(args,
parameters(0)).get.asInstanceOf[String]
+ logInfo(s"start execute truncate table procedure for $tableNameStr")
+
+ val partitionsStr = getArgValueOrDefault(args,
parameters(1)).getOrElse("").asInstanceOf[String]
+
+ val catalogTable = HoodieCLIUtils.getHoodieCatalogTable(sparkSession,
tableNameStr)
+
+ val (db, tableName) = getDbAndTableName(tableNameStr)
+
+ val catalog = sparkSession.sessionState.catalog
+ val table = catalog.getTableMetadata(TableIdentifier(tableName, Some(db)))
+ val tableId = table.identifier.quotedString
+
+ if (table.tableType == CatalogTableType.VIEW) {
+ throw new AnalysisException(
+ s"Operation not allowed: TRUNCATE TABLE on views: $tableId")
+ }
+
+ if (table.partitionColumnNames.isEmpty && partitionsStr.nonEmpty) {
+ throw new AnalysisException(
+ s"Operation not allowed: TRUNCATE TABLE ... PARTITION is not supported
" +
+ s"for tables that are not partitioned: $tableId")
+ }
+
+ val basePath = catalogTable.tableLocation
+ val properties = catalogTable.tableConfig.getProps
+
+ if (partitionsStr.isEmpty) {
+ val targetPath = new StoragePath(basePath)
+ val engineContext = new
HoodieSparkEngineContext(sparkSession.sparkContext)
+ val storage = HoodieStorageUtils.getStorage(
+ basePath,
HadoopFSUtils.getStorageConf(sparkSession.sessionState.newHadoopConf))
+
+ val startTime = System.currentTimeMillis()
+ FSUtils.deleteDir(engineContext, storage, targetPath,
sparkSession.sparkContext.defaultParallelism)
+
+ // ReInit hoodie.properties
+ val metaClient = HoodieTableMetaClient.newTableBuilder()
+ .fromProperties(properties)
+
.initTable(HadoopFSUtils.getStorageConf(sparkSession.sessionState.newHadoopConf),
catalogTable.tableLocation)
+
+ catalogTable.tableConfig.clearMetadataPartitions(metaClient)
+ logInfo(s"Success to execute truncate table procedure for
${tableNameStr}")
+ Seq(Row("SUCCESS", System.currentTimeMillis() - startTime))
+
+
+ } else {
+ val parameters = buildHoodieDropPartitionsConfig(sparkSession,
catalogTable, partitionsStr)
+ val (success, _, _, _, _, _) = HoodieSparkSqlWriter.write(
+ sparkSession.sqlContext,
+ SaveMode.Append,
+ parameters,
+ sparkSession.emptyDataFrame)
+ if (!success) {
+ throw new HoodieException("Truncate Hoodie Table procedure failed")
+ }
+ }
+
+ // 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.
+ sparkSession.catalog.refreshTable(table.identifier.quotedString)
+ logInfo(s"Finish execute truncate table procedure for $tableNameStr")
+ Seq.empty[Row]
+ }
+}
+
+object TruncateTableProcedure {
+ val NAME = "truncate_table"
+
+ def builder: Supplier[ProcedureBuilder] = new Supplier[ProcedureBuilder] {
+ override def get(): ProcedureBuilder = new TruncateTableProcedure
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestTruncateTableProcedure.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestTruncateTableProcedure.scala
new file mode 100644
index 00000000000..7c85011c5c8
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestTruncateTableProcedure.scala
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hudi.procedure
+
+import org.apache.hadoop.fs.Path
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata
+import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.hadoop.fs.HadoopFSUtils
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+
+import scala.collection.JavaConverters._
+
+class TestTruncateTableProcedure extends HoodieSparkProcedureTestBase {
+
+ test("Test Call truncate_table Procedureļ¼truncate table") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath + "/" + tableName
+ //Step1: create table and insert data
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | preCombineField = 'ts'
+ | )
+ """.stripMargin)
+
+ spark.sql(s"insert into $tableName select 1, 'a1', 10.0, 1000L")
+ spark.sql(s"insert into $tableName select 2, 'a2', 20.0, 1500L")
+ spark.sql(s"insert into $tableName select 3, 'a3', 30.0, 2000L")
+ spark.sql(s"insert into $tableName select 4, 'a4', 40.0, 2500L")
+
+ //Step2: call truncate_table procedure
+ spark.sql(s"""call truncate_table(table => '$tableName')""")
+
+ val fs = new
Path(tablePath).getFileSystem(spark.sparkContext.hadoopConfiguration)
+ val files = fs.listStatus(new Path(tablePath))
+
+ //Step3: check number of directories under tablePath, only .hoodie
+ assertTrue(files.size == 1)
+ }
+ }
+
+ test("Test Call truncate_table Procedureļ¼truncate given partitions") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val tablePath = tmp.getCanonicalPath + "/" + tableName
+ //Step1: create table and insert data
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long,
+ | year string,
+ | month string,
+ | day string
+ |) using hudi
+ |tblproperties (
+ | primaryKey = 'id',
+ | preCombineField = 'ts'
+ |)
+ |partitioned by(year, month, day)
+ |location '$tablePath'
+ |
+ """.stripMargin)
+ insertData(tableName)
+ //Step2: call truncate_table procedure, truncate given
partitions:year=2019/month=08/day=31,30,29
+ spark.sql(s"""call truncate_table(table => '$tableName', partitions =>
'year=2019/month=08/day=31,year=2019/month=08/day=30,year=2019/month=08/day=29')""")
+ val metaClient = getTableMetaClient(tablePath)
+ val replaceCommitInstant = metaClient.getActiveTimeline.getWriteTimeline
+ .getCompletedReplaceTimeline.getReverseOrderedInstants.findFirst()
+ .get()
+ val partitions = HoodieReplaceCommitMetadata
+
.fromBytes(metaClient.getActiveTimeline.getInstantDetails(replaceCommitInstant).get(),
classOf[HoodieReplaceCommitMetadata])
+ .getPartitionToReplaceFileIds
+ .keySet()
+ //Step3: check number of truncated partitions and location startWith
+ assertEquals(3, partitions.size())
+ assertTrue(partitions.asScala.forall(_.startsWith("year=2019/month=08")))
+ // clean
+ //Step4: call clean and check result: left only 1 record
+ spark.sql(s"""call run_clean(table => '$tableName', clean_policy =>
'KEEP_LATEST_FILE_VERSIONS', file_versions_retained => 1)""")
+ val result = spark.sql(s"""select * from $tableName""").collect()
+ assertEquals(1, result.length)
+ }
+ }
+
+ private def insertData(tableName: String): Unit = {
+ spark.sql(s"""insert into $tableName values (1, 'n1', 1, 1, '2019', '08',
'31')""")
+ spark.sql(s"""insert into $tableName values (2, 'n2', 2, 2, '2019', '08',
'30')""")
+ spark.sql(s"""insert into $tableName values (3, 'n3', 3, 3, '2019', '08',
'29')""")
+ spark.sql(s"""insert into $tableName values (4, 'n4', 4, 4, '2019', '07',
'31')""")
+ }
+ private def getTableMetaClient(tablePath: String): HoodieTableMetaClient = {
+ HoodieTableMetaClient.builder()
+ .setBasePath(tablePath)
+
.setConf(HadoopFSUtils.getStorageConf(spark.sparkContext.hadoopConfiguration))
+ .build()
+ }
+
+}
