weimingdiit commented on code in PR #2430:
URL: https://github.com/apache/auron/pull/2430#discussion_r3655410870
##########
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:
Good point. I added a precondition assertion for this test.
The test now inspects the planned changelog tasks before checking fallback
and verifies that all changelog tasks are AddedRowsScanTask, and that their
data file formats are exactly PARQUET and ORC. This makes sure the test is
exercising the intended mixed-format fallback branch, rather than passing due
to an unrelated changelog fallback.
--
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]
