github-actions[bot] commented on code in PR #65992: URL: https://github.com/apache/doris/pull/65992#discussion_r3643539072
########## regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.groovy: ########## @@ -0,0 +1,242 @@ +// 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_partition_evolution_position_dv", + "p0,external,iceberg,external_docker,external_docker_iceberg") { + 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_partition_evolution_position_dv" + String dbName = "iceberg_partition_evolution_position_dv_db" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + return spark_iceberg(""" + select snapshot_id + from demo.${dbName}.${tableName}.snapshots + order by committed_at desc + limit 1 + """)[0][0].toString() + } + + 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' + ) + """ + + try { + ["parquet", "orc"].each { String format -> + [2, 3].each { int formatVersion -> + String deleteKind = formatVersion == 2 ? "position" : "dv" + String tableName = "${deleteKind}_${format}_evolved" + + spark_iceberg_multi """ + create database if not exists demo.${dbName}; + drop table if exists demo.${dbName}.${tableName}; + create table demo.${dbName}.${tableName} ( + id int, + category string, + event_time timestamp, + payload struct<metric:int, label:string> + ) using iceberg + partitioned by (category, days(event_time)) + tblproperties ( + 'format-version'='${formatVersion}', + 'write.format.default'='${format}', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none' + ); + insert into demo.${dbName}.${tableName} + select /*+ coalesce(1) */ id, category, event_time, payload from values + (1, 'A', timestamp '2026-01-01 01:00:00', + named_struct('metric', 10, 'label', 'base-a')), + (2, 'A', timestamp '2026-01-01 02:00:00', + named_struct('metric', 20, 'label', 'base-delete')), + (3, 'B', timestamp '2026-01-02 01:00:00', + named_struct('metric', 30, 'label', 'base-b')), + (4, 'C', timestamp '2026-01-03 01:00:00', + named_struct('metric', 40, 'label', 'base-c')) + as t(id, category, event_time, payload); + """ + String baseSnapshot = latestSnapshotId(tableName) + sql """ + alter table `${catalogName}`.`${dbName}`.`${tableName}` + create tag ${tableName}_base as of version ${baseSnapshot} + """ + + // Scenario PE-D01: delete against the original spec before any evolution. + spark_iceberg """ + delete from demo.${dbName}.${tableName} where id = 2 + """ + String firstDeleteSnapshot = latestSnapshotId(tableName) + + // Scenario PE-D02: ADD partition field and complex child, then delete a row written + // with the new spec. Old/new delete files must remain associated with their specs. + spark_iceberg_multi """ + alter table demo.${dbName}.${tableName} add partition field bucket(8, id); + alter table demo.${dbName}.${tableName} add column payload.extra string; + insert into demo.${dbName}.${tableName} + select /*+ coalesce(1) */ id, category, event_time, payload from values + (5, 'A', timestamp '2026-02-01 01:00:00', + named_struct('metric', 50, 'label', 'add-a', 'extra', 'new-child')), + (6, 'A', timestamp '2026-02-01 02:00:00', + named_struct('metric', 60, 'label', 'add-delete', 'extra', 'new-child')) + as t(id, category, event_time, payload); + delete from demo.${dbName}.${tableName} where id = 6; + """ + String addedDeleteSnapshot = latestSnapshotId(tableName) + + // Scenario PE-D03: REPLACE temporal transform and rename/promote nested fields. + spark_iceberg_multi """ + alter table demo.${dbName}.${tableName} + replace partition field days(event_time) with months(event_time); + alter table demo.${dbName}.${tableName} + rename column payload.label to renamed_label; + alter table demo.${dbName}.${tableName} + alter column payload.metric type bigint; + insert into demo.${dbName}.${tableName} values + (7, 'A', timestamp '2026-03-01 01:00:00', + named_struct('metric', 7000000000, + 'renamed_label', 'replace-a', 'extra', 'renamed-child')), + (8, 'A', timestamp '2026-03-01 02:00:00', + named_struct('metric', 80, + 'renamed_label', 'replace-delete', 'extra', 'renamed-child')); + """ + + // Scenario PE-D04: DROP identity field before the final delete. The matching row + // is in a spec without category, while older files still expose category partitions. + spark_iceberg_multi """ + alter table demo.${dbName}.${tableName} drop partition field category; + delete from demo.${dbName}.${tableName} where id = 8; + drop table if exists demo.${dbName}.${tableName}_dimension; + create table demo.${dbName}.${tableName}_dimension (category string) + using iceberg tblproperties ('format-version'='2'); + insert into demo.${dbName}.${tableName}_dimension values ('A'); + """ + String finalSnapshot = latestSnapshotId(tableName) + sql """ + alter table `${catalogName}`.`${dbName}`.`${tableName}` + create tag ${tableName}_final as of version ${finalSnapshot} + """ + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${tableName}""" + + // Scenario PE-D05: static partition filters apply every delete exactly once. + List<List<String>> expectedCurrent = [["1"], ["5"], ["7"]] + assertEquals(expectedCurrent, stringRows(""" + select id from ${tableName} where category = 'A' order by id + """)) + assertEquals([["7", "7000000000"]], stringRows(""" + select id, payload.metric from ${tableName} + where payload.metric > 5000000000 order by id + """)) + + // Scenario PE-D06: runtime filter on the dropped identity partition column returns + // the same delete-aware rows with partition pruning enabled and disabled. + String rfQuery = """ + select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */ f.id + from ${tableName} f + join ${tableName}_dimension d on f.category = d.category + order by f.id + """ + sql """set runtime_filter_wait_infinitely=true""" Review Comment: **[P1] Ensure this checkpoint actually retains a runtime filter** These tiny external tables have no column statistics, while `enable_runtime_filter_prune` defaults to true and removes RFs whose probe/build stats are unknown. Toggling only partition pruning can therefore compare the same no-RF plan twice and still pass. Set `enable_runtime_filter_prune=false` as the dedicated RF suite does, then verify the filter reaches the fact scan (ideally with a positive pruning counter). ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.groovy: ########## @@ -0,0 +1,273 @@ +// 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_partition_evolution_filter_refs", + "p0,external,iceberg,external_docker,external_docker_iceberg") { + 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_partition_evolution_filter_refs" + String dbName = "iceberg_partition_evolution_filter_refs_db" + String identityTable = "identity_bucket_truncate_timeline" + String temporalTable = "temporal_transform_timeline" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_iceberg """ + select snapshot_id + from demo.${dbName}.${tableName}.snapshots + order by committed_at desc + limit 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + 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' + ) + """ + + try { + spark_iceberg_multi """ + create database if not exists demo.${dbName}; + drop table if exists demo.${dbName}.${identityTable}; + create table demo.${dbName}.${identityTable} ( + id int, + category string, + code string, + event_time timestamp, + payload struct<metric:int, label:string> + ) using iceberg + partitioned by (category, days(event_time)) + tblproperties ( + 'format-version'='2', + 'write.format.default'='parquet' + ); + insert into demo.${dbName}.${identityTable} values + (1, 'A', 'aa-1', timestamp '2026-01-01 01:00:00', + named_struct('metric', 10, 'label', 'base-a')), + (2, 'B', 'bb-1', timestamp '2026-01-02 01:00:00', + named_struct('metric', 20, 'label', 'base-b')); + """ + String identityBase = latestSnapshotId(identityTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create tag identity_base as of version ${identityBase} + """ + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create branch identity_base_branch as of version ${identityBase} + """ + + // Scenario PE-I01: ADD bucket partition field and add a complex-type child in the same + // timeline. Filters must evaluate old files whose spec has no bucket field. + spark_iceberg_multi """ + alter table demo.${dbName}.${identityTable} add partition field bucket(8, id); + alter table demo.${dbName}.${identityTable} add column payload.extra string; + insert into demo.${dbName}.${identityTable} values + (3, 'A', 'aa-2', timestamp '2026-02-01 01:00:00', + named_struct('metric', 30, 'label', 'bucket-a', 'extra', 'add-child')), + (4, 'C', 'cc-1', timestamp '2026-02-02 01:00:00', + named_struct('metric', 40, 'label', 'bucket-c', 'extra', 'add-child')); + """ + String identityAdded = latestSnapshotId(identityTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create tag identity_added as of version ${identityAdded} + """ + + // Scenario PE-I02: REPLACE bucket with truncate while renaming/promoting nested children. + // Predicates on both old and new partition source columns must scan every applicable spec. + spark_iceberg_multi """ + alter table demo.${dbName}.${identityTable} + replace partition field bucket(8, id) with truncate(2, code); + alter table demo.${dbName}.${identityTable} + rename column payload.label to renamed_label; + alter table demo.${dbName}.${identityTable} + alter column payload.metric type bigint; + insert into demo.${dbName}.${identityTable} values + (5, 'A', 'aa-3', timestamp '2026-03-01 01:00:00', + named_struct('metric', 5000000000, 'renamed_label', 'truncate-a', + 'extra', 'renamed-child')), + (6, 'D', 'dd-1', timestamp '2026-03-02 01:00:00', + named_struct('metric', 60, 'renamed_label', 'truncate-d', + 'extra', 'renamed-child')); + """ + String identityReplaced = latestSnapshotId(identityTable) + + // Scenario PE-I03: DROP identity and temporal fields, then drop/re-add a nested name. + // New unpartitioned-by-category files and old identity-partitioned files coexist. + spark_iceberg_multi """ + alter table demo.${dbName}.${identityTable} drop partition field category; + alter table demo.${dbName}.${identityTable} drop partition field days(event_time); + alter table demo.${dbName}.${identityTable} drop column payload.extra; + alter table demo.${dbName}.${identityTable} add column payload.extra bigint; + insert into demo.${dbName}.${identityTable} values + (7, 'A', 'aa-4', timestamp '2026-04-01 01:00:00', + named_struct('metric', 70, 'renamed_label', 'dropped-partition', + 'extra', 7000)), + (8, 'E', 'ee-1', timestamp '2026-04-02 01:00:00', + named_struct('metric', 80, 'renamed_label', 'dropped-partition', + 'extra', 8000)); + """ + String identityDropped = latestSnapshotId(identityTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create tag identity_dropped as of version ${identityDropped} + """ + + spark_iceberg_multi """ + drop table if exists demo.${dbName}.${temporalTable}; + create table demo.${dbName}.${temporalTable} ( + id int, + event_time timestamp, + payload string + ) using iceberg + partitioned by (years(event_time)) + tblproperties ( + 'format-version'='2', + 'write.format.default'='orc' + ); + insert into demo.${dbName}.${temporalTable} values + (11, timestamp '2024-01-01 01:00:00', 'year-2024'), + (12, timestamp '2025-01-01 01:00:00', 'year-2025'); + """ + String temporalYear = latestSnapshotId(temporalTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${temporalTable}` + create tag temporal_year as of version ${temporalYear} + """ + + // Scenario PE-T01: REPLACE year -> month -> day -> hour across ORC files. + // Range and equality filters validate every temporal transform boundary. + spark_iceberg_multi """ + alter table demo.${dbName}.${temporalTable} + replace partition field years(event_time) with months(event_time); + insert into demo.${dbName}.${temporalTable} values + (13, timestamp '2026-02-01 01:00:00', 'month-feb'), + (14, timestamp '2026-03-01 01:00:00', 'month-mar'); + alter table demo.${dbName}.${temporalTable} + replace partition field months(event_time) with days(event_time); + insert into demo.${dbName}.${temporalTable} values + (15, timestamp '2026-04-03 01:00:00', 'day-03'), + (16, timestamp '2026-04-04 01:00:00', 'day-04'); + alter table demo.${dbName}.${temporalTable} + replace partition field days(event_time) with hours(event_time); + insert into demo.${dbName}.${temporalTable} values + (17, timestamp '2026-05-01 08:00:00', 'hour-08'), + (18, timestamp '2026-05-01 09:00:00', 'hour-09'); + """ + String temporalHour = latestSnapshotId(temporalTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${temporalTable}` + create tag temporal_hour as of version ${temporalHour} + """ + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh catalog ${catalogName}""" + + // Scenario PE-F01: equality/range/IN/NULL-safe source-column filters span four specs. + assertEquals([["1"], ["3"], ["5"], ["7"]], stringRows(""" Review Comment: **[P1] Generate regression outputs for the fixed result contracts** This and the other six new suites encode deterministic ordered SQL rows with direct assertions, and the change adds no `.out` files. The repository testing contract requires fixed query results to use named `qt_`/`order_qt_` actions and runner-generated outputs. Please convert the stable row checks across the new suites; keep direct assertions only for dynamic process, artifact, and profile invariants. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.groovy: ########## @@ -0,0 +1,238 @@ +// 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 org.apache.doris.regression.action.ProfileAction + +suite("test_iceberg_partition_evolution_runtime_filter", Review Comment: **[P2] Serialize suites that discover profiles through the shared list** This suite (and `test_paimon_partition_schema_filter_refs`) polls the bounded FE-wide profile list after execution. A UUID prevents a wrong match but cannot recover a profile evicted by parallel query traffic; the adjacent `rf_partition_pruning` suite documents this same race and is `nonConcurrent`. Please serialize these two profile-dependent suites or use a direct profile/query handle, and adjust the blanket parallel-support claim. ########## regression-test/suites/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.groovy: ########## @@ -0,0 +1,218 @@ +// 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_paimon_partition_pk_delete_refs", + "p0,external,paimon,external_docker,external_docker_paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_partition_pk_delete_refs" + String dbName = "paimon_partition_pk_delete_refs_db" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + return spark_paimon(""" + select snapshot_id + from paimon.${dbName}.`${tableName}\$snapshots` + order by snapshot_id desc + limit 1 + """)[0][0].toString() + } + def createTag = { String tableName, String tagName -> + spark_paimon """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}' + ) + """ + } + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + 'type'='paimon', + 'warehouse'='s3://warehouse/wh', + 's3.endpoint'='http://${externalEnvIp}:${minioPort}', + 's3.access_key'='admin', + 's3.secret_key'='password', + 's3.path.style.access'='true', + 'meta.cache.paimon.table.ttl-second'='0' + ) + """ + + try { + ["parquet", "orc"].each { String format -> + String tableName = "partition_pk_dv_${format}" + String dimensionTable = "${tableName}_dimension" + spark_paimon_multi """ + create database if not exists paimon.${dbName}; + drop table if exists paimon.${dbName}.${tableName}; + create table paimon.${dbName}.${tableName} ( + id int not null, + part string not null, + old_name string, + note string, + payload struct<metric:int, label:string> + ) using paimon + partitioned by (part) + tblproperties ( + 'bucket'='1', + 'primary-key'='part,id', + 'file.format'='${format}', + 'deletion-vectors.enabled'='true' + ); + insert into paimon.${dbName}.${tableName} values + (1, 'p1', 'alpha', 'old-note-1', + named_struct('metric', 10, 'label', 'base-1')), + (2, 'p1', 'beta', 'old-note-2', + named_struct('metric', 20, 'label', 'base-delete')), + (3, 'p2', 'gamma', 'old-note-3', + named_struct('metric', 30, 'label', 'base-p2')); + """ + String baseSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_base") + + // Scenario PM-D01: add/rename fields, upsert one PK and delete another inside p1. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} add column payload.extra string; + alter table paimon.${dbName}.${tableName} rename column old_name to full_name; + insert into paimon.${dbName}.${tableName} + (id, part, full_name, note, payload) values + (1, 'p1', 'alpha-updated', 'new-note-1', + named_struct('metric', 11, 'label', 'updated-1', 'extra', 'extra-1')), + (4, 'p1', 'delta', 'delete-later', + named_struct('metric', 40, 'label', 'insert-4', 'extra', 'extra-4')); + delete from paimon.${dbName}.${tableName} + where part = 'p1' and id = 2; + """ + String firstDeleteSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_first_delete") + + // Scenario PM-D02: nested rename/type promotion and drop/re-add combine with another + // delete, insert and full compaction while the partition key stays fixed. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} + rename column payload.label to renamed_label; + alter table paimon.${dbName}.${tableName} + alter column payload.metric type bigint; + alter table paimon.${dbName}.${tableName} drop column note; + alter table paimon.${dbName}.${tableName} add column note bigint; + delete from paimon.${dbName}.${tableName} + where part = 'p1' and id = 4; + insert into paimon.${dbName}.${tableName} + (id, part, full_name, payload, note) values + (5, 'p1', 'epsilon', + named_struct('metric', 5000000000, + 'renamed_label', 'insert-5', 'extra', 'extra-5'), + 5000); + call paimon.sys.compact( + table => '${dbName}.${tableName}', + compact_strategy => 'full' + ); + drop table if exists paimon.${dbName}.${dimensionTable}; + create table paimon.${dbName}.${dimensionTable} (part string) + using paimon tblproperties ('file.format'='parquet'); + insert into paimon.${dbName}.${dimensionTable} values ('p1'); + """ + String finalSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_final") + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${tableName}""" + + // Scenario PM-D03: static partition filters apply PK upserts, deletes and DV state. + List<List<String>> expectedCurrent = [["1", "alpha-updated"], ["5", "epsilon"]] + assertEquals(expectedCurrent, stringRows(""" + select id, full_name from ${tableName} + where part = 'p1' order by id + """)) + assertEquals([["5", "5000000000", "5000"]], stringRows(""" + select id, payload.metric, note from ${tableName} + where part = 'p1' and payload.metric > 1000000000 + order by id + """)) + + // Scenario PM-D04: runtime-filter pruning on the partition key remains delete-aware. + String rfQuery = """ + select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */ + f.id, f.full_name + from ${tableName} f + join ${dimensionTable} d on f.part = d.part + order by f.id + """ + sql """set runtime_filter_wait_infinitely=true""" Review Comment: **[P1] Prove the PK runtime filter survives and prunes** With no column statistics, the default RF effectiveness pruner can remove this filter; both on/off queries then return `expectedCurrent` without exercising partition pruning. Apply the sibling suite's `enable_runtime_filter_prune=false` precondition and require a positive Paimon partition/file-range pruning counter for the enabled run. ########## regression-test/suites/external_table_p0/paimon/test_paimon_partition_mutation_atomicity.groovy: ########## @@ -0,0 +1,157 @@ +// 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_paimon_partition_mutation_atomicity", + "p0,external,paimon,external_docker,external_docker_paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_partition_mutation_atomicity" + String dbName = "paimon_partition_mutation_atomicity_db" + String tableName = "partition_contract" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def schemaCount = { + return spark_paimon(""" + select count(*) from paimon.${dbName}.`${tableName}\$schemas` + """)[0][0].toString().toInteger() + } + def assertSparkRejected = { String statement, String operation -> + String error = null + try { + spark_paimon(statement) + } catch (Exception e) { + error = e.getMessage() + } + assertNotNull(error, "Paimon must reject partition-key ${operation}") + assertTrue(error.toLowerCase().contains("partition"), Review Comment: **[P1] Match the actual partition-key rejection contract** Both the database and table identifiers already contain `partition`, so any Spark error that echoes `...partition_mutation_atomicity_db.partition_contract...` satisfies this check. For example, a parser-level unsupported-syntax error can leave the schema unchanged and make the type-change case pass without reaching Paimon's partition-key validation. Match the operation-specific message (the adjacent matrix uses `Cannot rename partition column`) rather than this generic token. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.groovy: ########## @@ -0,0 +1,238 @@ +// 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 org.apache.doris.regression.action.ProfileAction + +suite("test_iceberg_partition_evolution_runtime_filter", + "p0,external,iceberg,external_docker,external_docker_iceberg") { + 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_partition_evolution_runtime_filter" + String dbName = "iceberg_partition_evolution_runtime_filter_db" + String factTable = "evolved_fact" + String dimensionTable = "rf_dimension" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { + return spark_iceberg(""" + select snapshot_id + from demo.${dbName}.${factTable}.snapshots + order by committed_at desc + limit 1 + """)[0][0].toString() + } + def profileAction = new ProfileAction(context) + def profileCounterValues = { String profileText, String counterName -> + def values = [] + def matcher = profileText =~ ("(?m)^\\s*(?:-\\s*)?" + + java.util.regex.Pattern.quote(counterName) + ":\\s+([^\\n]+)") + while (matcher.find()) { + String valueText = matcher.group(1).toString() + def exact = valueText =~ /\(([0-9,]+)\)/ + def number = valueText =~ /([0-9,]+)/ + String rawValue = exact.find() ? exact.group(1) : (number.find() ? number.group(1) : null) + if (rawValue != null) { + values.add(Long.parseLong(rawValue.replace(",", ""))) + } + } + return values + } + def assertRuntimeFilterPruned = { String queryBody -> + String token = UUID.randomUUID().toString() + List<List<String>> rows = stringRows(""" + select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */ + '${token}', f.id + ${queryBody} + order by f.id + """) + // Scanner counters can arrive after the profile list first reports COMPLETE. + String profile = profileAction.getProfileBySql( + token, + ["RuntimeFilterPartitionPrunedRangeNum"], + 30000L, + 500L) + long fileRangesPruned = profileCounterValues( + profile, "RuntimeFilterPartitionPrunedRangeNum").sum(0L) + long partitionsPruned = profileCounterValues( + profile, "PartitionsPrunedByRuntimeFilter").sum(0L) + assertTrue(fileRangesPruned + partitionsPruned > 0L, + "Runtime filter did not prune any evolved Iceberg partition/file range; " + + profile.take(2000).replaceAll("\\s+", " ")) + return rows.collect { row -> [row[1]] } + } + + 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' + ) + """ + + try { + spark_iceberg_multi """ + create database if not exists demo.${dbName}; + drop table if exists demo.${dbName}.${factTable}; + create table demo.${dbName}.${factTable} ( + id int, + category string, + event_time timestamp, + payload struct<metric:int> + ) using iceberg + partitioned by (category, days(event_time)) + tblproperties ('format-version'='2'); + insert into demo.${dbName}.${factTable} values + (1, 'A', timestamp '2026-01-01 01:00:00', named_struct('metric', 10)), + (2, 'B', timestamp '2026-01-02 01:00:00', named_struct('metric', 20)), + (3, 'C', timestamp '2026-01-03 01:00:00', named_struct('metric', 30)); + """ + String baseSnapshot = latestSnapshotId() + sql """ + alter table `${catalogName}`.`${dbName}`.`${factTable}` + create tag rf_base as of version ${baseSnapshot} + """ + + // Scenario PE-RF01: add a partition field and evolve a nested payload between data files. + spark_iceberg_multi """ + alter table demo.${dbName}.${factTable} add partition field bucket(8, id); + alter table demo.${dbName}.${factTable} add column payload.label string; + insert into demo.${dbName}.${factTable} values + (4, 'A', timestamp '2026-02-01 01:00:00', + named_struct('metric', 40, 'label', 'add-spec')), + (5, 'D', timestamp '2026-02-02 01:00:00', + named_struct('metric', 50, 'label', 'add-spec')); + """ + String addedSnapshot = latestSnapshotId() + + // Scenario PE-RF02: replace a temporal transform. Runtime filters on event_time must + // translate against the transform of each file's own spec. + spark_iceberg_multi """ + alter table demo.${dbName}.${factTable} + replace partition field days(event_time) with months(event_time); + insert into demo.${dbName}.${factTable} values + (6, 'A', timestamp '2026-03-01 01:00:00', + named_struct('metric', 60, 'label', 'replace-spec')), + (7, 'E', timestamp '2026-03-02 01:00:00', + named_struct('metric', 70, 'label', 'replace-spec')); + """ + + // Scenario PE-RF03: drop the identity field. New files have no category partition value; + // runtime pruning may only discard older ranges and must still scan matching new rows. + spark_iceberg_multi """ + alter table demo.${dbName}.${factTable} drop partition field category; + insert into demo.${dbName}.${factTable} values + (8, 'A', timestamp '2026-04-01 01:00:00', + named_struct('metric', 80, 'label', 'drop-spec')), + (9, 'F', timestamp '2026-04-02 01:00:00', + named_struct('metric', 90, 'label', 'drop-spec')); + drop table if exists demo.${dbName}.${dimensionTable}; + create table demo.${dbName}.${dimensionTable} ( + category string, + lower_time timestamp, + upper_time timestamp + ) using iceberg + tblproperties ('format-version'='2'); + insert into demo.${dbName}.${dimensionTable} values + ('A', timestamp '2026-03-01 00:00:00', timestamp '2026-05-01 00:00:00'); + """ + String droppedSnapshot = latestSnapshotId() + sql """ + alter table `${catalogName}`.`${dbName}`.`${factTable}` + create tag rf_dropped as of version ${droppedSnapshot} + """ + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh catalog ${catalogName}""" + sql """set enable_profile=true""" + sql """set profile_level=2""" + // Small fixture tables have no column statistics. Keep the generated RF so this suite + // validates scanner-side partition pruning instead of optimizer selectivity heuristics. + sql """set enable_runtime_filter_prune=false""" + sql """set runtime_filter_wait_infinitely=true""" + sql """set runtime_filter_mode=GLOBAL""" + sql """set parallel_pipeline_task_num=1""" + sql """set disable_join_reorder=true""" + + String currentCategoryJoin = """ + from ${factTable} f + join ${dimensionTable} d on f.category = d.category + """ + String currentTemporalJoin = """ + from ${factTable} f + join ${dimensionTable} d + on f.event_time >= d.lower_time and f.event_time < d.upper_time + """ + + // Scenario PE-RF04: result parity with RF disabled protects correctness. + sql """set enable_runtime_filter_partition_prune=false""" + assertEquals([["1"], ["4"], ["6"], ["8"]], stringRows(""" + select f.id ${currentCategoryJoin} order by f.id + """)) + assertEquals([["6"], ["7"], ["8"], ["9"]], stringRows(""" + select f.id ${currentTemporalJoin} order by f.id + """)) + + // Scenario PE-RF05: current multi-spec scan must both return the same rows and show + // physical pruning in the profile. + sql """set enable_runtime_filter_partition_prune=true""" + assertEquals([["1"], ["4"], ["6"], ["8"]], Review Comment: **[P1] Do not count the category profile as the other RF cells** This is the only enabled, profile-verified call, and `category` existed in the original spec before later being dropped. No RF fixture adds that identity field, `currentTemporalJoin` runs only while partition pruning is disabled, and no query joins on `id`; moreover the current Iceberg split path supplies only identity-transform partition values to scanner RF pruning. The positive counter therefore proves only the already-present/drop-category path, not the documented add-identity/old-spec-new-field, bucket, or temporal cells. Add attributable enabled/profiled cases for supported paths, or narrow the matrix. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.groovy: ########## @@ -0,0 +1,242 @@ +// 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_partition_evolution_position_dv", + "p0,external,iceberg,external_docker,external_docker_iceberg") { + 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_partition_evolution_position_dv" + String dbName = "iceberg_partition_evolution_position_dv_db" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + return spark_iceberg(""" + select snapshot_id + from demo.${dbName}.${tableName}.snapshots + order by committed_at desc + limit 1 + """)[0][0].toString() + } + + 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' + ) + """ + + try { + ["parquet", "orc"].each { String format -> + [2, 3].each { int formatVersion -> + String deleteKind = formatVersion == 2 ? "position" : "dv" + String tableName = "${deleteKind}_${format}_evolved" + + spark_iceberg_multi """ + create database if not exists demo.${dbName}; + drop table if exists demo.${dbName}.${tableName}; + create table demo.${dbName}.${tableName} ( + id int, + category string, + event_time timestamp, + payload struct<metric:int, label:string> + ) using iceberg + partitioned by (category, days(event_time)) + tblproperties ( + 'format-version'='${formatVersion}', + 'write.format.default'='${format}', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none' + ); + insert into demo.${dbName}.${tableName} + select /*+ coalesce(1) */ id, category, event_time, payload from values + (1, 'A', timestamp '2026-01-01 01:00:00', + named_struct('metric', 10, 'label', 'base-a')), + (2, 'A', timestamp '2026-01-01 02:00:00', + named_struct('metric', 20, 'label', 'base-delete')), + (3, 'B', timestamp '2026-01-02 01:00:00', + named_struct('metric', 30, 'label', 'base-b')), + (4, 'C', timestamp '2026-01-03 01:00:00', + named_struct('metric', 40, 'label', 'base-c')) + as t(id, category, event_time, payload); + """ + String baseSnapshot = latestSnapshotId(tableName) + sql """ + alter table `${catalogName}`.`${dbName}`.`${tableName}` + create tag ${tableName}_base as of version ${baseSnapshot} + """ + + // Scenario PE-D01: delete against the original spec before any evolution. + spark_iceberg """ + delete from demo.${dbName}.${tableName} where id = 2 + """ + String firstDeleteSnapshot = latestSnapshotId(tableName) + + // Scenario PE-D02: ADD partition field and complex child, then delete a row written + // with the new spec. Old/new delete files must remain associated with their specs. + spark_iceberg_multi """ + alter table demo.${dbName}.${tableName} add partition field bucket(8, id); + alter table demo.${dbName}.${tableName} add column payload.extra string; + insert into demo.${dbName}.${tableName} + select /*+ coalesce(1) */ id, category, event_time, payload from values + (5, 'A', timestamp '2026-02-01 01:00:00', + named_struct('metric', 50, 'label', 'add-a', 'extra', 'new-child')), + (6, 'A', timestamp '2026-02-01 02:00:00', + named_struct('metric', 60, 'label', 'add-delete', 'extra', 'new-child')) + as t(id, category, event_time, payload); + delete from demo.${dbName}.${tableName} where id = 6; + """ + String addedDeleteSnapshot = latestSnapshotId(tableName) + + // Scenario PE-D03: REPLACE temporal transform and rename/promote nested fields. + spark_iceberg_multi """ + alter table demo.${dbName}.${tableName} + replace partition field days(event_time) with months(event_time); + alter table demo.${dbName}.${tableName} + rename column payload.label to renamed_label; + alter table demo.${dbName}.${tableName} + alter column payload.metric type bigint; + insert into demo.${dbName}.${tableName} values + (7, 'A', timestamp '2026-03-01 01:00:00', + named_struct('metric', 7000000000, + 'renamed_label', 'replace-a', 'extra', 'renamed-child')), + (8, 'A', timestamp '2026-03-01 02:00:00', + named_struct('metric', 80, + 'renamed_label', 'replace-delete', 'extra', 'renamed-child')); + """ + + // Scenario PE-D04: DROP identity field before the final delete. The matching row + // is in a spec without category, while older files still expose category partitions. + spark_iceberg_multi """ + alter table demo.${dbName}.${tableName} drop partition field category; Review Comment: **[P1] Create a data file after dropping `category`** IDs 7 and 8 were inserted before this DDL, so id 8's data file still belongs to the preceding spec that contains `category`; partition evolution does not rewrite it. Deleting id 8 immediately after the drop therefore does not construct the PE-D04 case described above, and the suite never applies a position delete/DV to a data file whose own spec lacks the field. Please insert and delete a victim written after this drop, and assert its data/delete `spec_id`. ########## regression-test/suites/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.groovy: ########## @@ -0,0 +1,218 @@ +// 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_paimon_partition_pk_delete_refs", + "p0,external,paimon,external_docker,external_docker_paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_partition_pk_delete_refs" + String dbName = "paimon_partition_pk_delete_refs_db" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + return spark_paimon(""" + select snapshot_id + from paimon.${dbName}.`${tableName}\$snapshots` + order by snapshot_id desc + limit 1 + """)[0][0].toString() + } + def createTag = { String tableName, String tagName -> + spark_paimon """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}' + ) + """ + } + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + 'type'='paimon', + 'warehouse'='s3://warehouse/wh', + 's3.endpoint'='http://${externalEnvIp}:${minioPort}', + 's3.access_key'='admin', + 's3.secret_key'='password', + 's3.path.style.access'='true', + 'meta.cache.paimon.table.ttl-second'='0' + ) + """ + + try { + ["parquet", "orc"].each { String format -> + String tableName = "partition_pk_dv_${format}" + String dimensionTable = "${tableName}_dimension" + spark_paimon_multi """ + create database if not exists paimon.${dbName}; + drop table if exists paimon.${dbName}.${tableName}; + create table paimon.${dbName}.${tableName} ( + id int not null, + part string not null, + old_name string, + note string, + payload struct<metric:int, label:string> + ) using paimon + partitioned by (part) + tblproperties ( + 'bucket'='1', + 'primary-key'='part,id', + 'file.format'='${format}', + 'deletion-vectors.enabled'='true' + ); + insert into paimon.${dbName}.${tableName} values + (1, 'p1', 'alpha', 'old-note-1', + named_struct('metric', 10, 'label', 'base-1')), + (2, 'p1', 'beta', 'old-note-2', + named_struct('metric', 20, 'label', 'base-delete')), + (3, 'p2', 'gamma', 'old-note-3', + named_struct('metric', 30, 'label', 'base-p2')); + """ + String baseSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_base") + + // Scenario PM-D01: add/rename fields, upsert one PK and delete another inside p1. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} add column payload.extra string; + alter table paimon.${dbName}.${tableName} rename column old_name to full_name; + insert into paimon.${dbName}.${tableName} + (id, part, full_name, note, payload) values + (1, 'p1', 'alpha-updated', 'new-note-1', + named_struct('metric', 11, 'label', 'updated-1', 'extra', 'extra-1')), + (4, 'p1', 'delta', 'delete-later', + named_struct('metric', 40, 'label', 'insert-4', 'extra', 'extra-4')); + delete from paimon.${dbName}.${tableName} + where part = 'p1' and id = 2; + """ + String firstDeleteSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_first_delete") + + // Scenario PM-D02: nested rename/type promotion and drop/re-add combine with another + // delete, insert and full compaction while the partition key stays fixed. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} + rename column payload.label to renamed_label; + alter table paimon.${dbName}.${tableName} + alter column payload.metric type bigint; + alter table paimon.${dbName}.${tableName} drop column note; Review Comment: **[P1] Assert the evolved fields around the PK delete timeline** No later query projects `payload.extra` or `payload.renamed_label`, and the only `note` assertion filters to newly inserted row 5. Surviving old row 1 is therefore never checked for NULL in the re-added BIGINT `note`, so field-ID leakage or a broken nested rename can pass. Add current and historical projections that cover row 1 and the old/new payload bindings. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.groovy: ########## @@ -0,0 +1,273 @@ +// 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_partition_evolution_filter_refs", + "p0,external,iceberg,external_docker,external_docker_iceberg") { + 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_partition_evolution_filter_refs" + String dbName = "iceberg_partition_evolution_filter_refs_db" + String identityTable = "identity_bucket_truncate_timeline" + String temporalTable = "temporal_transform_timeline" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_iceberg """ + select snapshot_id + from demo.${dbName}.${tableName}.snapshots + order by committed_at desc + limit 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + 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' + ) + """ + + try { + spark_iceberg_multi """ + create database if not exists demo.${dbName}; + drop table if exists demo.${dbName}.${identityTable}; + create table demo.${dbName}.${identityTable} ( + id int, + category string, + code string, + event_time timestamp, + payload struct<metric:int, label:string> + ) using iceberg + partitioned by (category, days(event_time)) + tblproperties ( + 'format-version'='2', + 'write.format.default'='parquet' + ); + insert into demo.${dbName}.${identityTable} values + (1, 'A', 'aa-1', timestamp '2026-01-01 01:00:00', + named_struct('metric', 10, 'label', 'base-a')), + (2, 'B', 'bb-1', timestamp '2026-01-02 01:00:00', + named_struct('metric', 20, 'label', 'base-b')); + """ + String identityBase = latestSnapshotId(identityTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create tag identity_base as of version ${identityBase} + """ + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create branch identity_base_branch as of version ${identityBase} + """ + + // Scenario PE-I01: ADD bucket partition field and add a complex-type child in the same + // timeline. Filters must evaluate old files whose spec has no bucket field. + spark_iceberg_multi """ + alter table demo.${dbName}.${identityTable} add partition field bucket(8, id); + alter table demo.${dbName}.${identityTable} add column payload.extra string; + insert into demo.${dbName}.${identityTable} values + (3, 'A', 'aa-2', timestamp '2026-02-01 01:00:00', + named_struct('metric', 30, 'label', 'bucket-a', 'extra', 'add-child')), + (4, 'C', 'cc-1', timestamp '2026-02-02 01:00:00', + named_struct('metric', 40, 'label', 'bucket-c', 'extra', 'add-child')); + """ + String identityAdded = latestSnapshotId(identityTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create tag identity_added as of version ${identityAdded} + """ + + // Scenario PE-I02: REPLACE bucket with truncate while renaming/promoting nested children. + // Predicates on both old and new partition source columns must scan every applicable spec. + spark_iceberg_multi """ + alter table demo.${dbName}.${identityTable} + replace partition field bucket(8, id) with truncate(2, code); + alter table demo.${dbName}.${identityTable} + rename column payload.label to renamed_label; + alter table demo.${dbName}.${identityTable} + alter column payload.metric type bigint; + insert into demo.${dbName}.${identityTable} values + (5, 'A', 'aa-3', timestamp '2026-03-01 01:00:00', + named_struct('metric', 5000000000, 'renamed_label', 'truncate-a', + 'extra', 'renamed-child')), + (6, 'D', 'dd-1', timestamp '2026-03-02 01:00:00', + named_struct('metric', 60, 'renamed_label', 'truncate-d', + 'extra', 'renamed-child')); + """ + String identityReplaced = latestSnapshotId(identityTable) + + // Scenario PE-I03: DROP identity and temporal fields, then drop/re-add a nested name. + // New unpartitioned-by-category files and old identity-partitioned files coexist. + spark_iceberg_multi """ + alter table demo.${dbName}.${identityTable} drop partition field category; + alter table demo.${dbName}.${identityTable} drop partition field days(event_time); + alter table demo.${dbName}.${identityTable} drop column payload.extra; + alter table demo.${dbName}.${identityTable} add column payload.extra bigint; + insert into demo.${dbName}.${identityTable} values + (7, 'A', 'aa-4', timestamp '2026-04-01 01:00:00', + named_struct('metric', 70, 'renamed_label', 'dropped-partition', + 'extra', 7000)), + (8, 'E', 'ee-1', timestamp '2026-04-02 01:00:00', + named_struct('metric', 80, 'renamed_label', 'dropped-partition', + 'extra', 8000)); + """ + String identityDropped = latestSnapshotId(identityTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${identityTable}` + create tag identity_dropped as of version ${identityDropped} + """ + + spark_iceberg_multi """ + drop table if exists demo.${dbName}.${temporalTable}; + create table demo.${dbName}.${temporalTable} ( + id int, + event_time timestamp, + payload string + ) using iceberg + partitioned by (years(event_time)) + tblproperties ( + 'format-version'='2', + 'write.format.default'='orc' + ); + insert into demo.${dbName}.${temporalTable} values + (11, timestamp '2024-01-01 01:00:00', 'year-2024'), + (12, timestamp '2025-01-01 01:00:00', 'year-2025'); + """ + String temporalYear = latestSnapshotId(temporalTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${temporalTable}` + create tag temporal_year as of version ${temporalYear} + """ + + // Scenario PE-T01: REPLACE year -> month -> day -> hour across ORC files. + // Range and equality filters validate every temporal transform boundary. + spark_iceberg_multi """ + alter table demo.${dbName}.${temporalTable} + replace partition field years(event_time) with months(event_time); + insert into demo.${dbName}.${temporalTable} values + (13, timestamp '2026-02-01 01:00:00', 'month-feb'), + (14, timestamp '2026-03-01 01:00:00', 'month-mar'); + alter table demo.${dbName}.${temporalTable} + replace partition field months(event_time) with days(event_time); + insert into demo.${dbName}.${temporalTable} values + (15, timestamp '2026-04-03 01:00:00', 'day-03'), + (16, timestamp '2026-04-04 01:00:00', 'day-04'); + alter table demo.${dbName}.${temporalTable} + replace partition field days(event_time) with hours(event_time); + insert into demo.${dbName}.${temporalTable} values + (17, timestamp '2026-05-01 08:00:00', 'hour-08'), + (18, timestamp '2026-05-01 09:00:00', 'hour-09'); + """ + String temporalHour = latestSnapshotId(temporalTable) + sql """ + alter table `${catalogName}`.`${dbName}`.`${temporalTable}` + create tag temporal_hour as of version ${temporalHour} + """ + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh catalog ${catalogName}""" + + // Scenario PE-F01: equality/range/IN/NULL-safe source-column filters span four specs. + assertEquals([["1"], ["3"], ["5"], ["7"]], stringRows(""" + select id from ${identityTable} where category = 'A' order by id + """)) + assertEquals([["1"], ["3"], ["5"], ["7"]], stringRows(""" + select id from ${identityTable} + where code in ('aa-1', 'aa-2', 'aa-3', 'aa-4') + order by id + """)) + assertEquals([["5"], ["6"], ["7"], ["8"]], stringRows(""" + select id from ${identityTable} + where event_time >= timestamp '2026-03-01 00:00:00' + order by id + """)) + assertEquals([["7", "7000"], ["8", "8000"]], stringRows(""" + select id, payload.extra from ${identityTable} + where payload.extra is not null + order by id + """)) + + // Scenario PE-R01: numeric snapshot, tag and branch retain their own data/spec timeline. + List<List<String>> identityBaseRows = [["1", "A"], ["2", "B"]] + assertEquals(identityBaseRows, stringRows(""" + select id, category + from ${identityTable} for version as of ${identityBase} + where category in ('A', 'B') + order by id + """)) + assertEquals(identityBaseRows, stringRows(""" + select id, category from ${identityTable}@tag(identity_base) + where category in ('A', 'B') order by id + """)) + assertEquals(identityBaseRows, stringRows(""" + select id, category from ${identityTable}@branch(identity_base_branch) + where category in ('A', 'B') order by id + """)) + assertEquals([["1"], ["3"]], stringRows(""" + select id from ${identityTable} for version as of ${identityAdded} + where category = 'A' order by id + """)) + assertEquals([["1"], ["3"], ["5"]], stringRows(""" + select id from ${identityTable} for version as of ${identityReplaced} + where category = 'A' order by id + """)) + assertEquals([["1"], ["3"], ["5"], ["7"]], stringRows(""" + select id from ${identityTable}@tag(identity_dropped) + where category = 'A' order by id + """)) + + // Scenario PE-F02/PE-R02: temporal filters use the spec selected by numeric/tag refs. + assertEquals([["11"], ["12"]], stringRows(""" + select id from ${temporalTable}@tag(temporal_year) + where event_time < timestamp '2026-01-01 00:00:00' + order by id + """)) + assertEquals([["17"], ["18"]], stringRows(""" Review Comment: **[P1] Make a current predicate require rows from every temporal spec** At the final snapshot this range legitimately matches only ids 17/18 from the newest hour spec. The earlier year assertion reads the pre-evolution tag, so these checks still pass if current planning incorrectly drops every year/month/day file. Add a current-snapshot predicate whose expected rows span the year, month, day, and hour files so each transform boundary is observable. ########## regression-test/suites/external_table_p0/iceberg_paimon_schema_time_travel_coverage.md: ########## @@ -11,6 +11,47 @@ The tests use explicit old/new field projections and negative bindings in addition to row-shape assertions. This prevents a rename with unchanged values from passing accidentally. +The partition-evolution extension also distinguishes result correctness from +physical pruning. Static predicates are validated across files written by +different partition specs, while runtime-filter cases require both equal +results with pruning disabled/enabled and a positive pruning counter. + +## Partition evolution matrix + +### Iceberg + +| Partition operation / transform | Static filter | Runtime filter | Snapshot/tag/branch | Position delete | Equality delete | Deletion vector | Complex schema in same timeline | Format / reader | +| --- | --- | --- | --- | --- | --- | --- | --- | --- | +| Add identity field | Covered | Covered | Covered | Covered | Covered | Covered | Covered | Parquet/ORC, V1/V2 | +| Drop identity field | Covered | Covered | Covered | Covered | Covered | Covered | Covered | Parquet/ORC, V1/V2 | +| Add/drop bucket | Covered | Covered by source-column join | Covered | Covered | Covered | Covered | Covered | Parquet/ORC, V1/V2 | +| Replace bucket with truncate | Covered | Result contract; transform is not RF-prunable | Covered | Covered by delete timeline | Covered by delete timeline | Covered by delete timeline | Covered | Parquet/ORC, V1/V2 | +| Year → month → day → hour | Covered | Covered for supported temporal transforms | Covered | Day → month covered | Day → month covered | Day → month covered | Covered | Parquet/ORC, V1/V2 | Review Comment: **[P1] Narrow or implement the format/reader cross-product** The only positive `bucket -> truncate` timeline is the Parquet table in `filter_refs`, and the only complete `year -> month -> day -> hour` timeline is its ORC table; neither toggles `enable_file_scanner_v2`. The other format/scanner loops exercise different transforms, so these two rows cannot claim blanket `Parquet/ORC, V1/V2` coverage. Run both full timelines under each format/scanner or record the exact exercised combinations. ########## regression-test/suites/external_table_p0/paimon/test_paimon_partition_schema_filter_refs.groovy: ########## @@ -0,0 +1,290 @@ +// 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 org.apache.doris.regression.action.ProfileAction + +suite("test_paimon_partition_schema_filter_refs", + "p0,external,paimon,external_docker,external_docker_paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_partition_schema_filter_refs" + String dbName = "paimon_partition_schema_filter_refs_db" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + return spark_paimon(""" + select snapshot_id + from paimon.${dbName}.`${tableName}\$snapshots` + order by snapshot_id desc + limit 1 + """)[0][0].toString() + } + def createTag = { String tableName, String tagName -> + spark_paimon """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}' + ) + """ + } + def profileAction = new ProfileAction(context) + def profileCounterValues = { String profileText, String counterName -> + def values = [] + def matcher = profileText =~ ("(?m)^\\s*(?:-\\s*)?" + + java.util.regex.Pattern.quote(counterName) + ":\\s+([^\\n]+)") + while (matcher.find()) { + def number = matcher.group(1).toString() =~ /([0-9,]+)/ + if (number.find()) { + values.add(Long.parseLong(number.group(1).replace(",", ""))) + } + } + return values + } + def assertRuntimeFilterPruned = { String tableName, String dimensionTable, + List<List<String>> expectedRows -> + String token = UUID.randomUUID().toString() + assertEquals(expectedRows, stringRows(""" + select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */ + '${token}', f.id + from ${tableName} f join ${dimensionTable} d on f.part = d.part + order by f.id + """).collect { row -> [row[1]] }) + // Scanner counters can arrive after the profile list first reports COMPLETE. + String profile = profileAction.getProfileBySql( + token, + ["RuntimeFilterPartitionPrunedRangeNum"], + 30000L, + 500L) + long fileRangesPruned = profileCounterValues( + profile, "RuntimeFilterPartitionPrunedRangeNum").sum(0L) + long partitionsPruned = profileCounterValues( + profile, "PartitionsPrunedByRuntimeFilter").sum(0L) + assertTrue(fileRangesPruned + partitionsPruned > 0L, + "Runtime filter did not prune any Paimon partition/file range; " + + profile.take(2000).replaceAll("\\s+", " ")) + } + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + 'type'='paimon', + 'warehouse'='s3://warehouse/wh', + 's3.endpoint'='http://${externalEnvIp}:${minioPort}', + 's3.access_key'='admin', + 's3.secret_key'='password', + 's3.path.style.access'='true', + 'meta.cache.paimon.table.ttl-second'='0' + ) + """ + + try { + ["parquet", "orc"].each { String format -> + String tableName = "partition_schema_${format}" + String dimensionTable = "${tableName}_dimension" + spark_paimon_multi """ + create database if not exists paimon.${dbName}; + drop table if exists paimon.${dbName}.${tableName}; + create table paimon.${dbName}.${tableName} ( + id int, + part string, + payload struct<metric:int, label:string>, + attrs map<string, struct<code:int>>, + events array<struct<score:int>> + ) using paimon + partitioned by (part) + tblproperties ('file.format'='${format}'); + insert into paimon.${dbName}.${tableName} values + (1, 'p1', named_struct('metric', 10, 'label', 'base-1'), + map('k', named_struct('code', 100)), + array(named_struct('score', 1000))), + (2, 'p2', named_struct('metric', 20, 'label', 'base-2'), + map('k', named_struct('code', 200)), + array(named_struct('score', 2000))); + """ + String baseSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_base") + spark_paimon """ + call paimon.sys.create_branch( + '${dbName}.${tableName}', + '${tableName}_base_branch', + '${tableName}_base' + ) + """ + + // Scenario PM-PE01: Paimon keeps a fixed partition key while STRUCT, MAP-value + // STRUCT and ARRAY-element STRUCT children are added. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} add column payload.extra string; + alter table paimon.${dbName}.${tableName} add column attrs.value.extra int; + alter table paimon.${dbName}.${tableName} add column events.element.extra int; + insert into paimon.${dbName}.${tableName} values + (3, 'p1', + named_struct('metric', 30, 'label', 'add-3', 'extra', 'payload-extra'), + map('k', named_struct('code', 300, 'extra', 301)), + array(named_struct('score', 3000, 'extra', 3001))), + (4, 'p3', + named_struct('metric', 40, 'label', 'add-4', 'extra', 'payload-extra'), + map('k', named_struct('code', 400, 'extra', 401)), + array(named_struct('score', 4000, 'extra', 4001))); + """ + String addedSnapshot = latestSnapshotId(tableName) + + // Scenario PM-PE02: rename and promote complex children without changing partition + // identity. Old snapshot/tag/branch schemas must remain independently readable. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} + rename column payload.label to renamed_label; + alter table paimon.${dbName}.${tableName} + rename column attrs.value.code to renamed_code; Review Comment: **[P1] Project the evolved MAP and ARRAY children through Doris** `attrs.value.code` and `events.element.score` are renamed here (and `.extra` was added earlier), but every Doris query below projects only `id`, `part`, or `payload.*`. The suite can therefore pass even if Doris cannot bind or decode either evolved MAP/ARRAY child. Add current and historical projections for the old/new names and extras, including negative binding at the wrong snapshot. ########## regression-test/suites/external_table_p0/iceberg_paimon_schema_time_travel_coverage.md: ########## @@ -11,6 +11,47 @@ The tests use explicit old/new field projections and negative bindings in addition to row-shape assertions. This prevents a rename with unchanged values from passing accidentally. +The partition-evolution extension also distinguishes result correctness from +physical pruning. Static predicates are validated across files written by +different partition specs, while runtime-filter cases require both equal +results with pruning disabled/enabled and a positive pruning counter. + +## Partition evolution matrix + +### Iceberg + +| Partition operation / transform | Static filter | Runtime filter | Snapshot/tag/branch | Position delete | Equality delete | Deletion vector | Complex schema in same timeline | Format / reader | +| --- | --- | --- | --- | --- | --- | --- | --- | --- | +| Add identity field | Covered | Covered | Covered | Covered | Covered | Covered | Covered | Parquet/ORC, V1/V2 | +| Drop identity field | Covered | Covered | Covered | Covered | Covered | Covered | Covered | Parquet/ORC, V1/V2 | +| Add/drop bucket | Covered | Covered by source-column join | Covered | Covered | Covered | Covered | Covered | Parquet/ORC, V1/V2 | +| Replace bucket with truncate | Covered | Result contract; transform is not RF-prunable | Covered | Covered by delete timeline | Covered by delete timeline | Covered by delete timeline | Covered | Parquet/ORC, V1/V2 | Review Comment: **[P1] Do not mark the bucket-to-truncate delete cells covered** The only `bucket(8, id) -> truncate(2, code)` replacement is in the filter/refs suite, which creates no deletes. The position/DV and equality-delete timelines replace only day with month (the latter later drops bucket), so all three “Covered by delete timeline” cells here are unimplemented. Add the transform to those delete fixtures or mark these cells accurately. ########## regression-test/suites/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.groovy: ########## @@ -0,0 +1,218 @@ +// 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_paimon_partition_pk_delete_refs", + "p0,external,paimon,external_docker,external_docker_paimon") { + String enabled = context.config.otherConfigs.get("enablePaimonTest") + if (enabled == null || !enabled.equalsIgnoreCase("true")) { + logger.info("disable paimon test") + return + } + + String minioPort = context.config.otherConfigs.get("iceberg_minio_port") + String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") + String catalogName = "test_paimon_partition_pk_delete_refs" + String dbName = "paimon_partition_pk_delete_refs_db" + + def stringRows = { String query -> + sql(query).collect { row -> + row.collect { value -> value == null ? null : value.toString() } + } + } + def latestSnapshotId = { String tableName -> + return spark_paimon(""" + select snapshot_id + from paimon.${dbName}.`${tableName}\$snapshots` + order by snapshot_id desc + limit 1 + """)[0][0].toString() + } + def createTag = { String tableName, String tagName -> + spark_paimon """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}' + ) + """ + } + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + 'type'='paimon', + 'warehouse'='s3://warehouse/wh', + 's3.endpoint'='http://${externalEnvIp}:${minioPort}', + 's3.access_key'='admin', + 's3.secret_key'='password', + 's3.path.style.access'='true', + 'meta.cache.paimon.table.ttl-second'='0' + ) + """ + + try { + ["parquet", "orc"].each { String format -> + String tableName = "partition_pk_dv_${format}" + String dimensionTable = "${tableName}_dimension" + spark_paimon_multi """ + create database if not exists paimon.${dbName}; + drop table if exists paimon.${dbName}.${tableName}; + create table paimon.${dbName}.${tableName} ( + id int not null, + part string not null, + old_name string, + note string, + payload struct<metric:int, label:string> + ) using paimon + partitioned by (part) + tblproperties ( + 'bucket'='1', + 'primary-key'='part,id', + 'file.format'='${format}', + 'deletion-vectors.enabled'='true' + ); + insert into paimon.${dbName}.${tableName} values + (1, 'p1', 'alpha', 'old-note-1', + named_struct('metric', 10, 'label', 'base-1')), + (2, 'p1', 'beta', 'old-note-2', + named_struct('metric', 20, 'label', 'base-delete')), + (3, 'p2', 'gamma', 'old-note-3', + named_struct('metric', 30, 'label', 'base-p2')); + """ + String baseSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_base") + + // Scenario PM-D01: add/rename fields, upsert one PK and delete another inside p1. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} add column payload.extra string; + alter table paimon.${dbName}.${tableName} rename column old_name to full_name; + insert into paimon.${dbName}.${tableName} + (id, part, full_name, note, payload) values + (1, 'p1', 'alpha-updated', 'new-note-1', + named_struct('metric', 11, 'label', 'updated-1', 'extra', 'extra-1')), + (4, 'p1', 'delta', 'delete-later', + named_struct('metric', 40, 'label', 'insert-4', 'extra', 'extra-4')); + delete from paimon.${dbName}.${tableName} + where part = 'p1' and id = 2; + """ + String firstDeleteSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_first_delete") + + // Scenario PM-D02: nested rename/type promotion and drop/re-add combine with another + // delete, insert and full compaction while the partition key stays fixed. + spark_paimon_multi """ + alter table paimon.${dbName}.${tableName} + rename column payload.label to renamed_label; + alter table paimon.${dbName}.${tableName} + alter column payload.metric type bigint; + alter table paimon.${dbName}.${tableName} drop column note; + alter table paimon.${dbName}.${tableName} add column note bigint; + delete from paimon.${dbName}.${tableName} + where part = 'p1' and id = 4; + insert into paimon.${dbName}.${tableName} + (id, part, full_name, payload, note) values + (5, 'p1', 'epsilon', + named_struct('metric', 5000000000, + 'renamed_label', 'insert-5', 'extra', 'extra-5'), + 5000); + call paimon.sys.compact( + table => '${dbName}.${tableName}', + compact_strategy => 'full' + ); + drop table if exists paimon.${dbName}.${dimensionTable}; + create table paimon.${dbName}.${dimensionTable} (part string) + using paimon tblproperties ('file.format'='parquet'); + insert into paimon.${dbName}.${dimensionTable} values ('p1'); + """ + String finalSnapshot = latestSnapshotId(tableName) + createTag(tableName, "${tableName}_final") + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${tableName}""" + + // Scenario PM-D03: static partition filters apply PK upserts, deletes and DV state. + List<List<String>> expectedCurrent = [["1", "alpha-updated"], ["5", "epsilon"]] + assertEquals(expectedCurrent, stringRows(""" + select id, full_name from ${tableName} + where part = 'p1' order by id + """)) + assertEquals([["5", "5000000000", "5000"]], stringRows(""" + select id, payload.metric, note from ${tableName} + where part = 'p1' and payload.metric > 1000000000 + order by id + """)) + + // Scenario PM-D04: runtime-filter pruning on the partition key remains delete-aware. + String rfQuery = """ + select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */ + f.id, f.full_name + from ${tableName} f + join ${dimensionTable} d on f.part = d.part + order by f.id + """ + sql """set runtime_filter_wait_infinitely=true""" + sql """set disable_join_reorder=true""" + sql """set enable_runtime_filter_partition_prune=false""" + assertEquals(expectedCurrent, stringRows(rfQuery)) + sql """set enable_runtime_filter_partition_prune=true""" + assertEquals(expectedCurrent, stringRows(rfQuery)) + + // Scenario PM-D05: numeric snapshots and tags preserve the matching schema/delete set. + assertEquals([["1", "alpha"], ["2", "beta"]], stringRows(""" + select id, old_name + from ${tableName} for version as of ${baseSnapshot} + where part = 'p1' order by id + """)) + assertEquals([["1", "alpha"], ["2", "beta"]], stringRows(""" + select id, old_name + from ${tableName}@tag(${tableName}_base) + where part = 'p1' order by id + """)) + assertEquals([["1", "alpha-updated"], ["4", "delta"]], stringRows(""" + select id, full_name + from ${tableName} for version as of ${firstDeleteSnapshot} + where part = 'p1' order by id + """)) + assertEquals(expectedCurrent, stringRows(""" + select id, full_name + from ${tableName}@tag(${tableName}_final) + where part = 'p1' order by id + """)) + + // Scenario PM-D06: JNI/native readers agree after partitioned PK deletes and compaction. + sql """set enable_paimon_cpp_reader=false""" + sql """set force_jni_scanner=true""" + List<List<String>> jniRows = stringRows(""" + select id, part, full_name from ${tableName} + where part in ('p1', 'p2') order by part, id + """) + sql """set force_jni_scanner=false""" Review Comment: **[P1] Assert the reader and DV path instead of allowing CPP fallback** With `force_jni_scanner=false`, Doris uses native only if `convertToRawFiles()` succeeds; otherwise this `enable_paimon_cpp_reader=true` setting routes the split to CPP, so row equality does not prove the advertised native path. The query is also current-only, does not project the nested/drop-readded fields, and the fixture never proves a physical DV exists. Assert nonzero native split stats plus a DV artifact, and compare those evolved fields and a historical ref under both reader settings (the sibling schema suite has the same current-only ambiguity). -- 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]
