slfan1989 commented on code in PR #2430:
URL: https://github.com/apache/auron/pull/2430#discussion_r3651786845
##########
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')")
Review Comment:
Could we assert that the changelog tasks actually contain both Parquet and
ORC files before checking fallback? Otherwise, this test may pass because of an
unrelated fallback and not exercise the mixed-format branch it intends to cover.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
