You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/07/18 11:56:46 UTC
(hudi) branch master updated: [HUDI-8970] Change scheduleandexecute in RunCompactionProcedure to compact all pending plans too (#12794)
This is an automated email from the ASF dual-hosted git repository.
voonhous 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 a806d3968488 [HUDI-8970] Change scheduleandexecute in RunCompactionProcedure to compact all pending plans too (#12794)
a806d3968488 is described below
commit a806d3968488a05a75c756e4dd436fde2f717755
Author: voonhous <vo...@gmail.com>
AuthorDate: Sat Jul 18 19:56:39 2026 +0800
[HUDI-8970] Change scheduleandexecute in RunCompactionProcedure to compact all pending plans too (#12794)
---
.../command/procedures/HoodieProcedureUtils.scala | 3 +-
.../procedures/RunCompactionProcedure.scala | 12 +++--
.../hudi/procedure/TestCompactionProcedure.scala | 54 ++++++++++++++++++++--
3 files changed, 60 insertions(+), 9 deletions(-)
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureUtils.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureUtils.scala
index f433c58179b5..30dbbb0f16a6 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureUtils.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureUtils.scala
@@ -80,8 +80,7 @@ object HoodieProcedureUtils {
}
/**
- * scheduleAndExecute: schedule a new plan and then execute it, if no plan is generated during
- * schedule, execute all pending plans
+ * scheduleAndExecute: schedule a new plan and then execute all pending plans regardless of plan scheduling outcome
*/
case object ScheduleAndExecute extends Operation {
override def value: String = "scheduleandexecute"
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
index 0f881cdc8547..772d8be1b5b7 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunCompactionProcedure.scala
@@ -72,7 +72,7 @@ class RunCompactionProcedure extends BaseProcedure with ProcedureBuilder with Sp
confs = confs ++ HoodieCLIUtils.extractOptions(getArgValueOrDefault(args, PARAMETERS(4)).get.asInstanceOf[String])
}
var specificInstants = getArgValueOrDefault(args, PARAMETERS(5))
- val limit = getArgValueOrDefault(args, PARAMETERS(6))
+ val limit = getArgValueOrDefault(args, PARAMETERS(6)).asInstanceOf[Option[Int]]
// For old version compatibility
if (op.equals("run")) {
@@ -92,7 +92,7 @@ class RunCompactionProcedure extends BaseProcedure with ProcedureBuilder with Sp
.toSeq.sortBy(f => f)
var (filteredPendingCompactionInstants, operation) = HoodieProcedureUtils.filterPendingInstantsAndGetOperation(
- pendingCompactionInstants, specificInstants.asInstanceOf[Option[String]], Option(op), limit.asInstanceOf[Option[Int]])
+ pendingCompactionInstants, specificInstants.asInstanceOf[Option[String]], Option(op), limit)
var client: SparkRDDWriteClient[_] = null
try {
@@ -108,10 +108,16 @@ class RunCompactionProcedure extends BaseProcedure with ProcedureBuilder with Sp
if (operation.isSchedule) {
val instantTime = client.scheduleCompaction(HOption.empty[java.util.Map[String, String]])
instantTime.ifPresent(instant => {
- filteredPendingCompactionInstants = Seq(instant)
+ // If schedule ONLY, return the instant scheduled else, return pending compaction instants too
+ filteredPendingCompactionInstants = if (operation.isExecute) filteredPendingCompactionInstants :+ instant else Seq(instant)
})
}
+ filteredPendingCompactionInstants = if (limit.isDefined) {
+ filteredPendingCompactionInstants.take(limit.get)
+ } else {
+ filteredPendingCompactionInstants
+ }
logInfo(s"Compaction instants to run: ${filteredPendingCompactionInstants.mkString(",")}.")
if (operation.isExecute) {
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
index fcc3cbdd9430..0624e370fdd4 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
@@ -344,7 +344,7 @@ class TestCompactionProcedure extends HoodieSparkProcedureTestBase {
}
}
- test("Test Call run_clustering with limit parameter") {
+ test("Test Call run_compaction with limit parameter") {
withSQLConf("hoodie.compact.inline" -> "false", "hoodie.compact.inline.max.delta.commits" -> "1") {
withTempDir { tmp =>
val tableName = generateTableName
@@ -368,7 +368,7 @@ class TestCompactionProcedure extends HoodieSparkProcedureTestBase {
val conf = new Configuration
val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
- assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
+ assert(metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size() == 0)
assert(metaClient.getActiveTimeline.filterPendingClusteringTimeline().empty())
spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
@@ -381,12 +381,58 @@ class TestCompactionProcedure extends HoodieSparkProcedureTestBase {
spark.sql(s"call run_compaction(table => '$tableName', op => 'schedule')")
metaClient.reloadActiveTimeline();
- assert(2 == metaClient.getActiveTimeline.filterPendingCompactionTimeline().getInstants.size())
+ assert(metaClient.getActiveTimeline.filterPendingCompactionTimeline().getInstants.size() == 2)
spark.sql(s"call run_compaction(table => '$tableName', op => 'execute', limit => 1)");
metaClient.reloadActiveTimeline();
- assert(1 == metaClient.getActiveTimeline.filterPendingCompactionTimeline().getInstants.size())
+ assert(metaClient.getActiveTimeline.filterPendingCompactionTimeline().getInstants.size() == 1)
+ }
+ }
+ }
+
+ test("Test Call run_compaction with scheduleandexecute op") {
+ withSQLConf("hoodie.compact.inline" -> "false", "hoodie.compact.inline.max.delta.commits" -> "1") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val basePath = s"${tmp.getCanonicalPath}/$tableName"
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | name string,
+ | price double,
+ | ts long
+ |) using hudi
+ | tblproperties (
+ | type = 'mor',
+ | primaryKey = 'id',
+ | preCombineField = 'ts'
+ | )
+ | location '${basePath}'
+ """.stripMargin)
+
+ val conf = new Configuration
+ val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
+
+ assert(metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size() == 0)
+ assert(metaClient.getActiveTimeline.filterPendingClusteringTimeline().empty())
+
+ spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
+ spark.sql(s"update $tableName set name = 'a2' where id = 1")
+
+ spark.sql(s"call run_compaction(table => '$tableName', op => 'schedule')")
+
+ metaClient.reloadActiveTimeline();
+ assert(metaClient.getActiveTimeline.filterPendingCompactionTimeline().getInstants.size() == 1)
+
+ spark.sql(s"insert into $tableName values(2, 'b1', 20, 3000)")
+ spark.sql(s"update $tableName set name = 'b3' where id = 2")
+ spark.sql(s"call run_compaction(table => '$tableName', op => 'scheduleandexecute')");
+
+ // scheduleandexecute should run compact on all pending compactions
+ metaClient.reloadActiveTimeline();
+ assert(metaClient.getActiveTimeline.filterPendingCompactionTimeline().getInstants.size() == 0)
}
}
}