github-actions[bot] commented on code in PR #63719:
URL: https://github.com/apache/doris/pull/63719#discussion_r3613062617
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_v2_to_v3_doris_spark_compare.groovy:
##########
@@ -31,7 +30,100 @@ suite("test_iceberg_v2_to_v3_doris_spark_compare",
"p0,external,iceberg,external
def formats = ["parquet", "orc"]
def tableNameForFormat = { baseName, format ->
- return format == "parquet" ? baseName : "${baseName}_orc"
+ return "${baseName}_${format}"
+ }
+
+ def createSparkUpgradeFixture = { tableName, format, partitioned,
datePrefix, deleteBeforeUpgrade ->
+ def partitionClause = partitioned ? "partitioned by (days(dt))" : ""
+ def deleteClause = deleteBeforeUpgrade ? """
+ delete from ${tableName}
+ where id = 2;
+ """ : ""
+ spark_iceberg_multi """
+ use demo.${dbName};
+
+ drop table if exists ${tableName};
+
+ create table ${tableName} (
+ id int,
+ tag string,
+ score int,
+ dt date
+ ) using iceberg
+ ${partitionClause}
+ tblproperties (
+ 'format-version' = '2',
+ 'write.format.default' = '${format}',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.update.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+ );
+
+ insert into ${tableName} values
+ (1, 'base', 100, date '${datePrefix}-01'),
+ (2, 'base', 200, date '${datePrefix}-02');
+
+ insert into ${tableName} values
+ (3, 'base', 300, date '${datePrefix}-03');
+
+ update ${tableName}
+ set tag = 'base_u', score = score + 10
+ where id = 1;
+
+ ${deleteClause}
+ alter table ${tableName}
+ set tblproperties ('format-version' = '3');
+ """
+ }
+
+ def createSparkReferenceFixture = { tableName, format ->
+ createSparkUpgradeFixture(tableName, format, true, "2024-02", false)
+ spark_iceberg_multi """
+ use demo.${dbName};
+
+ update ${tableName}
+ set tag = 'post_v3_u', score = score + 20
+ where id = 2;
+
+ insert into ${tableName} values
+ (4, 'post_v3_i', 400, date '2024-02-04');
+
+ call demo.system.rewrite_data_files(
+ table => 'demo.${dbName}.${tableName}',
+ options => map('target-file-size-bytes', '10485760',
'min-input-files', '1')
+ );
+ """
+ }
+
+ spark_iceberg """create database if not exists demo.${dbName}"""
Review Comment:
The Docker entrypoint still sources the unchanged 391-line
`iceberg/run29.sql`, which pre-creates 14 v2-to-v3 fixtures. This new loop then
builds 26 fixtures again: the `_orc` names drop and recreate the preinstalled
ORC tables, while the new `_parquet` names leave the preinstalled unsuffixed
Parquet tables unused. Please use one fixture strategy—remove the obsolete
`run29.sql` bootstrap when creating these dynamically, or reuse/extend it—so
every external P0 run does not pay for both workloads.
##########
regression-test/suites/external_table_p2/iceberg/test_iceberg_v3_row_lineage_random_cross_engine.groovy:
##########
@@ -0,0 +1,270 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import java.util.Random
+
+suite("test_iceberg_v3_row_lineage_random_cross_engine",
"p2,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+ String catalogName = "test_iceberg_v3_row_lineage_random_cross_engine"
+ String dbName = "test_row_lineage_random_cross_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String endpoint = "http://${externalEnvIp}:${minioPort}";
+
+ int rounds =
(context.config.otherConfigs.get("icebergRowLineageRandomRounds") ?:
"20").toInteger()
+ long seed =
(context.config.otherConfigs.get("icebergRowLineageRandomSeed") ?:
"61398").toLong()
+ def formats = ["parquet", "orc"]
+
+ def hasSparkIcebergJdbc = {
+ try {
+ spark_iceberg """select 1"""
+ return true
+ } catch (Exception e) {
+ logger.info("Check spark-iceberg JDBC failed: ${e.message}")
+ return false
+ }
+ }
+
+ def dorisChecksum = { tableName ->
+ def rows = sql("""
+ select count(*), coalesce(sum(id), 0), coalesce(sum(score), 0)
+ from ${tableName}
+ """)
+ return rows[0].collect { it.toString() }.join(",")
+ }
+
+ def sparkChecksum = { tableName ->
+ def rows = spark_iceberg("""
+ select concat_ws(',', cast(count(*) as string),
+ cast(coalesce(sum(id), 0) as string),
+ cast(coalesce(sum(score), 0) as string))
+ from demo.${dbName}.${tableName}
+ """)
Review Comment:
This helper is named as a business-row equality check, but the oracle only
compares `count`, `sum(id)`, and `sum(score)`. It cannot detect any divergence
in `name` or `dt` (both are changed by the randomized operations), and distinct
row sets can also collide on these sums. Please compare the complete ordered
`(id, name, score, dt)` results, or use a collision-resistant digest over every
column, so the randomized interoperability test can catch the row-level bugs it
is meant to cover.
##########
regression-test/suites/external_table_p2/iceberg/test_iceberg_v3_row_lineage_large_stability.groovy:
##########
@@ -0,0 +1,169 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_v3_row_lineage_large_stability",
"p2,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+
+ String catalogName = "test_iceberg_v3_row_lineage_large_stability"
+ String dbName = "test_row_lineage_large_stability_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String endpoint = "http://${externalEnvIp}:${minioPort}";
+
+ int rowCount =
(context.config.otherConfigs.get("icebergRowLineageLargeRows") ?:
"500000").toInteger()
+ def formats = ["parquet", "orc"]
+ def dataColumnNames = (1..498).collect { idx -> String.format("c%03d",
idx) }
+
+ def assertChecksum = { tableName, expectedRows ->
+ sql """refresh table ${dbName}.${tableName}"""
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ def sparkRows = spark_iceberg("""
+ select count(*), sum(id), sum(c001)
Review Comment:
This creates 498 payload columns, but every validation query projects only
`c001`; `c002` through `c498` are column-pruned in both engines. A wide-schema
mapping/rewrite bug that corrupts any later column therefore leaves the suite
green, so the advertised 500-column stability path is not actually validated.
Please include a deterministic full-row digest over all payload columns, or at
least validate representative early/middle/last columns (including `c498`)
before and after rewrite in both engines.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_doris_create_update_spark_read.groovy:
##########
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_v3_doris_create_update_spark_read",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+ String catalogName = "test_iceberg_v3_doris_create_update_spark_read"
+ String dbName = "test_v3_doris_create_update_spark_read_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+ def formats = ["parquet", "orc"]
+ def partitionFlags = [true, false]
+
+ def checksum = { rows ->
+ assertEquals(1, rows.size())
+ return rows[0].collect { col -> col == null ? null : col.toString() }
+ }
+
+ def assertDorisSparkChecksumEqual = { tableName ->
+ sql """refresh table ${dbName}.${tableName}"""
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ def sparkChecksum = checksum(spark_iceberg("""
+ select count(*), sum(id), sum(score),
+ max(case when id = 2 then tag end)
+ from demo.${dbName}.${tableName}
+ """))
+ def dorisChecksum = checksum(sql """
+ select count(*), sum(id), sum(score),
+ max(case when id = 2 then tag end)
+ from ${tableName}
+ """)
+ log.info("Doris checksum for ${tableName}: ${dorisChecksum}")
+ log.info("Spark checksum for ${tableName}: ${sparkChecksum}")
+ assertEquals(dorisChecksum, sparkChecksum)
+ }
+
+ def assertDorisSparkBusinessRowsEqual = { tableName ->
+ sql """refresh table ${dbName}.${tableName}"""
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ def sparkRows = spark_iceberg("""
+ select id, tag, score, dt
+ from demo.${dbName}.${tableName}
+ order by id
+ """)
+ def dorisRows = sql("""
+ select id, tag, score, dt
+ from ${tableName}
+ order by id
+ """)
+ log.info("Spark business rows for ${tableName}: ${sparkRows}")
+ log.info("Doris business rows for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ def assertDorisLineageReadable = { tableName ->
+ def rows = sql """
+ select id, _row_id, _last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """
+ log.info("Doris lineage rows for ${tableName}: ${rows}")
+ assertEquals(3, rows.size())
+ rows.each { row ->
+ assertTrue(row[1] != null, "_row_id should be non-null for
${tableName}, row=${row}")
+ assertTrue(row[2] != null,
+ "_last_updated_sequence_number should be non-null for
${tableName}, row=${row}")
+ }
+ }
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog if not exists ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1"
+ )
+ """
+
+ sql """switch ${catalogName}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+ sql """set enable_fallback_to_original_planner = false"""
+
+ boolean sparkReady = false
+ try {
+ spark_iceberg """select 1"""
+ sparkReady = true
+ formats.each { format ->
+ partitionFlags.each { partitioned ->
+ String partitionSuffix = partitioned ? "part" : "unpart"
+ String tableName =
"doris_v3_update_spark_read_${format}_${partitionSuffix}"
+ String partitionClause = partitioned ? "partition by list
(day(dt)) ()" : ""
+
+ sql """drop table if exists ${tableName}"""
+ sql """
+ create table ${tableName} (
+ id int,
+ tag string,
+ score int,
+ dt date
+ ) engine=iceberg
+ ${partitionClause}
+ properties (
+ "format-version" = "3",
+ "write.format.default" = "${format}",
+ "write.delete.mode" = "merge-on-read",
+ "write.update.mode" = "merge-on-read",
+ "write.merge.mode" = "merge-on-read"
+ )
+ """
+ sql """
+ insert into ${tableName} values
+ (1, 'doris_v3_i', 100, date '2024-03-01'),
+ (2, 'doris_v3_i', 200, date '2024-03-02'),
+ (3, 'doris_v3_i', 300, date '2024-03-03')
+ """
+ sql """
+ update ${tableName}
+ set tag = 'doris_v3_u', score = score + 20
+ where id = 2
+ """
+
+ def rows = sql """
+ select id, tag, score
+ from ${tableName}
+ order by id
+ """
+ assertEquals(3, rows.size())
+ assertEquals("doris_v3_u", rows[1][1].toString())
+ assertEquals(220, rows[1][2].toString().toInteger())
+ assertDorisSparkBusinessRowsEqual(tableName)
+ assertDorisLineageReadable(tableName)
+ assertDorisSparkChecksumEqual(tableName)
+ }
+ }
+ } catch (Exception e) {
+ if (!sparkReady) {
+ logger.info("spark-iceberg JDBC is unavailable, skip Doris v3
create/update Spark read test: ${e.message}")
Review Comment:
With `enableIcebergTest=true`, Spark is a required side of this
interoperability suite. If the initial JDBC probe fails, this catch logs and
returns successfully, so all four format/partition scenarios are skipped and a
broken Spark endpoint makes the suite green without testing anything. The new
P2 random-cross-engine suite repeats the same false-success pattern in
`hasSparkIcebergJdbc()`. Please let both probe failures fail their suites; if
Spark is genuinely optional, gate that at the harness/configuration level
instead of swallowing a runtime dependency failure.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_query_insert.groovy:
##########
@@ -138,6 +166,111 @@ suite("test_iceberg_v3_row_lineage_query_insert",
"p0,external,iceberg,external_
assertEquals(2, combinedPredicate.size())
assertEquals(expectedIds[1],
combinedPredicate[0][0].toString().toInteger())
assertEquals(expectedIds[2],
combinedPredicate[1][0].toString().toInteger())
+
+ def aggregateRows = sql("""
+ select count(*), min(_row_id), max(_last_updated_sequence_number)
+ from ${tableName}
+ where _row_id is not null
+ """)
+ log.info("Checking row lineage aggregate for ${tableName}:
result=${aggregateRows}")
+ assertEquals(expectedIds.size(),
aggregateRows[0][0].toString().toInteger())
+ assertTrue(aggregateRows[0][1] != null, "min(_row_id) should be
non-null for ${tableName}")
+ assertTrue(aggregateRows[0][2] != null,
+ "max(_last_updated_sequence_number) should be non-null for
${tableName}")
+ }
+
+ def assertV2DoesNotExposeRowLineageColumns = { tableName ->
+ sql("""set show_hidden_columns = true""")
+ def descRows = sql("""desc ${tableName}""")
+ def columns = collectDescColumns(descRows)
+ log.info("Checking v2 row lineage compatibility for ${tableName}:
desc=${descRows}")
+ assertTrue(!columns.contains("_row_id"),
+ "Iceberg v2 table should not expose _row_id for ${tableName},
got ${columns}")
+ assertTrue(!columns.contains("_last_updated_sequence_number"),
+ "Iceberg v2 table should not expose
_last_updated_sequence_number for ${tableName}, got ${columns}")
+
+ test {
+ sql """select id, _row_id from ${tableName}"""
+ exception "_row_id"
+ }
+
+ test {
+ sql """select id, _last_updated_sequence_number from
${tableName}"""
+ exception "_last_updated_sequence_number"
+ }
+
+ sql("""set show_hidden_columns = false""")
+ }
+
+ def assertRowLineagePredicatesAreNotPushedDown = { tableName, rowId,
sequenceNumber ->
+ def explainRowId = sql("""
+ explain verbose
+ select id
+ from ${tableName}
+ where id = 1 and _row_id = ${rowId}
+ """)
+ def explainSeq = sql("""
+ explain verbose
+ select id
+ from ${tableName}
+ where _last_updated_sequence_number >= ${sequenceNumber}
+ """)
+
+ [explainRowId, explainSeq].each { explainRows ->
+ String pushdownText = explainRows.collect { row ->
row.toString().toLowerCase() }
+ .findAll { line ->
line.contains("icebergpredicatepushdown") }
Review Comment:
The EXPLAIN result is returned one line per row: `IcebergScanNode` prints
the `icebergPredicatePushdown=` header first and the actual predicates on
following lines, and `StmtExecutor` splits on `\n`. Filtering to rows that
contain only the header therefore removes the rows where `_row_id` or
`_last_updated_sequence_number` would appear, so this test passes even if those
predicates are incorrectly pushed down. Please inspect the complete Iceberg
scan block (and assert the block is present for the mixed `id` predicate case)
before checking that the lineage columns are absent.
##########
regression-test/suites/external_table_p2/iceberg/test_iceberg_v3_row_lineage_large_stability.groovy:
##########
@@ -0,0 +1,169 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_v3_row_lineage_large_stability",
"p2,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+
+ String catalogName = "test_iceberg_v3_row_lineage_large_stability"
+ String dbName = "test_row_lineage_large_stability_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String endpoint = "http://${externalEnvIp}:${minioPort}";
+
+ int rowCount =
(context.config.otherConfigs.get("icebergRowLineageLargeRows") ?:
"500000").toInteger()
+ def formats = ["parquet", "orc"]
+ def dataColumnNames = (1..498).collect { idx -> String.format("c%03d",
idx) }
+
+ def assertChecksum = { tableName, expectedRows ->
+ sql """refresh table ${dbName}.${tableName}"""
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ def sparkRows = spark_iceberg("""
+ select count(*), sum(id), sum(c001)
+ from demo.${dbName}.${tableName}
+ """)
+ def dorisRows = sql("""
+ select count(*), sum(id), sum(c001)
+ from ${tableName}
+ """)
+ log.info("Spark checksum for ${tableName}: ${sparkRows}")
+ log.info("Doris checksum for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ assertEquals(expectedRows, dorisRows[0][0].toString().toLong())
+ assertTrue(dorisRows[0][1] != null, "sum(id) should be non-null for
${tableName}")
+ assertTrue(dorisRows[0][2] != null, "sum(c001) should be non-null for
${tableName}")
+ return dorisRows[0].collect { it == null ? null : it.toString() }
+ }
+
+ def assertPuffinDeleteFiles = { tableName ->
+ def deleteFiles = sql("""
+ select file_path, lower(file_format), record_count
+ from ${tableName}\$delete_files
+ order by file_path
+ """)
+ log.info("Large stability delete files for ${tableName}:
${deleteFiles}")
+ assertTrue(deleteFiles.size() > 0, "Large stability DML should create
delete files for ${tableName}")
+ deleteFiles.each { row ->
+ assertEquals("puffin", row[1].toString())
+ assertTrue(row[0].toString().toLowerCase().endsWith(".puffin"))
+ }
+ }
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog if not exists ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "${endpoint}",
+ "s3.region" = "us-east-1"
+ )
+ """
+
+ sql """switch ${catalogName}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+ sql """set enable_fallback_to_original_planner = false"""
+ sql """set show_hidden_columns = false"""
+
+ try {
+ formats.each { format ->
+ String tableName = "large_stability_${format}"
+ try {
+ sql """drop table if exists ${tableName}"""
+ String dataColumns = dataColumnNames.collect { col -> "${col}
int" }.join(",\n ")
+ String dataSelectExprs = dataColumnNames.withIndex().collect {
col, idx ->
+ "cast(number % ${1000 + idx} as int) as ${col}"
+ }.join(",\n ")
+ sql """
+ create table ${tableName} (
+ id int,
+ dt date,
+ ${dataColumns}
+ ) engine=iceberg
+ partition by list (day(dt)) ()
+ properties (
+ "format-version" = "3",
+ "write.format.default" = "${format}",
+ "write.delete.mode" = "merge-on-read",
+ "write.update.mode" = "merge-on-read",
+ "write.merge.mode" = "merge-on-read"
+ )
+ """
+
+ sql """
+ insert into ${tableName}
+ select
+ cast(number as int) as id,
+ cast(case when number % 2 = 0 then '2024-05-01' else
'2024-05-02' end as date) as dt,
+ ${dataSelectExprs}
+ from numbers("number" = "${rowCount}")
+ """
+
+ sql """
+ update ${tableName}
+ set c001 = c001 + 1
+ where dt = date '2024-05-01' and id % 10 = 0
+ """
+ sql """
+ delete from ${tableName}
+ where dt = date '2024-05-02' and id % 20 = 1
+ """
+
+ long deletedRows = rowCount <= 1 ? 0L : (((rowCount - 2) /
20).toLong() + 1L)
+ long expectedRows = rowCount - deletedRows
+ def beforeRewriteChecksum = assertChecksum(tableName,
expectedRows)
+ assertPuffinDeleteFiles(tableName)
+
+ def rewriteResult = sql("""
+ alter table ${catalogName}.${dbName}.${tableName}
+ execute rewrite_data_files(
+ "target-file-size-bytes" = "536870912",
+ "min-input-files" = "1"
+ )
+ """)
+ log.info("Large stability rewrite result for ${tableName}:
${rewriteResult}")
+ assertTrue(rewriteResult.size() > 0,
Review Comment:
A summary row is returned even when `rewrite_data_files` rewrites zero
files, so `rewriteResult.size() > 0` does not exercise the rewrite path (the
established v3 rewrite suite checks `rewriteResult[0][0] > 0`). This case also
records no lineage state before the action; replacing every row ID/sequence
with different non-null values would still pass the post-rewrite counts. Please
require a positive rewritten-file count and compare a deterministic `(id,
_row_id, _last_updated_sequence_number)` digest before and after the actual
rewrite.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_dml_mode_matrix.groovy:
##########
@@ -0,0 +1,279 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_v3_row_lineage_dml_mode_matrix",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+ String catalogName = "test_iceberg_v3_row_lineage_dml_mode_matrix"
+ String dbName = "test_row_lineage_dml_mode_matrix_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String endpoint = "http://${externalEnvIp}:${minioPort}";
+
+ def formats = ["parquet", "orc"]
+ def formatVersions = [2, 3]
+ def partitionFlags = [false, true]
+ def scenarios = [
+ [name: "m01_delete_cow", operation: "delete", property:
"write.delete.mode", mode: "copy-on-write"],
+ [name: "m02_delete_mor", operation: "delete", property:
"write.delete.mode", mode: "merge-on-read"],
+ [name: "m03_update_cow", operation: "update", property:
"write.update.mode", mode: "copy-on-write"],
+ [name: "m04_update_mor", operation: "update", property:
"write.update.mode", mode: "merge-on-read"],
+ [name: "m05_merge_cow", operation: "merge", property:
"write.merge.mode", mode: "copy-on-write"],
+ [name: "m06_merge_mor", operation: "merge", property:
"write.merge.mode", mode: "merge-on-read"]
+ ]
+ def initialRows = [
+ [1, "Alice", 10, "2024-01-01"],
+ [2, "Bob", 20, "2024-01-02"],
+ [3, "Carol", 30, "2024-01-03"]
+ ]
+
+ def createTable = { tableName, formatVersion, format, partitioned,
propertyName, mode ->
+ sql """drop table if exists ${tableName}"""
+ String partitionClause = partitioned ? "partition by list (day(dt))
()" : ""
+ sql """
+ create table ${tableName} (
+ id int,
+ name string,
+ score int,
+ dt date
+ ) engine=iceberg
+ ${partitionClause}
+ properties (
+ "format-version" = "${formatVersion}",
+ "write.format.default" = "${format}",
+ "${propertyName}" = "${mode}"
+ )
+ """
+ sql """
+ insert into ${tableName} values
+ (1, 'Alice', 10, date '2024-01-01'),
+ (2, 'Bob', 20, date '2024-01-02'),
+ (3, 'Carol', 30, date '2024-01-03')
+ """
+ }
+
+ def lineageMap = { tableName ->
+ def rows = sql("""
+ select id, _row_id, _last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ Map<Integer, List<Long>> result = [:]
+ rows.each { row ->
+ result[row[0].toString().toInteger()] =
[row[1].toString().toLong(), row[2].toString().toLong()]
+ }
+ log.info("Lineage map for ${tableName}: ${result}")
+ return result
+ }
+
+ def assertV3LineageNonNull = { tableName, expectedIds ->
+ def rows = sql("""
+ select id, _row_id, _last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ log.info("Checking v3 row lineage rows for ${tableName}: ${rows}")
+ assertEquals(expectedIds.size(), rows.size())
+ for (int i = 0; i < expectedIds.size(); i++) {
+ assertEquals(expectedIds[i], rows[i][0].toString().toInteger())
+ assertTrue(rows[i][1] != null, "_row_id should be non-null for
${tableName}, row=${rows[i]}")
+ assertTrue(rows[i][2] != null,
+ "_last_updated_sequence_number should be non-null for
${tableName}, row=${rows[i]}")
+ }
+ }
+
+ def assertDeleteFilesForVersion = { tableName, formatVersion, format,
scenario ->
+ def deleteFiles = sql("""
+ select file_path, lower(file_format), record_count
+ from ${tableName}\$delete_files
+ order by file_path
+ """)
+ log.info("Checking delete files for ${tableName},
scenario=${scenario.name}: ${deleteFiles}")
+ if (scenario.mode == "copy-on-write") {
+ assertEquals(0, deleteFiles.size(),
+ "COW scenario ${scenario.name} should not create delete files
for ${tableName}, actual=${deleteFiles}")
+ }
+ if (scenario.mode == "merge-on-read" && formatVersion == 3) {
Review Comment:
This condition requires a nonempty delete-file set only for v3. For every v2
merge-on-read case, an empty result skips all of the `file_format`,
`delete_pos`, and extension checks below, so all twelve v2 MOR combinations can
pass even if the operation rewrites data files like copy-on-write instead of
producing position deletes. Please require at least one delete file for every
successful MOR scenario, then retain the version-specific format checks.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_equality_delete_dv_interop.groovy:
##########
@@ -0,0 +1,317 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_v3_row_lineage_equality_delete_dv_interop",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+ String catalogName =
"test_iceberg_v3_row_lineage_equality_delete_dv_interop"
+ String dbName = "test_row_lineage_eq_delete_dv_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String endpoint = "http://${externalEnvIp}:${minioPort}";
+ String runSuffix = "${System.currentTimeMillis()}"
+
+ def fixtureMetadataFiles = [
+ parquet:
"s3a://warehouse/wh/multi_catalog/equality_delete_par_1/metadata/00011-f7268cac-dd2f-4061-a2c9-b5aa54ab52f4.metadata.json",
+ orc:
"s3a://warehouse/wh/multi_catalog/equality_delete_orc_1/metadata/00011-568385a2-14af-4237-b186-ce46d350be19.metadata.json"
+ ]
+
+ def normalizeRows = { rows ->
+ return rows.collect { row -> row.collect { col -> col == null ? null :
col.toString() } }
+ }
+
+ def unregisterSparkTable = { tableName ->
+ try {
+ spark_iceberg """
+ call demo.system.unregister_table(table =>
'${dbName}.${tableName}')
+ """
+ } catch (Throwable t) {
+ logger.info("Skip unregister ${dbName}.${tableName}: ${t.message}")
+ }
+ }
+
+ def refreshTable = { tableName ->
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ sql """refresh table ${dbName}.${tableName}"""
+ }
+
+ def assertSparkDorisBusinessRows = { tableName ->
+ refreshTable(tableName)
+ def sparkRows = spark_iceberg("""
+ select new_new_id, new_name, data, id
+ from demo.${dbName}.${tableName}
+ order by new_new_id, new_name, data, id
+ """)
+ def dorisRows = sql("""
+ select new_new_id, new_name, data, id
+ from ${tableName}
+ order by new_new_id, new_name, data, id
+ """)
+ log.info("Spark business rows for ${tableName}: ${sparkRows}")
+ log.info("Doris business rows for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ def assertDorisBusinessRows = { tableName, expectedRows ->
+ refreshTable(tableName)
+ def sparkRows = spark_iceberg("""
+ select new_new_id, new_name, data, id
+ from demo.${dbName}.${tableName}
+ order by new_new_id, new_name, data, id
+ """)
+ def dorisRows = sql("""
+ select new_new_id, new_name, data, id
+ from ${tableName}
+ order by new_new_id, new_name, data, id
+ """)
+ log.info("Spark business rows for ${tableName}: ${sparkRows}")
+ log.info("Doris business rows for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ def normalizedRows = normalizeRows(dorisRows)
+ log.info("Checking concrete Doris business rows for ${tableName}:
${normalizedRows}")
+ assertEquals(expectedRows, normalizedRows)
+ }
+
+ def assertSparkDorisAggregateRows = { tableName ->
+ refreshTable(tableName)
+ def sparkRows = spark_iceberg("""
+ select count(*), count(distinct new_new_id)
+ from demo.${dbName}.${tableName}
+ """)
+ def dorisRows = sql("""
+ select count(*), count(distinct new_new_id)
+ from ${tableName}
+ """)
+ log.info("Spark aggregate rows for ${tableName}: ${sparkRows}")
+ log.info("Doris aggregate rows for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ def assertDorisAggregateRows = { tableName, expectedRows ->
+ refreshTable(tableName)
+ def sparkRows = spark_iceberg("""
+ select count(*), count(distinct new_new_id)
+ from demo.${dbName}.${tableName}
+ """)
+ def dorisRows = sql("""
+ select count(*), count(distinct new_new_id)
+ from ${tableName}
+ """)
+ log.info("Spark aggregate rows for ${tableName}: ${sparkRows}")
+ log.info("Doris aggregate rows for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ def normalizedRows = normalizeRows(dorisRows)
+ log.info("Checking concrete Doris aggregate rows for ${tableName}:
${normalizedRows}")
+ assertEquals(expectedRows, normalizedRows)
+ }
+
+ def assertSparkDorisRowsByPredicate = { tableName, predicate ->
+ refreshTable(tableName)
+ def sparkRows = spark_iceberg("""
+ select new_new_id, new_name, data, id
+ from demo.${dbName}.${tableName}
+ where ${predicate}
+ order by new_new_id, new_name, data, id
+ """)
+ def dorisRows = sql("""
+ select new_new_id, new_name, data, id
+ from ${tableName}
+ where ${predicate}
+ order by new_new_id, new_name, data, id
+ """)
+ log.info("Spark rows for ${tableName} with predicate ${predicate}:
${sparkRows}")
+ log.info("Doris rows for ${tableName} with predicate ${predicate}:
${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ def assertDorisRowsByPredicate = { tableName, predicate, expectedRows ->
+ refreshTable(tableName)
+ def sparkRows = spark_iceberg("""
+ select new_new_id, new_name, data, id
+ from demo.${dbName}.${tableName}
+ where ${predicate}
+ order by new_new_id, new_name, data, id
+ """)
+ def dorisRows = sql("""
+ select new_new_id, new_name, data, id
+ from ${tableName}
+ where ${predicate}
+ order by new_new_id, new_name, data, id
+ """)
+ log.info("Spark rows for ${tableName} with predicate ${predicate}:
${sparkRows}")
+ log.info("Doris rows for ${tableName} with predicate ${predicate}:
${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ def normalizedRows = normalizeRows(dorisRows)
+ log.info("Checking concrete Doris rows for ${tableName} with predicate
${predicate}: ${normalizedRows}")
+ assertEquals(expectedRows, normalizedRows)
+ }
+
+ def assertLineageProjectionReadable = { tableName, expectNonNullLineage ->
+ def rows = sql("""
+ select new_new_id, _row_id, _last_updated_sequence_number
+ from ${tableName}
+ order by new_new_id
+ """)
+ logger.info("Row lineage projection for ${tableName},
expectNonNullLineage=${expectNonNullLineage}: ${rows}")
+ assertTrue(rows.size() > 0, "Row lineage projection should return
visible rows for ${tableName}")
+ rows.each { row ->
+ if (expectNonNullLineage) {
+ assertTrue(row[1] != null,
+ "_row_id should be non-null after v2-to-v3 write in
${tableName}, row=${row}")
+ assertTrue(row[2] != null,
+ "_last_updated_sequence_number should be non-null
after v2-to-v3 write in ${tableName}, row=${row}")
+ } else {
+ assertEquals(null, row[1],
+ "_row_id should be NULL before v2-to-v3 write in
${tableName}, row=${row}")
+ assertEquals(null, row[2],
+ "_last_updated_sequence_number should be NULL before
v2-to-v3 write in ${tableName}, row=${row}")
+ }
+ }
+ }
+
+ def deleteFileFormats = { tableName ->
+ refreshTable(tableName)
+ def sparkRows = spark_iceberg("""
+ select lower(file_format), count(*)
+ from demo.${dbName}.${tableName}.delete_files
+ group by lower(file_format)
+ order by lower(file_format)
+ """)
+ def dorisRows = sql("""
+ select lower(file_format), count(*)
+ from ${tableName}\$delete_files
+ group by lower(file_format)
+ order by lower(file_format)
+ """)
+ log.info("Spark delete file formats for ${tableName}: ${sparkRows}")
+ log.info("Doris delete file formats for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ return dorisRows
+ }
+
+ def assertOnlyEqualityDeleteFilesBeforeDorisDelete = { tableName ->
+ def deleteFiles = deleteFileFormats(tableName)
+ logger.info("Delete files before Doris v3 DELETE for ${tableName}:
${deleteFiles}")
+ assertTrue(deleteFiles.size() > 0, "Fixture should contain equality
delete files for ${tableName}")
+ assertTrue(!deleteFiles.any { row -> row[0].toString() == "puffin" },
+ "Fixture should not contain Puffin DV before Doris DELETE for
${tableName}: ${deleteFiles}")
+ }
+
+ def assertEqualityDeleteAndPuffinDvCoexist = { tableName ->
+ def deleteFiles = deleteFileFormats(tableName)
+ logger.info("Delete files after Doris v3 DELETE for ${tableName}:
${deleteFiles}")
+ assertTrue(deleteFiles.any { row -> row[0].toString() == "puffin" },
+ "Doris v3 DELETE should write Puffin DV for ${tableName}:
${deleteFiles}")
+ assertTrue(deleteFiles.any { row -> row[0].toString() != "puffin" },
Review Comment:
`file_format` does not identify the delete kind: both equality deletes and
ordinary position deletes can be Parquet/ORC. Therefore “Puffin plus any
non-Puffin” can pass even if the original equality deletes were replaced by, or
confused with, position-delete files. Please compare `content` (and
`equality_ids`/DV metadata where available) in both engines: require the
original entries to remain equality deletes and the new Puffin entries to be
position-delete DVs.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_schema_evolution.groovy:
##########
@@ -0,0 +1,234 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_v3_row_lineage_schema_evolution",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+ String catalogName = "test_iceberg_v3_row_lineage_schema_evolution"
+ String dbName = "test_row_lineage_schema_evolution_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String endpoint = "http://${externalEnvIp}:${minioPort}";
+
+ def formats = ["parquet", "orc"]
+ def partitionFlags = [true, false]
+
+ def descColumns = { tableName -> return sql("""desc
${tableName}""").collect { row -> row[0].toString().toLowerCase() }
+ }
+
+ def normalizeRows = { rows ->
+ return rows.collect { row -> row.collect { col -> col == null ? null :
col.toString() } }
+ }
+
+ def assertSparkDorisBusinessRows = { tableName, columns, orderBy,
expectedRows = null ->
+ sql """refresh table ${dbName}.${tableName}"""
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ def sparkRows = spark_iceberg("""
+ select ${columns}
+ from demo.${dbName}.${tableName}
+ order by ${orderBy}
+ """)
+ def dorisRows = sql("""
+ select ${columns}
+ from ${tableName}
+ order by ${orderBy}
+ """)
+ log.info("Spark business rows for ${tableName}: ${sparkRows}")
+ log.info("Doris business rows for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+
+ def normalizedRows = normalizeRows(dorisRows)
+ if (expectedRows != null) {
+ assertEquals(expectedRows, normalizedRows)
+ }
+ return normalizedRows
+ }
+
+ def lineageMap = { tableName ->
+ def rows = sql("""
+ select id, _row_id, _last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ Map<Integer, List<Long>> result = [:]
+ rows.each { row ->
+ assertTrue(row[1] != null, "_row_id should be non-null for
${tableName}, row=${row}")
+ assertTrue(row[2] != null,
+ "_last_updated_sequence_number should be non-null for
${tableName}, row=${row}")
+ result[row[0].toString().toInteger()] =
[row[1].toString().toLong(), row[2].toString().toLong()]
+ }
+ return result
+ }
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog if not exists ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "${endpoint}",
+ "s3.region" = "us-east-1"
+ )
+ """
+
+ sql """switch ${catalogName}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+ sql """set enable_fallback_to_original_planner = false"""
+ sql """set show_hidden_columns = false"""
+
+ try {
+ formats.each { format ->
+ partitionFlags.each { partitioned ->
+ String partitionSuffix = partitioned ? "part" : "unpart"
+ String tableName = "schema_evo_${format}_${partitionSuffix}"
+ String partitionClause = partitioned ? "partition by list
(day(dt)) ()" : ""
+ try {
+ sql """drop table if exists ${tableName}"""
+ sql """
+ create table ${tableName} (
+ id int,
+ name string,
+ score int,
+ dt date
+ ) engine=iceberg
+ ${partitionClause}
+ properties (
+ "format-version" = "3",
+ "write.format.default" = "${format}",
+ "write.update.mode" = "merge-on-read",
+ "write.merge.mode" = "merge-on-read"
+ )
+ """
+ sql """
+ insert into ${tableName} values
+ (1, 'a', 10, date '2024-10-01'),
+ (2, 'b', 20, date '2024-10-01')
+ """
+
+ Map<Integer, List<Long>> beforeLineage =
lineageMap(tableName)
+ def initialColumns = descColumns(tableName)
+ assertTrue(!initialColumns.contains("_row_id"))
+
assertTrue(!initialColumns.contains("_last_updated_sequence_number"))
+
+ sql """alter table ${tableName} add column extra string"""
+ sql """refresh table ${dbName}.${tableName}"""
+ def addColumns = descColumns(tableName)
+ assertTrue(addColumns.contains("extra"), "ADD COLUMN
should expose extra for ${tableName}")
+
+ def afterAddRows = sql("""
+ select id, name, score, extra, dt, _row_id,
_last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ log.info("Rows after ADD COLUMN for ${tableName}:
${afterAddRows}")
+ assertEquals(2, afterAddRows.size())
+ afterAddRows.each { row ->
+ assertTrue(row[3] == null, "New extra column should be
null for old rows in ${tableName}")
+ assertTrue(row[5] != null, "_row_id should remain
readable after ADD COLUMN in ${tableName}")
+ assertTrue(row[6] != null,
+ "_last_updated_sequence_number should remain
readable after ADD COLUMN in ${tableName}")
+ }
+ assertSparkDorisBusinessRows(tableName, "id, name, score,
extra, dt", "id", [
+ ["1", "a", "10", null, "2024-10-01"],
+ ["2", "b", "20", null, "2024-10-01"]
+ ])
+
+ sql """alter table ${tableName} drop column extra"""
+ sql """refresh table ${dbName}.${tableName}"""
+ def dropColumns = descColumns(tableName)
+ assertTrue(!dropColumns.contains("extra"), "DROP COLUMN
should remove extra for ${tableName}")
+
+ def afterDropRows = sql("""
+ select id, name, score, dt, _row_id,
_last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ log.info("Rows after DROP COLUMN for ${tableName}:
${afterDropRows}")
+ assertEquals(2, afterDropRows.size())
+ assertEquals("a", afterDropRows[0][1].toString())
+ assertEquals(10,
afterDropRows[0][2].toString().toInteger())
+ assertEquals("2024-10-01", afterDropRows[0][3].toString())
+ assertSparkDorisBusinessRows(tableName, "id, name, score,
dt", "id", [
+ ["1", "a", "10", "2024-10-01"],
+ ["2", "b", "20", "2024-10-01"]
+ ])
+
+ sql """update ${tableName} set score = score + 100 where
id = 1"""
+ Map<Integer, List<Long>> afterUpdateLineage =
lineageMap(tableName)
+ assertEquals(beforeLineage[1][0], afterUpdateLineage[1][0])
+ assertTrue(afterUpdateLineage[1][1] > beforeLineage[1][1],
+ "UPDATE after schema evolution should advance sequence
for id=1 in ${tableName}")
+ assertEquals(beforeLineage[2], afterUpdateLineage[2])
+
+ def finalRows = sql("""
+ select id, name, score, dt, _row_id,
_last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ assertEquals(110, finalRows[0][2].toString().toInteger())
+ assertEquals(20, finalRows[1][2].toString().toInteger())
+ assertSparkDorisBusinessRows(tableName, "id, name, score,
dt", "id", [
+ ["1", "a", "110", "2024-10-01"],
+ ["2", "b", "20", "2024-10-01"]
+ ])
+
+ spark_iceberg """refresh table
demo.${dbName}.${tableName}"""
+ spark_iceberg """
+ alter table demo.${dbName}.${tableName}
+ add columns (spark_added string)
+ """
+ sql """refresh table ${dbName}.${tableName}"""
+ def sparkAddColumns = descColumns(tableName)
+ assertTrue(sparkAddColumns.contains("spark_added"),
+ "Spark ADD COLUMN should be visible after Doris
REFRESH TABLE for ${tableName}")
+
+ def afterSparkAddRows = sql("""
+ select id, name, score, spark_added, dt, _row_id,
_last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ log.info("Rows after Spark ADD COLUMN for ${tableName}:
${afterSparkAddRows}")
+ assertEquals(2, afterSparkAddRows.size())
+ afterSparkAddRows.each { row ->
+ assertTrue(row[3] == null,
+ "Spark-added column should be null for existing
rows in ${tableName}")
+ assertTrue(row[5] != null, "_row_id should remain
readable after Spark schema evolution")
Review Comment:
`afterUpdateLineage` already captures the exact row IDs and sequence numbers
before this Spark metadata-only `ADD COLUMN`, but this loop checks only that
the refreshed lineage values are non-null. Reassigning every row's lineage to
different non-null values would therefore leave the suite green. Please compare
`lineageMap(tableName)` with `afterUpdateLineage` here so the schema-evolution
path proves that existing rows retain their exact lineage state.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_v3_row_lineage_uniqueness_stability.groovy:
##########
@@ -0,0 +1,188 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_v3_row_lineage_uniqueness_stability",
"p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("Iceberg test is disabled")
+ return
+ }
+
+ String catalogName = "test_iceberg_v3_row_lineage_uniqueness_stability"
+ String dbName = "test_row_lineage_uniqueness_stability_db"
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String endpoint = "http://${externalEnvIp}:${minioPort}";
+
+ def formats = ["parquet", "orc"]
+ def partitionFlags = [false, true]
+
+ def lineageMap = { tableName ->
+ def rows = sql("""
+ select id, _row_id, _last_updated_sequence_number
+ from ${tableName}
+ order by id
+ """)
+ Map<Integer, List<Long>> result = [:]
+ rows.each { row ->
+ assertTrue(row[1] != null, "_row_id should be non-null for
${tableName}, row=${row}")
+ assertTrue(row[2] != null,
+ "_last_updated_sequence_number should be non-null for
${tableName}, row=${row}")
+ result[row[0].toString().toInteger()] =
[row[1].toString().toLong(), row[2].toString().toLong()]
+ }
+ return result
+ }
+
+ def assertUniqueRowIds = { tableName ->
+ def counts = sql("""
+ select count(*), count(distinct _row_id)
+ from ${tableName}
+ where _row_id is not null
+ """)
+ log.info("Checking row_id uniqueness for ${tableName}: ${counts}")
+ assertEquals(counts[0][0].toString().toLong(),
counts[0][1].toString().toLong())
+ }
+
+ def assertSparkDorisBusinessRows = { tableName, expectedRows ->
+ sql """refresh table ${dbName}.${tableName}"""
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ def sparkRows = spark_iceberg("""
+ select id, name, score, dt
+ from demo.${dbName}.${tableName}
+ order by id
+ """)
+ def dorisRows = sql("""
+ select id, name, score, dt
+ from ${tableName}
+ order by id
+ """)
+ log.info("Spark business rows for ${tableName}: ${sparkRows}")
+ log.info("Doris business rows for ${tableName}: ${dorisRows}")
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+
+ def normalizedRows = dorisRows.collect {
+ [it[0].toString().toInteger(), it[1].toString(),
it[2].toString().toInteger(), it[3].toString()]
+ }
+ assertEquals(expectedRows, normalizedRows)
+ }
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog if not exists ${catalogName} properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "${endpoint}",
+ "s3.region" = "us-east-1"
+ )
+ """
+
+ sql """switch ${catalogName}"""
+ sql """create database if not exists ${dbName}"""
+ sql """use ${dbName}"""
+ sql """set enable_fallback_to_original_planner = false"""
+ sql """set show_hidden_columns = false"""
+
+ try {
+ formats.each { format ->
+ partitionFlags.each { partitioned ->
+ String partitionSuffix = partitioned ? "part" : "unpart"
+ String tableName = "unique_stable_${format}_${partitionSuffix}"
+ String partitionClause = partitioned ? "partition by list
(day(dt)) ()" : ""
+ try {
+ sql """drop table if exists ${tableName}"""
+ sql """
+ create table ${tableName} (
+ id int,
+ name string,
+ score int,
+ dt date
+ ) engine=iceberg
+ ${partitionClause}
+ properties (
+ "format-version" = "3",
+ "write.format.default" = "${format}",
+ "write.update.mode" = "merge-on-read",
+ "write.merge.mode" = "merge-on-read"
+ )
+ """
+ sql """insert into ${tableName} values (1, 'a', 10, date
'2024-08-01'), (2, 'b', 20, date '2024-08-02')"""
+ sql """insert into ${tableName} values (3, 'c', 30, date
'2024-08-03'), (4, 'd', 40, date '2024-08-04')"""
+ sql """insert into ${tableName} values (5, 'e', 50, date
'2024-08-05'), (6, 'f', 60, date '2024-08-06')"""
+
+ Map<Integer, List<Long>> initialLineage =
lineageMap(tableName)
+ assertUniqueRowIds(tableName)
+
+ sql """update ${tableName} set score = score + 10 where id
= 1"""
+ Map<Integer, List<Long>> afterUpdateLineage =
lineageMap(tableName)
+ assertUniqueRowIds(tableName)
+ assertEquals(initialLineage[1][0],
afterUpdateLineage[1][0])
+ assertTrue(afterUpdateLineage[1][1] > initialLineage[1][1],
+ "UPDATE should advance sequence for id=1 in
${tableName}")
+
+ sql """
+ merge into ${tableName} t
+ using (
+ select 4 as id, 'stable_m' as name, 404 as score,
date '2024-08-04' as dt
+ ) s
+ on t.id = s.id
+ when matched then update set name = s.name, score =
s.score
+ """
+ Map<Integer, List<Long>> afterMergeLineage =
lineageMap(tableName)
+ assertUniqueRowIds(tableName)
+ assertEquals(initialLineage[4][0], afterMergeLineage[4][0])
+ assertTrue(afterMergeLineage[4][1] > initialLineage[4][1],
+ "MERGE UPDATE should advance sequence for id=4 in
${tableName}")
+
+ def rewriteResult = sql("""
+ alter table ${catalogName}.${dbName}.${tableName}
+ execute rewrite_data_files(
+ "target-file-size-bytes" = "10485760",
+ "min-input-files" = "1"
+ )
+ """)
+ log.info("rewrite_data_files result for ${tableName}:
${rewriteResult}")
+ assertTrue(rewriteResult.size() > 0,
+ "rewrite_data_files should return summary rows for
${tableName}")
+
+ Map<Integer, List<Long>> afterRewriteLineage =
lineageMap(tableName)
+ assertUniqueRowIds(tableName)
+ assertEquals(afterUpdateLineage[1][0],
afterRewriteLineage[1][0])
Review Comment:
Both maps already contain the exact lineage state for all six visible rows,
but these assertions compare only two row IDs and one sequence number. A
rewrite can reassign IDs/sequences for the other rows (or ID 1's sequence) and
still pass uniqueness plus business-row checks. Please require
`afterRewriteLineage == afterMergeLineage` after also proving the action
rewrote at least one file.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
