[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 <slfan1989@apache.org>
Signed-off-by: weimingdiit <weimingdiit@gmail.com>

Author: weimingdiit

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(

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