Gabriel39 commented on code in PR #65992: URL: https://github.com/apache/doris/pull/65992#discussion_r3644273461
########## 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: 已在 profile 前显式设置 enable_runtime_filter_prune=false，并要求 RuntimeFilterPartitionPrunedRangeNum 与 PartitionsPrunedByRuntimeFilter 的合计大于 0。 ########## 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: 已将四个 profile 依赖 suite 标为 nonConcurrent，并在文档中说明其他独立 suite 仍可并发执行。 ########## 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: 已将稳定结果断言全部改为 qt/order_qt 风格，并提交 regression runner 生成的对应 .out 基线。 ########## 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: 已拆分可归因路径：category drop 与新增 identity region 均独立 profile 且要求正计数；bucket source 与 temporal 明确仅做结果契约，覆盖文档不再声称这些 transform 可物理 RF pruning。 ########## 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: 已显式设置 enable_runtime_filter_prune=false，并以 profile 的正 pruning counter 证明 Paimon PK 分区 RF 生效。 ########## 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: 已新增 payload.extra、payload.renamed_label 和 drop/re-add note 的 current/history 投影；旧存活 row 1 对新 BIGINT note 的值明确断言为 NULL。 ########## 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: 已新增 test_iceberg_partition_evolution_format_scanner.groovy，在 Parquet/ORC × scanner V1/V2 下运行完整 identity/bucket-to-truncate/drop 与 year-to-month-to-day-to-hour 的 current/tag timeline；文档也同步收窄 RF 与并发声明。 ########## 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: 已在 equality delete 和 position/DV timeline 中加入 bucket(8,id) 到 truncate(2,code) 的 replacement，并新增完整 format/reader timeline。 ########## 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: 已用 explain 中 paimonNativeReadSplits 与 SplitStat 精确验证 JNI/native 路径；通过 $files.deleteRowCount 证明物理 DV，并在两种 reader 下比较 current evolved fields 与 historical ref；sibling schema suite 也同步补齐。 ########## 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: 已新增 current/history 的 STRUCT、MAP、ARRAY 演进字段投影及 wrong-snapshot 负向绑定，并在 JNI/native 两种 reader 下验证。 ########## 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: 已增加一个 current predicate 跨越 year/month/day/hour 四种 spec；新的 format/scanner suite 也在 Parquet/ORC 下覆盖同一完整跨度。 ########## 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: 已将 rename、type change、drop 三类 Paimon partition mutation 收紧为实际稳定错误信息，并继续校验 schema count、current data 与历史 tag 的原子性。 ########## 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: 已补充 drop category 后在同一新 spec 写入 victim/survivor，再删除 victim，并分别断言当前 data/delete 的 spec_id 一致。 -- 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]
