[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 <slfan1989@apache.org>
Signed-off-by: Shilun Fan <slfan1989@apache.org>

Author: Sigma-Ma

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)
   }

thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala

@@ -24,15 +24,16 @@ 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
 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") {