This is an automated email from the ASF dual-hosted git repository.
wombatu-kun 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 8347be48ce31 fix(spark): make export_instants desc ordering work
(#19172)
8347be48ce31 is described below
commit 8347be48ce31ef3aee70964a9d0931e32b857136
Author: Y Ethan Guo <[email protected]>
AuthorDate: Sat Jul 4 17:03:33 2026 -0700
fix(spark): make export_instants desc ordering work (#19172)
---
.../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")
}
}
}
