(hudi) branch master updated: feat: Introduce a Spark procedure to trigger LSM timeline compaction (#18659)

[da...@apache.org]

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 f8d70cb58d7c feat: Introduce a Spark procedure to trigger LSM timeline compaction (#18659)
f8d70cb58d7c is described below

commit f8d70cb58d7c9ea6a96ece59ef4ce097c46c3921
Author: chaoyang <ch...@apache.org>
AuthorDate: Fri May 1 11:37:54 2026 +0800

    feat: Introduce a Spark procedure to trigger LSM timeline compaction (#18659)
    
    * feat: Introduce a Spark procedure to trigger LSM timeline compaction
    
    * fixup: address review on RunTimelineCompactionProcedure
    
    - Acquire the txn state-change lock around compactAndClean so the procedure cannot race with a concurrent archival/compaction over the LSM timeline manifest
    - Use TimeUnit.NANOSECONDS.toMillis for self-documenting unit conversion
---
 .../hudi/command/procedures/HoodieProcedures.scala |   1 +
 .../RunTimelineCompactionProcedure.scala           |  97 ++++++++++++++++++++
 .../TestRunTimelineCompactionProcedure.scala       | 100 +++++++++++++++++++++
 3 files changed, 198 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 4319ce77c1c2..508e37d47663 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
@@ -93,6 +93,7 @@ object HoodieProcedures {
       ,(ShowTablePropertiesProcedure.NAME, ShowTablePropertiesProcedure.builder)
       ,(HelpProcedure.NAME, HelpProcedure.builder)
       ,(ArchiveCommitsProcedure.NAME, ArchiveCommitsProcedure.builder)
+      ,(RunTimelineCompactionProcedure.NAME, RunTimelineCompactionProcedure.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/RunTimelineCompactionProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunTimelineCompactionProcedure.scala
new file mode 100644
index 000000000000..14b1bcb2d065
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunTimelineCompactionProcedure.scala
@@ -0,0 +1,97 @@
+/*
+ * 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
+import org.apache.hudi.client.SparkRDDWriteClient
+import org.apache.hudi.client.timeline.versioning.v2.LSMTimelineWriter
+import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator
+import org.apache.hudi.common.util.{Option => HOption}
+import org.apache.hudi.table.HoodieSparkTable
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
+
+import java.util.concurrent.TimeUnit
+import java.util.function.Supplier
+
+class RunTimelineCompactionProcedure extends BaseProcedure with ProcedureBuilder with Logging {
+
+  private val PARAMETERS = Array[ProcedureParameter](
+    ProcedureParameter.optional(0, "table", DataTypes.StringType),
+    ProcedureParameter.optional(1, "path", DataTypes.StringType)
+  )
+
+  private val OUTPUT_TYPE = new StructType(Array[StructField](
+    StructField("start_compaction_time", DataTypes.StringType, nullable = true, Metadata.empty),
+    StructField("time_taken_in_millis", DataTypes.LongType, nullable = true, Metadata.empty)
+  ))
+
+  override def parameters: Array[ProcedureParameter] = PARAMETERS
+
+  override def outputType: StructType = OUTPUT_TYPE
+
+  override def call(args: ProcedureArgs): Seq[Row] = {
+    super.checkArgs(PARAMETERS, args)
+
+    val tableName = getArgValueOrDefault(args, PARAMETERS(0))
+    val tablePath = getArgValueOrDefault(args, PARAMETERS(1))
+    val basePath = getBasePath(tableName, tablePath)
+
+    var client: SparkRDDWriteClient[_] = null
+    try {
+      client = HoodieCLIUtils.createHoodieWriteClient(sparkSession, basePath, Map.empty,
+        tableName.asInstanceOf[Option[String]])
+      val config = client.getConfig
+      val context = client.getEngineContext
+      val table = HoodieSparkTable.create(config, context)
+      val writer = LSMTimelineWriter.getInstance(config, table)
+      val txnManager = client.getTransactionManager
+
+      // Acquire the same state-change lock that TimelineArchiverV2 holds while it
+      // calls compactAndClean, so this procedure cannot race with a concurrent
+      // archival/compaction over the LSM timeline manifest.
+      val startInstant = HoodieInstantTimeGenerator.getCurrentInstantTimeStr
+      val startNanos = System.nanoTime()
+      txnManager.beginStateChange(HOption.empty(), HOption.empty())
+      try {
+        writer.compactAndClean(context)
+      } finally {
+        txnManager.endStateChange(HOption.empty())
+      }
+      val durationMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos)
+
+      Seq(Row(startInstant, durationMillis))
+    } finally {
+      if (client != null) {
+        client.close()
+      }
+    }
+  }
+
+  override def build: Procedure = new RunTimelineCompactionProcedure()
+}
+
+object RunTimelineCompactionProcedure {
+  val NAME: String = "run_timeline_compaction"
+
+  def builder: Supplier[ProcedureBuilder] = new Supplier[ProcedureBuilder] {
+    override def get(): RunTimelineCompactionProcedure = new RunTimelineCompactionProcedure()
+  }
+}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRunTimelineCompactionProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRunTimelineCompactionProcedure.scala
new file mode 100644
index 000000000000..0c5a485dcf77
--- /dev/null
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRunTimelineCompactionProcedure.scala
@@ -0,0 +1,100 @@
+/*
+ * 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
+
+class TestRunTimelineCompactionProcedure extends HoodieSparkProcedureTestBase {
+
+  test("Test Call run_timeline_compaction Procedure by Table") {
+    withTempDir { tmp =>
+      val tableName = generateTableName
+      spark.sql(
+        s"""
+           |create table $tableName (
+           | id int,
+           | name string,
+           | price double,
+           | ts long
+           | ) using hudi
+           | location '${tmp.getCanonicalPath}'
+           | tblproperties (
+           |   primaryKey = 'id',
+           |   type = 'cow',
+           |   orderingFields = 'ts',
+           |   hoodie.metadata.enable = 'false',
+           |   hoodie.keep.min.commits = '2',
+           |   hoodie.keep.max.commits = '3',
+           |   hoodie.cleaner.commits.retained = '1',
+           |   hoodie.timeline.compaction.batch.size = '2'
+           | )
+           |""".stripMargin)
+
+      // Generate enough commits so that subsequent archive_commits calls produce
+      // multiple L0 archived timeline files, which is the input that the timeline
+      // compaction procedure will then merge.
+      for (i <- 1 to 8) {
+        spark.sql(s"insert into $tableName values($i, 'a$i', ${i * 10}, ${i * 1000})")
+      }
+      spark.sql(s"call archive_commits(table => '$tableName'" +
+        s", min_commits => 2, max_commits => 3, retain_commits => 1, enable_metadata => false)")
+
+      val result = spark.sql(s"call run_timeline_compaction(table => '$tableName')").collect()
+      assertResult(1)(result.length)
+      assert(result(0).getString(0) != null && result(0).getString(0).nonEmpty,
+        s"start_compaction_time should be a non-empty instant string, got: ${result(0).getString(0)}")
+      assert(result(0).getLong(1) >= 0L,
+        s"time_taken_in_millis should be non-negative, got: ${result(0).getLong(1)}")
+
+      // Calling the procedure again on a quiesced table should still succeed
+      // (it is effectively a no-op when there is nothing left to compact).
+      val resultIdempotent = spark.sql(s"call run_timeline_compaction(table => '$tableName')").collect()
+      assertResult(1)(resultIdempotent.length)
+    }
+  }
+
+  test("Test Call run_timeline_compaction Procedure by Path") {
+    withTempDir { tmp =>
+      val tableName = generateTableName
+      spark.sql(
+        s"""
+           |create table $tableName (
+           | id int,
+           | name string,
+           | price double,
+           | ts long
+           | ) using hudi
+           | location '${tmp.getCanonicalPath}'
+           | tblproperties (
+           |   primaryKey = 'id',
+           |   type = 'cow',
+           |   orderingFields = 'ts',
+           |   hoodie.metadata.enable = 'false'
+           | )
+           |""".stripMargin)
+
+      spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+
+      val result = spark.sql(
+        s"call run_timeline_compaction(path => '${tmp.getCanonicalPath}')").collect()
+      assertResult(1)(result.length)
+      assert(result(0).getString(0) != null && result(0).getString(0).nonEmpty)
+      assert(result(0).getLong(1) >= 0L)
+    }
+  }
+}