github-actions[bot] commented on code in PR #65960: URL: https://github.com/apache/doris/pull/65960#discussion_r3637619902
########## regression-test/suites/external_table_p0/iceberg_paimon_schema_time_travel_coverage.md: ########## @@ -0,0 +1,119 @@ +# Iceberg / Paimon Schema Evolution × Time Travel P0 Coverage + +## Scope + +This document maps the P0 regression suites that validate Doris reads after +Iceberg or Paimon schema evolution. The matrix intentionally combines schema +changes with snapshot, tag, branch, time travel, delete/upsert and reader +variants instead of testing these dimensions in isolation. + +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. + +## Schema operations + +| ID | Operation | Iceberg | Paimon | P0 contract | +| --- | --- | --- | --- | --- | +| S01 | Add one nullable top-level field | Positive | Positive | Old refs reject the new field; new refs backfill NULL | +| S02 | Add multiple top-level fields | Positive | Positive | Order, types and NULL backfill are snapshot-local | +| S03 | Reorder / FIRST / AFTER | Positive | Positive | Field IDs remain stable across old and new refs | +| S04 | Rename top-level field | Positive | Positive | Old and new names bind only in their own schemas | +| S05 | Case-only or mixed-case rename | Positive | Positive | Case handling does not hide schema selection errors | +| S06 | Drop top-level field | Positive | Positive | Old refs retain the field; current refs reject it | +| S07 | Drop and re-add the same name | Positive | Positive | Old and new field IDs never share values | +| S08 | Compatible type promotion | Positive | Positive | Historical and current types remain correct | +| S09 | Add STRUCT child | Positive | Positive | Nested field is visible only after its schema version | +| S10 | Rename STRUCT child | Positive | Positive | Nested old/new paths have snapshot-local binding | +| S11 | Drop STRUCT child | Positive | Positive | Historical nested projections remain readable | +| S12 | Drop and re-add STRUCT child | Positive | Positive | Nested field IDs never leak values | +| S13 | Promote/reorder STRUCT child | Positive | Positive | Nested predicate and projection use the right type | +| S14 | MAP value STRUCT evolution | Positive | Positive | `element_at(...).field` uses the selected schema | +| S15 | ARRAY element STRUCT evolution | Positive | Positive | `array[i].field` uses the selected schema | +| S16 | Partition-column mutation | Positive/restricted | Negative format contract | Unsupported mutations are rejected atomically | +| S17 | Partition evolution | Positive | Format-constrained contract | Data and payload evolution preserve partition reads | +| S18 | PK-table non-key evolution | N/A | Positive | Upsert/delete/DV remain correct through evolution | +| S19 | Comment/default/nullability | Positive and negative | Positive and negative | Metadata-only changes and rejections are atomic | +| S20 | Narrowing, map-key, key/partition mutations | Negative | Negative | No schema, snapshot or cached state is polluted | + +## Historical references and actions + +| ID | Reference/action | Iceberg | Paimon | Coverage | +| --- | --- | --- | --- | --- | +| T00 | Latest/current | Yes | Yes | Current schema, explicit projection, predicate, aggregate | +| T01/T02 | Pre/post numeric snapshot | Yes | Yes | Old/new field binding and row visibility | +| T03 | Timestamp string | Yes | Yes | Resolves the expected pre-change schema | +| T04 | Epoch millis | Stable rejection | Yes | Iceberg syntax rejection; Paimon positive read | +| T05/T06/T07 | Pre/post tag forms | Yes | Yes | `FOR VERSION AS OF` and `@tag` | +| T08 | Pre-change branch | Negative product contract | Yes | Branch schema/data are compared with the format oracle | +| T09 | Independent branch evolution | Negative product contract | Negative product contract | Failure is isolated without crashing the shared BE | +| T10 | Expired snapshot retained by tag | Existing lifecycle + matrix | Yes | Retained tag never falls back to latest | +| T11 | Missing snapshot/tag | Yes | Yes | Stable error, never latest fallback | +| T12/T13 | Dual historical relations | Negative product contract | Negative product contract | Join, reverse join, UNION, nested UNION, CTE, subquery | +| T14 | Incremental read across change | N/A | Yes | JNI/CPP-supported paths use the end schema | +| T15 | Rollback to snapshot/timestamp | Yes | N/A | Data rolls back while Iceberg current schema semantics remain correct | +| T16 | Cherry-pick / fast-forward | Yes | Paimon fast-forward oracle | Current, tag and branch state are verified | + +## Delete and execution dimensions + +| Dimension | Iceberg | Paimon | +| --- | --- | --- | +| Position delete | v2, Parquet and ORC, before/after evolution | N/A | +| 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 | +| 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 | + +HMS and DLF are credential/environment variants rather than additional schema +semantics. Their existing catalog suites remain responsible for connectivity; +the deterministic P0 schema matrix uses REST/filesystem, and JDBC gets an +explicit rename plus old snapshot/tag smoke path. + +## Suite map + +| Suite | Responsibility | +| --- | --- | +| `iceberg/test_iceberg_schema_time_travel_matrix.groovy` | S01-S17 × T00-T13, nested evolution, partition evolution, cache/scanner variants | +| `iceberg/test_iceberg_schema_dual_relation_matrix.groovy` | Dual-snapshot join/UNION/CTE/subquery negative contracts | +| `iceberg/test_iceberg_schema_equality_delete_time_travel.groovy` | Equality delete × rename/promotion/drop-re-add | +| `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 | +| `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 | +| `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 | + +Each Groovy file contains `Scenario` comments identifying the matrix cell under +test. Jira keys are deliberately absent from Groovy source. Product issues link +back to the exact suite, scenario and file location from Jira instead. + +## Product contracts discovered by the matrix + +| Issue | Observed contract | Negative regression location | +| --- | --- | --- | +| DORIS-27425 | Iceberg branch can use latest schema instead of branch schema | `test_iceberg_schema_time_travel_matrix.groovy`, `test_iceberg_schema_ref_actions_matrix.groovy` | +| DORIS-27427 | Iceberg dual historical relations can share the wrong schema | `test_iceberg_schema_dual_relation_matrix.groovy` | +| DORIS-27428 | Paimon dual historical relations can share the wrong schema | `test_paimon_schema_dual_relation_matrix.groovy` | +| DORIS-27433 | Paimon branch schema init fails; a post-fast-forward scan can abort BE | `test_paimon_schema_branch_partition_matrix.groovy` | +| DORIS-27434 | Doris `DESC` omits Iceberg field comments | `test_iceberg_schema_metadata_atomicity_matrix.groovy` | + +## Validation status + +- The ten REST/filesystem matrix suites pass with 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. Review Comment: **[P1] Do not freeze known defects as successful P0 contracts** This document explicitly classifies these outcomes as current Doris bugs, while the new suites require the wrong exceptions/types to persist. `TestAction` fails when an expected exception disappears, so fixing a listed defect turns the green P0 suite red. Assert the intended successful rows/schema after the fixes, or quarantine these reproducers and keep the affected cells marked incomplete. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_time_travel_matrix.groovy: ########## @@ -0,0 +1,600 @@ +// 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_schema_time_travel_matrix", "p0,external,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_schema_time_travel_matrix" + String noCacheCatalogName = "test_paimon_schema_time_travel_matrix_no_cache" + String dbName = "paimon_schema_time_travel_matrix_db" + String topTable = "top_timeline" + String nestedTable = "nested_timeline" + String pkTable = "pk_dv_timeline" + String partitionTable = "partition_timeline" + + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_paimon """ + SELECT snapshot_id + FROM paimon.${dbName}.`${tableName}\$snapshots` + ORDER BY snapshot_id DESC + LIMIT 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + def createTag = { String tableName, String tagName, String snapshotId -> + spark_paimon """ + CALL paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}', + snapshot => ${snapshotId} + ) + """ + } + + def assertUnknownColumn = { String query, String columnName -> + test { + sql query + exception "'${columnName}'" + } + } + + sql """drop catalog if exists ${catalogName}""" + sql """drop catalog if exists ${noCacheCatalogName}""" + 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' + ) + """ + sql """ + CREATE CATALOG ${noCacheCatalogName} 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}.${topTable}; + CREATE TABLE paimon.${dbName}.${topTable} ( + id INT, + old_name STRING, + victim STRING, + metric INT + ) USING paimon + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${topTable} + VALUES (1, 'alpha', 'old-v1', 10); + """ + String topCp0 = latestSnapshotId(topTable) + createTag(topTable, "top_cp0", topCp0) + spark_paimon """ + CALL paimon.sys.create_branch( + '${dbName}.${topTable}', + 'top_cp0_branch', + 'top_cp0' + ) + """ + Thread.sleep(1100) + + // Scenario S01/S02/S03 x T00/T01/T02/T05/T08: + // add multiple columns and position one AFTER old_name. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + ADD COLUMNS (added STRING AFTER old_name, added_second BIGINT); + INSERT INTO paimon.${dbName}.${topTable} + (id, old_name, victim, metric, added, added_second) + VALUES (2, 'beta', 'old-v2', 20, 'added-v2', 200); + """ + String topCpAdd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_add", topCpAdd) + Thread.sleep(1100) + + // Scenario S04/S05 x T00-T08: + // explicit old/new column binding catches rename regressions hidden by SELECT *. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + RENAME COLUMN old_name TO MixedName; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, victim, metric, added, added_second) + VALUES (3, 'gamma', 'old-v3', 30, 'added-v3', 300); + """ + String topCpRename = latestSnapshotId(topTable) + createTag(topTable, "top_cp_rename", topCpRename) + Thread.sleep(1100) + + // Scenario S06 x T00/T01/T02/T05: the dropped field remains available only to old refs. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} DROP COLUMN victim; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second) + VALUES (4, 'delta', 40, 'added-v4', 400); + """ + String topCpDrop = latestSnapshotId(topTable) + createTag(topTable, "top_cp_drop", topCpDrop) + Thread.sleep(1100) + + // Scenario S07 x T00/T01/T02/T05/T12/T13: + // drop/re-add with a different type creates a new field ID and must not expose old values. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ADD COLUMN victim BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (5, 'epsilon', 50, 'added-v5', 500, 5000); + """ + String topCpReadd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_readd", topCpReadd) + Thread.sleep(1100) + + // Scenario S08 x T00/T01/T02/T05: compatible INT -> BIGINT promotion. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ALTER COLUMN metric TYPE BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (6, 'zeta', 6000000000, 'added-v6', 600, 6000); + """ + String topCpPromote = latestSnapshotId(topTable) + createTag(topTable, "top_cp_promote", topCpPromote) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${nestedTable}; + CREATE TABLE paimon.${dbName}.${nestedTable} ( + id INT, + payload STRUCT<old_child: INT, keep: INT>, + attributes MAP<STRING, STRUCT<old_child: INT, keep: INT>>, + events ARRAY<STRUCT<old_child: INT, keep: INT>> + ) USING paimon + TBLPROPERTIES ('file.format'='orc'); + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 1, + named_struct('old_child', 10, 'keep', 11), + map('a', named_struct('old_child', 20, 'keep', 21)), + array(named_struct('old_child', 30, 'keep', 31)) + ); + """ + String nestedCp0 = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp0", nestedCp0) + + // Scenario S09/S14/S15 x T00/T01/T02/T05: + // add children to STRUCT, MAP value struct and ARRAY element struct. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.added_child INT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 2, + named_struct('old_child', 110, 'keep', 111, 'added_child', 112), + map('a', named_struct('old_child', 120, 'keep', 121, 'added_child', 122)), + array(named_struct('old_child', 130, 'keep', 131, 'added_child', 132)) + ); + """ + String nestedCpAdd = latestSnapshotId(nestedTable) + + // Scenario S10/S14/S15 x T00/T01/T02/T05: rename every supported nested path. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN payload.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN attributes.value.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN events.element.old_child TO renamed_child; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 3, + named_struct('renamed_child', 210, 'keep', 211, 'added_child', 212), + map('a', named_struct('renamed_child', 220, 'keep', 221, 'added_child', 222)), + array(named_struct('renamed_child', 230, 'keep', 231, 'added_child', 232)) + ); + """ + String nestedCpRename = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp_rename", nestedCpRename) + + // Scenario S11/S12/S13 x T00/T01/T02/T05: + // nested drop/re-add and type promotion preserve field-ID isolation. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} DROP COLUMN payload.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN attributes.value.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN events.element.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ALTER COLUMN payload.keep TYPE BIGINT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 4, + named_struct('keep', 311, 'added_child', 312, 'renamed_child', 3100), + map('a', named_struct('keep', 321, 'added_child', 322, 'renamed_child', 3200)), + array(named_struct('keep', 331, 'added_child', 332, 'renamed_child', 3300)) + ); + """ + String nestedCpReadd = latestSnapshotId(nestedTable) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${pkTable}; + CREATE TABLE paimon.${dbName}.${pkTable} ( + id INT NOT NULL, + old_name STRING, + note STRING, + score INT + ) USING paimon + TBLPROPERTIES ( + 'bucket'='1', + 'primary-key'='id', + 'file.format'='parquet', + 'deletion-vectors.enabled'='true' + ); + INSERT INTO paimon.${dbName}.${pkTable} VALUES + (1, 'alpha', 'old-note-1', 10), + (2, 'beta', 'old-note-2', 20); + """ + String pkCp0 = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp0", pkCp0) + + // Scenario S01/S04/S18 x TC05: + // PK remains stable while a non-key field is added/renamed and rows are upserted/deleted. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN extra STRING; + ALTER TABLE paimon.${dbName}.${pkTable} RENAME COLUMN old_name TO full_name; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, note, score, extra) + VALUES (1, 'alpha-updated', 'new-note-1', 11, 'extra-1'); + DELETE FROM paimon.${dbName}.${pkTable} WHERE id = 2; + """ + String pkCpRenameDelete = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_rename_delete", pkCpRenameDelete) + + // Scenario S06/S07/S08/S18 x TC05: + // drop/re-add and promotion are combined with a later upsert and DV compaction. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} DROP COLUMN note; + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN note BIGINT; + ALTER TABLE paimon.${dbName}.${pkTable} ALTER COLUMN score TYPE BIGINT; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, score, extra, note) + VALUES (3, 'gamma', 3000000000, 'extra-3', 3000); + """ + String pkCpReadd = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_readd", pkCpReadd) + spark_paimon """ Review Comment: **[P1] Establish that this scan actually uses a deletion vector** The detailed checkpoint predates full compaction, while the later current query can read fully compacted ordinary files. Neither proves that a DV was produced or scanned. Assert deletion metadata or verbose split evidence with `hasDeletionVector=true` on the scan being tested before claiming the PK DV path. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_time_travel_matrix.groovy: ########## @@ -0,0 +1,600 @@ +// 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_schema_time_travel_matrix", "p0,external,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_schema_time_travel_matrix" + String noCacheCatalogName = "test_paimon_schema_time_travel_matrix_no_cache" + String dbName = "paimon_schema_time_travel_matrix_db" + String topTable = "top_timeline" + String nestedTable = "nested_timeline" + String pkTable = "pk_dv_timeline" + String partitionTable = "partition_timeline" + + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_paimon """ + SELECT snapshot_id + FROM paimon.${dbName}.`${tableName}\$snapshots` + ORDER BY snapshot_id DESC + LIMIT 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + def createTag = { String tableName, String tagName, String snapshotId -> + spark_paimon """ + CALL paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}', + snapshot => ${snapshotId} + ) + """ + } + + def assertUnknownColumn = { String query, String columnName -> + test { + sql query + exception "'${columnName}'" + } + } + + sql """drop catalog if exists ${catalogName}""" + sql """drop catalog if exists ${noCacheCatalogName}""" + 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' + ) + """ + sql """ + CREATE CATALOG ${noCacheCatalogName} 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' Review Comment: **[P1] Disable the schema cache in the cache-off catalog** This only disables `meta.cache.paimon.table.ttl-second`, but historical binding uses the separate schema entry registered by `PaimonExternalMetaCache`. Also set `meta.cache.paimon.schema.ttl-second=0`; otherwise this query does not exercise the advertised cache-off schema path. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_branch_partition_matrix.groovy: ########## @@ -0,0 +1,225 @@ +// 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_schema_branch_partition_matrix", "p0,external,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_schema_branch_partition_matrix" + String dbName = "paimon_schema_branch_partition_db" + String branchTable = "branch_schema_timeline" + String partitionTable = "partition_schema_timeline" + + 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}.${branchTable}; + create table paimon.${dbName}.${branchTable} ( + id int, + old_name string, + metric int + ) using paimon + tblproperties ('file.format'='parquet'); + insert into paimon.${dbName}.${branchTable} values (1, 'base-1', 10); + call paimon.sys.create_tag( + table => '${dbName}.${branchTable}', + tag => 'branch_base' + ); + call paimon.sys.create_branch( + '${dbName}.${branchTable}', + 'schema_branch', + 'branch_base' + ); + """ + + // Scenario T09-branch-evolution: branch owns an independent rename/type/add timeline. + spark_paimon_multi """ + alter table paimon.${dbName}.`${branchTable}\$branch_schema_branch` + rename column old_name to branch_name; + alter table paimon.${dbName}.`${branchTable}\$branch_schema_branch` + alter column metric type bigint; + alter table paimon.${dbName}.`${branchTable}\$branch_schema_branch` + add column branch_only string; + insert into paimon.${dbName}.`${branchTable}\$branch_schema_branch` + values (2, 'branch-2', 6000000000, 'branch-only-2'); + """ + + // Main evolves independently after the branch is created. + spark_paimon_multi """ + alter table paimon.${dbName}.${branchTable} add column main_only string; + insert into paimon.${dbName}.${branchTable} + values (3, 'main-3', 30, 'main-only-3'); + """ + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${branchTable}""" + + assertEquals([[1, "base-1", 10, null], [3, "main-3", 30, "main-only-3"]], sql(""" + select id, old_name, metric, main_only + from ${branchTable} + order by id + """)) + // Negative contract: Doris cannot initialize a Paimon branch with an independent schema. + test { + sql """ + select id, branch_name, metric, branch_only + from ${branchTable}@branch(schema_branch) + order by id + """ + exception "failed to initSchema" + } + test { + sql """select branch_only from ${branchTable}""" + exception "Unknown column 'branch_only'" + } + + // Scenario T09-fast-forward: branch schema and data replace main after Paimon fast-forward. + spark_paimon """ + call paimon.sys.fast_forward('${dbName}.${branchTable}', 'schema_branch') + """ + spark_paimon """refresh table paimon.${dbName}.${branchTable}""" + sql """refresh table ${branchTable}""" + assertEquals([ + [1, "base-1", 10L, null], + [2, "branch-2", 6000000000L, "branch-only-2"] + ], spark_paimon(""" + select id, branch_name, metric, branch_only + from paimon.${dbName}.${branchTable} + order by id + """)) + List<String> fastForwardColumns = sql("""desc ${branchTable}""") Review Comment: **[P1] Exercise the Doris scan after fast-forward** After fast-forward, only Spark validates the rows; Doris runs `DESC` plus an analyzer-time unknown-column check, neither of which reaches the BE path whose abort is documented above. `containsAll` also allows stale `main_only` to remain. Add the ordered Doris row query and assert the exact resulting schema. ########## regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy: ########## @@ -90,25 +92,63 @@ suite("test_paimon_jdbc_catalog", "p0,external") { } executeCommand("mkdir -p ${localDriverDir}", false, 60) - executeCommand("mkdir -p ${jdbcDriversDir}", true, 60) if (!new File(localDriverPath).exists()) { executeCommand("/usr/bin/curl --max-time 600 ${driverDownloadUrl} --output ${localDriverPath}", true, 660) } - executeCommand("cp -f ${localDriverPath} ${jdbcDriversDir}/${driverName}", true, 60) - String sparkContainerName = executeCommand("docker ps --filter name=spark-iceberg --format {{.Names}}", false, 30) + def clusterHostIps = new ArrayList() + String[][] backends = sql """show backends""" + for (def backend in backends) { + clusterHostIps.add(backend[1]) + } + String[][] frontends = sql """show frontends""" + for (def frontend in frontends) { + clusterHostIps.add(frontend[1]) + } + clusterHostIps = clusterHostIps.unique() + + Set<String> localHostIps = ["127.0.0.1", "localhost", "::1"] as Set + java.util.Collections.list(java.net.NetworkInterface.getNetworkInterfaces()).each { networkInterface -> + java.util.Collections.list(networkInterface.getInetAddresses()).each { address -> + localHostIps.add(address.getHostAddress().split("%")[0]) + } + } + localHostIps.add(java.net.InetAddress.getLocalHost().getHostName()) + localHostIps.add(java.net.InetAddress.getLocalHost().getCanonicalHostName()) + + for (def hostIp in clusterHostIps) { Review Comment: **[P1] Check the Spark fixture before mandatory cluster staging** Download and root SSH/SCP to every registered node happen before the explicit missing-container skip. An environment this suite intends to skip can therefore fail during staging, and a dead registered node has the same effect. Validate the fixture and live targets first, then distribute only when the suite will execute. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_metadata_atomicity_matrix.groovy: ########## @@ -0,0 +1,170 @@ +// 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_schema_metadata_atomicity_matrix", "p0,external,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_schema_metadata_atomicity_matrix" + String dbName = "paimon_schema_metadata_atomicity_db" + String tableName = "metadata_timeline" + + def createTag = { String tagName -> + spark_paimon """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}' + ) + """ + } + def snapshotCount = { + return spark_paimon(""" + select count(*) from paimon.${dbName}.`${tableName}\$snapshots` + """)[0][0].toString().toInteger() + } + def schemaCount = { + return spark_paimon(""" + select count(*) from paimon.${dbName}.`${tableName}\$schemas` + """)[0][0].toString().toInteger() + } + + 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 not null, + required_name string not null, + optional_value int, + info struct<value:int, keep:int> + ) using paimon + tblproperties ('file.format'='parquet'); + insert into paimon.${dbName}.${tableName} + values (1, 'name-1', 10, named_struct('value', 100, 'keep', 101)); + """ + createTag("metadata_before_change") + int initialSnapshots = snapshotCount() + + // Scenario S19-comment: top-level and nested comments create schema versions, not data snapshots. + spark_paimon """ + alter table paimon.${dbName}.${tableName} + alter column required_name comment 'top-level-comment' + """ + spark_paimon """ + alter table paimon.${dbName}.${tableName} + alter column info.value comment 'nested-comment' + """ + assertEquals(initialSnapshots, snapshotCount()) Review Comment: **[P2] Observe the metadata changes themselves** An unchanged data-snapshot count does not prove either comment was stored, and reading the same existing non-null row does not prove `DROP NOT NULL` took effect. Assert the exact top-level/nested comments and schema-version change, then write/read a post-change NULL while checking the tagged pre-change schema separately. ########## regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy: ########## @@ -90,25 +92,63 @@ suite("test_paimon_jdbc_catalog", "p0,external") { } executeCommand("mkdir -p ${localDriverDir}", false, 60) - executeCommand("mkdir -p ${jdbcDriversDir}", true, 60) if (!new File(localDriverPath).exists()) { executeCommand("/usr/bin/curl --max-time 600 ${driverDownloadUrl} --output ${localDriverPath}", true, 660) } - executeCommand("cp -f ${localDriverPath} ${jdbcDriversDir}/${driverName}", true, 60) - String sparkContainerName = executeCommand("docker ps --filter name=spark-iceberg --format {{.Names}}", false, 30) + def clusterHostIps = new ArrayList() + String[][] backends = sql """show backends""" + for (def backend in backends) { + clusterHostIps.add(backend[1]) + } + String[][] frontends = sql """show frontends""" + for (def frontend in frontends) { + clusterHostIps.add(frontend[1]) + } + clusterHostIps = clusterHostIps.unique() + + Set<String> localHostIps = ["127.0.0.1", "localhost", "::1"] as Set + java.util.Collections.list(java.net.NetworkInterface.getNetworkInterfaces()).each { networkInterface -> + java.util.Collections.list(networkInterface.getInetAddresses()).each { address -> + localHostIps.add(address.getHostAddress().split("%")[0]) + } + } + localHostIps.add(java.net.InetAddress.getLocalHost().getHostName()) + localHostIps.add(java.net.InetAddress.getLocalHost().getCanonicalHostName()) + + for (def hostIp in clusterHostIps) { + // Scenario: every FE/BE receives the JDBC driver so metadata failover and distributed + // scan scheduling do not depend on a driver installed only on the regression runner. + if (localHostIps.contains(hostIp)) { Review Comment: **[P2] Normalize host identity before choosing local versus SSH** `SHOW FRONTENDS/BACKENDS` returns configured host strings, but this set contains interface text plus only one JVM hostname/canonical name. A local DNS alias or equivalent IPv6 spelling therefore takes the root-SSH branch. Resolve each advertised host and compare normalized addresses, preferably in one helper shared by both JDBC suites. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_position_dv_time_travel.groovy: ########## @@ -0,0 +1,180 @@ +// 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_schema_position_dv_time_travel", + "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_schema_position_dv_time_travel" + String dbName = "iceberg_schema_position_dv_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' + ) + """ + + try { + spark_iceberg """create database if not exists demo.${dbName}""" + + [ + [table: "position_parquet", version: "2", format: "parquet"], + [table: "position_orc", version: "2", format: "orc"], + [table: "dv_parquet", version: "3", format: "parquet"], + [table: "dv_orc", version: "3", format: "orc"] + ].each { profile -> + String tableName = profile.table + spark_iceberg_multi """ + drop table if exists demo.${dbName}.${tableName}; + create table demo.${dbName}.${tableName} ( + id int, + old_name string, + victim string, + metric int, + info struct<old_child:int, keep:int> + ) using iceberg + tblproperties ( + 'format-version'='${profile.version}', + 'write.format.default'='${profile.format}', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.distribution-mode'='none', + 'write.target-file-size-bytes'='134217728' + ); + insert into demo.${dbName}.${tableName} + select /*+ COALESCE(1) */ id, old_name, victim, metric, info from values + (1, 'old-1', 'victim-1', 10, + named_struct('old_child', 100, 'keep', 101)), + (2, 'old-2', 'victim-2', 20, + named_struct('old_child', 200, 'keep', 201)), + (3, 'old-3', 'victim-3', 30, + named_struct('old_child', 300, 'keep', 301)) + as t(id, old_name, victim, metric, info); + """ + String beforeDelete = latestSnapshotId(tableName) + + // Scenario TC04-position/DV-before-change: delete visibility is snapshot-local. + spark_iceberg """ + delete from demo.${dbName}.${tableName} where id = 2 + """ + String afterFirstDelete = latestSnapshotId(tableName) + + spark_iceberg_multi """ + alter table demo.${dbName}.${tableName} rename column old_name to new_name; + alter table demo.${dbName}.${tableName} + rename column info.old_child to new_child; + alter table demo.${dbName}.${tableName} drop column victim; + alter table demo.${dbName}.${tableName} add column victim bigint; + alter table demo.${dbName}.${tableName} alter column metric type bigint; + insert into demo.${dbName}.${tableName} + (id, new_name, victim, metric, info) values + (4, 'new-4', 4000, 6000000000, + named_struct('new_child', 400, 'keep', 401)); + delete from demo.${dbName}.${tableName} where id = 3; + """ + String afterSchemaDelete = latestSnapshotId(tableName) + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${tableName}""" + + assertEquals([ + [1, "old-1", "victim-1", 10, 100], + [2, "old-2", "victim-2", 20, 200], + [3, "old-3", "victim-3", 30, 300] + ], sql(""" + select id, old_name, victim, metric, info.old_child + from ${tableName} for version as of ${beforeDelete} + order by id + """)) + assertEquals([[1], [3]], sql(""" + select id + from ${tableName} for version as of ${afterFirstDelete} + order by id + """)) + + // Scenario TC04-S04/S06/S07/S08/S10: delete files survive top-level and nested evolution. + assertEquals([ + [1, "old-1", null, 10L, 100], + [4, "new-4", 4000L, 6000000000L, 400] + ], sql(""" + select id, new_name, victim, metric, info.new_child + from ${tableName} for version as of ${afterSchemaDelete} + order by id + """)) + + // Scenario TC04-field-ID: re-added victim never exposes values from the dropped STRING field. + assertEquals([[1, null], [4, 4000L]], sql(""" + select id, victim + from ${tableName} for version as of ${afterSchemaDelete} + order by id + """)) + + // Scenario TC04-reader-diff: both scanners apply position deletes or DVs identically. + sql """set enable_file_scanner_v2=true""" + List<List<Object>> v2Rows = sql(""" + select id, new_name, victim, metric, info.new_child + from ${tableName} for version as of ${afterSchemaDelete} + where metric >= 10 + order by id + """) + sql """set enable_file_scanner_v2=false""" + List<List<Object>> legacyRows = sql(""" + select id, new_name, victim, metric, info.new_child + from ${tableName} for version as of ${afterSchemaDelete} + where metric >= 10 + order by id + """) + assertEquals(v2Rows, legacyRows) + + // Scenario TC04-delete-file-kind: v2 produces position deletes and v3 produces DVs. + List<List<Object>> deleteFiles = spark_iceberg(""" + select content, file_format + from demo.${dbName}.${tableName}.delete_files + """) + assertTrue(deleteFiles.size() > 0, Review Comment: **[P1] Distinguish v3 DVs from ordinary position deletes** The query retrieves `content` and `file_format`, but the assertion only checks that some delete file exists, which every position-delete profile satisfies. Require Puffin/`.puffin` metadata for v3 and the expected non-Puffin position-delete kind for v2. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_metadata_atomicity_matrix.groovy: ########## @@ -0,0 +1,163 @@ +// 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_schema_metadata_atomicity_matrix", + "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_schema_metadata_atomicity_matrix" + String dbName = "iceberg_schema_metadata_atomicity_db" + String tableName = "metadata_timeline" + + def snapshotCount = { + return sql("""select count(*) from ${tableName}\$snapshots""")[0][0].toString().toInteger() + } + def schemaCount = { + return sql("""select count(*) from ${tableName}\$metadata_log_entries""")[0][0] + .toString().toInteger() + } + def allDescText = { + return sql("""desc ${tableName}""").flatten().collect { + it == null ? "" : it.toString() + }.join(" ") + } + + 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 """create database if not exists ${dbName}""" + sql """use ${dbName}""" + + try { + sql """drop table if exists ${tableName}""" + sql """ + create table ${tableName} ( + id int not null, + required_name string not null, + optional_value int, + info struct<value:int, keep:int>, + attrs map<string, int> + ) engine=iceberg + properties ('format-version'='2', 'write.format.default'='parquet') + """ + sql """ + insert into ${tableName} values + (1, 'name-1', 10, named_struct('value', 100, 'keep', 101), map('k', 1000)) + """ + sql """alter table ${tableName} create tag metadata_before_change""" + int initialSnapshots = snapshotCount() + + // Scenario S19-comment: top-level and nested comments are metadata-only schema changes. + sql """alter table ${tableName} modify column required_name comment 'top-level-comment'""" + sql """alter table ${tableName} modify column info.value comment 'nested-comment'""" + String descAfterComment = allDescText() + List<List<Object>> sparkDescription = spark_iceberg(""" + describe table extended demo.${dbName}.${tableName} + """) + String sparkDescriptionText = sparkDescription.flatten().collect { + it == null ? "" : it.toString() + }.join(" ") + assertTrue(sparkDescriptionText.contains("top-level-comment")) + // Negative contract: Doris DESC currently omits Iceberg field comments. + assertFalse(descAfterComment.contains("top-level-comment")) + assertFalse(descAfterComment.contains("nested-comment")) + assertEquals(initialSnapshots, snapshotCount()) + + // Scenario S19-nullability: relaxing required to optional preserves data and historical refs. + sql """alter table ${tableName} modify column required_name string null""" Review Comment: **[P2] Verify that nullability actually changed** Both queries read only the original non-null row, so an accepted no-op would pass. Assert current optionality versus the pre-change tag's required field and add a post-change NULL write/read. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_time_travel_matrix.groovy: ########## @@ -0,0 +1,600 @@ +// 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_schema_time_travel_matrix", "p0,external,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_schema_time_travel_matrix" + String noCacheCatalogName = "test_paimon_schema_time_travel_matrix_no_cache" + String dbName = "paimon_schema_time_travel_matrix_db" + String topTable = "top_timeline" + String nestedTable = "nested_timeline" + String pkTable = "pk_dv_timeline" + String partitionTable = "partition_timeline" + + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_paimon """ + SELECT snapshot_id + FROM paimon.${dbName}.`${tableName}\$snapshots` + ORDER BY snapshot_id DESC + LIMIT 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + def createTag = { String tableName, String tagName, String snapshotId -> + spark_paimon """ + CALL paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}', + snapshot => ${snapshotId} + ) + """ + } + + def assertUnknownColumn = { String query, String columnName -> + test { + sql query + exception "'${columnName}'" + } + } + + sql """drop catalog if exists ${catalogName}""" + sql """drop catalog if exists ${noCacheCatalogName}""" + 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' + ) + """ + sql """ + CREATE CATALOG ${noCacheCatalogName} 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}.${topTable}; + CREATE TABLE paimon.${dbName}.${topTable} ( + id INT, + old_name STRING, + victim STRING, + metric INT + ) USING paimon + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${topTable} + VALUES (1, 'alpha', 'old-v1', 10); + """ + String topCp0 = latestSnapshotId(topTable) + createTag(topTable, "top_cp0", topCp0) + spark_paimon """ + CALL paimon.sys.create_branch( + '${dbName}.${topTable}', + 'top_cp0_branch', + 'top_cp0' + ) + """ + Thread.sleep(1100) + + // Scenario S01/S02/S03 x T00/T01/T02/T05/T08: + // add multiple columns and position one AFTER old_name. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + ADD COLUMNS (added STRING AFTER old_name, added_second BIGINT); + INSERT INTO paimon.${dbName}.${topTable} + (id, old_name, victim, metric, added, added_second) + VALUES (2, 'beta', 'old-v2', 20, 'added-v2', 200); + """ + String topCpAdd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_add", topCpAdd) + Thread.sleep(1100) + + // Scenario S04/S05 x T00-T08: + // explicit old/new column binding catches rename regressions hidden by SELECT *. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + RENAME COLUMN old_name TO MixedName; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, victim, metric, added, added_second) + VALUES (3, 'gamma', 'old-v3', 30, 'added-v3', 300); + """ + String topCpRename = latestSnapshotId(topTable) + createTag(topTable, "top_cp_rename", topCpRename) + Thread.sleep(1100) + + // Scenario S06 x T00/T01/T02/T05: the dropped field remains available only to old refs. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} DROP COLUMN victim; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second) + VALUES (4, 'delta', 40, 'added-v4', 400); + """ + String topCpDrop = latestSnapshotId(topTable) + createTag(topTable, "top_cp_drop", topCpDrop) + Thread.sleep(1100) + + // Scenario S07 x T00/T01/T02/T05/T12/T13: + // drop/re-add with a different type creates a new field ID and must not expose old values. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ADD COLUMN victim BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (5, 'epsilon', 50, 'added-v5', 500, 5000); + """ + String topCpReadd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_readd", topCpReadd) + Thread.sleep(1100) + + // Scenario S08 x T00/T01/T02/T05: compatible INT -> BIGINT promotion. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ALTER COLUMN metric TYPE BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (6, 'zeta', 6000000000, 'added-v6', 600, 6000); + """ + String topCpPromote = latestSnapshotId(topTable) + createTag(topTable, "top_cp_promote", topCpPromote) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${nestedTable}; + CREATE TABLE paimon.${dbName}.${nestedTable} ( + id INT, + payload STRUCT<old_child: INT, keep: INT>, + attributes MAP<STRING, STRUCT<old_child: INT, keep: INT>>, + events ARRAY<STRUCT<old_child: INT, keep: INT>> + ) USING paimon + TBLPROPERTIES ('file.format'='orc'); + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 1, + named_struct('old_child', 10, 'keep', 11), + map('a', named_struct('old_child', 20, 'keep', 21)), + array(named_struct('old_child', 30, 'keep', 31)) + ); + """ + String nestedCp0 = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp0", nestedCp0) + + // Scenario S09/S14/S15 x T00/T01/T02/T05: + // add children to STRUCT, MAP value struct and ARRAY element struct. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.added_child INT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 2, + named_struct('old_child', 110, 'keep', 111, 'added_child', 112), + map('a', named_struct('old_child', 120, 'keep', 121, 'added_child', 122)), + array(named_struct('old_child', 130, 'keep', 131, 'added_child', 132)) + ); + """ + String nestedCpAdd = latestSnapshotId(nestedTable) + + // Scenario S10/S14/S15 x T00/T01/T02/T05: rename every supported nested path. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN payload.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN attributes.value.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN events.element.old_child TO renamed_child; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 3, + named_struct('renamed_child', 210, 'keep', 211, 'added_child', 212), + map('a', named_struct('renamed_child', 220, 'keep', 221, 'added_child', 222)), + array(named_struct('renamed_child', 230, 'keep', 231, 'added_child', 232)) + ); + """ + String nestedCpRename = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp_rename", nestedCpRename) + + // Scenario S11/S12/S13 x T00/T01/T02/T05: + // nested drop/re-add and type promotion preserve field-ID isolation. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} DROP COLUMN payload.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN attributes.value.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN events.element.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ALTER COLUMN payload.keep TYPE BIGINT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 4, + named_struct('keep', 311, 'added_child', 312, 'renamed_child', 3100), + map('a', named_struct('keep', 321, 'added_child', 322, 'renamed_child', 3200)), + array(named_struct('keep', 331, 'added_child', 332, 'renamed_child', 3300)) + ); + """ + String nestedCpReadd = latestSnapshotId(nestedTable) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${pkTable}; + CREATE TABLE paimon.${dbName}.${pkTable} ( + id INT NOT NULL, + old_name STRING, + note STRING, + score INT + ) USING paimon + TBLPROPERTIES ( + 'bucket'='1', + 'primary-key'='id', + 'file.format'='parquet', + 'deletion-vectors.enabled'='true' + ); + INSERT INTO paimon.${dbName}.${pkTable} VALUES + (1, 'alpha', 'old-note-1', 10), + (2, 'beta', 'old-note-2', 20); + """ + String pkCp0 = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp0", pkCp0) + + // Scenario S01/S04/S18 x TC05: + // PK remains stable while a non-key field is added/renamed and rows are upserted/deleted. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN extra STRING; + ALTER TABLE paimon.${dbName}.${pkTable} RENAME COLUMN old_name TO full_name; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, note, score, extra) + VALUES (1, 'alpha-updated', 'new-note-1', 11, 'extra-1'); + DELETE FROM paimon.${dbName}.${pkTable} WHERE id = 2; + """ + String pkCpRenameDelete = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_rename_delete", pkCpRenameDelete) + + // Scenario S06/S07/S08/S18 x TC05: + // drop/re-add and promotion are combined with a later upsert and DV compaction. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} DROP COLUMN note; + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN note BIGINT; + ALTER TABLE paimon.${dbName}.${pkTable} ALTER COLUMN score TYPE BIGINT; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, score, extra, note) + VALUES (3, 'gamma', 3000000000, 'extra-3', 3000); + """ + String pkCpReadd = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_readd", pkCpReadd) + spark_paimon """ + CALL paimon.sys.compact(table => '${dbName}.${pkTable}', compact_strategy => 'full') + """ + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${partitionTable}; + CREATE TABLE paimon.${dbName}.${partitionTable} ( + id INT, + old_partition STRING, + payload STRING + ) USING paimon + PARTITIONED BY (old_partition) + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${partitionTable} + VALUES (1, 'p1', 'old'), (2, 'p2', 'old'); + """ + String partitionCp0 = latestSnapshotId(partitionTable) + createTag(partitionTable, "partition_cp0", partitionCp0) + + // Scenario S16-negative: Paimon must reject partition-column rename atomically. + String partitionRenameError = null + try { + spark_paimon """ + ALTER TABLE paimon.${dbName}.${partitionTable} + RENAME COLUMN old_partition TO new_partition + """ + } catch (Exception e) { + partitionRenameError = e.getMessage() + } + assertNotNull(partitionRenameError) + assertTrue(partitionRenameError.contains("Cannot rename partition column")) + + // Scenario S17 x TC03: rename a payload field while retaining partition pruning. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${partitionTable} + RENAME COLUMN payload TO new_payload; + INSERT INTO paimon.${dbName}.${partitionTable} + VALUES (3, 'p3', 'new'); + """ + String partitionCpRename = latestSnapshotId(partitionTable) + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh catalog ${catalogName}""" + + // Scenario TC01: validate latest schema/data, explicit new binding, predicate and aggregate. + assertEquals([[1, null], [2, null], [3, null], [4, null], [5, 5000L], [6, 6000L]], + sql("""select id, victim from ${topTable} order by id""")) + assertEquals([[6, 6000000000L]], + sql("""select id, metric from ${topTable} where metric > 5000000000""")) + assertEquals([[6L, 6000000150L]], + sql("""select count(*), sum(metric) from ${topTable}""")) + assertUnknownColumn("""select old_name from ${topTable}""", "old_name") + + // Scenario T01/T05/T06/T08: old snapshot/tag/branch uses the old schema. + List<List<Object>> topCp0Rows = [[1, "alpha", "old-v1", 10]] + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for version as of ${topCp0} + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for version as of 'top_cp0' + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable}@tag(top_cp0) + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable}@branch(top_cp0_branch) + order by id + """)) + assertUnknownColumn(""" + select MixedName from ${topTable} for version as of ${topCp0} + """, "MixedName") + + // Scenario T03/T04: time string and epoch millis resolve to the pre-change schema. + List<List<Object>> cp0TimeRows = sql(""" + select date_format(date_add(commit_time, interval 1 second), '%Y-%m-%d %H:%i:%s'), + cast(unix_timestamp(commit_time) * 1000 + 999 as bigint) + from ${topTable}\$snapshots + where snapshot_id = ${topCp0} + """) + assertEquals(1, cp0TimeRows.size()) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for time as of "${cp0TimeRows[0][0]}" + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for time as of ${cp0TimeRows[0][1]} + order by id + """)) + + // Scenario S01-S08: checkpoint projections prove name/type/field-ID isolation. + assertEquals([[1, "alpha", null], [2, "beta", "added-v2"]], + sql(""" + select id, old_name, added + from ${topTable} for version as of ${topCpAdd} + order by id + """)) + assertEquals([[1, "alpha"], [2, "beta"], [3, "gamma"]], + sql(""" + select id, MixedName + from ${topTable} for version as of ${topCpRename} + where metric >= 10 + order by id + """)) + assertUnknownColumn(""" + select old_name from ${topTable} for version as of ${topCpRename} + """, "old_name") + assertUnknownColumn(""" + select victim from ${topTable} for version as of ${topCpDrop} + """, "victim") + assertEquals([[1, null], [2, null], [3, null], [4, null], [5, 5000L]], + sql(""" + select id, victim + from ${topTable} for version as of ${topCpReadd} + order by id + """)) + assertEquals([[6, 6000000000L]], + sql(""" + select id, metric + from ${topTable} for version as of ${topCpPromote} + where metric > 5000000000 + """)) + + // Scenario TC02: nested refs validate STRUCT/MAP/ARRAY projection and predicate. + assertEquals([[1, 10, 20, 30]], + sql(""" + select id, payload.old_child, + element_at(attributes, 'a').old_child, + events[1].old_child + from ${nestedTable} for version as of ${nestedCp0} + where payload.old_child = 10 + """)) + assertEquals([[1, null, null, null], [2, 112, 122, 132]], + sql(""" + select id, payload.added_child, + element_at(attributes, 'a').added_child, + events[1].added_child + from ${nestedTable} for version as of ${nestedCpAdd} + order by id + """)) + assertEquals([[1, 10, 20, 30], [2, 110, 120, 130], [3, 210, 220, 230]], + sql(""" + select id, payload.renamed_child, + element_at(attributes, 'a').renamed_child, + events[1].renamed_child + from ${nestedTable} for version as of 'nested_cp_rename' + order by id + """)) + assertUnknownColumn(""" + select payload.old_child + from ${nestedTable} for version as of ${nestedCpRename} + """, "old_child") + assertEquals([[1, null], [2, null], [3, null], [4, 3100L]], + sql(""" + select id, payload.renamed_child + from ${nestedTable} for version as of ${nestedCpReadd} + order by id + """)) + + // Scenario TC05/S18: PK upsert/delete/DV results remain correct at old and new refs. + assertEquals([[1, "alpha", "old-note-1", 10], [2, "beta", "old-note-2", 20]], + sql(""" + select id, old_name, note, score + from ${pkTable} for version as of 'pk_cp0' + order by id + """)) + assertEquals([[1, "alpha-updated", "new-note-1", 11, "extra-1"]], + sql(""" + select id, full_name, note, score, extra + from ${pkTable} for version as of ${pkCpRenameDelete} + order by id + """)) + assertEquals([[1, "alpha-updated", null], [3, "gamma", 3000L]], + sql(""" + select id, full_name, note + from ${pkTable} for version as of ${pkCpReadd} + order by id + """)) + assertUnknownColumn(""" + select old_name from ${pkTable} for version as of ${pkCpRenameDelete} + """, "old_name") + + // Scenario T14: incremental reads crossing a rename checkpoint bind the end schema. + List<List<Object>> incrementalJni + List<List<Object>> incrementalCpp + sql """set force_jni_scanner=false""" + sql """set enable_paimon_cpp_reader=false""" + incrementalJni = sql(""" + select id, full_name, score + from ${pkTable}@incr( + 'startSnapshotId'='${pkCp0}', + 'endSnapshotId'='${pkCpRenameDelete}' + ) + order by id + """) + sql """set enable_paimon_cpp_reader=true""" + incrementalCpp = sql(""" + select id, full_name, score + from ${pkTable}@incr( + 'startSnapshotId'='${pkCp0}', + 'endSnapshotId'='${pkCpRenameDelete}' + ) + order by id + """) + assertEquals(incrementalJni, incrementalCpp) Review Comment: **[P1] Prove the incremental result and reader paths** Equality permits both sides to be empty or equally wrong, and `incrementalJni` runs with `force_jni_scanner=false`; native raw-file conversion can also win before the JNI/C++ selector. Assert concrete rows and the end schema, then use plan/profile evidence or a fixture that proves each intended reader actually ran. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_time_travel_matrix.groovy: ########## @@ -0,0 +1,600 @@ +// 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_schema_time_travel_matrix", "p0,external,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_schema_time_travel_matrix" + String noCacheCatalogName = "test_paimon_schema_time_travel_matrix_no_cache" + String dbName = "paimon_schema_time_travel_matrix_db" + String topTable = "top_timeline" + String nestedTable = "nested_timeline" + String pkTable = "pk_dv_timeline" + String partitionTable = "partition_timeline" + + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_paimon """ + SELECT snapshot_id + FROM paimon.${dbName}.`${tableName}\$snapshots` + ORDER BY snapshot_id DESC + LIMIT 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + def createTag = { String tableName, String tagName, String snapshotId -> + spark_paimon """ + CALL paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}', + snapshot => ${snapshotId} + ) + """ + } + + def assertUnknownColumn = { String query, String columnName -> + test { + sql query + exception "'${columnName}'" + } + } + + sql """drop catalog if exists ${catalogName}""" + sql """drop catalog if exists ${noCacheCatalogName}""" + 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' + ) + """ + sql """ + CREATE CATALOG ${noCacheCatalogName} 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}.${topTable}; + CREATE TABLE paimon.${dbName}.${topTable} ( + id INT, + old_name STRING, + victim STRING, + metric INT + ) USING paimon + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${topTable} + VALUES (1, 'alpha', 'old-v1', 10); + """ + String topCp0 = latestSnapshotId(topTable) + createTag(topTable, "top_cp0", topCp0) + spark_paimon """ + CALL paimon.sys.create_branch( + '${dbName}.${topTable}', + 'top_cp0_branch', + 'top_cp0' + ) + """ + Thread.sleep(1100) + + // Scenario S01/S02/S03 x T00/T01/T02/T05/T08: + // add multiple columns and position one AFTER old_name. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + ADD COLUMNS (added STRING AFTER old_name, added_second BIGINT); Review Comment: **[P2] Observe every advertised schema cell** `added_second` and its placement are never queried/described, promoted `payload.keep` is never read afterward, and the re-added MAP/ARRAY children are not queried at their re-add checkpoint. These S02/S03/S12/S13 cells can regress without changing an assertion; add explicit value, type, order, and historical checks. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_jdbc_catalog.groovy: ########## @@ -97,9 +105,18 @@ suite("test_iceberg_jdbc_catalog", "p0,external") { executeCommand("/usr/bin/curl --max-time 600 ${mysql_driver_download_url} --output ${local_mysql_driver_path}", true) } for (def ip in host_ips) { - executeCommand("ssh -o StrictHostKeyChecking=no root@${ip} \"mkdir -p ${jdbc_drivers_dir}\"", false) - scpFiles("root", ip, local_driver_path, jdbc_drivers_dir, false) - scpFiles("root", ip, local_mysql_driver_path, jdbc_drivers_dir, false) + // Scenario: every FE/BE receives the JDBC drivers so distributed scans and FE failover + // do not depend on the regression runner sharing a filesystem with one cluster node. + if (localHostIps.contains(ip)) { + // A local test node must not require root SSH merely to install a JDBC driver. + executeCommand("mkdir -p ${jdbc_drivers_dir}", true) + executeCommand("cp -f ${local_driver_path} ${jdbc_drivers_dir}/${driver_name}", true) Review Comment: **[P1] Publish the shared driver atomically or serialize installation** Both `p0,external` JDBC suites are normal parallel suites and can check/download/copy `postgresql-42.5.0.jar` to the same live filename concurrently. One catalog can load the JAR while the other truncates and rewrites it. Use a lock/reuse strategy or stage to a unique temporary file and atomically rename. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_time_travel_matrix.groovy: ########## @@ -0,0 +1,600 @@ +// 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_schema_time_travel_matrix", "p0,external,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_schema_time_travel_matrix" + String noCacheCatalogName = "test_paimon_schema_time_travel_matrix_no_cache" + String dbName = "paimon_schema_time_travel_matrix_db" + String topTable = "top_timeline" + String nestedTable = "nested_timeline" + String pkTable = "pk_dv_timeline" + String partitionTable = "partition_timeline" + + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_paimon """ + SELECT snapshot_id + FROM paimon.${dbName}.`${tableName}\$snapshots` + ORDER BY snapshot_id DESC + LIMIT 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + def createTag = { String tableName, String tagName, String snapshotId -> + spark_paimon """ + CALL paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}', + snapshot => ${snapshotId} + ) + """ + } + + def assertUnknownColumn = { String query, String columnName -> + test { + sql query + exception "'${columnName}'" + } + } + + sql """drop catalog if exists ${catalogName}""" + sql """drop catalog if exists ${noCacheCatalogName}""" + 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' + ) + """ + sql """ + CREATE CATALOG ${noCacheCatalogName} 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}.${topTable}; + CREATE TABLE paimon.${dbName}.${topTable} ( + id INT, + old_name STRING, + victim STRING, + metric INT + ) USING paimon + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${topTable} + VALUES (1, 'alpha', 'old-v1', 10); + """ + String topCp0 = latestSnapshotId(topTable) + createTag(topTable, "top_cp0", topCp0) + spark_paimon """ + CALL paimon.sys.create_branch( + '${dbName}.${topTable}', + 'top_cp0_branch', + 'top_cp0' + ) + """ + Thread.sleep(1100) + + // Scenario S01/S02/S03 x T00/T01/T02/T05/T08: + // add multiple columns and position one AFTER old_name. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + ADD COLUMNS (added STRING AFTER old_name, added_second BIGINT); + INSERT INTO paimon.${dbName}.${topTable} + (id, old_name, victim, metric, added, added_second) + VALUES (2, 'beta', 'old-v2', 20, 'added-v2', 200); + """ + String topCpAdd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_add", topCpAdd) + Thread.sleep(1100) + + // Scenario S04/S05 x T00-T08: + // explicit old/new column binding catches rename regressions hidden by SELECT *. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + RENAME COLUMN old_name TO MixedName; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, victim, metric, added, added_second) + VALUES (3, 'gamma', 'old-v3', 30, 'added-v3', 300); + """ + String topCpRename = latestSnapshotId(topTable) + createTag(topTable, "top_cp_rename", topCpRename) + Thread.sleep(1100) + + // Scenario S06 x T00/T01/T02/T05: the dropped field remains available only to old refs. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} DROP COLUMN victim; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second) + VALUES (4, 'delta', 40, 'added-v4', 400); + """ + String topCpDrop = latestSnapshotId(topTable) + createTag(topTable, "top_cp_drop", topCpDrop) + Thread.sleep(1100) + + // Scenario S07 x T00/T01/T02/T05/T12/T13: + // drop/re-add with a different type creates a new field ID and must not expose old values. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ADD COLUMN victim BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (5, 'epsilon', 50, 'added-v5', 500, 5000); + """ + String topCpReadd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_readd", topCpReadd) + Thread.sleep(1100) + + // Scenario S08 x T00/T01/T02/T05: compatible INT -> BIGINT promotion. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ALTER COLUMN metric TYPE BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (6, 'zeta', 6000000000, 'added-v6', 600, 6000); + """ + String topCpPromote = latestSnapshotId(topTable) + createTag(topTable, "top_cp_promote", topCpPromote) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${nestedTable}; + CREATE TABLE paimon.${dbName}.${nestedTable} ( + id INT, + payload STRUCT<old_child: INT, keep: INT>, + attributes MAP<STRING, STRUCT<old_child: INT, keep: INT>>, + events ARRAY<STRUCT<old_child: INT, keep: INT>> + ) USING paimon + TBLPROPERTIES ('file.format'='orc'); + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 1, + named_struct('old_child', 10, 'keep', 11), + map('a', named_struct('old_child', 20, 'keep', 21)), + array(named_struct('old_child', 30, 'keep', 31)) + ); + """ + String nestedCp0 = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp0", nestedCp0) + + // Scenario S09/S14/S15 x T00/T01/T02/T05: + // add children to STRUCT, MAP value struct and ARRAY element struct. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.added_child INT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 2, + named_struct('old_child', 110, 'keep', 111, 'added_child', 112), + map('a', named_struct('old_child', 120, 'keep', 121, 'added_child', 122)), + array(named_struct('old_child', 130, 'keep', 131, 'added_child', 132)) + ); + """ + String nestedCpAdd = latestSnapshotId(nestedTable) + + // Scenario S10/S14/S15 x T00/T01/T02/T05: rename every supported nested path. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN payload.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN attributes.value.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN events.element.old_child TO renamed_child; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 3, + named_struct('renamed_child', 210, 'keep', 211, 'added_child', 212), + map('a', named_struct('renamed_child', 220, 'keep', 221, 'added_child', 222)), + array(named_struct('renamed_child', 230, 'keep', 231, 'added_child', 232)) + ); + """ + String nestedCpRename = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp_rename", nestedCpRename) + + // Scenario S11/S12/S13 x T00/T01/T02/T05: + // nested drop/re-add and type promotion preserve field-ID isolation. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} DROP COLUMN payload.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN attributes.value.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN events.element.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ALTER COLUMN payload.keep TYPE BIGINT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 4, + named_struct('keep', 311, 'added_child', 312, 'renamed_child', 3100), + map('a', named_struct('keep', 321, 'added_child', 322, 'renamed_child', 3200)), + array(named_struct('keep', 331, 'added_child', 332, 'renamed_child', 3300)) + ); + """ + String nestedCpReadd = latestSnapshotId(nestedTable) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${pkTable}; + CREATE TABLE paimon.${dbName}.${pkTable} ( + id INT NOT NULL, + old_name STRING, + note STRING, + score INT + ) USING paimon + TBLPROPERTIES ( + 'bucket'='1', + 'primary-key'='id', + 'file.format'='parquet', + 'deletion-vectors.enabled'='true' + ); + INSERT INTO paimon.${dbName}.${pkTable} VALUES + (1, 'alpha', 'old-note-1', 10), + (2, 'beta', 'old-note-2', 20); + """ + String pkCp0 = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp0", pkCp0) + + // Scenario S01/S04/S18 x TC05: + // PK remains stable while a non-key field is added/renamed and rows are upserted/deleted. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN extra STRING; + ALTER TABLE paimon.${dbName}.${pkTable} RENAME COLUMN old_name TO full_name; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, note, score, extra) + VALUES (1, 'alpha-updated', 'new-note-1', 11, 'extra-1'); + DELETE FROM paimon.${dbName}.${pkTable} WHERE id = 2; + """ + String pkCpRenameDelete = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_rename_delete", pkCpRenameDelete) + + // Scenario S06/S07/S08/S18 x TC05: + // drop/re-add and promotion are combined with a later upsert and DV compaction. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} DROP COLUMN note; + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN note BIGINT; + ALTER TABLE paimon.${dbName}.${pkTable} ALTER COLUMN score TYPE BIGINT; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, score, extra, note) + VALUES (3, 'gamma', 3000000000, 'extra-3', 3000); + """ + String pkCpReadd = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_readd", pkCpReadd) + spark_paimon """ + CALL paimon.sys.compact(table => '${dbName}.${pkTable}', compact_strategy => 'full') + """ + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${partitionTable}; + CREATE TABLE paimon.${dbName}.${partitionTable} ( + id INT, + old_partition STRING, + payload STRING + ) USING paimon + PARTITIONED BY (old_partition) + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${partitionTable} + VALUES (1, 'p1', 'old'), (2, 'p2', 'old'); + """ + String partitionCp0 = latestSnapshotId(partitionTable) + createTag(partitionTable, "partition_cp0", partitionCp0) + + // Scenario S16-negative: Paimon must reject partition-column rename atomically. + String partitionRenameError = null + try { + spark_paimon """ + ALTER TABLE paimon.${dbName}.${partitionTable} + RENAME COLUMN old_partition TO new_partition + """ + } catch (Exception e) { + partitionRenameError = e.getMessage() + } + assertNotNull(partitionRenameError) + assertTrue(partitionRenameError.contains("Cannot rename partition column")) + + // Scenario S17 x TC03: rename a payload field while retaining partition pruning. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${partitionTable} + RENAME COLUMN payload TO new_payload; + INSERT INTO paimon.${dbName}.${partitionTable} + VALUES (3, 'p3', 'new'); + """ + String partitionCpRename = latestSnapshotId(partitionTable) + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh catalog ${catalogName}""" + + // Scenario TC01: validate latest schema/data, explicit new binding, predicate and aggregate. + assertEquals([[1, null], [2, null], [3, null], [4, null], [5, 5000L], [6, 6000L]], + sql("""select id, victim from ${topTable} order by id""")) + assertEquals([[6, 6000000000L]], + sql("""select id, metric from ${topTable} where metric > 5000000000""")) + assertEquals([[6L, 6000000150L]], + sql("""select count(*), sum(metric) from ${topTable}""")) + assertUnknownColumn("""select old_name from ${topTable}""", "old_name") + + // Scenario T01/T05/T06/T08: old snapshot/tag/branch uses the old schema. + List<List<Object>> topCp0Rows = [[1, "alpha", "old-v1", 10]] + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for version as of ${topCp0} + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for version as of 'top_cp0' + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable}@tag(top_cp0) + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable}@branch(top_cp0_branch) + order by id + """)) + assertUnknownColumn(""" + select MixedName from ${topTable} for version as of ${topCp0} + """, "MixedName") + + // Scenario T03/T04: time string and epoch millis resolve to the pre-change schema. + List<List<Object>> cp0TimeRows = sql(""" + select date_format(date_add(commit_time, interval 1 second), '%Y-%m-%d %H:%i:%s'), + cast(unix_timestamp(commit_time) * 1000 + 999 as bigint) + from ${topTable}\$snapshots + where snapshot_id = ${topCp0} + """) + assertEquals(1, cp0TimeRows.size()) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for time as of "${cp0TimeRows[0][0]}" + order by id + """)) + assertEquals(topCp0Rows, sql(""" + select id, old_name, victim, metric + from ${topTable} for time as of ${cp0TimeRows[0][1]} + order by id + """)) + + // Scenario S01-S08: checkpoint projections prove name/type/field-ID isolation. + assertEquals([[1, "alpha", null], [2, "beta", "added-v2"]], + sql(""" + select id, old_name, added + from ${topTable} for version as of ${topCpAdd} + order by id + """)) + assertEquals([[1, "alpha"], [2, "beta"], [3, "gamma"]], + sql(""" + select id, MixedName + from ${topTable} for version as of ${topCpRename} + where metric >= 10 + order by id + """)) + assertUnknownColumn(""" + select old_name from ${topTable} for version as of ${topCpRename} + """, "old_name") + assertUnknownColumn(""" + select victim from ${topTable} for version as of ${topCpDrop} + """, "victim") + assertEquals([[1, null], [2, null], [3, null], [4, null], [5, 5000L]], + sql(""" + select id, victim + from ${topTable} for version as of ${topCpReadd} + order by id + """)) + assertEquals([[6, 6000000000L]], + sql(""" + select id, metric + from ${topTable} for version as of ${topCpPromote} + where metric > 5000000000 + """)) + + // Scenario TC02: nested refs validate STRUCT/MAP/ARRAY projection and predicate. + assertEquals([[1, 10, 20, 30]], + sql(""" + select id, payload.old_child, + element_at(attributes, 'a').old_child, + events[1].old_child + from ${nestedTable} for version as of ${nestedCp0} + where payload.old_child = 10 + """)) + assertEquals([[1, null, null, null], [2, 112, 122, 132]], + sql(""" + select id, payload.added_child, + element_at(attributes, 'a').added_child, + events[1].added_child + from ${nestedTable} for version as of ${nestedCpAdd} + order by id + """)) + assertEquals([[1, 10, 20, 30], [2, 110, 120, 130], [3, 210, 220, 230]], + sql(""" + select id, payload.renamed_child, + element_at(attributes, 'a').renamed_child, + events[1].renamed_child + from ${nestedTable} for version as of 'nested_cp_rename' + order by id + """)) + assertUnknownColumn(""" + select payload.old_child + from ${nestedTable} for version as of ${nestedCpRename} + """, "old_child") + assertEquals([[1, null], [2, null], [3, null], [4, 3100L]], + sql(""" + select id, payload.renamed_child + from ${nestedTable} for version as of ${nestedCpReadd} + order by id + """)) + + // Scenario TC05/S18: PK upsert/delete/DV results remain correct at old and new refs. + assertEquals([[1, "alpha", "old-note-1", 10], [2, "beta", "old-note-2", 20]], + sql(""" + select id, old_name, note, score + from ${pkTable} for version as of 'pk_cp0' + order by id + """)) + assertEquals([[1, "alpha-updated", "new-note-1", 11, "extra-1"]], + sql(""" + select id, full_name, note, score, extra + from ${pkTable} for version as of ${pkCpRenameDelete} + order by id + """)) + assertEquals([[1, "alpha-updated", null], [3, "gamma", 3000L]], + sql(""" + select id, full_name, note + from ${pkTable} for version as of ${pkCpReadd} + order by id + """)) + assertUnknownColumn(""" + select old_name from ${pkTable} for version as of ${pkCpRenameDelete} + """, "old_name") + + // Scenario T14: incremental reads crossing a rename checkpoint bind the end schema. + List<List<Object>> incrementalJni + List<List<Object>> incrementalCpp + sql """set force_jni_scanner=false""" + sql """set enable_paimon_cpp_reader=false""" + incrementalJni = sql(""" + select id, full_name, score + from ${pkTable}@incr( + 'startSnapshotId'='${pkCp0}', + 'endSnapshotId'='${pkCpRenameDelete}' + ) + order by id + """) + sql """set enable_paimon_cpp_reader=true""" + incrementalCpp = sql(""" + select id, full_name, score + from ${pkTable}@incr( + 'startSnapshotId'='${pkCp0}', + 'endSnapshotId'='${pkCpRenameDelete}' + ) + order by id + """) + assertEquals(incrementalJni, incrementalCpp) + + // Scenario TC03/S16: partition pruning and renamed payloads bind to their own snapshots. + assertEquals([[1, "p1", "old"], [2, "p2", "old"]], + sql(""" + select id, old_partition, payload + from ${partitionTable} for version as of ${partitionCp0} + where old_partition = 'p1' or old_partition = 'p2' Review Comment: **[P2] Make the pruning predicate exclude a partition** The checkpoint contains only `p1` and `p2`, and this predicate selects both, so removing it produces the same rows. Restrict the query to one partition and assert selected-partition evidence in `EXPLAIN`; otherwise the claimed pruning cell passes even when pruning is ignored. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_time_travel_matrix.groovy: ########## @@ -0,0 +1,732 @@ +// 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_schema_time_travel_matrix", + "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_schema_time_travel_matrix" + String noCacheCatalogName = "test_iceberg_schema_time_travel_matrix_no_cache" + String dbName = "iceberg_schema_time_travel_matrix_db" + String topTable = "top_timeline" + String nestedTable = "nested_timeline" + String dorisNestedTable = "doris_nested_timeline" + String deleteTable = "delete_partition_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() + } + + def assertUnknownColumn = { String query, String columnName -> + test { + sql query + exception "'${columnName}'" + } + } + + sql """drop catalog if exists ${catalogName}""" + sql """drop catalog if exists ${noCacheCatalogName}""" + 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 """ + CREATE CATALOG ${noCacheCatalogName} 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}.${topTable}; + CREATE TABLE demo.${dbName}.${topTable} ( + id INT, + old_name STRING, + victim STRING, + metric INT + ) USING iceberg + TBLPROPERTIES ( + 'format-version'='2', + 'write.format.default'='parquet' + ); + INSERT INTO demo.${dbName}.${topTable} + VALUES (1, 'alpha', 'old-v1', 10); + """ + String topCp0 = latestSnapshotId(topTable) + sql """ALTER TABLE `${catalogName}`.`${dbName}`.`${topTable}` CREATE TAG top_cp0""" + sql """ALTER TABLE `${catalogName}`.`${dbName}`.`${topTable}` CREATE BRANCH top_cp0_branch""" + Thread.sleep(1100) + + // Scenario S01/S02/S03 x T00/T01/T02/T05/T08: + // add and reorder columns while keeping the pre-change snapshot, tag and branch readable. + spark_iceberg_multi """ + ALTER TABLE demo.${dbName}.${topTable} + ADD COLUMNS (added STRING AFTER old_name, added_second BIGINT); Review Comment: **[P2] Add value and type oracles for the created cells** `added_second` has no historical NULL-backfill/current-value/type assertion, and the promoted nested siblings are omitted from later projections. Add explicit historical/current checks using beyond-INT values where relevant; these advertised cells can currently regress without changing a test result. ########## regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_equality_delete_time_travel.groovy: ########## @@ -0,0 +1,145 @@ +// 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_schema_equality_delete_time_travel", + "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_schema_equality_delete_time_travel" + + 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 """use multi_catalog""" + + try { + ["par", "orc"].each { format -> + String tableName = "equality_delete_${format}_1" + List<String> snapshotIds = sql(""" + select snapshot_id + from ${tableName}\$snapshots + order by committed_at, snapshot_id + """).collect { row -> row[0].toString() } + assertTrue(snapshotIds.size() >= 7, + "Equality-delete fixture must retain the schema timeline for ${tableName}") + + String oldSnapshot = snapshotIds[0] + String renamedSnapshot = snapshotIds[2] + String latestSnapshot = snapshotIds.last() + + // Scenario TC04-EQ-S04: equality deletes before rename stay bound to the old field ID. + assertEquals([ + [1, "smith", "a"], + [2, "danny", "b"], + [3, "alice", "c"], + [4, "bob", "d"] + ], sql(""" + select id, name, data + from ${tableName} for version as of ${oldSnapshot} + order by id + """)) + test { + sql """ + select new_new_id + from ${tableName} for version as of ${oldSnapshot} + """ + exception "Unknown column 'new_new_id'" + } + + // Scenario TC04-EQ-S04/S08: the renamed-key snapshot applies equality deletes by field ID. Review Comment: **[P1] Add the type promotion this checkpoint claims** Both referenced equality-delete fixture histories only rename fields and later add another `INT`; neither contains an S08 type transition. Add a real compatible promotion with a beyond-INT value and assert the pre/post rows, or remove equality-delete × promotion from the coverage claim. ########## regression-test/suites/external_table_p0/paimon/test_paimon_schema_time_travel_matrix.groovy: ########## @@ -0,0 +1,600 @@ +// 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_schema_time_travel_matrix", "p0,external,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_schema_time_travel_matrix" + String noCacheCatalogName = "test_paimon_schema_time_travel_matrix_no_cache" + String dbName = "paimon_schema_time_travel_matrix_db" + String topTable = "top_timeline" + String nestedTable = "nested_timeline" + String pkTable = "pk_dv_timeline" + String partitionTable = "partition_timeline" + + def latestSnapshotId = { String tableName -> + List<List<Object>> rows = spark_paimon """ + SELECT snapshot_id + FROM paimon.${dbName}.`${tableName}\$snapshots` + ORDER BY snapshot_id DESC + LIMIT 1 + """ + assertEquals(1, rows.size()) + return rows[0][0].toString() + } + + def createTag = { String tableName, String tagName, String snapshotId -> + spark_paimon """ + CALL paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => '${tagName}', + snapshot => ${snapshotId} + ) + """ + } + + def assertUnknownColumn = { String query, String columnName -> + test { + sql query + exception "'${columnName}'" + } + } + + sql """drop catalog if exists ${catalogName}""" + sql """drop catalog if exists ${noCacheCatalogName}""" + 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' + ) + """ + sql """ + CREATE CATALOG ${noCacheCatalogName} 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}.${topTable}; + CREATE TABLE paimon.${dbName}.${topTable} ( + id INT, + old_name STRING, + victim STRING, + metric INT + ) USING paimon + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${topTable} + VALUES (1, 'alpha', 'old-v1', 10); + """ + String topCp0 = latestSnapshotId(topTable) + createTag(topTable, "top_cp0", topCp0) + spark_paimon """ + CALL paimon.sys.create_branch( + '${dbName}.${topTable}', + 'top_cp0_branch', + 'top_cp0' + ) + """ + Thread.sleep(1100) + + // Scenario S01/S02/S03 x T00/T01/T02/T05/T08: + // add multiple columns and position one AFTER old_name. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + ADD COLUMNS (added STRING AFTER old_name, added_second BIGINT); + INSERT INTO paimon.${dbName}.${topTable} + (id, old_name, victim, metric, added, added_second) + VALUES (2, 'beta', 'old-v2', 20, 'added-v2', 200); + """ + String topCpAdd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_add", topCpAdd) + Thread.sleep(1100) + + // Scenario S04/S05 x T00-T08: + // explicit old/new column binding catches rename regressions hidden by SELECT *. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} + RENAME COLUMN old_name TO MixedName; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, victim, metric, added, added_second) + VALUES (3, 'gamma', 'old-v3', 30, 'added-v3', 300); + """ + String topCpRename = latestSnapshotId(topTable) + createTag(topTable, "top_cp_rename", topCpRename) + Thread.sleep(1100) + + // Scenario S06 x T00/T01/T02/T05: the dropped field remains available only to old refs. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} DROP COLUMN victim; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second) + VALUES (4, 'delta', 40, 'added-v4', 400); + """ + String topCpDrop = latestSnapshotId(topTable) + createTag(topTable, "top_cp_drop", topCpDrop) + Thread.sleep(1100) + + // Scenario S07 x T00/T01/T02/T05/T12/T13: + // drop/re-add with a different type creates a new field ID and must not expose old values. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ADD COLUMN victim BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (5, 'epsilon', 50, 'added-v5', 500, 5000); + """ + String topCpReadd = latestSnapshotId(topTable) + createTag(topTable, "top_cp_readd", topCpReadd) + Thread.sleep(1100) + + // Scenario S08 x T00/T01/T02/T05: compatible INT -> BIGINT promotion. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${topTable} ALTER COLUMN metric TYPE BIGINT; + INSERT INTO paimon.${dbName}.${topTable} + (id, MixedName, metric, added, added_second, victim) + VALUES (6, 'zeta', 6000000000, 'added-v6', 600, 6000); + """ + String topCpPromote = latestSnapshotId(topTable) + createTag(topTable, "top_cp_promote", topCpPromote) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${nestedTable}; + CREATE TABLE paimon.${dbName}.${nestedTable} ( + id INT, + payload STRUCT<old_child: INT, keep: INT>, + attributes MAP<STRING, STRUCT<old_child: INT, keep: INT>>, + events ARRAY<STRUCT<old_child: INT, keep: INT>> + ) USING paimon + TBLPROPERTIES ('file.format'='orc'); + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 1, + named_struct('old_child', 10, 'keep', 11), + map('a', named_struct('old_child', 20, 'keep', 21)), + array(named_struct('old_child', 30, 'keep', 31)) + ); + """ + String nestedCp0 = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp0", nestedCp0) + + // Scenario S09/S14/S15 x T00/T01/T02/T05: + // add children to STRUCT, MAP value struct and ARRAY element struct. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.added_child INT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.added_child INT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 2, + named_struct('old_child', 110, 'keep', 111, 'added_child', 112), + map('a', named_struct('old_child', 120, 'keep', 121, 'added_child', 122)), + array(named_struct('old_child', 130, 'keep', 131, 'added_child', 132)) + ); + """ + String nestedCpAdd = latestSnapshotId(nestedTable) + + // Scenario S10/S14/S15 x T00/T01/T02/T05: rename every supported nested path. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN payload.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN attributes.value.old_child TO renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + RENAME COLUMN events.element.old_child TO renamed_child; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 3, + named_struct('renamed_child', 210, 'keep', 211, 'added_child', 212), + map('a', named_struct('renamed_child', 220, 'keep', 221, 'added_child', 222)), + array(named_struct('renamed_child', 230, 'keep', 231, 'added_child', 232)) + ); + """ + String nestedCpRename = latestSnapshotId(nestedTable) + createTag(nestedTable, "nested_cp_rename", nestedCpRename) + + // Scenario S11/S12/S13 x T00/T01/T02/T05: + // nested drop/re-add and type promotion preserve field-ID isolation. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${nestedTable} DROP COLUMN payload.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN attributes.value.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} + DROP COLUMN events.element.renamed_child; + ALTER TABLE paimon.${dbName}.${nestedTable} ADD COLUMN payload.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN attributes.value.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ADD COLUMN events.element.renamed_child BIGINT; + ALTER TABLE paimon.${dbName}.${nestedTable} + ALTER COLUMN payload.keep TYPE BIGINT; + INSERT INTO paimon.${dbName}.${nestedTable} VALUES ( + 4, + named_struct('keep', 311, 'added_child', 312, 'renamed_child', 3100), + map('a', named_struct('keep', 321, 'added_child', 322, 'renamed_child', 3200)), + array(named_struct('keep', 331, 'added_child', 332, 'renamed_child', 3300)) + ); + """ + String nestedCpReadd = latestSnapshotId(nestedTable) + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${pkTable}; + CREATE TABLE paimon.${dbName}.${pkTable} ( + id INT NOT NULL, + old_name STRING, + note STRING, + score INT + ) USING paimon + TBLPROPERTIES ( + 'bucket'='1', + 'primary-key'='id', + 'file.format'='parquet', + 'deletion-vectors.enabled'='true' + ); + INSERT INTO paimon.${dbName}.${pkTable} VALUES + (1, 'alpha', 'old-note-1', 10), + (2, 'beta', 'old-note-2', 20); + """ + String pkCp0 = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp0", pkCp0) + + // Scenario S01/S04/S18 x TC05: + // PK remains stable while a non-key field is added/renamed and rows are upserted/deleted. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN extra STRING; + ALTER TABLE paimon.${dbName}.${pkTable} RENAME COLUMN old_name TO full_name; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, note, score, extra) + VALUES (1, 'alpha-updated', 'new-note-1', 11, 'extra-1'); + DELETE FROM paimon.${dbName}.${pkTable} WHERE id = 2; + """ + String pkCpRenameDelete = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_rename_delete", pkCpRenameDelete) + + // Scenario S06/S07/S08/S18 x TC05: + // drop/re-add and promotion are combined with a later upsert and DV compaction. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${pkTable} DROP COLUMN note; + ALTER TABLE paimon.${dbName}.${pkTable} ADD COLUMN note BIGINT; + ALTER TABLE paimon.${dbName}.${pkTable} ALTER COLUMN score TYPE BIGINT; + INSERT INTO paimon.${dbName}.${pkTable} + (id, full_name, score, extra, note) + VALUES (3, 'gamma', 3000000000, 'extra-3', 3000); + """ + String pkCpReadd = latestSnapshotId(pkTable) + createTag(pkTable, "pk_cp_readd", pkCpReadd) + spark_paimon """ + CALL paimon.sys.compact(table => '${dbName}.${pkTable}', compact_strategy => 'full') + """ + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.${partitionTable}; + CREATE TABLE paimon.${dbName}.${partitionTable} ( + id INT, + old_partition STRING, + payload STRING + ) USING paimon + PARTITIONED BY (old_partition) + TBLPROPERTIES ('file.format'='parquet'); + INSERT INTO paimon.${dbName}.${partitionTable} + VALUES (1, 'p1', 'old'), (2, 'p2', 'old'); + """ + String partitionCp0 = latestSnapshotId(partitionTable) + createTag(partitionTable, "partition_cp0", partitionCp0) + + // Scenario S16-negative: Paimon must reject partition-column rename atomically. + String partitionRenameError = null + try { + spark_paimon """ + ALTER TABLE paimon.${dbName}.${partitionTable} + RENAME COLUMN old_partition TO new_partition + """ + } catch (Exception e) { + partitionRenameError = e.getMessage() + } + assertNotNull(partitionRenameError) + assertTrue(partitionRenameError.contains("Cannot rename partition column")) + + // Scenario S17 x TC03: rename a payload field while retaining partition pruning. + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.${partitionTable} + RENAME COLUMN payload TO new_payload; + INSERT INTO paimon.${dbName}.${partitionTable} + VALUES (3, 'p3', 'new'); + """ + String partitionCpRename = latestSnapshotId(partitionTable) + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh catalog ${catalogName}""" + + // Scenario TC01: validate latest schema/data, explicit new binding, predicate and aggregate. + assertEquals([[1, null], [2, null], [3, null], [4, null], [5, 5000L], [6, 6000L]], Review Comment: **[P1] Commit generated expected results for fixed row matrices** The new standalone suites use many hard-coded deterministic `assertEquals` row sets and add no `.out` files. Root `AGENTS.md` requires determined regression results to use named `qt`/`order_qt` cases with auto-generated outputs. Keep direct assertions only for runtime-derived metadata and invariants. ########## regression-test/suites/external_table_p0/paimon/test_paimon_jdbc_catalog.groovy: ########## @@ -90,25 +92,63 @@ suite("test_paimon_jdbc_catalog", "p0,external") { } executeCommand("mkdir -p ${localDriverDir}", false, 60) - executeCommand("mkdir -p ${jdbcDriversDir}", true, 60) if (!new File(localDriverPath).exists()) { executeCommand("/usr/bin/curl --max-time 600 ${driverDownloadUrl} --output ${localDriverPath}", true, 660) } - executeCommand("cp -f ${localDriverPath} ${jdbcDriversDir}/${driverName}", true, 60) - String sparkContainerName = executeCommand("docker ps --filter name=spark-iceberg --format {{.Names}}", false, 30) + def clusterHostIps = new ArrayList() + String[][] backends = sql """show backends""" + for (def backend in backends) { + clusterHostIps.add(backend[1]) + } + String[][] frontends = sql """show frontends""" + for (def frontend in frontends) { + clusterHostIps.add(frontend[1]) + } + clusterHostIps = clusterHostIps.unique() + + Set<String> localHostIps = ["127.0.0.1", "localhost", "::1"] as Set + java.util.Collections.list(java.net.NetworkInterface.getNetworkInterfaces()).each { networkInterface -> + java.util.Collections.list(networkInterface.getInetAddresses()).each { address -> + localHostIps.add(address.getHostAddress().split("%")[0]) + } + } + localHostIps.add(java.net.InetAddress.getLocalHost().getHostName()) + localHostIps.add(java.net.InetAddress.getLocalHost().getCanonicalHostName()) + + for (def hostIp in clusterHostIps) { + // Scenario: every FE/BE receives the JDBC driver so metadata failover and distributed + // scan scheduling do not depend on a driver installed only on the regression runner. + if (localHostIps.contains(hostIp)) { + executeCommand("mkdir -p ${jdbcDriversDir}", true, 60) + executeCommand("cp -f ${localDriverPath} ${jdbcDriversDir}/${driverName}", true, 60) + } else { + executeCommand( + "ssh -o BatchMode=yes -o StrictHostKeyChecking=no root@${hostIp} \"mkdir -p ${jdbcDriversDir}\"", + true, + 60 + ) + scpFiles("root", hostIp, localDriverPath, jdbcDriversDir, false) + } Review Comment: **[P1] Bracket IPv6 literals in the SCP authority** The helper builds `root@${host}:${path}`. For a remote IPv6 literal, OpenSSH treats the first colon as the host/path separator, so SSH `mkdir` can succeed and this copy then targets the wrong host. Use `root@[${host}]:${path}` for IPv6, ideally in the shared staging helper. -- 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]
