This is an automated email from the ASF dual-hosted git repository.
slfan1989 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new 44472751 [AURON #2429] Add mixed-format fallback coverage for Iceberg
changelog native scan (#2430)
44472751 is described below
commit 44472751550f3cd03b54683f7cf1cb800eff6922
Author: Ming Wei <[email protected]>
AuthorDate: Fri Jul 31 08:33:41 2026 +0800
[AURON #2429] Add mixed-format fallback coverage for Iceberg changelog
native scan (#2430)
**Which issue does this PR close?**
Closes #2429
**Rationale for this change**
Auron native Iceberg changelog scan requires all planned changelog tasks
to use a single supported file format.
Without validating the planned tasks, a mixed-format fallback test could
pass because of an unrelated changelog fallback and fail to exercise the
intended file-format guard.
**What changes are included in this PR?**
Adds fallback coverage for an Iceberg changelog range containing both
Parquet and ORC data files.
The test:
- verifies that all planned changelog tasks are `AddedRowsScanTask`
- verifies that the task file formats are exactly Parquet and ORC
- compares the result with Spark execution
- verifies that `NativeIcebergTableScan` is not used
No execution logic is changed.
**Are there any user-facing changes?**
No user-facing changes. This is a test-only change.
**How was this patch tested?**
UT.
---------
Co-authored-by: Shilun Fan <[email protected]>
Signed-off-by: weimingdiit <[email protected]>
---
.../iceberg/AuronIcebergIntegrationSuite.scala | 97 +++++++++++++++++++++-
1 file changed, 96 insertions(+), 1 deletion(-)
diff --git
a/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
b/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
index 0032e289..d2953f04 100644
---
a/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
+++
b/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
@@ -22,8 +22,9 @@ import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
-import org.apache.iceberg.{FileFormat, FileScanTask, MetadataColumns}
+import org.apache.iceberg.{AddedRowsScanTask, ChangelogScanTask, FileFormat,
FileScanTask, MetadataColumns, ScanTask}
import org.apache.iceberg.data.{GenericAppenderFactory, Record}
import org.apache.iceberg.deletes.PositionDelete
import org.apache.iceberg.spark.Spark3Util
@@ -738,6 +739,55 @@ class AuronIcebergIntegrationSuite
}
}
+ test("iceberg changelog scan falls back for mixed file formats") {
+ withTable("local.db.t_changelog_mixed_formats") {
+ withTempView("t_changelog_mixed_formats_changes") {
+ sql("""
+ |create table local.db.t_changelog_mixed_formats (id int, v
string)
+ |using iceberg
+ |tblproperties ('format-version' = '2')
+ |""".stripMargin)
+ sql("insert into local.db.t_changelog_mixed_formats values (0,
'seed')")
+ val startSnapshotId =
currentSnapshotId("local.db.t_changelog_mixed_formats")
+ sql("insert into local.db.t_changelog_mixed_formats values (1,
'parquet')")
+ sql("""
+ |alter table local.db.t_changelog_mixed_formats
+ |set tblproperties ('write.format.default' = 'orc')
+ |""".stripMargin)
+ sql("insert into local.db.t_changelog_mixed_formats values (2, 'orc')")
+ val endSnapshotId =
currentSnapshotId("local.db.t_changelog_mixed_formats")
+ createChangelogView(
+ "local.db.t_changelog_mixed_formats",
+ "t_changelog_mixed_formats_changes",
+ startSnapshotId,
+ endSnapshotId)
+
+ val query =
+ """
+ |select id, v, _change_type, _change_ordinal, _commit_snapshot_id
+ |from t_changelog_mixed_formats_changes
+ |order by id
+ |""".stripMargin
+ var expected: Seq[Row] = Nil
+ withSQLConf("spark.auron.enable" -> "false") {
+ expected = sql(query).collect().toSeq
+ }
+ val changelogTasks = changelogScanTasks(query)
+ assert(changelogTasks.forall(_.isInstanceOf[AddedRowsScanTask]),
changelogTasks)
+ val formats = changelogTasks.collect { case task: AddedRowsScanTask =>
+ task.file().format()
+ }
+ assert(formats.toSet == Set(FileFormat.PARQUET, FileFormat.ORC),
formats)
+ withSQLConf("spark.auron.enable" -> "true",
"spark.auron.enable.iceberg.scan" -> "true") {
+ val df = sql(query)
+ checkAnswer(df, expected)
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativeIcebergTableScan"))
+ }
+ }
+ }
+ }
+
test("iceberg scan falls back when reading unsupported metadata columns") {
withTable("local.db.t4_pos") {
sql("create table local.db.t4_pos using iceberg as select 1 as id, 'a'
as v")
@@ -899,6 +949,51 @@ class AuronIcebergIntegrationSuite
nativeScan
}
+ private def changelogScanTasks(sqlText: String): Seq[ChangelogScanTask] = {
+ val scan = sql(sqlText).queryExecution.sparkPlan
+ .collectFirst {
+ case batchScan: BatchScanExec if isChangelogScan(batchScan.scan) =>
+ batchScan
+ }
+ .getOrElse {
+ throw new AssertionError(s"Cannot find changelog BatchScanExec for
query: $sqlText")
+ }
+
+ val tasks =
+ scan.scan.toBatch.planInputPartitions().toSeq.flatMap { partition =>
+ icebergScanTasks(partition)
+ }
+ assert(tasks.forall(_.isInstanceOf[ChangelogScanTask]), tasks)
+ tasks.collect { case task: ChangelogScanTask => task }
+ }
+
+ private def isChangelogScan(scan: org.apache.spark.sql.connector.read.Scan):
Boolean =
+ scan.getClass.getName ==
"org.apache.iceberg.spark.source.SparkChangelogScan"
+
+ private def icebergScanTasks(partition: AnyRef): Seq[ScanTask] = {
+ try {
+ val taskGroupField = partition.getClass.getDeclaredField("taskGroup")
+ taskGroupField.setAccessible(true)
+ val taskGroup = taskGroupField.get(partition)
+
+ val tasksMethod = taskGroup.getClass.getDeclaredMethod("tasks")
+ tasksMethod.setAccessible(true)
+ tasksMethod
+ .invoke(taskGroup)
+ .asInstanceOf[java.util.Collection[_]]
+ .asScala
+ .collect { case task: ScanTask =>
+ task
+ }
+ .toSeq
+ } catch {
+ case NonFatal(t) =>
+ throw new AssertionError(
+ s"Cannot read Iceberg scan tasks from ${partition.getClass.getName}",
+ t)
+ }
+ }
+
private def icebergScanPlan(df: DataFrame) =
df.queryExecution.sparkPlan.collectFirst { case scan: BatchScanExec =>
IcebergScanSupport.plan(scan)
