This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7809a73814e [regression](test) expand Iceberg and Paimon partition
evolution coverage (#65992)
7809a73814e is described below
commit 7809a73814eb5f6e7b0b888290db9fbfd25046b6
Author: Gabriel <[email protected]>
AuthorDate: Fri Jul 24 19:55:51 2026 +0800
[regression](test) expand Iceberg and Paimon partition evolution coverage
(#65992)
## What changed
- add five Iceberg P0 suites for partition evolution with static/runtime
filters, snapshot/tag/branch reads, position deletes, equality deletes,
deletion vectors, and a complete format/scanner transform timeline
- add three Paimon P0 suites for fixed-partition complex schema
evolution, partitioned primary-key delete/upsert/compaction, and atomic
rejection of unsupported partition-key mutations
- keep independent dimensions in separate Groovy suites for parallel
execution, while marking four profile-dependent suites `nonConcurrent`
- add deterministic runner-generated `.out` files and update the
Iceberg/Paimon schema evolution and time-travel coverage matrix
## Coverage highlights
- Iceberg add/drop/replace partition fields and identity, bucket,
truncate, year/month/day/hour transforms
- complex STRUCT/MAP/ARRAY changes combined with partition filtering and
historical references
- runtime-filter pruning with attributable positive profile-counter
assertions on supported identity partition paths
- Parquet/ORC, file scanner V1/V2, Iceberg v2 position/equality deletes,
and v3 deletion vectors
- Paimon Parquet/ORC, asserted JNI/native split paths, physical DV
metadata, primary-key upsert/delete/compaction, snapshot/tag/branch
- topology-independent execution on a distributed Doris cluster
## Validation
- master: `f1460f89230441bc1b6b1872d66bd32a526b25b1`
- one FE and two live BEs
- eight partition-evolution extension suites: 8 passed, 0 failed, 0
fatal, 0 skipped
- documented P0 matrix: 20 suites, including 8 partition-evolution
extensions
- no Doris production code changed
- no additional Doris product issue was reproduced
---
...iceberg_partition_evolution_equality_delete.out | 36 ++
...est_iceberg_partition_evolution_filter_refs.out | 71 ++++
..._iceberg_partition_evolution_format_scanner.out | 153 +++++++++
...est_iceberg_partition_evolution_position_dv.out | 201 +++++++++++
..._iceberg_partition_evolution_runtime_filter.out | 44 +++
.../test_paimon_partition_mutation_atomicity.out | 14 +
.../test_paimon_partition_pk_delete_refs.out | 109 ++++++
.../test_paimon_partition_schema_filter_refs.out | 145 ++++++++
...berg_partition_evolution_equality_delete.groovy | 302 +++++++++++++++++
..._iceberg_partition_evolution_filter_refs.groovy | 275 ++++++++++++++++
...eberg_partition_evolution_format_scanner.groovy | 211 ++++++++++++
..._iceberg_partition_evolution_position_dv.groovy | 318 ++++++++++++++++++
...eberg_partition_evolution_runtime_filter.groovy | 277 ++++++++++++++++
.../iceberg_paimon_schema_time_travel_coverage.md | 85 ++++-
...test_paimon_partition_mutation_atomicity.groovy | 148 +++++++++
.../test_paimon_partition_pk_delete_refs.groovy | 346 +++++++++++++++++++
...test_paimon_partition_schema_filter_refs.groovy | 366 +++++++++++++++++++++
17 files changed, 3089 insertions(+), 12 deletions(-)
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_equality_delete.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_equality_delete.out
new file mode 100644
index 00000000000..5c4002a33f0
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_equality_delete.out
@@ -0,0 +1,36 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !current_filter --
+1
+5
+7
+
+-- !current_nested --
+7 7000000000
+
+-- !base_tag --
+1
+2
+
+-- !first_delete_snapshot --
+1
+
+-- !added_snapshot --
+1
+5
+6
+
+-- !final_tag --
+1
+5
+7
+
+-- !scanner_v2 --
+1
+5
+7
+
+-- !scanner_v1 --
+1
+5
+7
+
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.out
new file mode 100644
index 00000000000..e4a7f2ca565
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.out
@@ -0,0 +1,71 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !current_identity_filter --
+1
+3
+5
+7
+
+-- !current_truncate_source_filter --
+1
+3
+5
+7
+
+-- !current_temporal_source_filter --
+5
+6
+7
+8
+
+-- !current_readded_nested_field --
+7 7000
+8 8000
+
+-- !identity_base_snapshot --
+1 A
+2 B
+
+-- !identity_base_tag --
+1 A
+2 B
+
+-- !identity_base_branch --
+1 A
+2 B
+
+-- !identity_added_snapshot --
+1
+3
+
+-- !identity_replaced_snapshot --
+1
+3
+5
+
+-- !identity_dropped_tag --
+1
+3
+5
+7
+
+-- !temporal_year_tag --
+11
+12
+
+-- !temporal_current_all_specs --
+11
+12
+13
+14
+15
+16
+17
+18
+
+-- !temporal_hour_snapshot --
+17
+18
+
+-- !temporal_hour_tag --
+18
+
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_format_scanner.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_format_scanner.out
new file mode 100644
index 00000000000..c8cd07156d5
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_format_scanner.out
@@ -0,0 +1,153 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !identity_parquet_v2_current --
+1 \N
+2 \N
+3 \N
+4 \N
+5 \N
+6 \N
+7 7000
+8 8000
+
+-- !identity_parquet_v2_base_tag --
+1 A base-a
+2 B base-b
+
+-- !identity_parquet_v2_final_tag --
+1 base-a
+3 bucket-a
+5 truncate-a
+7 drop-a
+
+-- !temporal_parquet_v2_current --
+11
+12
+13
+14
+15
+16
+17
+18
+
+-- !temporal_parquet_v2_final_tag --
+13
+14
+15
+16
+17
+18
+
+-- !identity_parquet_v1_current --
+1 \N
+2 \N
+3 \N
+4 \N
+5 \N
+6 \N
+7 7000
+8 8000
+
+-- !identity_parquet_v1_base_tag --
+1 A base-a
+2 B base-b
+
+-- !identity_parquet_v1_final_tag --
+1 base-a
+3 bucket-a
+5 truncate-a
+7 drop-a
+
+-- !temporal_parquet_v1_current --
+11
+12
+13
+14
+15
+16
+17
+18
+
+-- !temporal_parquet_v1_final_tag --
+13
+14
+15
+16
+17
+18
+
+-- !identity_orc_v2_current --
+1 \N
+2 \N
+3 \N
+4 \N
+5 \N
+6 \N
+7 7000
+8 8000
+
+-- !identity_orc_v2_base_tag --
+1 A base-a
+2 B base-b
+
+-- !identity_orc_v2_final_tag --
+1 base-a
+3 bucket-a
+5 truncate-a
+7 drop-a
+
+-- !temporal_orc_v2_current --
+11
+12
+13
+14
+15
+16
+17
+18
+
+-- !temporal_orc_v2_final_tag --
+13
+14
+15
+16
+17
+18
+
+-- !identity_orc_v1_current --
+1 \N
+2 \N
+3 \N
+4 \N
+5 \N
+6 \N
+7 7000
+8 8000
+
+-- !identity_orc_v1_base_tag --
+1 A base-a
+2 B base-b
+
+-- !identity_orc_v1_final_tag --
+1 base-a
+3 bucket-a
+5 truncate-a
+7 drop-a
+
+-- !temporal_orc_v1_current --
+11
+12
+13
+14
+15
+16
+17
+18
+
+-- !temporal_orc_v1_final_tag --
+13
+14
+15
+16
+17
+18
+
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.out
new file mode 100644
index 00000000000..bba7dbde155
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.out
@@ -0,0 +1,201 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !position_parquet_current_filter --
+1
+5
+7
+10
+
+-- !position_parquet_current_nested --
+7 7000000000
+
+-- !position_parquet_rf_disabled --
+1
+5
+7
+10
+
+-- !position_parquet_base_snapshot --
+1
+2
+3
+4
+
+-- !position_parquet_base_tag --
+1
+2
+
+-- !position_parquet_first_delete_snapshot --
+1
+
+-- !position_parquet_added_delete_snapshot --
+1
+5
+
+-- !position_parquet_final_tag --
+1
+5
+7
+10
+
+-- !position_parquet_scanner_v2 --
+1
+5
+7
+10
+
+-- !position_parquet_scanner_v1 --
+1
+5
+7
+10
+
+-- !dv_parquet_current_filter --
+1
+5
+7
+10
+
+-- !dv_parquet_current_nested --
+7 7000000000
+
+-- !dv_parquet_rf_disabled --
+1
+5
+7
+10
+
+-- !dv_parquet_base_snapshot --
+1
+2
+3
+4
+
+-- !dv_parquet_base_tag --
+1
+2
+
+-- !dv_parquet_first_delete_snapshot --
+1
+
+-- !dv_parquet_added_delete_snapshot --
+1
+5
+
+-- !dv_parquet_final_tag --
+1
+5
+7
+10
+
+-- !dv_parquet_scanner_v2 --
+1
+5
+7
+10
+
+-- !dv_parquet_scanner_v1 --
+1
+5
+7
+10
+
+-- !position_orc_current_filter --
+1
+5
+7
+10
+
+-- !position_orc_current_nested --
+7 7000000000
+
+-- !position_orc_rf_disabled --
+1
+5
+7
+10
+
+-- !position_orc_base_snapshot --
+1
+2
+3
+4
+
+-- !position_orc_base_tag --
+1
+2
+
+-- !position_orc_first_delete_snapshot --
+1
+
+-- !position_orc_added_delete_snapshot --
+1
+5
+
+-- !position_orc_final_tag --
+1
+5
+7
+10
+
+-- !position_orc_scanner_v2 --
+1
+5
+7
+10
+
+-- !position_orc_scanner_v1 --
+1
+5
+7
+10
+
+-- !dv_orc_current_filter --
+1
+5
+7
+10
+
+-- !dv_orc_current_nested --
+7 7000000000
+
+-- !dv_orc_rf_disabled --
+1
+5
+7
+10
+
+-- !dv_orc_base_snapshot --
+1
+2
+3
+4
+
+-- !dv_orc_base_tag --
+1
+2
+
+-- !dv_orc_first_delete_snapshot --
+1
+
+-- !dv_orc_added_delete_snapshot --
+1
+5
+
+-- !dv_orc_final_tag --
+1
+5
+7
+10
+
+-- !dv_orc_scanner_v2 --
+1
+5
+7
+10
+
+-- !dv_orc_scanner_v1 --
+1
+5
+7
+10
+
diff --git
a/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.out
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.out
new file mode 100644
index 00000000000..78518f51a42
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.out
@@ -0,0 +1,44 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !category_rf_disabled --
+1
+4
+6
+8
+
+-- !added_identity_rf_disabled --
+1
+4
+6
+8
+
+-- !bucket_source_rf_disabled --
+4
+
+-- !temporal_rf_disabled --
+6
+7
+8
+9
+
+-- !bucket_source_rf_enabled --
+4
+
+-- !temporal_rf_enabled --
+6
+7
+8
+9
+
+-- !base_snapshot_rf --
+1
+
+-- !added_snapshot_rf --
+1
+4
+
+-- !dropped_tag_rf --
+1
+4
+6
+8
+
diff --git
a/regression-test/data/external_table_p0/paimon/test_paimon_partition_mutation_atomicity.out
b/regression-test/data/external_table_p0/paimon/test_paimon_partition_mutation_atomicity.out
new file mode 100644
index 00000000000..e1fcf8c5234
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/test_paimon_partition_mutation_atomicity.out
@@ -0,0 +1,14 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !current_partition_filter --
+1
+2
+4
+
+-- !current_added_payload --
+4 added
+
+-- !base_tag_partition_filter --
+1 A 1 base-a1
+2 A 2 base-a2
+3 B 1 base-b1
+
diff --git
a/regression-test/data/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.out
b/regression-test/data/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.out
new file mode 100644
index 00000000000..0d8ef36d633
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.out
@@ -0,0 +1,109 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !parquet_current_partition_filter --
+1 alpha-updated
+5 epsilon
+
+-- !parquet_current_promoted_and_readded --
+5 5000000000 5000
+
+-- !parquet_current_evolved_fields --
+1 updated-1 extra-1 \N
+5 insert-5 extra-5 5000
+
+-- !parquet_pre_dv_old_row_null --
+1 updated-1 extra-1 \N
+
+-- !parquet_rf_disabled --
+1 alpha-updated
+5 epsilon
+
+-- !parquet_base_snapshot --
+1 alpha old-note-1 base-1
+2 beta old-note-2 base-delete
+
+-- !parquet_base_tag --
+1 alpha old-note-1 base-1
+2 beta old-note-2 base-delete
+
+-- !parquet_first_delete_snapshot --
+1 alpha-updated new-note-1 updated-1 extra-1
+4 delta delete-later insert-4 extra-4
+
+-- !parquet_final_tag --
+1 alpha-updated \N updated-1 extra-1
+5 epsilon 5000 insert-5 extra-5
+
+-- !parquet_jni_current_dv --
+1 p1 alpha-updated \N updated-1 extra-1
+5 p1 epsilon 5000 insert-5 extra-5
+3 p2 gamma \N base-p2 \N
+
+-- !parquet_jni_historical --
+1 p1 alpha old-note-1 base-1
+2 p1 beta old-note-2 base-delete
+3 p2 gamma old-note-3 base-p2
+
+-- !parquet_native_current_dv --
+1 p1 alpha-updated \N updated-1 extra-1
+5 p1 epsilon 5000 insert-5 extra-5
+3 p2 gamma \N base-p2 \N
+
+-- !parquet_native_historical --
+1 p1 alpha old-note-1 base-1
+2 p1 beta old-note-2 base-delete
+3 p2 gamma old-note-3 base-p2
+
+-- !orc_current_partition_filter --
+1 alpha-updated
+5 epsilon
+
+-- !orc_current_promoted_and_readded --
+5 5000000000 5000
+
+-- !orc_current_evolved_fields --
+1 updated-1 extra-1 \N
+5 insert-5 extra-5 5000
+
+-- !orc_pre_dv_old_row_null --
+1 updated-1 extra-1 \N
+
+-- !orc_rf_disabled --
+1 alpha-updated
+5 epsilon
+
+-- !orc_base_snapshot --
+1 alpha old-note-1 base-1
+2 beta old-note-2 base-delete
+
+-- !orc_base_tag --
+1 alpha old-note-1 base-1
+2 beta old-note-2 base-delete
+
+-- !orc_first_delete_snapshot --
+1 alpha-updated new-note-1 updated-1 extra-1
+4 delta delete-later insert-4 extra-4
+
+-- !orc_final_tag --
+1 alpha-updated \N updated-1 extra-1
+5 epsilon 5000 insert-5 extra-5
+
+-- !orc_jni_current_dv --
+1 p1 alpha-updated \N updated-1 extra-1
+5 p1 epsilon 5000 insert-5 extra-5
+3 p2 gamma \N base-p2 \N
+
+-- !orc_jni_historical --
+1 p1 alpha old-note-1 base-1
+2 p1 beta old-note-2 base-delete
+3 p2 gamma old-note-3 base-p2
+
+-- !orc_native_current_dv --
+1 p1 alpha-updated \N updated-1 extra-1
+5 p1 epsilon 5000 insert-5 extra-5
+3 p2 gamma \N base-p2 \N
+
+-- !orc_native_historical --
+1 p1 alpha old-note-1 base-1
+2 p1 beta old-note-2 base-delete
+3 p2 gamma old-note-3 base-p2
+
diff --git
a/regression-test/data/external_table_p0/paimon/test_paimon_partition_schema_filter_refs.out
b/regression-test/data/external_table_p0/paimon/test_paimon_partition_schema_filter_refs.out
new file mode 100644
index 00000000000..ad3c6a41095
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/test_paimon_partition_schema_filter_refs.out
@@ -0,0 +1,145 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !parquet_current_partition_filter --
+1
+3
+5
+7
+
+-- !parquet_current_promoted_metric --
+5 5000000000
+
+-- !parquet_current_readded_payload --
+7 7000
+
+-- !parquet_current_all_complex_children --
+1 base-1 \N 100 \N 1000 \N
+3 add-3 \N 300 301 3000 3001
+5 rename-5 \N 500 501 5000 5001
+7 readd-7 7000 700 701 7000 7001
+
+-- !parquet_rf_disabled --
+1
+3
+5
+7
+
+-- !parquet_base_snapshot --
+1 p1 base-1
+2 p2 base-2
+
+-- !parquet_base_tag --
+1 p1 base-1
+2 p2 base-2
+
+-- !parquet_base_branch --
+1 p1 base-1
+2 p2 base-2
+
+-- !parquet_added_snapshot_complex --
+1 base-1 100 \N 1000 \N
+3 add-3 300 301 3000 3001
+
+-- !parquet_final_tag --
+1
+3
+5
+7
+
+-- !parquet_base_snapshot_complex --
+1 100 1000
+2 200 2000
+
+-- !parquet_jni_current_complex --
+1 base-1 \N 100 1000
+3 add-3 \N 300 3000
+4 add-4 \N 400 4000
+5 rename-5 \N 500 5000
+7 readd-7 7000 700 7000
+
+-- !parquet_jni_historical_complex --
+1 base-1 100 1000
+2 base-2 200 2000
+
+-- !parquet_native_current_complex --
+1 base-1 \N 100 1000
+3 add-3 \N 300 3000
+4 add-4 \N 400 4000
+5 rename-5 \N 500 5000
+7 readd-7 7000 700 7000
+
+-- !parquet_native_historical_complex --
+1 base-1 100 1000
+2 base-2 200 2000
+
+-- !orc_current_partition_filter --
+1
+3
+5
+7
+
+-- !orc_current_promoted_metric --
+5 5000000000
+
+-- !orc_current_readded_payload --
+7 7000
+
+-- !orc_current_all_complex_children --
+1 base-1 \N 100 \N 1000 \N
+3 add-3 \N 300 301 3000 3001
+5 rename-5 \N 500 501 5000 5001
+7 readd-7 7000 700 701 7000 7001
+
+-- !orc_rf_disabled --
+1
+3
+5
+7
+
+-- !orc_base_snapshot --
+1 p1 base-1
+2 p2 base-2
+
+-- !orc_base_tag --
+1 p1 base-1
+2 p2 base-2
+
+-- !orc_base_branch --
+1 p1 base-1
+2 p2 base-2
+
+-- !orc_added_snapshot_complex --
+1 base-1 100 \N 1000 \N
+3 add-3 300 301 3000 3001
+
+-- !orc_final_tag --
+1
+3
+5
+7
+
+-- !orc_base_snapshot_complex --
+1 100 1000
+2 200 2000
+
+-- !orc_jni_current_complex --
+1 base-1 \N 100 1000
+3 add-3 \N 300 3000
+4 add-4 \N 400 4000
+5 rename-5 \N 500 5000
+7 readd-7 7000 700 7000
+
+-- !orc_jni_historical_complex --
+1 base-1 100 1000
+2 base-2 200 2000
+
+-- !orc_native_current_complex --
+1 base-1 \N 100 1000
+3 add-3 \N 300 3000
+4 add-4 \N 400 4000
+5 rename-5 \N 500 5000
+7 readd-7 7000 700 7000
+
+-- !orc_native_historical_complex --
+1 base-1 100 1000
+2 base-2 200 2000
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_equality_delete.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_equality_delete.groovy
new file mode 100644
index 00000000000..fab5dc7176d
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_equality_delete.groovy
@@ -0,0 +1,302 @@
+// 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_equality_delete",
+ "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_equality_delete"
+ String dbName = "iceberg_partition_evolution_equality_delete_db"
+ String tableName = "equality_delete_evolved"
+
+ def stringRows = { String query ->
+ sql(query).collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ }
+ }
+ def latestSnapshotId = {
+ // The Java helper commits through the REST catalog directly, so
invalidate Spark's
+ // cached table metadata before recording the snapshot used by
time-travel assertions.
+ spark_iceberg """refresh table demo.${dbName}.${tableName}"""
+ return spark_iceberg("""
+ select snapshot_id
+ from demo.${dbName}.${tableName}.snapshots
+ order by committed_at desc
+ limit 1
+ """)[0][0].toString()
+ }
+ def executeCommand = { String command, int timeoutSeconds = 300 ->
+ StringBuilder stdout = new StringBuilder()
+ StringBuilder stderr = new StringBuilder()
+ def process = new ProcessBuilder("/bin/bash", "-c", command).start()
+ process.consumeProcessOutput(stdout, stderr)
+ process.waitForOrKill(timeoutSeconds * 1000)
+ assertEquals(0, process.exitValue(),
+ "Command failed\nstdout:\n${stdout}\nstderr:\n${stderr}")
+ return stdout.toString()
+ }
+ String dockerCommand =
context.config.otherConfigs.get("externalDockerCommand") ?: "docker"
+ String sparkContainer =
context.config.otherConfigs.get("icebergSparkContainer")
+ if (sparkContainer == null || sparkContainer.isEmpty()) {
+ String containers = executeCommand(
+ "${dockerCommand} ps --format '{{.ID}}\t{{.Names}}'", 30)
+ def matches = []
+ containers.readLines().each { String line ->
+ String containerId = line.split(/\t/, 2)[0]
+ String probe = "${dockerCommand} exec ${containerId} bash -lc " +
+ "'test -f /mnt/SUCCESS && command -v spark-sql >/dev/null'"
+ try {
+ executeCommand(probe, 30)
+ matches.add(containerId)
+ } catch (Throwable ignored) {
+ // A shared external environment contains multiple services;
only Spark has the
+ // Iceberg jars needed to construct a real equality-delete
file.
+ }
+ }
+ assertEquals(1, matches.size(), "Expected exactly one usable Spark
Iceberg container")
+ sparkContainer = matches[0]
+ }
+ def runInSparkContainer = { String command ->
+ executeCommand("${dockerCommand} exec ${sparkContainer} bash -lc
'${command}'", 300)
+ }
+
+ String javaSource = '''
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.deletes.EqualityDeleteWriter;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.Parquet;
+
+public class AppendEvolutionEqualityDelete {
+ public static void main(String[] args) throws Exception {
+ Map<String, String> props = new HashMap<>();
+ props.put("type", "rest");
+ props.put("uri", "http://rest:8181";);
+ props.put("warehouse", "s3://warehouse/wh/");
+ props.put("io-impl", "org.apache.iceberg.aws.s3.S3FileIO");
+ props.put("s3.endpoint", "http://minio:9000";);
+ props.put("s3.path-style-access", "true");
+ props.put("s3.region", "us-east-1");
+ Catalog catalog = CatalogUtil.buildIcebergCatalog("demo", props, null);
+ Table table = catalog.loadTable(TableIdentifier.of(args[0], args[1]));
+ if (!table.spec().isUnpartitioned()) {
+ throw new IllegalStateException("Equality-delete checkpoints must
use an unpartitioned spec");
+ }
+ String fieldName = args[2];
+ Schema equalitySchema = table.schema().select(fieldName);
+ int fieldId = table.schema().findField(fieldName).fieldId();
+ OutputFile output = table.io().newOutputFile(
+ table.location() + "/data/evolution-equality-delete-" + args[3]
+ + "-" + System.currentTimeMillis() + ".parquet");
+ EqualityDeleteWriter<Record> writer = Parquet.writeDeletes(output)
+ .forTable(table)
+ .rowSchema(equalitySchema)
+ .withSpec(PartitionSpec.unpartitioned())
+ .createWriterFunc(GenericParquetWriter::create)
+ .equalityFieldIds(fieldId)
+ .overwrite()
+ .buildEqualityWriter();
+ GenericRecord record = GenericRecord.create(equalitySchema);
+ record.setField(fieldName, Integer.valueOf(args[3]));
+ writer.write(record);
+ writer.close();
+ DeleteFile deleteFile = writer.toDeleteFile();
+ table.newRowDelta().addDeletes(deleteFile).commit();
+ }
+}
+'''
+ String encodedJava = javaSource.getBytes("UTF-8").encodeBase64().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 {
+ runInSparkContainer(
+ "echo ${encodedJava} | base64 -d
>/tmp/AppendEvolutionEqualityDelete.java && "
+ + "javac -cp \"/opt/spark/jars/*\" "
+ + "/tmp/AppendEvolutionEqualityDelete.java")
+ 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,
+ code string,
+ event_time timestamp,
+ payload struct<metric:int, label:string>
+ ) using iceberg
+ tblproperties (
+ 'format-version'='2',
+ 'write.format.default'='parquet'
+ );
+ insert into demo.${dbName}.${tableName} values
+ (1, 'A', 'aa-1', timestamp '2026-01-01 01:00:00',
+ named_struct('metric', 10, 'label', 'base-a')),
+ (2, 'A', 'aa-2', timestamp '2026-01-01 02:00:00',
+ named_struct('metric', 20, 'label', 'base-delete')),
+ (3, 'B', 'bb-1', timestamp '2026-01-02 01:00:00',
+ named_struct('metric', 30, 'label', 'base-b')),
+ (4, 'C', 'cc-1', timestamp '2026-01-03 01:00:00',
+ named_struct('metric', 40, 'label', 'base-c'));
+ """
+ String baseSnapshot = latestSnapshotId()
+ sql """
+ alter table `${catalogName}`.`${dbName}`.`${tableName}`
+ create tag equality_base as of version ${baseSnapshot}
+ """
+
+ // Scenario PE-EQ01: an equality delete written under the original
unpartitioned spec
+ // must remain effective after later partition specs are added.
+ runInSparkContainer(
+ "java -cp \"/tmp:/opt/spark/jars/*\"
AppendEvolutionEqualityDelete "
+ + "${dbName} ${tableName} id 2")
+ String firstDeleteSnapshot = latestSnapshotId()
+
+ // Scenario PE-EQ02: ADD identity/day/bucket fields and a nested child
after the delete.
+ spark_iceberg_multi """
+ alter table demo.${dbName}.${tableName} add partition field
category;
+ alter table demo.${dbName}.${tableName} add partition field
days(event_time);
+ 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} values
+ (5, 'A', 'aa-3', timestamp '2026-02-01 01:00:00',
+ named_struct('metric', 50, 'label', 'added-a', 'extra',
'new-child')),
+ (6, 'A', 'aa-4', timestamp '2026-02-01 02:00:00',
+ named_struct('metric', 60, 'label', 'added-delete',
'extra', 'new-child'));
+ """
+ String addedSnapshot = latestSnapshotId()
+
+ // Scenario PE-EQ03: REPLACE day -> month while renaming/promoting
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}
+ replace partition field bucket(8, id) with truncate(2, code);
+ 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', 'aa-5', timestamp '2026-03-01 01:00:00',
+ named_struct('metric', 7000000000,
+ 'renamed_label', 'replace-a', 'extra',
'renamed-child')),
+ (8, 'A', 'aa-6', timestamp '2026-03-01 02:00:00',
+ named_struct('metric', 80,
+ 'renamed_label', 'replace-delete', 'extra',
'renamed-child'));
+ """
+
+ // Scenario PE-EQ04: return to an unpartitioned spec and write
equality deletes after
+ // evolution. This covers delete files both before and after the
multi-spec interval.
+ spark_iceberg_multi """
+ alter table demo.${dbName}.${tableName} drop partition field
category;
+ alter table demo.${dbName}.${tableName} drop partition field
months(event_time);
+ alter table demo.${dbName}.${tableName} drop partition field
truncate(2, code);
+ """
+ runInSparkContainer(
+ "java -cp \"/tmp:/opt/spark/jars/*\"
AppendEvolutionEqualityDelete "
+ + "${dbName} ${tableName} id 6")
+ runInSparkContainer(
+ "java -cp \"/tmp:/opt/spark/jars/*\"
AppendEvolutionEqualityDelete "
+ + "${dbName} ${tableName} id 8")
+ String finalSnapshot = latestSnapshotId()
+ sql """
+ alter table `${catalogName}`.`${dbName}`.`${tableName}`
+ create tag equality_final as of version ${finalSnapshot}
+ """
+
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+ sql """refresh catalog ${catalogName}"""
+
+ // Scenario PE-EQ05: partition-source filters honor equality deletes
across every spec.
+ qt_current_filter """
+ select id from ${tableName} where category = 'A' order by id
+ """
+ qt_current_nested """
+ select id, payload.metric from ${tableName}
+ where event_time >= timestamp '2026-03-01 00:00:00'
+ and payload.metric > 5000000000
+ order by id
+ """
+
+ // Scenario PE-EQ06: numeric snapshots and tags select the matching
delete/spec state.
+ qt_base_tag """
+ select id from ${tableName}@tag(equality_base)
+ where category = 'A' order by id
+ """
+ qt_first_delete_snapshot """
+ select id from ${tableName} for version as of
${firstDeleteSnapshot}
+ where category = 'A' order by id
+ """
+ qt_added_snapshot """
+ select id from ${tableName} for version as of ${addedSnapshot}
+ where category = 'A' order by id
+ """
+ qt_final_tag """
+ select id from ${tableName}@tag(equality_final)
+ where category = 'A' order by id
+ """
+
+ // Scenario PE-EQ07: both scanner implementations apply the same
equality deletes.
+ sql """set enable_file_scanner_v2=true"""
+ qt_scanner_v2 """
+ select id from ${tableName} where category = 'A' order by id
+ """
+ sql """set enable_file_scanner_v2=false"""
+ qt_scanner_v1 """
+ select id from ${tableName} where category = 'A' order by id
+ """
+
+ List<List<String>> equalityFiles = stringRows("""
+ select file_format from ${tableName}\$all_files
+ where content = 2 order by file_path
+ """)
+ assertEquals(3, equalityFiles.size(),
+ "${tableName} must contain all three equality-delete files")
+ } finally {
+ sql """set enable_file_scanner_v2=true"""
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.groovy
new file mode 100644
index 00000000000..bc31278706f
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_filter_refs.groovy
@@ -0,0 +1,275 @@
+// 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 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.
+ qt_current_identity_filter """
+ select id from ${identityTable} where category = 'A' order by id
+ """
+ qt_current_truncate_source_filter """
+ select id from ${identityTable}
+ where code in ('aa-1', 'aa-2', 'aa-3', 'aa-4')
+ order by id
+ """
+ qt_current_temporal_source_filter """
+ select id from ${identityTable}
+ where event_time >= timestamp '2026-03-01 00:00:00'
+ order by id
+ """
+ qt_current_readded_nested_field """
+ 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.
+ qt_identity_base_snapshot """
+ select id, category
+ from ${identityTable} for version as of ${identityBase}
+ where category in ('A', 'B')
+ order by id
+ """
+ qt_identity_base_tag """
+ select id, category from ${identityTable}@tag(identity_base)
+ where category in ('A', 'B') order by id
+ """
+ qt_identity_base_branch """
+ select id, category from
${identityTable}@branch(identity_base_branch)
+ where category in ('A', 'B') order by id
+ """
+ qt_identity_added_snapshot """
+ select id from ${identityTable} for version as of ${identityAdded}
+ where category = 'A' order by id
+ """
+ qt_identity_replaced_snapshot """
+ select id from ${identityTable} for version as of
${identityReplaced}
+ where category = 'A' order by id
+ """
+ qt_identity_dropped_tag """
+ 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.
+ qt_temporal_year_tag """
+ select id from ${temporalTable}@tag(temporal_year)
+ where event_time < timestamp '2026-01-01 00:00:00'
+ order by id
+ """
+ // This current predicate deliberately crosses files written by year,
month, day and hour
+ // specs so losing any older transform boundary changes the fixed
result contract.
+ qt_temporal_current_all_specs """
+ select id from ${temporalTable}
+ where event_time >= timestamp '2024-01-01 00:00:00'
+ and event_time < timestamp '2026-05-02 00:00:00'
+ order by id
+ """
+ qt_temporal_hour_snapshot """
+ select id from ${temporalTable} for version as of ${temporalHour}
+ where event_time >= timestamp '2026-05-01 00:00:00'
+ and event_time < timestamp '2026-05-02 00:00:00'
+ order by id
+ """
+ qt_temporal_hour_tag """
+ select id from ${temporalTable}@tag(temporal_hour)
+ where event_time = timestamp '2026-05-01 09:00:00'
+ """
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_format_scanner.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_format_scanner.groovy
new file mode 100644
index 00000000000..7b492b67f0e
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_format_scanner.groovy
@@ -0,0 +1,211 @@
+// 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_format_scanner",
+ "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_format_scanner"
+ String dbName = "iceberg_partition_evolution_format_scanner_db"
+
+ 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',
+ 'meta.cache.iceberg.schema.ttl-second'='0'
+ )
+ """
+
+ try {
+ ["parquet", "orc"].each { String format ->
+ String identityTable = "identity_bucket_truncate_${format}"
+ String temporalTable = "temporal_transform_${format}"
+
+ // Scenario PE-X01: run the complete identity + bucket -> truncate
+ drop timeline
+ // for each data format, including a nested
add/rename/promotion/drop-readd sequence.
+ 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'='${format}'
+ );
+ 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 ${identityTable}_base as of version ${identityBase}
+ """
+ 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', 'added')),
+ (4, 'C', 'cc-1', timestamp '2026-02-02 01:00:00',
+ named_struct('metric', 40, 'label', 'bucket-c',
'extra', 'added'));
+ 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')),
+ (6, 'D', 'dd-1', timestamp '2026-03-02 01:00:00',
+ named_struct('metric', 60, 'renamed_label',
'truncate-d',
+ 'extra', 'renamed'));
+ 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', 'drop-a',
'extra', 7000)),
+ (8, 'E', 'ee-1', timestamp '2026-04-02 01:00:00',
+ named_struct('metric', 80, 'renamed_label', 'drop-e',
'extra', 8000));
+ """
+ String identityFinal = latestSnapshotId(identityTable)
+ sql """
+ alter table `${catalogName}`.`${dbName}`.`${identityTable}`
+ create tag ${identityTable}_final as of version
${identityFinal}
+ """
+
+ // Scenario PE-X02: run the complete year -> month -> day -> hour
replacement timeline
+ // in the same format so each transform boundary is observable
under both scanners.
+ 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'='${format}'
+ );
+ 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');
+ 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 temporalFinal = latestSnapshotId(temporalTable)
+ sql """
+ alter table `${catalogName}`.`${dbName}`.`${temporalTable}`
+ create tag ${temporalTable}_final as of version
${temporalFinal}
+ """
+
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+ sql """refresh catalog ${catalogName}"""
+
+ [true, false].each { boolean scannerV2 ->
+ String scanner = scannerV2 ? "v2" : "v1"
+ sql """set enable_file_scanner_v2=${scannerV2}"""
+
+ // Scenario PE-X03: every format/scanner cell reads current
data spanning every
+ // identity/bucket/truncate/drop spec and preserves nested
field IDs.
+ "qt_identity_${format}_${scanner}_current"("""
+ select id, payload.extra from ${identityTable}
+ where category = 'A' or code in ('bb-1', 'cc-1', 'dd-1',
'ee-1')
+ order by id
+ """)
+ "qt_identity_${format}_${scanner}_base_tag"("""
+ select id, category, payload.label
+ from ${identityTable}@tag(${identityTable}_base)
+ order by id
+ """)
+ "qt_identity_${format}_${scanner}_final_tag"("""
+ select id, payload.renamed_label
+ from ${identityTable}@tag(${identityTable}_final)
+ where category = 'A'
+ order by id
+ """)
+
+ // Scenario PE-X04: the current predicate spans
year/month/day/hour files and the
+ // historical tag is read by every format/scanner combination.
+ "qt_temporal_${format}_${scanner}_current"("""
+ select id from ${temporalTable}
+ where event_time >= timestamp '2024-01-01 00:00:00'
+ and event_time < timestamp '2026-05-02 00:00:00'
+ order by id
+ """)
+ "qt_temporal_${format}_${scanner}_final_tag"("""
+ select id from ${temporalTable}@tag(${temporalTable}_final)
+ where event_time >= timestamp '2026-02-01 00:00:00'
+ order by id
+ """)
+ }
+ }
+ } finally {
+ sql """set enable_file_scanner_v2=true"""
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.groovy
new file mode 100644
index 00000000000..4ed60374d04
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_position_dv.groovy
@@ -0,0 +1,318 @@
+// 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_position_dv",
+
"p0,external,iceberg,external_docker,external_docker_iceberg,nonConcurrent") {
+ 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()
+ }
+ 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 tableName, String dimensionTable
->
+ String token = UUID.randomUUID().toString()
+ List<List<String>> rows = stringRows("""
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */
+ '${token}', f.id
+ from ${tableName} f
+ join ${dimensionTable} d on f.category = d.category
+ order by f.id
+ """)
+ 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 a delete-aware 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 {
+ ["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,
+ code 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, code, event_time,
payload from values
+ (1, 'A', 'aa-1', timestamp '2026-01-01 01:00:00',
+ named_struct('metric', 10, 'label', 'base-a')),
+ (2, 'A', 'aa-2', timestamp '2026-01-01 02:00:00',
+ named_struct('metric', 20, 'label',
'base-delete')),
+ (3, 'B', 'bb-1', timestamp '2026-01-02 01:00:00',
+ named_struct('metric', 30, 'label', 'base-b')),
+ (4, 'C', 'cc-1', timestamp '2026-01-03 01:00:00',
+ named_struct('metric', 40, 'label', 'base-c'))
+ as t(id, category, code, 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, code, event_time,
payload from values
+ (5, 'A', 'aa-3', timestamp '2026-02-01 01:00:00',
+ named_struct('metric', 50, 'label', 'add-a',
'extra', 'new-child')),
+ (6, 'A', 'aa-4', timestamp '2026-02-01 02:00:00',
+ named_struct('metric', 60, 'label', 'add-delete',
'extra', 'new-child'))
+ as t(id, category, code, 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}
+ replace partition field bucket(8, id) with truncate(2,
code);
+ 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', 'aa-5', timestamp '2026-03-01 01:00:00',
+ named_struct('metric', 7000000000,
+ 'renamed_label', 'replace-a', 'extra',
'renamed-child')),
+ (8, 'A', 'aa-6', timestamp '2026-03-01 02:00:00',
+ named_struct('metric', 80,
+ 'renamed_label', 'replace-delete', 'extra',
'renamed-child'));
+ """
+
+ // Scenario PE-D04: DROP identity field, write a victim with
the resulting spec,
+ // then delete that victim. The data and delete artifacts must
both use the spec
+ // that lacks 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;
+ insert into demo.${dbName}.${tableName}
+ select /*+ coalesce(1) */ id, category, code, event_time,
payload from values
+ (9, 'A', 'aa-7', timestamp '2026-04-01 01:00:00',
+ named_struct('metric', 90,
+ 'renamed_label', 'drop-delete', 'extra',
'new-spec')),
+ (10, 'A', 'aa-8', timestamp '2026-04-01 02:00:00',
+ named_struct('metric', 100,
+ 'renamed_label', 'drop-survivor', 'extra',
'new-spec'))
+ as t(id, category, code, event_time, payload);
+ delete from demo.${dbName}.${tableName} where id = 9;
+ 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');
+ """
+ List<List<Object>> dataSpecRows = spark_iceberg """
+ select max(spec_id)
+ from demo.${dbName}.${tableName}.all_files
+ where content = 0
+ """
+ List<List<Object>> deleteSpecRows = spark_iceberg """
+ select max(spec_id)
+ from demo.${dbName}.${tableName}.all_files
+ where content = 1
+ """
+ assertEquals(dataSpecRows[0][0], deleteSpecRows[0][0],
+ "The post-drop victim data and delete artifacts must
use the same spec")
+ 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}"""
+ String actionSuffix = "${deleteKind}_${format}"
+
+ // Scenario PE-D05: static partition filters apply every
delete exactly once.
+ List<List<String>> expectedCurrent = [["1"], ["5"], ["7"],
["10"]]
+ "qt_${actionSuffix}_current_filter"("""
+ select id from ${tableName} where category = 'A' order by
id
+ """)
+ "qt_${actionSuffix}_current_nested"("""
+ 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 pruning disabled, then
proves that the enabled
+ // query retains an RF and prunes at least one partition/file
range.
+ 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"""
+ sql """set disable_join_reorder=true"""
+ sql """set enable_runtime_filter_prune=false"""
+ sql """set runtime_filter_mode=GLOBAL"""
+ sql """set parallel_pipeline_task_num=1"""
+ sql """set enable_profile=true"""
+ sql """set profile_level=2"""
+ sql """set enable_runtime_filter_partition_prune=false"""
+ "qt_${actionSuffix}_rf_disabled"(rfQuery)
+ sql """set enable_runtime_filter_partition_prune=true"""
+ assertEquals(expectedCurrent,
+ assertRuntimeFilterPruned(tableName,
"${tableName}_dimension"))
+
+ // Scenario PE-D07: numeric snapshots and tags preserve
historical rows and deletes.
+ "qt_${actionSuffix}_base_snapshot"("""
+ select id from ${tableName} for version as of
${baseSnapshot}
+ where category in ('A', 'B', 'C') order by id
+ """)
+ "qt_${actionSuffix}_base_tag"("""
+ select id from ${tableName}@tag(${tableName}_base)
+ where category = 'A' order by id
+ """)
+ "qt_${actionSuffix}_first_delete_snapshot"("""
+ select id from ${tableName} for version as of
${firstDeleteSnapshot}
+ where category = 'A' order by id
+ """)
+ "qt_${actionSuffix}_added_delete_snapshot"("""
+ select id from ${tableName} for version as of
${addedDeleteSnapshot}
+ where category = 'A' order by id
+ """)
+ "qt_${actionSuffix}_final_tag"("""
+ select id from ${tableName}@tag(${tableName}_final)
+ where category = 'A' order by id
+ """)
+
+ // Scenario PE-D08: legacy and V2 scanners must agree on
multi-spec delete planning.
+ sql """set enable_file_scanner_v2=true"""
+ "qt_${actionSuffix}_scanner_v2"("""
+ select id from ${tableName} where category = 'A' order by
id
+ """)
+ sql """set enable_file_scanner_v2=false"""
+ "qt_${actionSuffix}_scanner_v1"("""
+ select id from ${tableName} where category = 'A' order by
id
+ """)
+ sql """set enable_file_scanner_v2=true"""
+
+ // Scenario PE-D09: verify the fixture really contains
physical delete artifacts.
+ // Iceberg may replace or consolidate older artifacts across
delete commits, while
+ // the snapshot assertions above still validate every
pre/post-evolution phase.
+ List<List<String>> deleteFiles = stringRows("""
+ select file_format
+ from ${tableName}\$all_files
+ where content = 1
+ order by file_format
+ """)
+ assertTrue(deleteFiles.size() >= 1,
+ "${tableName} must contain at least one physical
delete artifact")
+ if (formatVersion == 3) {
+ assertTrue(deleteFiles.any { row ->
row[0].equalsIgnoreCase("PUFFIN") },
+ "${tableName} must contain Iceberg deletion
vectors")
+ }
+ }
+ }
+ } finally {
+ sql """set enable_file_scanner_v2=true"""
+ sql """set enable_runtime_filter_prune=true"""
+ sql """set enable_runtime_filter_partition_prune=true"""
+ sql """set disable_join_reorder=false"""
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.groovy
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.groovy
new file mode 100644
index 00000000000..729ba0418b5
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/test_iceberg_partition_evolution_runtime_filter.groovy
@@ -0,0 +1,277 @@
+// 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,nonConcurrent") {
+ 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,
+ region 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', 'r1', timestamp '2026-01-01 01:00:00',
+ named_struct('metric', 10)),
+ (2, 'B', 'r2', timestamp '2026-01-02 01:00:00',
+ named_struct('metric', 20)),
+ (3, 'C', 'r3', 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 identity and bucket fields and evolve a
nested payload between
+ // data files. Old files lack both partition values but still contain
the source columns.
+ spark_iceberg_multi """
+ alter table demo.${dbName}.${factTable} add partition field region;
+ 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', 'r1', timestamp '2026-02-01 01:00:00',
+ named_struct('metric', 40, 'label', 'add-spec')),
+ (5, 'D', 'r4', 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', 'r1', timestamp '2026-03-01 01:00:00',
+ named_struct('metric', 60, 'label', 'replace-spec')),
+ (7, 'E', 'r5', 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', 'r1', timestamp '2026-04-01 01:00:00',
+ named_struct('metric', 80, 'label', 'drop-spec')),
+ (9, 'F', 'r6', 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,
+ region string,
+ id int,
+ lower_time timestamp,
+ upper_time timestamp
+ ) using iceberg
+ tblproperties ('format-version'='2');
+ insert into demo.${dbName}.${dimensionTable} values
+ ('A', 'r1', 4,
+ 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 addedIdentityJoin = """
+ from ${factTable} f
+ join ${dimensionTable} d on f.region = d.region
+ """
+ String bucketSourceJoin = """
+ from ${factTable} f
+ join ${dimensionTable} d on f.id = d.id
+ """
+ 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"""
+ qt_category_rf_disabled """
+ select f.id ${currentCategoryJoin} order by f.id
+ """
+ qt_added_identity_rf_disabled """
+ select f.id ${addedIdentityJoin} order by f.id
+ """
+ qt_bucket_source_rf_disabled """
+ select f.id ${bucketSourceJoin} order by f.id
+ """
+ qt_temporal_rf_disabled """
+ select f.id ${currentTemporalJoin} order by f.id
+ """
+
+ // Scenario PE-RF05: independently profile the original identity field
that is later
+ // dropped and the identity field added after old files were written.
+ sql """set enable_runtime_filter_partition_prune=true"""
+ assertEquals([["1"], ["4"], ["6"], ["8"]],
+ assertRuntimeFilterPruned(currentCategoryJoin))
+ assertEquals([["1"], ["4"], ["6"], ["8"]],
+ assertRuntimeFilterPruned(addedIdentityJoin))
+
+ // Scenario PE-RF06: bucket and temporal source-column RFs retain
result correctness.
+ // Scanner-side RF pruning currently consumes identity partition
values only, so these
+ // cells intentionally do not claim a positive physical-pruning
counter.
+ qt_bucket_source_rf_enabled """
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */
f.id
+ ${bucketSourceJoin}
+ order by f.id
+ """
+ qt_temporal_rf_enabled """
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */
f.id
+ ${currentTemporalJoin}
+ order by f.id
+ """
+
+ // Scenario PE-RF07: numeric snapshot and tag retain runtime-filter
correctness.
+ qt_base_snapshot_rf """
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */
f.id
+ from ${factTable} for version as of ${baseSnapshot} f
+ join ${dimensionTable} d on f.category = d.category
+ order by f.id
+ """
+ qt_added_snapshot_rf """
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */
f.id
+ from ${factTable} for version as of ${addedSnapshot} f
+ join ${dimensionTable} d on f.region = d.region
+ order by f.id
+ """
+ qt_dropped_tag_rf """
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */
f.id
+ from ${factTable}@tag(rf_dropped) f
+ join ${dimensionTable} d on f.category = d.category
+ order by f.id
+ """
+ } finally {
+ sql """set enable_runtime_filter_prune=true"""
+ sql """set enable_runtime_filter_partition_prune=true"""
+ sql """set disable_join_reorder=false"""
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg_paimon_schema_time_travel_coverage.md
b/regression-test/suites/external_table_p0/iceberg_paimon_schema_time_travel_coverage.md
index 8bc1e9520b0..31a1cfd1ea2 100644
---
a/regression-test/suites/external_table_p0/iceberg_paimon_schema_time_travel_coverage.md
+++
b/regression-test/suites/external_table_p0/iceberg_paimon_schema_time_travel_coverage.md
@@ -11,6 +11,49 @@ 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. Runtime filters on identity partition fields require
+equal results with pruning disabled/enabled and a positive pruning counter;
+bucket-source and temporal-transform joins assert result parity because those
+transform paths are not physically runtime-filter-prunable.
+
+## 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 | Result contract; source transform is not
RF-prunable | 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 by one current predicate spanning all
four specs | Result contract; temporal transforms are not RF-prunable | Covered
| Day → month covered | Day → month covered | Day → month covered | Covered |
Parquet/ORC, V1/V2 |
+| Old spec lacks newly added field | Covered | Covered | Covered | Covered |
Covered | Covered | Covered | Parquet/ORC, V1/V2 |
+| New spec lacks dropped field | Covered | Covered | Covered | Covered |
Covered | Covered | Covered | Parquet/ORC, V1/V2 |
+| Partition-field alias replacement | Covered by existing position-delete
suite | Not RF-prunable separately | Existing metadata coverage | Covered | N/A
| N/A | N/A | Parquet |
+
+The Iceberg delete timelines write deletes both before and after spec changes.
+Queries then combine partition predicates, runtime filters and historical
+references so deleted rows cannot be resurrected or over-deleted when Doris
+plans files from several specs.
+
+### Paimon
+
+Paimon does not support changing the partition-key set after table creation.
+Its P0 contract therefore combines supported schema reordering and payload
+evolution with fixed partition keys, and verifies that unsupported key
+mutations fail atomically.
+
+| Paimon partition contract | Static filter | Runtime filter |
Snapshot/tag/branch | Delete/upsert/DV | Complex schema | Format / reader |
+| --- | --- | --- | --- | --- | --- | --- |
+| Reorder partition columns in schema | Covered | N/A | Covered by pre-reorder
tag | N/A | Add payload field | Parquet |
+| Rename partition key | Rejected atomically | N/A | Historical tag unchanged
| Data unchanged | Schema count unchanged | Parquet |
+| Change partition-key type | Rejected atomically | N/A | Historical tag
unchanged | Data unchanged | Schema count unchanged | Parquet |
+| Drop partition key | Rejected atomically | N/A | Historical tag unchanged |
Data unchanged | Schema count unchanged | Parquet |
+| Fixed key + STRUCT/MAP/ARRAY evolution | Covered | Covered with positive
pruning counter | Covered | Append timeline | Current and historical nested
projections plus wrong-schema negatives | Parquet/ORC; JNI and native split
paths asserted |
+| Partitioned PK table | Covered | Covered with positive pruning counter |
Covered | Upsert/delete/compaction plus physical DV artifact | Current and
historical evolved fields, including drop/re-add NULL isolation | Parquet/ORC;
JNI and native split paths asserted |
+
## Schema operations
| ID | Operation | Iceberg | Paimon | P0 contract |
@@ -62,7 +105,7 @@ from passing accidentally.
| Equality delete | Rename, promotion, drop/re-add, old/new snapshots | N/A |
| Deletion vector | v3, Parquet and ORC, before/after evolution | PK-table DV
path |
| Row operations | Delete visibility around every checkpoint | Upsert, delete
and compaction |
-| Readers | File scanner V1/V2 | JNI/native/CPP-supported paths |
+| Readers | Complete identity/bucket/truncate/drop and year/month/day/hour
timelines under file scanner V1/V2, for both Parquet and ORC | JNI and native
paths are proved by split statistics for current and historical projections |
| Cache | REST cache on/off | Filesystem metadata cache on/off |
| Catalog smoke | REST full matrix and JDBC rename/time-travel | Filesystem
full matrix and JDBC rename/time-travel |
| Cluster topology | External endpoints are cluster-reachable; JDBC drivers
are installed on every FE/BE | External endpoints are cluster-reachable; JDBC
drivers are installed on every FE/BE |
@@ -82,10 +125,18 @@ explicit rename plus old snapshot/tag smoke path.
| `iceberg/test_iceberg_schema_position_dv_time_travel.groovy` | Position
delete and DV × top-level/nested evolution, Parquet/ORC |
| `iceberg/test_iceberg_schema_ref_actions_matrix.groovy` | Rollback,
cherry-pick, fast-forward and branch action semantics |
| `iceberg/test_iceberg_schema_metadata_atomicity_matrix.groovy` |
Comment/default/nullability/narrowing/map-key atomicity |
+| `iceberg/test_iceberg_partition_evolution_filter_refs.groovy` |
Add/drop/replace specs, all transform families, static filters and
snapshot/tag/branch |
+| `iceberg/test_iceberg_partition_evolution_runtime_filter.groovy` |
Runtime-filter pruning across specs with missing/new partition fields |
+| `iceberg/test_iceberg_partition_evolution_position_dv.groovy` | Position
delete and deletion vector before/after spec changes, filters and refs |
+| `iceberg/test_iceberg_partition_evolution_equality_delete.groovy` | Real
equality-delete files before/after add/replace/drop specs |
+| `iceberg/test_iceberg_partition_evolution_format_scanner.groovy` | Full
partition-transform timeline for Parquet/ORC under scanner V1/V2, current and
tagged refs |
| `paimon/test_paimon_schema_time_travel_matrix.groovy` | S01-S18 × T00-T14,
PK upsert/delete/DV, cache/readers |
| `paimon/test_paimon_schema_dual_relation_matrix.groovy` | Dual-snapshot
join/UNION/CTE/subquery negative contracts |
| `paimon/test_paimon_schema_branch_partition_matrix.groovy` | Independent
branch evolution, fast-forward and partition restrictions |
| `paimon/test_paimon_schema_metadata_atomicity_matrix.groovy` |
Comment/default/nullability/narrowing atomicity |
+| `paimon/test_paimon_partition_schema_filter_refs.groovy` | Fixed partition
key with complex evolution, pruning and historical refs |
+| `paimon/test_paimon_partition_pk_delete_refs.groovy` | Partitioned PK
upsert/delete/DV/compaction with filters and refs |
+| `paimon/test_paimon_partition_mutation_atomicity.groovy` | Supported reorder
plus atomic rejection of key rename/type/drop |
| `iceberg/test_iceberg_jdbc_catalog.groovy` | JDBC catalog rename × numeric
snapshot/tag smoke |
| `paimon/test_paimon_jdbc_catalog.groovy` | JDBC catalog rename × numeric
snapshot/tag smoke |
@@ -105,15 +156,25 @@ back to the exact suite, scenario and file location from
Jira instead.
## Validation status
-- The ten REST/filesystem matrix suites pass with no failed, fatal or skipped
+- All eight partition-evolution extension suites pass against master
+ `f1460f89230441bc1b6b1872d66bd32a526b25b1`: no failed, fatal or skipped
suite.
-- The two JDBC catalog suites pass with no failed, fatal or skipped suite.
-- The validation covers current and historical schema binding, nested
- evolution, deletes, branches, tags, dual historical relations, metadata
- atomicity, reader/cache variants, catalog variants and distributed cluster
- scheduling.
-
-The requested schema-change × historical-operation correctness matrix has no
-unimplemented P0 cell. Unsupported format operations and currently incorrect
-Doris behavior are represented by stable negative regression contracts rather
-than being marked as missing.
+- The matrix contains twenty P0 suites: ten original REST/filesystem suites,
+ two JDBC catalog suites and eight partition-evolution extension suites.
+- Independent REST/filesystem suites support parallel execution. The four
+ profile-dependent partition extensions are marked `nonConcurrent` so profile
+ lookup and session-counter assertions cannot consume another suite's query;
+ the two existing JDBC catalog suites are validated separately because they
+ initialize shared external catalog fixtures.
+- Distributed validation used one FE and two live BEs. The cases use
+ cluster-reachable external endpoints and contain no fixed backend,
+ backend-local path or single-node scheduling assumption.
+- No additional Doris product issue was reproduced by the partition-evolution
+ extension. Test-fixture and Runtime Filter session preconditions found during
+ development were corrected before the final run and were not recorded as
+ product defects.
+
+The requested schema-change × partition-operation × historical-reference
+matrix has no unimplemented P0 cell. Unsupported format operations and
+currently incorrect Doris behavior are represented by stable negative
+regression contracts rather than being marked as missing.
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_partition_mutation_atomicity.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_partition_mutation_atomicity.groovy
new file mode 100644
index 00000000000..e86c5af0e9d
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_partition_mutation_atomicity.groovy
@@ -0,0 +1,148 @@
+// 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 schemaCount = {
+ return spark_paimon("""
+ select count(*) from paimon.${dbName}.`${tableName}\$schemas`
+ """)[0][0].toString().toInteger()
+ }
+ def assertSparkRejected = { String statement, String operation, String
expectedMessage ->
+ String error = null
+ try {
+ spark_paimon(statement)
+ } catch (Exception e) {
+ error = e.getMessage()
+ }
+ assertNotNull(error, "Paimon must reject partition-key ${operation}")
+ assertTrue(error.contains(expectedMessage),
+ "Partition-key ${operation} should report
'${expectedMessage}': ${error}")
+ }
+
+ 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 {
+ 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_a string,
+ payload string,
+ part_b int
+ ) using paimon
+ partitioned by (part_a, part_b)
+ tblproperties ('file.format'='parquet');
+ insert into paimon.${dbName}.${tableName} values
+ (1, 'A', 'base-a1', 1),
+ (2, 'A', 'base-a2', 2),
+ (3, 'B', 'base-b1', 1);
+ call paimon.sys.create_tag(
+ table => '${dbName}.${tableName}',
+ tag => 'partition_contract_base'
+ );
+ """
+
+ // Scenario PM-M01-positive: partition columns may move in schema
order without changing
+ // their partition identity, values, tag or filter semantics.
+ spark_paimon_multi """
+ alter table paimon.${dbName}.${tableName} alter column part_b
first;
+ alter table paimon.${dbName}.${tableName} alter column part_a
after id;
+ alter table paimon.${dbName}.${tableName} add column added_payload
string after payload;
+ insert into paimon.${dbName}.${tableName}
+ (part_b, id, part_a, payload, added_payload)
+ values (3, 4, 'A', 'after-reorder', 'added');
+ """
+ int schemasAfterSupportedChanges = schemaCount()
+
+ // Scenario PM-M02-negative: rename, type change and drop of either
partition key must be
+ // rejected atomically. These are format restrictions, not missing
Doris capabilities.
+ assertSparkRejected("""
+ alter table paimon.${dbName}.${tableName}
+ rename column part_a to renamed_part_a
+ """, "rename", "Cannot rename partition column")
+ assertSparkRejected("""
+ alter table paimon.${dbName}.${tableName}
+ alter column part_b type bigint
+ """, "type change", "Cannot update partition column")
+ assertSparkRejected("""
+ alter table paimon.${dbName}.${tableName} drop column part_a
+ """, "drop", "Cannot drop partition key or primary key")
+ assertSparkRejected("""
+ alter table paimon.${dbName}.${tableName} drop column part_b
+ """, "drop", "Cannot drop partition key or primary key")
+ assertEquals(schemasAfterSupportedChanges, schemaCount(),
+ "Rejected partition mutations must not create Paimon schemas")
+
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+ sql """refresh table ${tableName}"""
+
+ // Scenario PM-M03: current data and static partition filtering
survive accepted reorder
+ // and all rejected mutations.
+ qt_current_partition_filter """
+ select id from ${tableName}
+ where part_a = 'A' and part_b in (1, 2, 3)
+ order by id
+ """
+ qt_current_added_payload """
+ select id, added_payload from ${tableName}
+ where part_a = 'A' and part_b = 3
+ """
+
+ // Scenario PM-M04: the pre-reorder tag retains its original schema
and partitions.
+ qt_base_tag_partition_filter """
+ select id, part_a, part_b, payload
+ from ${tableName}@tag(partition_contract_base)
+ where part_a in ('A', 'B')
+ order by id
+ """
+ test {
+ sql """
+ select added_payload
+ from ${tableName}@tag(partition_contract_base)
+ """
+ exception "Unknown column 'added_payload'"
+ }
+ } finally {
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.groovy
new file mode 100644
index 00000000000..85007d147ba
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_partition_pk_delete_refs.groovy
@@ -0,0 +1,346 @@
+// 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_pk_delete_refs",
+
"p0,external,paimon,external_docker,external_docker_paimon,nonConcurrent") {
+ 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}'
+ )
+ """
+ }
+ 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 tableName, String dimensionTable
->
+ String token = UUID.randomUUID().toString()
+ List<List<String>> rows = stringRows("""
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER') */
+ '${token}', f.id, f.full_name
+ from ${tableName} f
+ join ${dimensionTable} d on f.part = d.part
+ order by f.id
+ """)
+ 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 a Paimon PK partition/file
range; "
+ + profile.take(2000).replaceAll("\\s+", " "))
+ return rows.collect { row -> [row[1], row[2]] }
+ }
+ def getExplainText = { String query ->
+ return sql("explain verbose ${query}").collect { row ->
row[0].toString() }.join("\n")
+ }
+ def assertNativePath = { String query, String label ->
+ String explainText = getExplainText(query)
+ def splitMatcher = (explainText =~
/paimonNativeReadSplits=(\d+)\/(\d+)/)
+ assertTrue(splitMatcher.find(), "Expected paimonNativeReadSplits for
${label}")
+ long nativeSplits = Long.parseLong(splitMatcher.group(1))
+ long totalSplits = Long.parseLong(splitMatcher.group(2))
+ assertTrue(totalSplits > 0 && nativeSplits > 0,
+ "Expected native splits for ${label}, native=${nativeSplits},
total=${totalSplits}")
+ assertTrue(explainText.contains("SplitStat [type=NATIVE"),
+ "Expected a NATIVE split for ${label}")
+ }
+ def assertJniPath = { String query, String label ->
+ String explainText = getExplainText(query)
+ def splitMatcher = (explainText =~
/paimonNativeReadSplits=(\d+)\/(\d+)/)
+ assertTrue(splitMatcher.find(), "Expected paimonNativeReadSplits for
${label}")
+ long nativeSplits = Long.parseLong(splitMatcher.group(1))
+ long totalSplits = Long.parseLong(splitMatcher.group(2))
+ assertTrue(totalSplits > 0 && nativeSplits == 0,
+ "Expected JNI-only splits for ${label},
native=${nativeSplits}, total=${totalSplits}")
+ assertTrue(explainText.contains("SplitStat [type=JNI"),
+ "Expected a JNI split for ${label}")
+ }
+
+ 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')),
+ (90, 'p_dv', 'dv-victim', 'old-note-90',
+ named_struct('metric', 90, 'label', 'dv-victim'));
+ """
+ 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'
+ );
+ call paimon.sys.create_tag(
+ table => '${dbName}.${tableName}',
+ tag => '${tableName}_pre_dv'
+ );
+ -- Prevent synchronous batch compaction from materializing the
sacrificial
+ -- delete, so the fixture retains a physical DV artifact for
reader checks.
+ alter table paimon.${dbName}.${tableName}
+ set tblproperties ('write-only'='true');
+ delete from paimon.${dbName}.${tableName}
+ where part = 'p_dv' and id = 90;
+ 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');
+ """
+ long deletedRowsInFiles = spark_paimon("""
+ select coalesce(sum(deleteRowCount), 0)
+ from paimon.${dbName}.`${tableName}\$files`
+ """)[0][0].toString().toLong()
+ assertTrue(deletedRowsInFiles > 0,
+ "${tableName} must retain a physical deletion-vector row")
+ String finalSnapshot = latestSnapshotId(tableName)
+ createTag(tableName, "${tableName}_final")
+
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+ sql """refresh table ${tableName}"""
+ String actionSuffix = format
+
+ // Scenario PM-D03: static partition filters apply PK upserts,
deletes and DV state.
+ List<List<String>> expectedCurrent = [["1", "alpha-updated"],
["5", "epsilon"]]
+ "qt_${actionSuffix}_current_partition_filter"("""
+ select id, full_name from ${tableName}
+ where part = 'p1' order by id
+ """)
+ "qt_${actionSuffix}_current_promoted_and_readded"("""
+ select id, payload.metric, note from ${tableName}
+ where part = 'p1' and payload.metric > 1000000000
+ order by id
+ """)
+ // Surviving pre-readd row 1 must see NULL for the new BIGINT note
field, while the
+ // nested rename and added child stay bound by field ID across the
delete timeline.
+ "qt_${actionSuffix}_current_evolved_fields"("""
+ select id, payload.renamed_label, payload.extra, note
+ from ${tableName}
+ where part = 'p1'
+ order by id
+ """)
+ "qt_${actionSuffix}_pre_dv_old_row_null"("""
+ select id, payload.renamed_label, payload.extra, note
+ from ${tableName}@tag(${tableName}_pre_dv)
+ where part = 'p1' and id = 1
+ """)
+ test {
+ sql """select payload.label from ${tableName} where part =
'p1'"""
+ exception "label"
+ }
+
+ // 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_prune=false"""
+ sql """set runtime_filter_mode=GLOBAL"""
+ sql """set parallel_pipeline_task_num=1"""
+ sql """set enable_profile=true"""
+ sql """set profile_level=2"""
+ sql """set enable_runtime_filter_partition_prune=false"""
+ "qt_${actionSuffix}_rf_disabled"(rfQuery)
+ sql """set enable_runtime_filter_partition_prune=true"""
+ assertEquals(expectedCurrent, assertRuntimeFilterPruned(tableName,
dimensionTable))
+
+ // Scenario PM-D05: numeric snapshots and tags preserve the
matching schema/delete set.
+ "qt_${actionSuffix}_base_snapshot"("""
+ select id, old_name, note, payload.label
+ from ${tableName} for version as of ${baseSnapshot}
+ where part = 'p1' order by id
+ """)
+ "qt_${actionSuffix}_base_tag"("""
+ select id, old_name, note, payload.label
+ from ${tableName}@tag(${tableName}_base)
+ where part = 'p1' order by id
+ """)
+ "qt_${actionSuffix}_first_delete_snapshot"("""
+ select id, full_name, note, payload.label, payload.extra
+ from ${tableName} for version as of ${firstDeleteSnapshot}
+ where part = 'p1' order by id
+ """)
+ "qt_${actionSuffix}_final_tag"("""
+ select id, full_name, note, payload.renamed_label,
payload.extra
+ from ${tableName}@tag(${tableName}_final)
+ where part = 'p1' order by id
+ """)
+ test {
+ sql """
+ select payload.renamed_label
+ from ${tableName} for version as of ${baseSnapshot}
+ """
+ exception "renamed_label"
+ }
+
+ // Scenario PM-D06: JNI/native readers agree for current DV state
and historical
+ // projections, and explain must prove the requested reader path
rather than fallback.
+ String currentReaderQuery = """
+ select id, part, full_name, note, payload.renamed_label,
payload.extra
+ from ${tableName}
+ where part in ('p1', 'p2')
+ order by part, id
+ """
+ String historicalReaderQuery = """
+ select id, part, old_name, note, payload.label
+ from ${tableName} for version as of ${baseSnapshot}
+ where part in ('p1', 'p2')
+ order by part, id
+ """
+ sql """set enable_paimon_cpp_reader=false"""
+ sql """set force_jni_scanner=true"""
+ assertJniPath(currentReaderQuery, "${tableName} current DV")
+ assertJniPath(historicalReaderQuery, "${tableName} historical")
+ "qt_${actionSuffix}_jni_current_dv"(currentReaderQuery)
+ "qt_${actionSuffix}_jni_historical"(historicalReaderQuery)
+ sql """set force_jni_scanner=false"""
+ sql """set enable_paimon_cpp_reader=true"""
+ assertNativePath(currentReaderQuery, "${tableName} current DV")
+ assertNativePath(historicalReaderQuery, "${tableName} historical")
+ "qt_${actionSuffix}_native_current_dv"(currentReaderQuery)
+ "qt_${actionSuffix}_native_historical"(historicalReaderQuery)
+ assertEquals(finalSnapshot, latestSnapshotId(tableName))
+ }
+ } finally {
+ sql """set enable_paimon_cpp_reader=false"""
+ sql """set force_jni_scanner=false"""
+ sql """set enable_runtime_filter_prune=true"""
+ sql """set enable_runtime_filter_partition_prune=true"""
+ sql """set disable_join_reorder=false"""
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_partition_schema_filter_refs.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_partition_schema_filter_refs.groovy
new file mode 100644
index 00000000000..979fbb7784c
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_partition_schema_filter_refs.groovy
@@ -0,0 +1,366 @@
+// 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,nonConcurrent") {
+ 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+", " "))
+ }
+ def getExplainText = { String query ->
+ return sql("explain verbose ${query}").collect { row ->
row[0].toString() }.join("\n")
+ }
+ def assertNativePath = { String query, String label ->
+ String explainText = getExplainText(query)
+ def splitMatcher = (explainText =~
/paimonNativeReadSplits=(\d+)\/(\d+)/)
+ assertTrue(splitMatcher.find(), "Expected paimonNativeReadSplits for
${label}")
+ long nativeSplits = Long.parseLong(splitMatcher.group(1))
+ long totalSplits = Long.parseLong(splitMatcher.group(2))
+ assertTrue(totalSplits > 0 && nativeSplits > 0,
+ "Expected native splits for ${label}, native=${nativeSplits},
total=${totalSplits}")
+ assertTrue(explainText.contains("SplitStat [type=NATIVE"),
+ "Expected a NATIVE split for ${label}")
+ }
+ def assertJniPath = { String query, String label ->
+ String explainText = getExplainText(query)
+ def splitMatcher = (explainText =~
/paimonNativeReadSplits=(\d+)\/(\d+)/)
+ assertTrue(splitMatcher.find(), "Expected paimonNativeReadSplits for
${label}")
+ long nativeSplits = Long.parseLong(splitMatcher.group(1))
+ long totalSplits = Long.parseLong(splitMatcher.group(2))
+ assertTrue(totalSplits > 0 && nativeSplits == 0,
+ "Expected JNI-only splits for ${label},
native=${nativeSplits}, total=${totalSplits}")
+ assertTrue(explainText.contains("SplitStat [type=JNI"),
+ "Expected a JNI split for ${label}")
+ }
+
+ 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;
+ alter table paimon.${dbName}.${tableName}
+ rename column events.element.score to renamed_score;
+ alter table paimon.${dbName}.${tableName}
+ alter column payload.metric type bigint;
+ insert into paimon.${dbName}.${tableName} values
+ (5, 'p1',
+ named_struct('metric', 5000000000,
+ 'renamed_label', 'rename-5', 'extra',
'payload-extra'),
+ map('k', named_struct('renamed_code', 500, 'extra',
501)),
+ array(named_struct('renamed_score', 5000, 'extra',
5001))),
+ (6, 'p4',
+ named_struct('metric', 60,
+ 'renamed_label', 'rename-6', 'extra',
'payload-extra'),
+ map('k', named_struct('renamed_code', 600, 'extra',
601)),
+ array(named_struct('renamed_score', 6000, 'extra',
6001)));
+ """
+
+ // Scenario PM-PE03: drop/re-add a nested name to verify field IDs
never leak values
+ // through partition-filtered scans.
+ spark_paimon_multi """
+ alter table paimon.${dbName}.${tableName} drop column
payload.extra;
+ alter table paimon.${dbName}.${tableName} add column
payload.extra bigint;
+ insert into paimon.${dbName}.${tableName} values
+ (7, 'p1',
+ named_struct('metric', 70,
+ 'renamed_label', 'readd-7', 'extra', 7000),
+ map('k', named_struct('renamed_code', 700, 'extra',
701)),
+ array(named_struct('renamed_score', 7000, 'extra',
7001)));
+ 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}"""
+ String actionSuffix = format
+
+ // Scenario PM-F01: static partition filtering combines every
complex-schema version.
+ "qt_${actionSuffix}_current_partition_filter"("""
+ select id from ${tableName} where part = 'p1' order by id
+ """)
+ "qt_${actionSuffix}_current_promoted_metric"("""
+ select id, payload.metric from ${tableName}
+ where part = 'p1' and payload.metric > 1000000000
+ order by id
+ """)
+ "qt_${actionSuffix}_current_readded_payload"("""
+ select id, payload.extra from ${tableName}
+ where part = 'p1' and payload.extra is not null
+ order by id
+ """)
+ // Project every evolved STRUCT/MAP/ARRAY child so a reader cannot
pass by decoding
+ // only the top-level partition and payload fields.
+ "qt_${actionSuffix}_current_all_complex_children"("""
+ select id, payload.renamed_label, payload.extra,
+ element_at(attrs, 'k').renamed_code,
+ element_at(attrs, 'k').extra,
+ events[1].renamed_score,
+ events[1].extra
+ from ${tableName}
+ where part = 'p1'
+ order by id
+ """)
+
+ // Scenario PM-RF01: runtime-filter partition pruning stays
result-equivalent across
+ // complex schema versions.
+ String rfQuery = """
+ select /*+ SET_VAR(runtime_filter_type='IN_OR_BLOOM_FILTER')
*/ f.id
+ from ${tableName} f join ${dimensionTable} d on f.part = d.part
+ order by f.id
+ """
+ sql """set runtime_filter_wait_infinitely=true"""
+ // Small fixture tables have no column statistics. Keep the
generated RF so this
+ // suite validates scanner-side partition pruning rather than RF
selectivity pruning.
+ sql """set enable_runtime_filter_prune=false"""
+ sql """set runtime_filter_mode=GLOBAL"""
+ sql """set parallel_pipeline_task_num=1"""
+ sql """set disable_join_reorder=true"""
+ sql """set enable_profile=true"""
+ sql """set profile_level=2"""
+ sql """set enable_runtime_filter_partition_prune=false"""
+ "qt_${actionSuffix}_rf_disabled"(rfQuery)
+ sql """set enable_runtime_filter_partition_prune=true"""
+ assertRuntimeFilterPruned(
+ tableName, dimensionTable, [["1"], ["3"], ["5"], ["7"]])
+
+ // Scenario PM-R01: numeric snapshot, tag and branch bind their
historical schemas.
+ "qt_${actionSuffix}_base_snapshot"("""
+ select id, part, payload.label
+ from ${tableName} for version as of ${baseSnapshot}
+ where part in ('p1', 'p2') order by id
+ """)
+ "qt_${actionSuffix}_base_tag"("""
+ select id, part, payload.label
+ from ${tableName}@tag(${tableName}_base)
+ where part in ('p1', 'p2') order by id
+ """)
+ "qt_${actionSuffix}_base_branch"("""
+ select id, part, payload.label
+ from ${tableName}@branch(${tableName}_base_branch)
+ where part in ('p1', 'p2') order by id
+ """)
+ "qt_${actionSuffix}_added_snapshot_complex"("""
+ select id, payload.label,
+ element_at(attrs, 'k').code,
+ element_at(attrs, 'k').extra,
+ events[1].score,
+ events[1].extra
+ from ${tableName} for version as of ${addedSnapshot}
+ where part = 'p1'
+ order by id
+ """)
+ "qt_${actionSuffix}_final_tag"("""
+ select id from ${tableName}@tag(${tableName}_final)
+ where part = 'p1' order by id
+ """)
+ "qt_${actionSuffix}_base_snapshot_complex"("""
+ select id, element_at(attrs, 'k').code, events[1].score
+ from ${tableName} for version as of ${baseSnapshot}
+ order by id
+ """)
+ test {
+ sql """
+ select element_at(attrs, 'k').renamed_code
+ from ${tableName} for version as of ${baseSnapshot}
+ """
+ exception "renamed_code"
+ }
+ test {
+ sql """
+ select events[1].score
+ from ${tableName}@tag(${tableName}_final)
+ """
+ exception "score"
+ }
+
+ // Scenario PM-RD01: JNI and native readers agree for
partition-filtered historical
+ // and current complex projections, and the explain plan proves
each requested path.
+ String currentReaderQuery = """
+ select id, payload.renamed_label, payload.extra,
+ element_at(attrs, 'k').renamed_code,
+ events[1].renamed_score
+ from ${tableName}
+ where part in ('p1', 'p3')
+ order by id
+ """
+ String historicalReaderQuery = """
+ select id, payload.label, element_at(attrs, 'k').code,
events[1].score
+ from ${tableName} for version as of ${baseSnapshot}
+ where part in ('p1', 'p2')
+ order by id
+ """
+ sql """set enable_paimon_cpp_reader=false"""
+ sql """set force_jni_scanner=true"""
+ assertJniPath(currentReaderQuery, "${tableName} current")
+ assertJniPath(historicalReaderQuery, "${tableName} historical")
+ "qt_${actionSuffix}_jni_current_complex"(currentReaderQuery)
+ "qt_${actionSuffix}_jni_historical_complex"(historicalReaderQuery)
+ sql """set force_jni_scanner=false"""
+ sql """set enable_paimon_cpp_reader=true"""
+ assertNativePath(currentReaderQuery, "${tableName} current")
+ assertNativePath(historicalReaderQuery, "${tableName} historical")
+ "qt_${actionSuffix}_native_current_complex"(currentReaderQuery)
+
"qt_${actionSuffix}_native_historical_complex"(historicalReaderQuery)
+ assertEquals(finalSnapshot, latestSnapshotId(tableName))
+ }
+ } finally {
+ sql """set enable_paimon_cpp_reader=false"""
+ sql """set force_jni_scanner=false"""
+ sql """set enable_runtime_filter_prune=true"""
+ sql """set enable_runtime_filter_partition_prune=true"""
+ sql """set disable_join_reorder=false"""
+ sql """drop catalog if exists ${catalogName}"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
