You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by sl...@apache.org on 2026/09/15 01:51:19 UTC
(auron) branch master updated: [AURON #2511] Prune full-data-file Iceberg changelog delete tasks (#2523)
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 bcd086bf0 [AURON #2511] Prune full-data-file Iceberg changelog delete tasks (#2523)
bcd086bf0 is described below
commit bcd086bf086c24048edd251f22fc6f7bfefed29e
Author: Ma Zhengxuan <49...@users.noreply.github.com>
AuthorDate: Tue Sep 15 09:51:14 2026 +0800
[AURON #2511] Prune full-data-file Iceberg changelog delete tasks (#2523)
# Which issue does this PR close?
Closes #2511
# Rationale for this change
Native Iceberg changelog scans retain full-data-file DELETE tasks even
when supported metadata predicates exclude them. Task pruning currently
evaluates these predicates only for INSERT tasks.
# What changes are included in this PR?
- Apply the existing metadata predicate evaluator to all supported
native changelog data-file tasks, including DELETE tasks with no
existing delete files.
- Add coverage for mixed INSERT/DELETE ranges, snapshot and ordinal
predicates, and empty scans.
- Verify explicit expected rows against Spark execution and assert
retained scan task counts and file metrics.
- Preserve task eligibility checks, post-scan filters, and operator
boundaries. The existing changelog-view test continues to retain its
task because its filter cannot safely reach the scan.
# Are there any user-facing changes?
Eligible native Iceberg changelog scans can read fewer DELETE tasks. No
public API or configuration changes.
# How was this patch tested?
Passed locally on Spark 3.5.8 / Iceberg 1.10.1 / JDK 17:
- Module compilation and Spotless check.
- Three focused tests covering full-data-file DELETE scans, metadata
pruning, and mixed INSERT/DELETE scans.
- `./dev/reformat --check`.
- Full `AuronIcebergIntegrationSuite` on Spark 3.4, 3.5, and 4.0.
# Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
Generated-by: OpenAI Codex (GPT-6)
Co-authored-by: Shilun Fan <sl...@apache.org>
Signed-off-by: Shilun Fan <sl...@apache.org>
---
.../sql/auron/iceberg/IcebergScanSupport.scala | 17 ++--
.../iceberg/AuronIcebergIntegrationSuite.scala | 108 ++++++++++++++++++++-
2 files changed, 112 insertions(+), 13 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 564dbb202..eca7440b9 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
@@ -618,17 +618,12 @@ object IcebergScanSupport extends Logging {
changelogTaskPredicate(condition, partitionSchema)
.map(predicate =>
tasks.filter { task =>
- task.changelogTask match {
- case _: AddedRowsScanTask =>
- val values = metadataPartitionValues(
- task.file.location(),
- task.file.specId(),
- Some(task.changelogTask),
- partitionSchema)
- predicate(values)
- case _ =>
- true
- }
+ val values = metadataPartitionValues(
+ task.file.location(),
+ task.file.specId(),
+ Some(task.changelogTask),
+ partitionSchema)
+ predicate(values)
})
.getOrElse(tasks)
}
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 d166cfb66..61870ed78 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
@@ -24,7 +24,7 @@ import java.util.concurrent.TimeUnit
import scala.collection.JavaConverters._
import scala.util.control.NonFatal
-import org.apache.iceberg.{AddedRowsScanTask, ChangelogScanTask, FileFormat, FileScanTask, MetadataColumns, ScanTask}
+import org.apache.iceberg.{AddedRowsScanTask, ChangelogScanTask, DeletedDataFileScanTask, FileFormat, FileScanTask, MetadataColumns, ScanTask}
import org.apache.iceberg.data.{GenericAppenderFactory, Record}
import org.apache.iceberg.deletes.PositionDelete
import org.apache.iceberg.spark.Spark3Util
@@ -32,7 +32,8 @@ import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.auron.AuronColumnarOverrides
import org.apache.spark.sql.auron.iceberg.{IcebergConvertProvider, IcebergScanSupport}
-import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualTo, In, Literal, Not, Or}
+import org.apache.spark.sql.catalyst.CatalystTypeConverters
+import org.apache.spark.sql.catalyst.expressions.{And, Attribute, EqualTo, Expression, In, Literal, Not, Or}
import org.apache.spark.sql.catalyst.trees.TreeNodeTag
import org.apache.spark.sql.execution.{FilterExec, FormattedMode, SparkPlan}
import org.apache.spark.sql.execution.ExplainUtils.collectFirst
@@ -818,12 +819,115 @@ class AuronIcebergIntegrationSuite
withSQLConf("spark.auron.enable" -> "true", "spark.auron.enable.iceberg.scan" -> "true") {
val df = sql(metadataFilterQuery)
checkAnswer(df, Seq.empty)
+ // The changelog view's row processing prevents pushing this filter down to the scan.
assert(executedNativeIcebergTableScanExec(df).metrics("numFiles").value == 1L)
}
}
}
}
+ test(
+ "iceberg native changelog scan prunes full-data-file delete tasks by metadata predicates") {
+ val tableName = "local.db.t_changelog_delete_pruning"
+ withTable(tableName) {
+ sql(s"""
+ |create table $tableName (id int, p int)
+ |using iceberg
+ |partitioned by (p)
+ |tblproperties ('format-version' = '2')
+ |""".stripMargin)
+ sql(s"insert into $tableName values (1, 1), (2, 2)")
+ val startSnapshotId = currentSnapshotId(tableName)
+ sql(s"delete from $tableName where p = 1")
+ val firstDeleteSnapshotId = currentSnapshotId(tableName)
+ sql(s"insert into $tableName values (3, 3)")
+ val insertSnapshotId = currentSnapshotId(tableName)
+ sql(s"delete from $tableName where p = 2")
+ val endSnapshotId = currentSnapshotId(tableName)
+
+ val scan = rawChangelogScan(tableName, startSnapshotId, endSnapshotId)
+ val tasks = scan.scan.toBatch.planInputPartitions().toSeq.flatMap(icebergScanTasks)
+ assert(tasks.size == 3)
+ assert(tasks.count(_.isInstanceOf[AddedRowsScanTask]) == 1)
+ val deleteTasks = tasks.collect { case task: DeletedDataFileScanTask => task }
+ assert(deleteTasks.size == 2)
+ assert(deleteTasks.forall(_.existingDeletes().isEmpty))
+
+ val firstDelete = Row(1, 1, "DELETE", 0, firstDeleteSnapshotId)
+ val insert = Row(3, 3, "INSERT", 1, insertSnapshotId)
+ val secondDelete = Row(2, 2, "DELETE", 2, endSnapshotId)
+ val cases: Seq[(Map[String, Attribute] => Expression, Seq[Row])] = Seq(
+ (
+ attributes => EqualTo(attributes("_change_type"), Literal("DELETE")),
+ Seq(firstDelete, secondDelete)),
+ (attributes => EqualTo(attributes("_change_type"), Literal("INSERT")), Seq(insert)),
+ (
+ attributes => EqualTo(attributes("_commit_snapshot_id"), Literal(endSnapshotId)),
+ Seq(secondDelete)),
+ (
+ attributes =>
+ In(
+ attributes("_commit_snapshot_id"),
+ Seq(Literal(firstDeleteSnapshotId), Literal(insertSnapshotId))),
+ Seq(firstDelete, insert)),
+ (attributes => EqualTo(attributes("_change_ordinal"), Literal(0)), Seq(firstDelete)),
+ (
+ attributes => In(attributes("_change_ordinal"), Seq(Literal(1), Literal(2))),
+ Seq(insert, secondDelete)),
+ (
+ attributes =>
+ And(
+ EqualTo(attributes("_change_type"), Literal("DELETE")),
+ EqualTo(attributes("_commit_snapshot_id"), Literal(firstDeleteSnapshotId))),
+ Seq(firstDelete)),
+ (
+ attributes => EqualTo(attributes("_commit_snapshot_id"), Literal(startSnapshotId)),
+ Seq.empty),
+ (
+ attributes =>
+ And(
+ EqualTo(attributes("_change_type"), Literal("INSERT")),
+ EqualTo(attributes("_change_ordinal"), Literal(0))),
+ Seq.empty))
+
+ cases.foreach { case (condition, expected) =>
+ def filteredScan: FilterExec = {
+ val scan = rawChangelogScan(tableName, startSnapshotId, endSnapshotId)
+ FilterExec(
+ condition(scan.output.map(attribute => attribute.name -> attribute).toMap),
+ scan)
+ }
+
+ def checkRows(plan: SparkPlan): Unit = {
+ val schema = plan.schema
+ val rows = plan
+ .execute()
+ .mapPartitions { rows =>
+ val toScala = CatalystTypeConverters.createToScalaConverter(schema)
+ rows.map(row => toScala(row).asInstanceOf[Row])
+ }
+ .collect()
+ .toSeq
+ assert(rows.sortBy(_.getInt(0)) == expected.sortBy(_.getInt(0)), plan.treeString)
+ }
+
+ withSQLConf("spark.auron.enable" -> "false") {
+ checkRows(filteredScan)
+ }
+ withSQLConf("spark.auron.enable" -> "true", "spark.auron.enable.iceberg.scan" -> "true") {
+ val transformed =
+ AuronColumnarOverrides(spark).preColumnarTransitions.apply(filteredScan)
+ val nativeScan = transformed.collectFirst { case native: NativeIcebergTableScanExec =>
+ native
+ }.get
+ assert(nativeScan.staticPlan.scanTasks.size == expected.size)
+ checkRows(transformed)
+ assert(nativeScan.metrics("numFiles").value == expected.size)
+ }
+ }
+ }
+ }
+
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") {