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 6d342bd48..1a27718c2 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 @@ -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")