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 4c2abaaeb [AURON #2434] Support full-data-file Iceberg changelog
deletes (#2435)
4c2abaaeb is described below
commit 4c2abaaebf841ed3399401b8a48179392985bdbe
Author: Ming Wei <[email protected]>
AuthorDate: Tue Aug 4 20:25:59 2026 +0800
[AURON #2434] Support full-data-file Iceberg changelog deletes (#2435)
**Which issue does this PR close?**
Closes #2434
**Rationale for this change**
Auron native Iceberg changelog scan currently supports insert-only
changelog tasks.
Full-data-file delete changelog tasks can be executed natively without
row-level delete-file handling. Auron can scan the deleted data file
directly and materialize changelog metadata from the Iceberg changelog
task.
Supporting this case improves native Iceberg changelog scan coverage
while keeping more complex delete and update semantics on Spark's
reader.
**What changes are included in this PR?**
Allows native Iceberg changelog scan to accept `DeletedDataFileScanTask`
when:
- the changelog operation is `DELETE`
- `existingDeletes()` is empty
Keeps existing native support for `AddedRowsScanTask` when:
- the changelog operation is `INSERT`
- `deletes()` is empty
Keeps fallback behavior for unsupported changelog tasks, including
row-level deletes, position/equality deletes, update changelog tasks,
mixed file formats, and delete tasks with existing delete files.
Adds coverage for:
- full-data-file delete changelog native scan
- a changelog range that contains both insert tasks and full-data-file
delete tasks
**Are there any user-facing changes?**
No user-facing API changes. More Iceberg changelog scans can now be
executed natively by Auron.
**How was this patch tested?**
UT.
---------
Co-authored-by: Shilun Fan <[email protected]>
Signed-off-by: weimingdiit <[email protected]>
---
.../sql/auron/iceberg/IcebergScanSupport.scala | 60 ++++++++++-------
.../iceberg/AuronIcebergIntegrationSuite.scala | 76 ++++++++++++++++++----
2 files changed, 100 insertions(+), 36 deletions(-)
diff --git
a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
b/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
index e087e1617..9521d5a7f 100644
---
a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
+++
b/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
@@ -22,7 +22,7 @@ import scala.collection.JavaConverters._
import scala.util.control.NonFatal
import org.apache.commons.lang3.reflect.MethodUtils
-import org.apache.iceberg.{AddedRowsScanTask, ChangelogOperation,
ChangelogScanTask, FileFormat, FileScanTask, MetadataColumns, ScanTask}
+import org.apache.iceberg.{AddedRowsScanTask, ChangelogOperation,
ChangelogScanTask, DataFile, DeletedDataFileScanTask, FileFormat, FileScanTask,
MetadataColumns, ScanTask}
import org.apache.iceberg.expressions.{And => IcebergAnd, BoundPredicate,
Expression => IcebergExpression, Not => IcebergNot, Or => IcebergOr,
UnboundPredicate}
import org.apache.iceberg.spark.source.AuronIcebergSourceUtil
import org.apache.spark.internal.Logging
@@ -267,22 +267,14 @@ object IcebergScanSupport extends Logging {
return None
}
- val addedRowsTasks = changelogTasks.collect { case task: AddedRowsScanTask
=> task }
- // First native changelog support is insert-only. Delete and update images
need Iceberg
- // delete-file handling, so keep them on Spark's reader for now.
- if (addedRowsTasks.size != changelogTasks.size) {
+ // Native changelog scan can read data-file rows directly for insert tasks
and
+ // full-data-file delete tasks. Row-level deletes still need Iceberg
delete-file handling.
+ val nativeChangelogTasks =
changelogTasks.flatMap(toNativeChangelogDataFileTask)
+ if (nativeChangelogTasks.size != changelogTasks.size) {
return None
}
- if (!addedRowsTasks.forall(_.operation() == ChangelogOperation.INSERT)) {
- return None
- }
-
- if (!addedRowsTasks.forall(task => deletesEmpty(task.deletes()))) {
- return None
- }
-
- val formats = addedRowsTasks.map(_.file().format()).distinct
+ val formats = nativeChangelogTasks.map(_.file.format()).distinct
if (formats.size > 1) {
return None
}
@@ -298,7 +290,7 @@ object IcebergScanSupport extends Logging {
}
val pruningPredicates =
collectPruningPredicates(scan.asInstanceOf[AnyRef], readSchema)
- val nativeTasks = addedRowsTasks.map(task => toNativeScanTask(task,
partitionSchema))
+ val nativeTasks = nativeChangelogTasks.map(task => toNativeScanTask(task,
partitionSchema))
Some(
IcebergScanPlan(
nativeTasks,
@@ -528,6 +520,12 @@ object IcebergScanSupport extends Logging {
private case class IcebergPartitionView(tasks: Seq[ScanTask])
+ private case class NativeChangelogDataFileTask(
+ file: DataFile,
+ start: Long,
+ length: Long,
+ changelogTask: ChangelogScanTask)
+
private def icebergPartition(partition: InputPartition):
Option[IcebergPartitionView] = {
val className = partition.getClass.getName
// Only accept Iceberg SparkInputPartition to access task groups.
@@ -559,6 +557,21 @@ object IcebergScanSupport extends Logging {
}
}
+ private def toNativeChangelogDataFileTask(
+ task: ChangelogScanTask): Option[NativeChangelogDataFileTask] = {
+ task match {
+ case added: AddedRowsScanTask
+ if added.operation() == ChangelogOperation.INSERT &&
+ deletesEmpty(added.deletes()) =>
+ Some(NativeChangelogDataFileTask(added.file(), added.start(),
added.length(), added))
+ case deleted: DeletedDataFileScanTask if
deletesEmpty(deleted.existingDeletes()) =>
+ Some(
+ NativeChangelogDataFileTask(deleted.file(), deleted.start(),
deleted.length(), deleted))
+ case _ =>
+ None
+ }
+ }
+
private def toNativeScanTask(
task: FileScanTask,
partitionSchema: StructType): IcebergNativeScanTask = {
@@ -572,15 +585,18 @@ object IcebergScanSupport extends Logging {
}
private def toNativeScanTask(
- task: AddedRowsScanTask,
+ task: NativeChangelogDataFileTask,
partitionSchema: StructType): IcebergNativeScanTask = {
- val file = task.file()
IcebergNativeScanTask(
- file.location(),
- task.start(),
- task.length(),
- file.fileSizeInBytes(),
- metadataPartitionValues(file.location(), file.specId(), Some(task),
partitionSchema))
+ task.file.location(),
+ task.start,
+ task.length,
+ task.file.fileSizeInBytes(),
+ metadataPartitionValues(
+ task.file.location(),
+ task.file.specId(),
+ Some(task.changelogTask),
+ partitionSchema))
}
private def metadataPartitionValues(
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 d2953f040..8d3dd1b0d 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
@@ -598,6 +598,48 @@ class AuronIcebergIntegrationSuite
}
}
+ test("iceberg native scan supports full-data-file delete changelog scan") {
+ withTable("local.db.t_changelog_full_file_delete") {
+ withTempView("t_changelog_full_file_delete_changes") {
+ sql("""
+ |create table local.db.t_changelog_full_file_delete (id int, v
string, p int)
+ |using iceberg
+ |partitioned by (p)
+ |tblproperties ('format-version' = '2')
+ |""".stripMargin)
+ sql("""
+ |insert into local.db.t_changelog_full_file_delete
+ |values (1, 'a', 1), (2, 'b', 2)
+ |""".stripMargin)
+ val startSnapshotId =
currentSnapshotId("local.db.t_changelog_full_file_delete")
+ sql("delete from local.db.t_changelog_full_file_delete where p = 1")
+ val endSnapshotId =
currentSnapshotId("local.db.t_changelog_full_file_delete")
+ createChangelogView(
+ "local.db.t_changelog_full_file_delete",
+ "t_changelog_full_file_delete_changes",
+ startSnapshotId,
+ endSnapshotId)
+
+ val query =
+ """
+ |select id, v, p, _change_type
+ |from t_changelog_full_file_delete_changes
+ |order by id
+ |""".stripMargin
+ val expected = Seq(Row(1, "a", 1, "DELETE"))
+ withSQLConf("spark.auron.enable" -> "false") {
+ checkAnswer(sql(query), expected)
+ }
+ withSQLConf("spark.auron.enable" -> "true",
"spark.auron.enable.iceberg.scan" -> "true") {
+ val df = sql(query)
+ checkAnswer(df, expected)
+ val nativeScan = executedNativeIcebergTableScanExec(df)
+ assert(nativeScan.staticPlan.scanTasks.size == 1)
+ }
+ }
+ }
+ }
+
test("iceberg native changelog scan remains correct in dynamic pruning
join") {
withTable("local.db.t_changelog_dpp", "local.db.t_changelog_dpp_dim") {
withTempView("t_changelog_dpp_changes") {
@@ -701,39 +743,45 @@ class AuronIcebergIntegrationSuite
}
}
- test("iceberg changelog scan falls back when delete changes exist") {
- withTable("local.db.t_changelog_delete") {
- withTempView("t_changelog_delete_changes") {
+ test("iceberg native scan supports mixed insert and full-data-file delete
changelog scan") {
+ withTable("local.db.t_changelog_mixed_delete") {
+ withTempView("t_changelog_mixed_delete_changes") {
sql("""
- |create table local.db.t_changelog_delete (id int, v string)
+ |create table local.db.t_changelog_mixed_delete (id int, v
string, p int)
|using iceberg
+ |partitioned by (p)
|tblproperties ('format-version' = '2')
|""".stripMargin)
- sql("insert into local.db.t_changelog_delete values (1, 'a'), (2,
'b')")
- val startSnapshotId = currentSnapshotId("local.db.t_changelog_delete")
- sql("delete from local.db.t_changelog_delete where id = 1")
- val endSnapshotId = currentSnapshotId("local.db.t_changelog_delete")
+ sql("""
+ |insert into local.db.t_changelog_mixed_delete
+ |values (1, 'a', 1), (2, 'b', 2)
+ |""".stripMargin)
+ val startSnapshotId =
currentSnapshotId("local.db.t_changelog_mixed_delete")
+ sql("delete from local.db.t_changelog_mixed_delete where p = 1")
+ sql("insert into local.db.t_changelog_mixed_delete values (3, 'c', 3)")
+ val endSnapshotId =
currentSnapshotId("local.db.t_changelog_mixed_delete")
createChangelogView(
- "local.db.t_changelog_delete",
- "t_changelog_delete_changes",
+ "local.db.t_changelog_mixed_delete",
+ "t_changelog_mixed_delete_changes",
startSnapshotId,
endSnapshotId)
val query =
"""
- |select id, v, _change_type, _change_ordinal, _commit_snapshot_id
- |from t_changelog_delete_changes
+ |select id, v, p, _change_type, _change_ordinal,
_commit_snapshot_id
+ |from t_changelog_mixed_delete_changes
|order by id, _change_type
|""".stripMargin
var expected: Seq[Row] = Nil
withSQLConf("spark.auron.enable" -> "false") {
expected = sql(query).collect().toSeq
}
+ assert(expected.exists(row => row.getString(3) == "DELETE"))
+ assert(expected.exists(row => row.getString(3) == "INSERT"))
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"))
+ executedNativeIcebergTableScanExec(df)
}
}
}
