github-actions[bot] commented on code in PR #66021: URL: https://github.com/apache/doris/pull/66021#discussion_r3649191231
########## regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_concurrent_merge_invariants.groovy: ########## @@ -0,0 +1,167 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import java.util.Collections +import java.util.concurrent.CountDownLatch + +suite("test_iceberg_write_concurrent_merge_invariants", + "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_write_concurrent_merge_invariants" + String dbName = "iceberg_write_concurrent_merge_invariants_db" + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + "type" = "iceberg", + "iceberg.catalog.type" = "rest", + "uri" = "http://${externalEnvIp}:${restPort}";, + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";, + "s3.region" = "us-east-1", + "meta.cache.iceberg.table.ttl-second" = "0", + "meta.cache.iceberg.schema.ttl-second" = "0" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + + sql """drop table if exists concurrent_merge""" + sql """ + create table concurrent_merge ( + id int not null, + region string, + payload string + ) + partition by list (region) () + properties ( + "format-version" = "2", + "write.format.default" = "parquet", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read", + "write.merge.isolation-level" = "serializable" + ) + """ + sql """insert into concurrent_merge values (1, 'A', 'base')""" + long snapshotsBefore = (sql """select count(*) from concurrent_merge\$snapshots""")[0][0] as long + + // WC02-S01: Start two conflicting MERGE statements at the same barrier. + // The exact winner is intentionally unspecified; cardinality, snapshot + // accounting and cross-engine visibility are deterministic invariants. + CountDownLatch start = new CountDownLatch(1) + List<String> successes = Collections.synchronizedList(new ArrayList<String>()) + List<String> failures = Collections.synchronizedList(new ArrayList<String>()) + + def first = thread { + start.await() + try { + sql """ + merge into ${catalogName}.${dbName}.concurrent_merge t + using (select 1 as id, 'B' as region, 'winner-one' as payload) s + on t.id = s.id + when matched then update set region = s.region, payload = s.payload + """ + successes.add("one") + } catch (Exception e) { + failures.add(e.getMessage()) Review Comment: [P1] Reject unrelated failures from the concurrent MERGE The one-loser outcome is valid only for an Iceberg validation/optimistic-commit conflict, but this catch accepts every `Exception`. If one session hits an unrelated planner/RPC/BE/catalog error and the other commits, all later assertions still pass. Re-throw unexpected exceptions and accept only the specific conflict class/message. ########## regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_merge_duplicate_source_negative.groovy: ########## @@ -0,0 +1,102 @@ +// 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_write_merge_duplicate_source_negative", + "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 knownBugEnabled = context.config.otherConfigs.get("enableIcebergKnownBugTest") + if (knownBugEnabled == null || !knownBugEnabled.equalsIgnoreCase("true")) { + logger.info("skip isolated Iceberg known-bug 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_write_merge_duplicate_source_negative" + String dbName = "iceberg_write_merge_duplicate_source_negative_db" + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + "type" = "iceberg", + "iceberg.catalog.type" = "rest", + "uri" = "http://${externalEnvIp}:${restPort}";, + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";, + "s3.region" = "us-east-1" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + + sql """drop table if exists duplicate_source_target""" + sql """ + create table duplicate_source_target ( + id int, + region string, + payload string + ) + partition by list (region) () + properties ( + "format-version" = "2", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read" + ) + """ + sql """insert into duplicate_source_target values (1, 'A', 'committed')""" + + long snapshotsBefore = + (sql """select count(*) from duplicate_source_target\$snapshots""")[0][0] as long + long filesBefore = + (sql """select count(*) from duplicate_source_target\$files""")[0][0] as long + + // Negative scenario: Iceberg MERGE cardinality permits only one source row + // to update a target row. The entire statement must fail before publishing. + test { + sql """ + merge into duplicate_source_target t + using ( + select 1 as id, 'B' as region, 'first-update' as payload + union all + select 1, 'C', 'second-update' + ) s + on t.id = s.id + when matched then update set + region = s.region, + payload = s.payload + """ + exception "more than one" + } + assertEquals(snapshotsBefore, + (sql """select count(*) from duplicate_source_target\$snapshots""")[0][0] as long) + assertEquals(filesBefore, + (sql """select count(*) from duplicate_source_target\$files""")[0][0] as long) + order_qt_duplicate_source_atomic_state """ Review Comment: [P1] Commit the expected outputs for the opt-in suites When this flag is enabled in a normal validation run, execution reaches this `order_qt_*`, but the suite has no matching `.out`; the framework throws `Missing outputFile` before it can validate the postcondition. The same omission affects `test_iceberg_write_merge_truncate_negative` and `test_iceberg_write_nullable_truncate_negative`. Please generate and commit all three output files/tags so these isolated regressions can pass once the product behavior is fixed. ########## regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_concurrent_merge_invariants.groovy: ########## @@ -0,0 +1,167 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +import java.util.Collections +import java.util.concurrent.CountDownLatch + +suite("test_iceberg_write_concurrent_merge_invariants", + "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_write_concurrent_merge_invariants" + String dbName = "iceberg_write_concurrent_merge_invariants_db" + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + "type" = "iceberg", + "iceberg.catalog.type" = "rest", + "uri" = "http://${externalEnvIp}:${restPort}";, + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";, + "s3.region" = "us-east-1", + "meta.cache.iceberg.table.ttl-second" = "0", + "meta.cache.iceberg.schema.ttl-second" = "0" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + + sql """drop table if exists concurrent_merge""" + sql """ + create table concurrent_merge ( + id int not null, + region string, + payload string + ) + partition by list (region) () + properties ( + "format-version" = "2", + "write.format.default" = "parquet", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read", + "write.merge.isolation-level" = "serializable" + ) + """ + sql """insert into concurrent_merge values (1, 'A', 'base')""" + long snapshotsBefore = (sql """select count(*) from concurrent_merge\$snapshots""")[0][0] as long + + // WC02-S01: Start two conflicting MERGE statements at the same barrier. + // The exact winner is intentionally unspecified; cardinality, snapshot + // accounting and cross-engine visibility are deterministic invariants. + CountDownLatch start = new CountDownLatch(1) + List<String> successes = Collections.synchronizedList(new ArrayList<String>()) + List<String> failures = Collections.synchronizedList(new ArrayList<String>()) + + def first = thread { + start.await() + try { + sql """ + merge into ${catalogName}.${dbName}.concurrent_merge t + using (select 1 as id, 'B' as region, 'winner-one' as payload) s + on t.id = s.id + when matched then update set region = s.region, payload = s.payload + """ + successes.add("one") + } catch (Exception e) { + failures.add(e.getMessage()) + } + } + def second = thread { + start.await() + try { + sql """ + merge into ${catalogName}.${dbName}.concurrent_merge t + using (select 1 as id, 'C' as region, 'winner-two' as payload) s + on t.id = s.id + when matched then update set region = s.region, payload = s.payload + """ + successes.add("two") + } catch (Exception e) { + failures.add(e.getMessage()) + } + } + start.countDown() Review Comment: [P1] Synchronize both workers before dispatch This latch only releases workers; it never proves that both have reached `await()` before `countDown()`. One closure can start after the other statement has already finished, and every assertion still passes (the append case repeats the pattern). Use two dedicated workers (or ensure pool capacity) plus a readiness barrier before release; if commit overlap itself is the contract, coordinate at the server/commit phase as well. ########## regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy: ########## @@ -0,0 +1,224 @@ +// 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_write_evolution_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_write_evolution_refs" + String dbName = "iceberg_write_evolution_refs_db" + + def latestSnapshotId = { + return (sql """ + select snapshot_id + from evolution_refs\$snapshots + order by committed_at desc + limit 1 + """)[0][0].toString() + } + + def assertSparkMatchesDoris = { String relation, String projection -> + sql """refresh table ${dbName}.evolution_refs""" + spark_iceberg """refresh table demo.${dbName}.evolution_refs""" + def sparkRows = spark_iceberg """ + select ${projection} + from demo.${dbName}.evolution_refs${relation} + order by id + """ + def dorisRows = sql """ + select ${projection} + from evolution_refs${relation} + order by id + """ + assertSparkDorisResultEquals(sparkRows, dorisRows) + } + + 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" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + sql """set enable_fallback_to_original_planner = false""" + + sql """drop table if exists evolution_refs""" + sql """ + create table evolution_refs ( + id int not null, + region string, + bucket_key string not null, + event_time datetime, + amount decimal(12, 2), + payload struct<metric:int, label:string> + ) + partition by list (region, bucket(4, bucket_key), day(event_time)) () + properties ( + "format-version" = "2", + "write.format.default" = "parquet", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read" + ) + """ + + // W01-S01: Doris writes the first snapshot using identity, string bucket and day transforms. + sql """ + insert into evolution_refs values + (1, 'CN', 'alpha', '2026-01-01 08:00:00', 10.10, struct(10, 'base-cn')), + (2, 'US', 'beta', '2026-01-02 09:00:00', 20.20, struct(20, 'base-us')), + (3, null, 'null-key', null, 30.30, struct(30, null)) + """ + String baseSnapshot = latestSnapshotId() + sql """alter table evolution_refs create tag base_tag as of version ${baseSnapshot}""" + sql """alter table evolution_refs create branch base_branch as of version ${baseSnapshot}""" + assertSparkMatchesDoris("", "id, region, bucket_key, event_time, amount") + + // W01-S02: Schema and partition spec evolve together before the next Doris write. + // Renaming the partition source column must preserve its Iceberg field id. + sql """alter table evolution_refs add column note string""" + sql """alter table evolution_refs rename column region zone""" + sql """ + alter table evolution_refs modify column payload struct< + metric:bigint, + label:string, + extra:string + > + """ + sql """ + alter table evolution_refs + replace partition key day(event_time) with month(event_time) as event_month + """ + sql """ + alter table evolution_refs + replace partition key bucket(4, bucket_key) with bucket(8, id) as id_bucket + """ + sql """alter table evolution_refs drop partition key region""" + sql """alter table evolution_refs add partition key truncate(2, bucket_key) as bucket_prefix""" + + sql """ + insert into evolution_refs values + (4, 'CN-east', 'gamma', '2026-02-01 10:00:00', 40.40, + struct(4000000000, 'new-cn', 'after-evolution'), 'new-spec'), + (5, 'DE-west', 'delta', '2026-03-02 11:00:00', 50.50, + struct(50, 'new-de', null), null), + (6, null, 'epsilon', null, 60.60, + struct(60, null, 'null-partition'), 'null-zone') + """ + String evolvedSnapshot = latestSnapshotId() + sql """alter table evolution_refs create tag evolved_tag as of version ${evolvedSnapshot}""" + + // W01-S03: Source-column filters must cover files written with both partition specs. + order_qt_current_rows """ + select id, zone, bucket_key, event_time, amount, payload.metric, payload.extra, note + from evolution_refs + order by id + """ + order_qt_cross_spec_zone_filter """ + select id from evolution_refs + where zone = 'CN' or zone like 'CN-%' + order by id + """ + order_qt_cross_spec_time_filter """ + select id from evolution_refs + where event_time is null or event_time >= timestamp '2026-02-01 00:00:00' + order by id + """ + order_qt_partition_specs """ + select spec_id, sum(record_count) + from evolution_refs\$partitions + group by spec_id + order by spec_id + """ + assertSparkMatchesDoris("", "id, zone, bucket_key, event_time, amount") + + // W01-S04: Numeric snapshot and tag retain both base data and the historical schema. + order_qt_base_snapshot """ + select id, region + from evolution_refs for version as of ${baseSnapshot} + order by id + """ + order_qt_base_tag """ + select id, region + from evolution_refs@tag(base_tag) + order by id + """ + order_qt_evolved_tag """ + select id, zone, note + from evolution_refs@tag(evolved_tag) + order by id + """ + + // W01-S05: A branch created before both evolutions accepts the current schema/spec. + // Its commit and full overwrite must not change main or the protected base tag. + sql """ + insert into evolution_refs@branch(base_branch) + (id, zone, bucket_key, event_time, amount, payload, note) + values + (7, 'JP-east', 'branch-a', '2026-04-01 12:00:00', 70.70, + struct(70, 'branch', 'current-schema'), 'branch-insert') + """ + order_qt_branch_after_insert """ + select id, zone, note + from evolution_refs@branch(base_branch) + order by id + """ + order_qt_main_unchanged_after_branch_insert """ + select id from evolution_refs order by id + """ + + sql """ + insert overwrite table evolution_refs@branch(base_branch) Review Comment: [P1] Overwrite an existing partition in these scenarios This write moves from April to May, so it targets a previously empty current-spec partition and an append produces the same output. The branch-boundary suite likewise uses a new identity partition, and the overwrite-evolution suite creates spec 5 immediately before its overwrite without seeding that spec. Seed and overwrite the same current partition, then assert the seeded row disappears while unrelated main/tag data remains. ########## regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md: ########## @@ -0,0 +1,120 @@ +<!-- +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. +--> + +# Iceberg 写入 P0 覆盖矩阵 + +## 范围与判定原则 + +本文档覆盖 Doris 向 Iceberg 表写入时的正确性、兼容性和失败原子性。矩阵既检查单项能力，也检查 schema change、Partition Evolution、snapshot/tag/branch、行级 delete/update/merge、表模型、分区与 bucket、数据类型和 NULL 语义之间的交互。 + +所有正向写入场景都以 Doris 确定性查询结果和 Spark 读取同一 Iceberg 表的结果一致作为双重 oracle；涉及历史引用时，另行校验 snapshot、tag 和 branch 的隔离性。 + +覆盖状态含义： + +- 已覆盖（验证通过）：P0 suite 对该场景有确定性结果断言，且已在多 BE 环境验证。 +- 预期拒绝：Doris 明确不支持该操作，P0 suite 验证错误信息与失败原子性。 +- 已覆盖（隔离负向）：已形成可复现产品问题的 regression；默认 P0 隔离运行，避免杀死共享 BE 或提交不可读文件。 + +## 风险点 + +| 编号 | 风险描述 | 来源 | 影响面 | 级别 | +| --- | --- | --- | --- | --- | +| R01 | schema change 后 writer 仍按旧列位置或旧 field id 写入，造成静默错列 | 白盒：Iceberg field id 与 Doris slot 映射 | 数据正确性 | P0 | +| R02 | Partition Evolution 后新文件落入旧 spec、分区值计算错误，或跨 spec 过滤漏数 | 黑盒 + 白盒：多 partition spec 并存 | 写入与查询正确性 | P0 | +| R03 | 字符串、数值、日期时间、decimal、布尔和 NULL 作为 identity/bucket/truncate/time transform 源时行为不一致 | 黑盒：类型与边界输入 | 分区路由、裁剪 | P0 | +| R04 | schema/partition 演进后 snapshot、tag、branch 绑定了错误 schema 或数据版本 | 黑盒：历史读与引用 | time travel 正确性 | P0 | +| R05 | MOR 的 delete/update/merge 跨新旧 spec 时生成错误 delete file；COW 被拒绝后仍发布快照 | 白盒：row-level DML commit | 数据丢失、失败原子性 | P0 | +| R06 | Duplicate、Unique MOW、Unique MOR、Aggregate 源表语义在 INSERT SELECT 时被改变 | 黑盒：不同 Doris 表模型 | 跨表写入正确性 | P0 | +| R07 | RANGE/LIST/无分区源表以及 HASH/RANDOM/AUTO bucket 在多 BE 执行时产生重复或丢行 | 黑盒 + 白盒：分布式 exchange 与 sink writer | 分布式写入正确性 | P0 | +| R08 | primitive、ARRAY、MAP、STRUCT 及嵌套 NULL 在 schema change 前后写入错误 | 黑盒：复杂类型与 NULL | 数据正确性、兼容性 | P0 | +| R09 | NULL 写入 Iceberg required 列未报错、部分数据或空快照被提交 | 白盒：required 校验与 commit | 约束、失败原子性 | P0 | +| R10 | INSERT OVERWRITE 在演进后的当前 spec、branch 或 NULL 分区上误删其他分区 | 黑盒：覆盖写 | 数据丢失 | P0 | +| R11 | 单 BE 可通过但多 BE 并发 sink 出现文件名、commit 或分区冲突 | 白盒：并行 writer 与统一 commit | 分布式稳定性 | P0 | +| R12 | nullable STRING 或 DML 产生的 Nullable block 经过 truncate transform 时 BE FATAL | 白盒：partition transformer 列类型约束 | 集群可用性 | P0 | +| R13 | MERGE 的多个源行匹配同一目标行时未执行基数校验，错误提交重复数据 | 黑盒 + 白盒：MERGE cardinality 与 commit | 数据正确性 | P0 | +| R14 | branch 写入污染 main，或 tag/不支持的 branch 行级 DML 失败后仍发布快照 | 黑盒：reference write 边界 | 历史引用、失败原子性 | P0 | +| R15 | 当前 spec 覆盖写未正确清理旧 spec delete file，或失败覆盖写留下部分快照 | 白盒：overwrite commit 与 delete file | 数据丢失、失败原子性 | P0 | +| R16 | CTAS 对复杂类型、NULL、分区 transform、文件格式和失败清理的行为不一致 | 黑盒：DDL + writer 一体提交 | schema、文件格式、原子性 | P0 | +| R17 | sort order、distribution mode、多次文件 flush 和并发 commit 组合导致乱序、丢行或重复提交 | 白盒：exchange、sort writer、optimistic commit | 分布式正确性、稳定性 | P0 | +| R18 | STRING identity/bucket/truncate 对空串、中文、emoji、组合字符和 NULL 的物理分区值计算错误 | 黑盒：UTF-8 transform metadata | 分区路由、裁剪 | P0 | + +## 组合覆盖 + +| 维度 | 场景 | 状态 | P0 suite | +| --- | --- | --- | --- | +| 基础写入 | Parquet/ORC、primitive/复杂类型、INSERT/OVERWRITE | 已覆盖 | `test_iceberg_write_insert`、`test_iceberg_insert_overwrite` | +| Partition transform | identity、bucket、truncate、year/month/day/hour | 已覆盖 | `test_iceberg_write_transform_partitions`、`test_iceberg_static_partition_overwrite` | +| schema + partition 演进 | add/rename/drop/type promotion 与 ADD/REPLACE/DROP partition field 后继续写入和过滤 | 已覆盖（验证通过） | `test_iceberg_write_evolution_refs` | +| 复杂类型演进 | ARRAY/MAP/STRUCT promotion、STRUCT 新增字段、旧文件与新写入并存 | 已覆盖（验证通过） | `test_iceberg_write_complex_evolution` | +| 历史版本 | 演进前后 snapshot、tag、branch；branch 独立写入和覆盖写 | 已覆盖（验证通过） | `test_iceberg_write_evolution_refs` | +| MOR | partition evolution 后 DELETE/UPDATE/MERGE，校验当前、delete files 与历史版本 | 已覆盖（验证通过） | `test_iceberg_write_dml_modes_evolution` | +| COW | partition evolution 后 DELETE/UPDATE/MERGE 拒绝，且数据和 snapshot 数不变 | 预期拒绝 | `test_iceberg_write_dml_modes_evolution` | +| Doris 源表模型 | Duplicate、Unique MOW、Unique MOR、Aggregate | 已覆盖（验证通过） | `test_iceberg_write_source_models` | +| Doris 源分区 | 无分区、RANGE、LIST | 已覆盖（验证通过） | `test_iceberg_write_source_models` | +| Doris 源 bucket | HASH 固定 bucket、RANDOM bucket、HASH AUTO bucket | 已覆盖（验证通过） | `test_iceberg_write_source_models` | +| 分区源类型 | STRING/INT/BIGINT/DATE/DATETIME/DECIMAL 的 bucket 与适用 transform；BOOLEAN identity 与非法 bucket | 已覆盖（验证通过） | `test_iceberg_write_partition_types_null` | +| NULL 分区 | identity NULL、数值/decimal bucket 与 truncate NULL、time transform NULL、多列组合 NULL | 已覆盖（验证通过） | `test_iceberg_write_partition_types_null` | +| nullable STRING truncate | nullable STRING 经过 truncate transform 的 INSERT，以及 NOT NULL 源列经 UPDATE block 写入 | 已覆盖（隔离负向） | `test_iceberg_write_nullable_truncate_negative` | +| MERGE 完整语义 | 条件 MATCHED、DELETE/UPDATE、多个条件 NOT MATCHED、NULL-safe 与普通 NULL key | 已覆盖（验证通过） | `test_iceberg_write_merge_semantics` | +| MERGE 基数约束 | 多个源行匹配同一目标行必须整句失败且不发布快照 | 已覆盖（隔离负向） | `test_iceberg_write_merge_duplicate_source_negative` | +| MERGE + STRING truncate | required truncate 源列经 MERGE nullable projection 写入 | 已覆盖（隔离负向） | `test_iceberg_write_merge_truncate_negative` | +| branch/tag 写入边界 | branch INSERT/OVERWRITE 隔离；tag 写入和 branch DELETE/UPDATE/MERGE 明确拒绝 | 已覆盖（验证通过） | `test_iceberg_write_branch_dml_boundary` | +| nullable 数据 | 顶层 NULL、ARRAY NULL 元素、MAP NULL value、STRUCT NULL child | 已覆盖并增强 | `test_iceberg_write_insert`、`test_iceberg_write_complex_evolution` | +| required 列正向与 schema change | required 列合法写入、nullable 列写 NULL、增加 required 列与 nullable→required 拒绝 | 已覆盖（验证通过） | `test_iceberg_write_nullability_atomicity` | +| required 列写 NULL | VALUES 与分布式 INSERT SELECT 混合批次写 NULL | 已覆盖（隔离负向） | `test_iceberg_write_required_null_values_negative`、`test_iceberg_write_required_null_select_negative` | +| 覆盖写 | 当前 spec、静态分区、branch、空输入、连续多次 partition evolution、NULL 当前分区 | 已覆盖并增强 | `test_iceberg_static_partition_overwrite`、`test_iceberg_write_evolution_refs`、`test_iceberg_write_overwrite_evolution` | +| 覆盖写 + delete files | MOR DELETE/UPDATE/MERGE 后覆盖写，演进前后 delete files 与历史 tag 共存 | 已覆盖（验证通过） | `test_iceberg_write_overwrite_delete_files` | +| 覆盖写失败原子性 | main/branch 分布式严格类型转换失败、快照/文件/数据不变、修正后重试 | 已覆盖（验证通过） | `test_iceberg_write_overwrite_atomicity` | +| STRING 物理 transform | identity、nullable bucket、required truncate 的 UTF-8 边界值及 transform width evolution | 已覆盖（验证通过） | `test_iceberg_write_string_transform_metadata` | +| CTAS | 复杂类型、嵌套 NULL、identity+bucket、ORC 压缩、失败建表清理 | 已覆盖（验证通过） | `test_iceberg_write_ctas_format_boundary` | +| 文件格式边界 | Parquet/ORC 正向写入；Avro 表写入明确拒绝并保持快照和文件不变 | 已覆盖（正向 + 预期拒绝） | `test_iceberg_write_ctas_format_boundary` | +| 排序与分布属性 | 多列 sort order、NULL ordering、none/hash/range distribution、强制多文件 flush | 已覆盖（验证通过） | `test_iceberg_write_order_distribution_properties` | +| 并发写入 | 同行冲突 MERGE 的串行化不变量、非冲突分布式 append | 已覆盖（验证通过） | `test_iceberg_write_concurrent_merge_invariants` | +| 分布式执行 | 多 bucket 源表、多分区 Iceberg sink、多 BE writer、suite 间无共享 catalog/database | 已覆盖（验证通过） | 所有本次新增 suite | +| Spark 交叉验证 | Doris 写入后由 Spark 与 Doris 查询同一 Iceberg 表并逐行比较，含行数据和物理分区 metadata | 已覆盖（验证通过） | 十五个正向 suite | Review Comment: [P1] Align the coverage matrix with the executed oracles This summary says all fifteen positive suites have Spark/Doris cross-validation, but the branch-boundary and overwrite-atomicity suites have none, and the concurrent appends plus evolved branch writes occur after the last comparison. Separately, the nullable-truncate row claims an UPDATE-produced Nullable block although that suite contains no UPDATE. Add those scenarios/oracles or mark them uncovered instead of reporting complete P0 coverage. ########## regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_order_distribution_properties.groovy: ########## @@ -0,0 +1,205 @@ +// 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_write_order_distribution_properties", + "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_write_order_distribution_properties" + String dbName = "iceberg_write_order_distribution_properties_db" + + sql """drop catalog if exists ${catalogName}""" + sql """ + create catalog ${catalogName} properties ( + "type" = "iceberg", + "iceberg.catalog.type" = "rest", + "uri" = "http://${externalEnvIp}:${restPort}";, + "s3.access_key" = "admin", + "s3.secret_key" = "password", + "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";, + "s3.region" = "us-east-1", + "meta.cache.iceberg.table.ttl-second" = "0", + "meta.cache.iceberg.schema.ttl-second" = "0" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + + sql """drop table if exists ordered_evolution""" + sql """ + create table ordered_evolution ( + id int, + region string, + payload string, + score int + ) + order by (region asc nulls last, id desc nulls first) + partition by list (region) () + properties ( + "format-version" = "2", + "write.format.default" = "parquet", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read", + "write.distribution-mode" = "range" + ) + """ + + // WP01-S01: The planned Iceberg write contains the declared global order, + // including direction and NULL ordering. + explain { + sql """ + insert into ordered_evolution + select number, + if(number % 4 = 0, null, concat('R', number % 3)), + concat('payload-', number), + number + from numbers('number' = '32') + """ + contains "ORDER BY (`region` ASC NULLS LAST, `id` DESC NULLS FIRST)" + } + + // WP01-S02: Force several sorted writer flushes on a distributed source. + // This exercises global order, NULL partitions and file rollover together. + sql """set iceberg_write_target_file_size_bytes = 51200""" + sql """ + insert into ordered_evolution + select number, + if(number % 7 = 0, null, concat('R', number % 5)), + concat('payload-', number, '-', repeat('x', 64)), + number + from numbers('number' = '10000') + """ + def filesAfterInsert = sql """ + select count(*), sum(record_count) + from ordered_evolution\$files + """ + assertTrue((filesAfterInsert[0][0] as long) > 1L) Review Comment: [P2] Make the file-count oracle depend on rollover The input already spans six Iceberg partitions, so `$files` is greater than one even if `iceberg_write_target_file_size_bytes` is ignored and every partition writes exactly one file. Assert that at least one physical partition owns multiple data files, or use an unpartitioned sink, so the check specifically proves target-size rollover. ########## regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_dml_modes_evolution.groovy: ########## @@ -0,0 +1,249 @@ +// 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_write_dml_modes_evolution", + "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_write_dml_modes_evolution" + String dbName = "iceberg_write_dml_modes_evolution_db" + + def assertSparkMatchesDoris = { String tableName -> + sql """refresh table ${dbName}.${tableName}""" + spark_iceberg """refresh table demo.${dbName}.${tableName}""" + def sparkRows = spark_iceberg """ + select id, region, bucket_key, event_time, score, status + from demo.${dbName}.${tableName} + order by id + """ + def dorisRows = sql """ + select id, region, bucket_key, event_time, score, status + from ${tableName} + order by id + """ + assertSparkDorisResultEquals(sparkRows, dorisRows) + } + + 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" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + sql """set enable_fallback_to_original_planner = false""" + + sql """drop table if exists mor_evolution""" + sql """ + create table mor_evolution ( + id int not null, + region string, + bucket_key string not null, + event_time datetime, + score int + ) + partition by list (region, bucket(4, bucket_key), day(event_time)) () + properties ( + "format-version" = "2", + "write.format.default" = "parquet", + "write.delete.mode" = "merge-on-read", + "write.update.mode" = "merge-on-read", + "write.merge.mode" = "merge-on-read" + ) + """ + + // W03-S01: The MOR baseline includes NULL in every partition transform family. + sql """ + insert into mor_evolution values + (1, 'A', 'alpha', '2026-01-01 01:00:00', 10), + (2, 'B', 'beta', '2026-01-02 02:00:00', 20), + (3, null, 'null-key', null, 30), + (4, 'A', 'delta', '2026-01-04 04:00:00', 40) + """ + String morBaseSnapshot = (sql """ + select snapshot_id from mor_evolution\$snapshots + order by committed_at desc limit 1 + """)[0][0].toString() + sql """alter table mor_evolution create tag mor_base as of version ${morBaseSnapshot}""" + + // W03-S02: Evolve schema and partition spec, then write more files before row-level DML. + sql """alter table mor_evolution add column status string""" + sql """ + alter table mor_evolution + replace partition key day(event_time) with month(event_time) as event_month + """ + sql """ + alter table mor_evolution + replace partition key bucket(4, bucket_key) with bucket(8, id) as id_bucket + """ + sql """ + insert into mor_evolution values + (5, 'B', 'echo', '2026-02-01 05:00:00', 50, 'new-spec'), + (6, null, 'foxtrot', null, 60, 'new-null'), + (7, 'C', 'golf', '2026-03-01 07:00:00', 70, null) + """ + String morBeforeDmlSnapshot = (sql """ + select snapshot_id from mor_evolution\$snapshots + order by committed_at desc limit 1 + """)[0][0].toString() + sql """alter table mor_evolution create tag mor_before_dml as of version ${morBeforeDmlSnapshot}""" + + // W03-S03: DELETE spans old/new specs and removes NULL partition rows. + sql """delete from mor_evolution where region is null""" + + // W03-S04: UPDATE changes partition source values in files from both specs. + sql """ + update mor_evolution + set region = concat(region, '-updated'), + score = score + 100, + status = 'updated' + where region = 'A' + """ + + // W03-S05: MERGE deletes, updates and inserts across different transformed partitions. + sql """ + merge into mor_evolution t + using ( + select 2 as id, 'B-merged' as region, 'beta-merged' as bucket_key, + timestamp '2026-04-02 02:00:00' as event_time, 220 as score, + 'U' as op + union all + select 5, 'B', 'echo', timestamp '2026-02-01 05:00:00', 50, 'D' + union all + select 8, 'D', 'hotel', timestamp '2026-05-01 08:00:00', 80, 'I' + ) s + on t.id = s.id + when matched and s.op = 'D' then delete + when matched then update set + region = s.region, + bucket_key = s.bucket_key, + event_time = s.event_time, + score = s.score, + status = 'merged' + when not matched then insert (id, region, bucket_key, event_time, score, status) + values (s.id, s.region, s.bucket_key, s.event_time, s.score, 'inserted') + """ + + order_qt_mor_current """ + select id, region, bucket_key, event_time, score, status + from mor_evolution + order by id + """ + order_qt_mor_base_tag """ + select id, region, bucket_key, event_time, score + from mor_evolution@tag(mor_base) + order by id + """ + order_qt_mor_before_dml_tag """ + select id, region, bucket_key, event_time, score, status + from mor_evolution@tag(mor_before_dml) + order by id + """ + order_qt_mor_delete_files """ + select spec_id, count(*), sum(record_count) + from mor_evolution\$delete_files + group by spec_id + order by spec_id + """ + assertSparkMatchesDoris("mor_evolution") + + // W03-S06: COW accepts INSERT after partition evolution but Doris explicitly rejects + // DELETE/UPDATE/MERGE. Each rejection must leave both data and snapshot count unchanged. + sql """drop table if exists cow_evolution""" + sql """ + create table cow_evolution ( + id int not null, + region string, + bucket_key string not null, + event_time datetime, + score int, + status string + ) + partition by list (region, bucket(4, bucket_key), day(event_time)) () + properties ( + "format-version" = "2", + "write.format.default" = "orc", + "write.delete.mode" = "copy-on-write", + "write.update.mode" = "copy-on-write", + "write.merge.mode" = "copy-on-write" + ) + """ + sql """ + insert into cow_evolution values + (1, 'A', 'alpha', '2026-01-01 01:00:00', 10, 'base'), + (2, null, 'null-key', null, 20, 'null-partition') + """ + sql """ + alter table cow_evolution + replace partition key bucket(4, bucket_key) with bucket(8, id) as id_bucket + """ + sql """ + alter table cow_evolution + replace partition key day(event_time) with month(event_time) as event_month + """ + sql """ + insert into cow_evolution values + (3, 'B', 'beta', '2026-02-01 03:00:00', 30, 'new-spec') + """ + + long cowSnapshots = (sql """select count(*) from cow_evolution\$snapshots""")[0][0] as long + test { + sql """delete from cow_evolution where region is null""" + exception "Doris does not support DELETE on Iceberg copy-on-write tables" Review Comment: [P2] Check both error substrings explicitly `TestAction` has a single `exception` field, so the second call overwrites this one; the advertised `Doris does not support ...` text is never asserted (the tag test repeats the pattern). Use `check` to require both substrings, or assert one distinctive contiguous message. ########## regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_required_null_select_negative.groovy: ########## @@ -0,0 +1,92 @@ +// 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_write_required_null_select_negative", + "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 + } + + // This opt-in switch isolates a write that can publish an unreadable Iceberg data file. + String knownBugTestEnabled = context.config.otherConfigs.get("enableIcebergKnownBugTest") + if (knownBugTestEnabled == null || !knownBugTestEnabled.equalsIgnoreCase("true")) { + logger.info("skip isolated Iceberg known-bug regression") + 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_write_required_null_select_negative" + String dbName = "iceberg_write_required_null_select_negative_db" + String internalDb = "iceberg_write_required_null_select_negative_internal_db" + + sql """drop database if exists internal.${internalDb} force""" + sql """create database internal.${internalDb}""" + sql """ + create table internal.${internalDb}.nullable_source ( + id int, + required_text string + ) + duplicate key(id) + distributed by hash(id) buckets 3 + properties ("replication_num" = "1") + """ + sql """ + insert into internal.${internalDb}.nullable_source values + (1, 'valid'), + (2, null) + """ + + 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" + ) + """ + sql """switch ${catalogName}""" + sql """drop database if exists ${dbName} force""" + sql """create database ${dbName}""" + sql """use ${dbName}""" + + sql """ + create table required_select ( + id int not null, + required_text string not null + ) + partition by list (bucket(8, id)) () + properties ("format-version" = "2") + """ + + // W07-S02: A mixed distributed INSERT SELECT must reject the whole statement atomically. + test { + sql """ + insert into required_select + select id, required_text + from internal.${internalDb}.nullable_source + """ + exception "null" Review Comment: [P1] Assert that the failed required-field write publishes nothing `exception "null"` alone does not prove statement atomicity: a late error after a snapshot/file was published would still pass this test. Capture the empty table's snapshot/file counts (and visible rows) before the write and assert they remain unchanged afterward; apply the same postcondition to the VALUES variant. ########## regression-test/suites/external_table_p0/iceberg/write/ICEBERG_WRITE_P0_COVERAGE.md: ########## @@ -0,0 +1,120 @@ +<!-- +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. +--> + +# Iceberg 写入 P0 覆盖矩阵 + +## 范围与判定原则 + +本文档覆盖 Doris 向 Iceberg 表写入时的正确性、兼容性和失败原子性。矩阵既检查单项能力，也检查 schema change、Partition Evolution、snapshot/tag/branch、行级 delete/update/merge、表模型、分区与 bucket、数据类型和 NULL 语义之间的交互。 + +所有正向写入场景都以 Doris 确定性查询结果和 Spark 读取同一 Iceberg 表的结果一致作为双重 oracle；涉及历史引用时，另行校验 snapshot、tag 和 branch 的隔离性。 + +覆盖状态含义： + +- 已覆盖（验证通过）：P0 suite 对该场景有确定性结果断言，且已在多 BE 环境验证。 +- 预期拒绝：Doris 明确不支持该操作，P0 suite 验证错误信息与失败原子性。 +- 已覆盖（隔离负向）：已形成可复现产品问题的 regression；默认 P0 隔离运行，避免杀死共享 BE 或提交不可读文件。 + +## 风险点 + +| 编号 | 风险描述 | 来源 | 影响面 | 级别 | +| --- | --- | --- | --- | --- | +| R01 | schema change 后 writer 仍按旧列位置或旧 field id 写入，造成静默错列 | 白盒：Iceberg field id 与 Doris slot 映射 | 数据正确性 | P0 | +| R02 | Partition Evolution 后新文件落入旧 spec、分区值计算错误，或跨 spec 过滤漏数 | 黑盒 + 白盒：多 partition spec 并存 | 写入与查询正确性 | P0 | +| R03 | 字符串、数值、日期时间、decimal、布尔和 NULL 作为 identity/bucket/truncate/time transform 源时行为不一致 | 黑盒：类型与边界输入 | 分区路由、裁剪 | P0 | +| R04 | schema/partition 演进后 snapshot、tag、branch 绑定了错误 schema 或数据版本 | 黑盒：历史读与引用 | time travel 正确性 | P0 | +| R05 | MOR 的 delete/update/merge 跨新旧 spec 时生成错误 delete file；COW 被拒绝后仍发布快照 | 白盒：row-level DML commit | 数据丢失、失败原子性 | P0 | +| R06 | Duplicate、Unique MOW、Unique MOR、Aggregate 源表语义在 INSERT SELECT 时被改变 | 黑盒：不同 Doris 表模型 | 跨表写入正确性 | P0 | +| R07 | RANGE/LIST/无分区源表以及 HASH/RANDOM/AUTO bucket 在多 BE 执行时产生重复或丢行 | 黑盒 + 白盒：分布式 exchange 与 sink writer | 分布式写入正确性 | P0 | +| R08 | primitive、ARRAY、MAP、STRUCT 及嵌套 NULL 在 schema change 前后写入错误 | 黑盒：复杂类型与 NULL | 数据正确性、兼容性 | P0 | +| R09 | NULL 写入 Iceberg required 列未报错、部分数据或空快照被提交 | 白盒：required 校验与 commit | 约束、失败原子性 | P0 | +| R10 | INSERT OVERWRITE 在演进后的当前 spec、branch 或 NULL 分区上误删其他分区 | 黑盒：覆盖写 | 数据丢失 | P0 | +| R11 | 单 BE 可通过但多 BE 并发 sink 出现文件名、commit 或分区冲突 | 白盒：并行 writer 与统一 commit | 分布式稳定性 | P0 | +| R12 | nullable STRING 或 DML 产生的 Nullable block 经过 truncate transform 时 BE FATAL | 白盒：partition transformer 列类型约束 | 集群可用性 | P0 | +| R13 | MERGE 的多个源行匹配同一目标行时未执行基数校验，错误提交重复数据 | 黑盒 + 白盒：MERGE cardinality 与 commit | 数据正确性 | P0 | +| R14 | branch 写入污染 main，或 tag/不支持的 branch 行级 DML 失败后仍发布快照 | 黑盒：reference write 边界 | 历史引用、失败原子性 | P0 | +| R15 | 当前 spec 覆盖写未正确清理旧 spec delete file，或失败覆盖写留下部分快照 | 白盒：overwrite commit 与 delete file | 数据丢失、失败原子性 | P0 | +| R16 | CTAS 对复杂类型、NULL、分区 transform、文件格式和失败清理的行为不一致 | 黑盒：DDL + writer 一体提交 | schema、文件格式、原子性 | P0 | +| R17 | sort order、distribution mode、多次文件 flush 和并发 commit 组合导致乱序、丢行或重复提交 | 白盒：exchange、sort writer、optimistic commit | 分布式正确性、稳定性 | P0 | +| R18 | STRING identity/bucket/truncate 对空串、中文、emoji、组合字符和 NULL 的物理分区值计算错误 | 黑盒：UTF-8 transform metadata | 分区路由、裁剪 | P0 | + +## 组合覆盖 + +| 维度 | 场景 | 状态 | P0 suite | +| --- | --- | --- | --- | +| 基础写入 | Parquet/ORC、primitive/复杂类型、INSERT/OVERWRITE | 已覆盖 | `test_iceberg_write_insert`、`test_iceberg_insert_overwrite` | +| Partition transform | identity、bucket、truncate、year/month/day/hour | 已覆盖 | `test_iceberg_write_transform_partitions`、`test_iceberg_static_partition_overwrite` | +| schema + partition 演进 | add/rename/drop/type promotion 与 ADD/REPLACE/DROP partition field 后继续写入和过滤 | 已覆盖（验证通过） | `test_iceberg_write_evolution_refs` | +| 复杂类型演进 | ARRAY/MAP/STRUCT promotion、STRUCT 新增字段、旧文件与新写入并存 | 已覆盖（验证通过） | `test_iceberg_write_complex_evolution` | +| 历史版本 | 演进前后 snapshot、tag、branch；branch 独立写入和覆盖写 | 已覆盖（验证通过） | `test_iceberg_write_evolution_refs` | +| MOR | partition evolution 后 DELETE/UPDATE/MERGE，校验当前、delete files 与历史版本 | 已覆盖（验证通过） | `test_iceberg_write_dml_modes_evolution` | +| COW | partition evolution 后 DELETE/UPDATE/MERGE 拒绝，且数据和 snapshot 数不变 | 预期拒绝 | `test_iceberg_write_dml_modes_evolution` | +| Doris 源表模型 | Duplicate、Unique MOW、Unique MOR、Aggregate | 已覆盖（验证通过） | `test_iceberg_write_source_models` | +| Doris 源分区 | 无分区、RANGE、LIST | 已覆盖（验证通过） | `test_iceberg_write_source_models` | +| Doris 源 bucket | HASH 固定 bucket、RANDOM bucket、HASH AUTO bucket | 已覆盖（验证通过） | `test_iceberg_write_source_models` | +| 分区源类型 | STRING/INT/BIGINT/DATE/DATETIME/DECIMAL 的 bucket 与适用 transform；BOOLEAN identity 与非法 bucket | 已覆盖（验证通过） | `test_iceberg_write_partition_types_null` | Review Comment: [P1] Make the transform coverage depend on physical values Several rows marked verified are not observable in the cited tests: numeric/temporal cases assert only source rows and spec totals; evolved STRING widths are never read from `$partitions`; the replaced-bucket filter predicates only unchanged `p_string`; CTAS's three distinct identity regions make its partition count independent of `bucket(4,id)` (and LZ4 is unchecked); and the truncate opt-ins only full-scan logical rows. Add metadata/predicate/property assertions that fail when each transform or writer property is wrong, or narrow these coverage claims. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
