This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 18a2eb849fea7babf2dc56656686ee35d244c393 Author: Y Ethan Guo <[email protected]> AuthorDate: Sat Jul 4 17:03:33 2026 -0700 fix(spark): make export_instants desc ordering work (#19172) (cherry picked from commit 8347be48ce31ef3aee70964a9d0931e32b857136) --- .../procedures/ExportInstantsProcedure.scala | 14 ++- .../procedure/TestExportInstantsProcedure.scala | 122 +++++++++++++++++---- 2 files changed, 110 insertions(+), 26 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ExportInstantsProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ExportInstantsProcedure.scala index 5ba9960d1d3c..3281e9de653b 100644 --- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ExportInstantsProcedure.scala +++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ExportInstantsProcedure.scala @@ -87,17 +87,19 @@ class ExportInstantsProcedure extends BaseProcedure with ProcedureBuilder with L if (!new File(localFolder).isDirectory) throw new HoodieException(localFolder + " is not a valid local directory") - // The non archived instants can be listed from the Timeline. - val nonArchivedInstants: util.List[HoodieInstant] = metaClient + // The non archived instants can be listed from the Timeline. Wrap in a mutable list because the + // desc branch below reverses it in place via Collections.reverse, which fails on the read-only + // list produced by asJava over an immutable Scala collection. + val nonArchivedInstants: util.List[HoodieInstant] = new util.ArrayList[HoodieInstant](metaClient .getActiveTimeline .filterCompletedInstants.getInstants.iterator().asScala .filter((i: HoodieInstant) => actionSet.contains(i.getAction)) - .toList.asJava + .toList.asJava) - // Archived instants are in the commit archive files + // Archived instants are in the commit archive files (also mutable, for the same reason). val statuses: Array[FileStatus] = HadoopFSUtils.getFs(basePath, jsc.hadoopConfiguration()).globStatus(archivePath) - val archivedStatuses = List(statuses: _*) - .sortWith((f1, f2) => (f1.getModificationTime - f2.getModificationTime).toInt > 0).asJava + val archivedStatuses = new util.ArrayList[FileStatus](List(statuses: _*) + .sortWith((f1, f2) => (f1.getModificationTime - f2.getModificationTime).toInt > 0).asJava) if (desc) { Collections.reverse(nonArchivedInstants) diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestExportInstantsProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestExportInstantsProcedure.scala index 372c6cc5e7c5..2b5a9357473f 100644 --- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestExportInstantsProcedure.scala +++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestExportInstantsProcedure.scala @@ -17,34 +17,116 @@ package org.apache.spark.sql.hudi.procedure +import org.apache.spark.sql.Row + +import java.io.File + class TestExportInstantsProcedure extends HoodieSparkProcedureTestBase { + private def createCowTable(tableName: String, path: String): Unit = { + spark.sql( + s""" + |create table $tableName ( + | id int, + | name string, + | price double, + | ts long + |) using hudi + | location '$path' + | tblproperties ( + | primaryKey = 'id', + | orderingFields = 'ts' + | ) + """.stripMargin) + } + + private def newExportDir(tmp: File, name: String): File = { + val dir = new File(tmp, name) + assert(dir.mkdirs(), s"Failed to create export dir $dir") + dir + } + + private def exportedCount(result: Array[Row]): Int = { + assertResult(1)(result.length) + val detail = result.head.getString(0) + val matched = "Exported (\\d+) Instants".r.findFirstMatchIn(detail) + assert(matched.isDefined, s"Unexpected export detail: $detail") + matched.get.group(1).toInt + } + test("Test Call export_instants Procedure") { withTempDir { tmp => val tableName = generateTableName - // create table - spark.sql( - s""" - |create table $tableName ( - | id int, - | name string, - | price double, - | ts long - |) using hudi - | location '${tmp.getCanonicalPath}/$tableName' - | tblproperties ( - | primaryKey = 'id', - | orderingFields = 'ts' - | ) - """.stripMargin) + createCowTable(tableName, s"${tmp.getCanonicalPath}/$tableName") + + spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000") + + val exportDir = newExportDir(tmp, "export_basic") + val result = spark.sql( + s"""call export_instants(table => '$tableName', local_folder => '${exportDir.getCanonicalPath}')""").collect() + + // A single insert produces one exportable commit instant that is written to disk. + assertResult(1)(exportedCount(result)) + assertResult(1)(exportDir.listFiles().count(_.getName.endsWith(".commit"))) + } + } + + test("Test Call export_instants Procedure with desc ordering") { + withTempDir { tmp => + val tableName = generateTableName + createCowTable(tableName, s"${tmp.getCanonicalPath}/$tableName") + + spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000") + spark.sql(s"insert into $tableName select 2, 'a2', 20, 2000") + spark.sql(s"insert into $tableName select 3, 'a3', 30, 3000") + + val exportDir = newExportDir(tmp, "export_desc") + val result = spark.sql( + s"""call export_instants(table => '$tableName', + | local_folder => '${exportDir.getCanonicalPath}', desc => true)""".stripMargin).collect() + + // The desc branch reverses the active instants and exports all three commits to disk. + assertResult(3)(exportedCount(result)) + assertResult(3)(exportDir.listFiles().count(_.getName.endsWith(".commit"))) + } + } + + test("Test Call export_instants Procedure filters by action") { + withTempDir { tmp => + val tableName = generateTableName + createCowTable(tableName, s"${tmp.getCanonicalPath}/$tableName") + + spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000") + spark.sql(s"insert into $tableName select 2, 'a2', 20, 2000") - // insert data to table + // Restricting to an action that is not present exports nothing. + val cleanDir = newExportDir(tmp, "export_clean_only") + val cleanResult = spark.sql( + s"""call export_instants(table => '$tableName', + | local_folder => '${cleanDir.getCanonicalPath}', actions => 'clean')""".stripMargin).collect() + assertResult(0)(exportedCount(cleanResult)) + assertResult(0)(cleanDir.listFiles().count(_.getName.endsWith(".clean"))) + + // Restricting to the commit action exports exactly the commit instants. + val commitDir = newExportDir(tmp, "export_commit_only") + val commitResult = spark.sql( + s"""call export_instants(table => '$tableName', + | local_folder => '${commitDir.getCanonicalPath}', actions => 'commit')""".stripMargin).collect() + assertResult(2)(exportedCount(commitResult)) + assertResult(2)(commitDir.listFiles().count(_.getName.endsWith(".commit"))) + } + } + + test("Test Call export_instants Procedure with an invalid local folder") { + withTempDir { tmp => + val tableName = generateTableName + createCowTable(tableName, s"${tmp.getCanonicalPath}/$tableName") spark.sql(s"insert into $tableName select 1, 'a1', 10, 1000") - val result = spark.sql(s"""call export_instants(table => '$tableName', local_folder => '${tmp.getCanonicalPath}/$tableName')""").limit(1).collect() - assertResult(1) { - result.length - } + val notADir = new File(tmp, "does_not_exist").getCanonicalPath + checkExceptionContain( + s"""call export_instants(table => '$tableName', local_folder => '$notADir')""")( + "is not a valid local directory") } } }
