github-actions[bot] commented on code in PR #65135:
URL: https://github.com/apache/doris/pull/65135#discussion_r3567827838
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -394,6 +423,41 @@ private void setIcebergParams(TFileRangeDesc rangeDesc,
IcebergSplit icebergSpli
rangeDesc.setTableFormatParams(tableFormatFileDesc);
}
+ private void setIcebergPositionDeleteSysTableParams(TFileRangeDesc
rangeDesc, IcebergSplit icebergSplit,
+ TTableFormatFileDesc tableFormatFileDesc, TIcebergFileDesc
fileDesc) {
+ rangeDesc.setFormatType(icebergSplit.getPositionDeleteFileFormat());
+ tableFormatFileDesc.setTableLevelRowCount(-1);
Review Comment:
This native `$position_deletes` path needs a BE support/smooth-upgrade guard
before it emits Parquet/ORC/Puffin ranges. The range format is switched to the
delete file format here and the only signal that it is a system-table range is
the top-level `iceberg_params.content` set below; old BEs do not have the new
`is_iceberg_position_deletes_sys_table` routing, so they will take the ordinary
Iceberg Parquet/ORC reader path against the metadata-table slots if
`FederationBackendPolicy` assigns the split to them during rolling/smooth
upgrade. Other new plan shapes already skip `Backend.isSmoothUpgradeSrc()` BEs
before dispatch; this should similarly fall back, gate on all selected BEs
supporting the new route, or fail during planning instead of sending a range
old BEs can misread.
##########
regression-test/suites/external_table_p0/iceberg/test_iceberg_position_deletes_sys_table.groovy:
##########
@@ -0,0 +1,659 @@
+// 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_position_deletes_sys_table", "p0,external") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test.")
+ return
+ }
+
+ 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 catalogName = "test_iceberg_position_deletes_sys_table"
+ String dbName = "position_deletes_sys_table_db"
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ CREATE CATALOG ${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",
+ "meta.cache.iceberg.table.ttl-second" = "0",
+ "meta.cache.iceberg.schema.ttl-second" = "0"
+ );"""
+
+ spark_iceberg_multi """
+ SET spark.sql.shuffle.partitions=1;
+ CREATE DATABASE IF NOT EXISTS demo.${dbName};
+ DROP TABLE IF EXISTS demo.${dbName}.pd_unpartitioned;
+ CREATE TABLE demo.${dbName}.pd_unpartitioned (
+ id INT,
+ name STRING
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_unpartitioned
+ SELECT /*+ COALESCE(1) */ id, name FROM VALUES
+ (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd') AS t(id, name);
+ DELETE FROM demo.${dbName}.pd_unpartitioned WHERE id = 2;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_orc_unpartitioned;
+ CREATE TABLE demo.${dbName}.pd_orc_unpartitioned (
+ id INT,
+ name STRING
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.format.default'='orc',
+ 'write.delete.format.default'='orc',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_orc_unpartitioned
+ SELECT /*+ COALESCE(1) */ id, name FROM VALUES
+ (1, 'a'), (2, 'b'), (3, 'c') AS t(id, name);
+ DELETE FROM demo.${dbName}.pd_orc_unpartitioned WHERE id = 2;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_partitioned;
+ CREATE TABLE demo.${dbName}.pd_partitioned (
+ id INT,
+ name STRING,
+ dt STRING
+ ) USING iceberg
+ PARTITIONED BY (dt)
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_partitioned
+ SELECT /*+ COALESCE(1) */ id, name, dt FROM VALUES
+ (1, 'a', '2026-06-26'),
+ (2, 'b', '2026-06-26'),
+ (3, 'c', '2026-06-27'),
+ (4, 'd', '2026-06-27') AS t(id, name, dt);
+ DELETE FROM demo.${dbName}.pd_partitioned WHERE id = 2;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_int_partitioned;
+ CREATE TABLE demo.${dbName}.pd_int_partitioned (
+ id INT,
+ name STRING,
+ p INT
+ ) USING iceberg
+ PARTITIONED BY (p)
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_int_partitioned
+ SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES
+ (1, 'a', 10),
+ (2, 'b', 10),
+ (3, 'c', 20) AS t(id, name, p);
+ DELETE FROM demo.${dbName}.pd_int_partitioned WHERE id = 2;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_date_partitioned;
+ CREATE TABLE demo.${dbName}.pd_date_partitioned (
+ id INT,
+ name STRING,
+ dt DATE
+ ) USING iceberg
+ PARTITIONED BY (dt)
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_date_partitioned
+ SELECT /*+ COALESCE(1) */ id, name, CAST(dt AS DATE) FROM VALUES
+ (1, 'a', '2026-01-02'),
+ (2, 'b', '2026-01-02'),
+ (3, 'c', '2026-03-04') AS t(id, name, dt);
+ DELETE FROM demo.${dbName}.pd_date_partitioned WHERE id = 2;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_partition_evolution;
+ CREATE TABLE demo.${dbName}.pd_partition_evolution (
+ id INT,
+ name STRING,
+ p INT
+ ) USING iceberg
+ PARTITIONED BY (p)
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_partition_evolution
+ SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES
+ (1, 'a', 10),
+ (2, 'b', 10),
+ (3, 'c', 20),
+ (4, 'd', 20) AS t(id, name, p);
+ DELETE FROM demo.${dbName}.pd_partition_evolution WHERE id = 2;
+ ALTER TABLE demo.${dbName}.pd_partition_evolution ADD PARTITION FIELD
bucket(1, id);
+ INSERT INTO demo.${dbName}.pd_partition_evolution
+ SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES
+ (5, 'e', 10),
+ (6, 'f', 10) AS t(id, name, p);
+ DELETE FROM demo.${dbName}.pd_partition_evolution WHERE id = 6;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_partition_rename;
+ CREATE TABLE demo.${dbName}.pd_partition_rename (
+ id INT,
+ name STRING,
+ p INT
+ ) USING iceberg
+ PARTITIONED BY (p)
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_partition_rename
+ SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES
+ (1, 'a', 10),
+ (2, 'b', 10),
+ (3, 'c', 20),
+ (4, 'd', 20) AS t(id, name, p);
+ DELETE FROM demo.${dbName}.pd_partition_rename WHERE id = 2;
+ -- Replacing an identity transform with the same transform and an
alias renames the
+ -- partition field while preserving its Iceberg field ID.
+ ALTER TABLE demo.${dbName}.pd_partition_rename REPLACE PARTITION FIELD
p WITH p AS p2;
+ INSERT INTO demo.${dbName}.pd_partition_rename
+ SELECT /*+ COALESCE(1) */ id, name, p FROM VALUES
+ (5, 'e', 10),
+ (6, 'f', 10) AS t(id, name, p);
+ DELETE FROM demo.${dbName}.pd_partition_rename WHERE id = 6;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_v3_unpartitioned;
+ CREATE TABLE demo.${dbName}.pd_v3_unpartitioned (
+ id INT,
+ name STRING
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_v3_unpartitioned
+ SELECT /*+ COALESCE(1) */ id, name FROM VALUES
+ (1, 'a'), (2, 'b'), (3, 'c') AS t(id, name);
+ DELETE FROM demo.${dbName}.pd_v3_unpartitioned WHERE id = 2;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_v3_partitioned;
+ CREATE TABLE demo.${dbName}.pd_v3_partitioned (
+ id INT,
+ name STRING,
+ dt STRING
+ ) USING iceberg
+ PARTITIONED BY (dt)
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_v3_partitioned
+ SELECT /*+ COALESCE(1) */ id, name, dt FROM VALUES
+ (1, 'a', '2026-07-01'),
+ (2, 'b', '2026-07-01'),
+ (3, 'c', '2026-07-02') AS t(id, name, dt);
+ DELETE FROM demo.${dbName}.pd_v3_partitioned WHERE id = 2;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_schema_time_travel;
+ CREATE TABLE demo.${dbName}.pd_schema_time_travel (
+ id INT,
+ name STRING
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='2',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read',
+ 'write.distribution-mode'='none',
+ 'write.target-file-size-bytes'='134217728'
+ );
+ INSERT INTO demo.${dbName}.pd_schema_time_travel
+ SELECT /*+ COALESCE(1) */ id, name FROM VALUES
+ (1, 'a'), (2, 'b'), (3, 'c') AS t(id, name);
+ DELETE FROM demo.${dbName}.pd_schema_time_travel WHERE id = 2;
+ ALTER TABLE demo.${dbName}.pd_schema_time_travel ADD COLUMN note
STRING;
+ INSERT INTO demo.${dbName}.pd_schema_time_travel
+ SELECT /*+ COALESCE(1) */ id, name, note FROM VALUES
+ (4, 'd', 'new'), (5, 'e', 'new') AS t(id, name, note);
+ DELETE FROM demo.${dbName}.pd_schema_time_travel WHERE id = 4;
+ DROP TABLE IF EXISTS demo.${dbName}.pd_no_deletes;
+ CREATE TABLE demo.${dbName}.pd_no_deletes (
+ id INT,
+ name STRING
+ ) USING iceberg
+ TBLPROPERTIES ('format-version'='2');
+ INSERT INTO demo.${dbName}.pd_no_deletes VALUES (1, 'a'), (2, 'b');
+ """
+
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+
+ def assertPositionDeletesSchema = { String tableName, boolean partitioned,
List<String> extraColumns = [] ->
+ List<List<Object>> descRows = sql """desc
${tableName}\$position_deletes"""
+ List<String> columns = descRows.collect { it[0].toString() }
+ List<String> expectedColumns = ["file_path", "pos", "row"]
+ if (partitioned) {
+ expectedColumns.add("partition")
+ }
+ expectedColumns.addAll(["spec_id", "delete_file_path"])
+ expectedColumns.addAll(extraColumns)
+ assertEquals(expectedColumns, columns)
+ }
+
+ def countRows = { String query ->
+ List<List<Object>> rows = sql query
+ assertEquals(1, rows.size())
+ return ((Number) rows[0][0]).longValue()
+ }
+
+ def assertSparkDorisPositionDeletes = { String tableName, List<String>
columns ->
+ String columnList = columns.join(", ")
+ String orderBy = "file_path, pos, delete_file_path"
+ List<List<Object>> sparkRows = spark_iceberg """
+ select ${columnList}
+ from demo.${dbName}.${tableName}.position_deletes
+ order by ${orderBy}
+ """
+ List<List<Object>> dorisRows = sql """
+ select ${columnList}
+ from ${tableName}\$position_deletes
+ order by ${orderBy}
+ """
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ }
+
+ def assertSparkDorisTableRows = { String tableName, List<List<Object>>
expectedRows ->
+ spark_iceberg """REFRESH TABLE demo.${dbName}.${tableName}"""
+ List<List<Object>> sparkRows = spark_iceberg """
+ select id, name, dt
+ from demo.${dbName}.${tableName}
+ order by id
+ """
+ List<List<Object>> dorisRows = sql """
+ select id, name, dt
+ from ${tableName}
+ order by id
+ """
+ assertSparkDorisResultEquals(sparkRows, dorisRows)
+ assertEquals(expectedRows, dorisRows)
+ }
+
+ List<String> v3ExtraColumns = ["content_offset", "content_size_in_bytes"]
+ List<String> commonCompareColumns = ["file_path", "pos", "spec_id",
"delete_file_path"]
+ List<String> v3CompareColumns = commonCompareColumns + v3ExtraColumns
+
+ assertPositionDeletesSchema("pd_unpartitioned", false)
+ long unpartitionedCount = countRows("""select count(*) from
pd_unpartitioned\$position_deletes""")
+ assertTrue(unpartitionedCount > 0)
+ assertSparkDorisPositionDeletes("pd_unpartitioned", commonCompareColumns)
+ assertEquals(unpartitionedCount, countRows("""select count(pos) from
pd_unpartitioned\$position_deletes"""))
+ assertEquals(unpartitionedCount, (long) sql(
+ """select * from pd_unpartitioned\$position_deletes""").size())
+ assertEquals(unpartitionedCount, (long) sql(
+ """select pos from pd_unpartitioned\$position_deletes""").size())
+ assertEquals(unpartitionedCount, (long) sql(
+ """select delete_file_path, pos from
pd_unpartitioned\$position_deletes""").size())
+ assertEquals(unpartitionedCount, (long) sql(
+ """select file_path, pos, delete_file_path from
pd_unpartitioned\$position_deletes where pos >= 0""").size())
+ assertEquals(unpartitionedCount, countRows(
+ """select count(*) from pd_unpartitioned\$position_deletes where
spec_id = 0"""))
+ assertEquals(unpartitionedCount, countRows(
+ """select count(*) from pd_unpartitioned\$position_deletes where
delete_file_path is not null"""))
+ assertEquals(0L, countRows("""select count(*) from
pd_unpartitioned\$position_deletes where pos < 0"""))
+ List<List<Object>> unpartitionedRows = sql """select `row` from
pd_unpartitioned\$position_deletes"""
+ assertEquals(unpartitionedCount, (long) unpartitionedRows.size())
+
+ assertPositionDeletesSchema("pd_orc_unpartitioned", false)
+ long orcUnpartitionedCount = countRows(
+ """select count(*) from pd_orc_unpartitioned\$position_deletes""")
+ assertTrue(orcUnpartitionedCount > 0)
+ assertSparkDorisPositionDeletes("pd_orc_unpartitioned",
commonCompareColumns)
+ assertEquals(orcUnpartitionedCount, (long) sql(
+ """select `row` from
pd_orc_unpartitioned\$position_deletes""").size())
+ assertTrue(unpartitionedRows.every { it[0] == null })
+ try {
Review Comment:
This assertion is still checking `unpartitionedRows`, which is the Parquet
result populated above. The ORC query on the preceding lines only has its row
count checked, so a regression in `pd_orc_unpartitioned$position_deletes`.`row`
values would still pass this test. Please store the ORC `row` result in its own
variable and assert that variable here.
--
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]
