[AURON #2429] Add missing fallback coverage for Iceberg changelog native scan

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

@@ -713,6 +713,92 @@ class AuronIcebergIntegrationSuite
     }
   }
 
+  test("iceberg changelog scan falls back for unsupported changelog operations") {
+    withTable("local.db.t_changelog_unsupported_operation") {
+      withTempView("t_changelog_unsupported_operation_changes") {
+        sql("""
+              |create table local.db.t_changelog_unsupported_operation (id int, v string)
+              |using iceberg
+              |tblproperties (
+              |  'format-version' = '2',
+              |  'write.delete.mode' = 'merge-on-read'
+              |)
+              |""".stripMargin)
+        sql("insert into local.db.t_changelog_unsupported_operation values (0, 'seed')")
+        val startSnapshotId = currentSnapshotId("local.db.t_changelog_unsupported_operation")
+        sql("insert into local.db.t_changelog_unsupported_operation values (1, 'a'), (2, 'b')")
+        sql("delete from local.db.t_changelog_unsupported_operation where id = 1")
+        val endSnapshotId = currentSnapshotId("local.db.t_changelog_unsupported_operation")
+        createChangelogView(
+          "local.db.t_changelog_unsupported_operation",
+          "t_changelog_unsupported_operation_changes",
+          startSnapshotId,
+          endSnapshotId)
+
+        val query =
+          """
+            |select id, v, _change_type, _change_ordinal, _commit_snapshot_id
+            |from t_changelog_unsupported_operation_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(2) != "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"))
+        }
+      }
+    }
+  }
+
+  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
+        }
+        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")