From c70ea3b38ab50d758105151443127a7043e02dea Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 13 Sep 2026 11:31:32 +0800 Subject: [PATCH 01/11] fix(paimon): preserve schema and timestamp precision --- .../paimon/PaimonConnectorMetadata.java | 25 ++--- .../paimon/PaimonPredicateConverter.java | 12 +-- .../paimon/PaimonScanPlanProvider.java | 8 +- .../PaimonConnectorMetadataMvccTest.java | 14 +-- .../paimon/PaimonPredicateConverterTest.java | 12 +++ .../paimon/PaimonScanPlanProviderTest.java | 93 +++++++++++++++++++ 6 files changed, 137 insertions(+), 27 deletions(-) diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java index 62113194c76895..9d353024dc0451 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java @@ -619,12 +619,13 @@ public boolean usesStatementSnapshotForOptions( * loads the branch as its OWN table (independent schema/snapshots, via the 3-arg branch * Identifier through {@link PaimonTableHandle#withBranch}) and pins its LATEST snapshot — * branches have NO in-branch time-travel (legacy {@code PaimonExternalTable} reads the - * branch's {@code latestSnapshot()} only). The branch identity is carried to + * branch's {@code latestSnapshot()} only). Its schema id remains {@code -1} so binding uses + * the branch's current schema, which can advance without a new data snapshot. The branch + * identity is carried to * {@link #applySnapshot} via an internal sentinel ({@code CoreOptions.BRANCH} key, NOT a * scan-copy option); no {@code scan.snapshot-id} is pinned (the branch reads its own latest). - * An empty branch (no snapshot) pins {@code snapshotId=-1} and {@code schemaId=-1}: a benign - * divergence from legacy's {@code schemaId=0L} — the resulting schema is identical (both - * resolve to the branch's current schema), mirroring the INCREMENTAL empty-table -1 note. + * An empty branch also pins {@code snapshotId=-1}; both empty and non-empty branches bind + * against the current branch schema. * * *

CONTRACT DIFFERENCE (intentional, documented): legacy {@code PaimonUtil} THREW a @@ -733,9 +734,9 @@ public Optional resolveTimeTravel( // latestSnapshot() only). Table branchTable = resolveTable(paimonHandle.withBranch(branchName)); long snapshotId = catalogOps.latestSnapshotId(branchTable).orElse(-1L); - long schemaId = snapshotId < 0 - ? -1L - : catalogOps.snapshotSchemaId(branchTable, snapshotId).orElse(-1L); + // A schema-only ALTER advances the branch schema without creating a data snapshot. + // Keep the data fence but bind the branch's current schema instead of that snapshot's old schema. + long schemaId = -1L; // Carry the branch identity to applySnapshot via an internal sentinel // (CoreOptions.BRANCH key). Branch is a handle-IDENTITY change, not a scan-copy // option: applySnapshot reads this sentinel and routes it to handle.withBranch (it is @@ -905,8 +906,8 @@ private long parseTimestampMillis(ConnectorSession session, ConnectorTimeTravelS *

Threads the FULL {@code snapshot.getProperties()} map: this may be * {@code scan.snapshot-id=} (snapshot-id / timestamp time-travel) OR * {@code scan.tag-name=} (tag time-travel), whichever {@link #resolveTimeTravel} pinned. - * When {@code properties} is empty (the {@link #beginQuerySnapshot} latest-pin path, which - * carries no properties) it falls back to {@code scan.snapshot-id=} for B5a parity. + * When {@code properties} is empty (the {@link #beginQuerySnapshot} latest-pin path), it pins + * {@code scan.snapshot-id=} while retaining the current bound schema generation. * *

BRANCH is special: when the snapshot carries the {@code CoreOptions.BRANCH} sentinel (set by * {@link #resolveTimeTravel}'s BRANCH case), it is a handle-IDENTITY change, not a scan option — @@ -959,8 +960,10 @@ public ConnectorTableHandle applySnapshot(ConnectorSession session, return paimonHandle.withScanOptions( PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), -1L)); } - Map scanOptions = Collections.singletonMap( - CoreOptions.SCAN_SNAPSHOT_ID.key(), String.valueOf(snapshot.getSnapshotId())); + // The latest statement snapshot fences data only. Its schema was bound from the current table, + // which may be newer after a schema-only ALTER that created no data snapshot. + Map scanOptions = PaimonScanParams.pinOptionsToSnapshot( + Collections.emptyMap(), snapshot.getSnapshotId()); return paimonHandle.withScanOptions(scanOptions); } diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java index 5fb90b3beb9732..c03c9675645b81 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java @@ -42,7 +42,6 @@ import java.math.BigDecimal; import java.time.LocalDate; import java.time.LocalDateTime; -import java.time.ZoneOffset; import java.util.ArrayList; import java.util.List; import java.util.stream.Collectors; @@ -329,15 +328,10 @@ private Object convertLiteralValue(ConnectorLiteral literal, DataType paimonType } return null; case TIMESTAMP_WITHOUT_TIME_ZONE: - // Zone-free type: interpret the literal's wall-clock in UTC to match paimon's - // stored min/max file/partition stats (computed by reading the wall clock as UTC). - // Mirrors legacy PaimonValueConverter#visit(TimestampType), which uses a fixed - // GMT Calendar. Using the session zone here would shift the epoch-millis vs the - // stored stats and risk false file/partition pruning = silent data loss. + // Preserve the complete wall-clock value: narrowing it to epoch milliseconds can + // make Paimon prune every file matching a non-millisecond-aligned predicate. if (value instanceof LocalDateTime) { - LocalDateTime dt = (LocalDateTime) value; - long millis = dt.toInstant(ZoneOffset.UTC).toEpochMilli(); - return Timestamp.fromEpochMillis(millis); + return Timestamp.fromLocalDateTime((LocalDateTime) value); } return null; case TIMESTAMP_WITH_LOCAL_TIME_ZONE: diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java index 3f40af1f56bc5b..75bb1ce7797462 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java @@ -346,7 +346,13 @@ Table resolveScanTable(PaimonTableHandle paimonHandle) { Map scanOptions = paimonHandle.getScanOptions(); Table finalTable = table; if (scanOptions != null && !scanOptions.isEmpty()) { - if (PaimonScanParams.isOptionsPin(scanOptions)) { + if (table instanceof FileStoreTable + && PaimonScanParams.preservesBoundSchema(scanOptions)) { + // A statement fence owns data visibility, not schema time travel. Reusing Table.copy + // here would roll schema-only ALTERs back to the data snapshot's older schema. + finalTable = PaimonScanParams.applyOptionsWithoutTimeTravel( + (FileStoreTable) table, scanOptions); + } else if (PaimonScanParams.isOptionsPin(scanOptions)) { // An @options pin owns the whole scan-startup state: applyOptions strips the internal // markers and nulls out the absent members of paimon's inherited read-state family, so a // scan.mode / tag persisted on the base table cannot leak into this relation's read. diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java index 61ec28c61d952b..ac9cd639e417c6 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java @@ -518,7 +518,8 @@ public void resolveBranchFoundLoadsBranchTableAndPinsItsLatestSnapshot() { ConnectorMvccSnapshot snap = metadataWith(ops) .resolveTimeTravel(null, handle, ConnectorTimeTravelSpec.branch("b1")).get(); - // WHY: @branch must pin the BRANCH's LATEST snapshot + its schemaId, carry the branch identity + // WHY: @branch must pin the BRANCH's latest data snapshot while retaining its current schema, + // because a schema-only ALTER does not create a data snapshot. It must also carry the branch identity // via the CoreOptions.BRANCH sentinel (NOT scan.snapshot-id — the branch reads its own latest), // and validate the branch on the BASE table. Branches have no in-branch time-travel (legacy // reads the branch's latestSnapshot() only). MUTATION: pinning scan.snapshot-id -> the no-key @@ -526,8 +527,8 @@ public void resolveBranchFoundLoadsBranchTableAndPinsItsLatestSnapshot() { // base table instead of the branch -> the lastMvccTable assertion red. Assertions.assertEquals(7L, snap.getSnapshotId(), "@branch must pin the BRANCH's latest snapshot id"); - Assertions.assertEquals(3L, snap.getSchemaId(), - "@branch must stamp the BRANCH's latest snapshot schemaId"); + Assertions.assertEquals(-1L, snap.getSchemaId(), + "@branch must use the current-schema sentinel independently of its latest data snapshot"); Assertions.assertEquals("b1", snap.getProperties().get(CoreOptions.BRANCH.key()), "@branch must carry the branch name under the CoreOptions.BRANCH sentinel key"); Assertions.assertNull(snap.getProperties().get("scan.snapshot-id"), @@ -541,10 +542,11 @@ public void resolveBranchFoundLoadsBranchTableAndPinsItsLatestSnapshot() { "the branch table must be loaded via a 3-arg branch Identifier"); Assertions.assertNull(ops.lastGetTableId.getSystemTableName(), "a branch load must NOT carry a system-table name"); - // The latest-snapshot / schemaId lookups ran against the BRANCH table, not the base. (The last - // seam call before this assertion is snapshotSchemaId, which captured lastMvccTable.) + // The latest-snapshot lookup ran against the BRANCH table, not the base. Assertions.assertSame(branch, ops.lastMvccTable, - "latestSnapshotId/snapshotSchemaId must run against the BRANCH table"); + "latestSnapshotId must run against the BRANCH table"); + Assertions.assertFalse(ops.log.contains("snapshotSchemaId:7"), + "a branch schema-only ALTER must not be hidden by the data snapshot's old schema id"); // branchExists validation ran against the BASE table (legacy resolvePaimonBranch). Assertions.assertEquals("b1", ops.lastBranchExistsArg, "branchExists must be asked about the requested branch name"); diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java index 6fcf5564d590c5..4dde877319eac0 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java @@ -88,6 +88,18 @@ public void ntzPushedWithUtcSemantics() { "NTZ literal must be the wall clock converted via fixed UTC (legacy GMT parity)"); } + @Test + public void ntzPushPreservesMicroseconds() { + RowType rowType = RowType.builder().field("ts", DataTypes.TIMESTAMP(6)).build(); + LocalDateTime literal = LocalDateTime.of(2024, 1, 1, 0, 0, 0, 123_456_000); + + List predicates = convertEq(rowType, "ts", literal); + + LeafPredicate leaf = (LeafPredicate) predicates.get(0); + Assertions.assertEquals(Timestamp.fromLocalDateTime(literal), leaf.literals().get(0), + "an NTZ predicate literal must retain precision below one millisecond"); + } + @Test public void ltzNotPushed() { RowType rowType = RowType.builder() diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java index b7fe37f2ccc09e..edf264a577d528 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanPlanProviderTest.java @@ -23,6 +23,7 @@ import org.apache.doris.connector.spi.ConnectorType; import org.apache.doris.connector.spi.DorisConnectorException; import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; +import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef; import org.apache.doris.connector.spi.pushdown.ConnectorComparison; import org.apache.doris.connector.spi.pushdown.ConnectorExpression; @@ -896,6 +897,98 @@ public void resolveScanTableAppliesSnapshotPinViaCopy() { "the scan path must layer the handle's scanOptions via Table.copy(scanOptions)"); } + @Test + public void resolveScanTableKeepsCurrentSchemaForStatementSnapshot(@TempDir Path warehouse) + throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder() + .column("id", DataTypes.INT()) + .primaryKey("id") + .option("bucket", "1") + .build(), false); + FileStoreTable firstGeneration = (FileStoreTable) catalog.getTable(id); + BatchWriteBuilder writeBuilder = firstGeneration.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite()) { + write.write(GenericRow.of(1)); + List messages = write.prepareCommit(); + try (BatchTableCommit commit = writeBuilder.newCommit()) { + commit.commit(messages); + } + } + long dataSnapshotId = firstGeneration.latestSnapshot() + .orElseThrow(AssertionError::new).id(); + new SchemaManager(firstGeneration.fileIO(), firstGeneration.location()) + .commitChanges(SchemaChange.addColumn("added", DataTypes.INT())); + FileStoreTable latestGeneration = (FileStoreTable) catalog.getTable(id); + + RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); + PaimonTableHandle handle = new PaimonTableHandle( + "db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(latestGeneration); + PaimonTableHandle pinned = (PaimonTableHandle) new PaimonConnectorMetadata( + ops, PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext()) + .applySnapshot(null, handle, ConnectorMvccSnapshot.builder() + .snapshotId(dataSnapshotId) + .build()); + + Table scanTable = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), ops).resolveScanTable(pinned); + + Assertions.assertTrue(scanTable.rowType().getFieldNames().contains("added"), + "pinning data visibility must not roll a normal query back to the snapshot's old schema"); + Assertions.assertEquals(String.valueOf(dataSnapshotId), + scanTable.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key())); + } + } + + @Test + public void resolveScanTableKeepsCurrentSchemaForReaderOnlyOptions(@TempDir Path warehouse) + throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder() + .column("id", DataTypes.INT()) + .primaryKey("id") + .option("bucket", "1") + .build(), false); + FileStoreTable firstGeneration = (FileStoreTable) catalog.getTable(id); + BatchWriteBuilder writeBuilder = firstGeneration.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite()) { + write.write(GenericRow.of(1)); + List messages = write.prepareCommit(); + try (BatchTableCommit commit = writeBuilder.newCommit()) { + commit.commit(messages); + } + } + long dataSnapshotId = firstGeneration.latestSnapshot() + .orElseThrow(AssertionError::new).id(); + new SchemaManager(firstGeneration.fileIO(), firstGeneration.location()) + .commitChanges(SchemaChange.addColumn("added", DataTypes.INT())); + FileStoreTable latestGeneration = (FileStoreTable) catalog.getTable(id); + + PaimonTableHandle handle = new PaimonTableHandle( + "db", "t", Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(latestGeneration); + Map scanOptions = PaimonScanParams.markAsOptions( + PaimonScanParams.pinOptionsToSnapshot( + Collections.singletonMap("scan.plan-sort-partition", "true"), + dataSnapshotId)); + + Table scanTable = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingPaimonCatalogOps()) + .resolveScanTable(handle.withScanOptions(scanOptions)); + + Assertions.assertTrue(scanTable.rowType().getFieldNames().contains("added"), + "reader-only OPTIONS must retain the current bound schema while pinning data visibility"); + Assertions.assertEquals("true", scanTable.options().get("scan.plan-sort-partition")); + } + } + @Test public void resolveScanTableWithoutScanOptionsDoesNotCopy() { RecordingPaimonCatalogOps ops = new RecordingPaimonCatalogOps(); From 4ae6c48c3265cce9a18f85cb99ecac6ff86b1067 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 13 Sep 2026 11:46:24 +0800 Subject: [PATCH 02/11] test(paimon): cover schema-only snapshot reads --- ...imon_schema_only_snapshot_precision.groovy | 124 ++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy new file mode 100644 index 00000000000000..4b5627a76b7b05 --- /dev/null +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy @@ -0,0 +1,124 @@ +// 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_only_snapshot_precision", "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_only_snapshot_precision" + String dbName = "paimon_schema_only_snapshot_precision_db" + String tableName = "schema_only_timeline" + String branchName = "schema_only_branch" + + def latestSnapshotId = { + List> 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() + } + + 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, + old_name string, + event_time timestamp + ) using paimon + tblproperties ('file.format'='parquet'); + insert into paimon.${dbName}.${tableName} + values (1, 'base', timestamp '2024-01-01 00:00:00.123456'); + """ + String dataSnapshotId = latestSnapshotId() + spark_paimon_multi """ + call paimon.sys.create_tag( + table => '${dbName}.${tableName}', + tag => 'schema_base' + ); + call paimon.sys.create_branch( + '${dbName}.${tableName}', + '${branchName}', + 'schema_base' + ); + alter table paimon.${dbName}.`${tableName}\$branch_${branchName}` + rename column old_name to branch_name; + alter table paimon.${dbName}.${tableName} + rename column old_name to current_name; + """ + + // A schema-only rename must leave the data snapshot unchanged; otherwise these queries + // would not exercise the split between current schema binding and snapshot-pinned data. + assertEquals(dataSnapshotId, latestSnapshotId()) + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table ${tableName}""" + + assertEquals([[1, "base"]], sql(""" + select id, current_name from ${tableName} order by id + """)) + assertEquals([[1, "base"]], sql(""" + select id, current_name + from ${tableName}@options('scan.plan-sort-partition'='true') + order by id + """)) + assertEquals([[1, "base"]], sql(""" + select id, branch_name + from ${tableName}@branch(${branchName}) + order by id + """)) + + // Sub-millisecond precision must survive FE predicate conversion or Paimon's file + // statistics can reject the only matching file before either reader sees it. + sql """set force_jni_scanner=false""" + assertEquals([[1]], sql(""" + select id from ${tableName} + where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) + """)) + sql """set force_jni_scanner=true""" + assertEquals([[1]], sql(""" + select id from ${tableName} + where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) + """)) + } finally { + sql """set force_jni_scanner=false""" + sql """drop catalog if exists ${catalogName}""" + } +} From 4ffdc1fc1d5b4dbf94ad6dac896b9c24a9976bb3 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 13 Sep 2026 14:42:34 +0800 Subject: [PATCH 03/11] [fix](paimon) Retain bound schema and branch fences through planning ### What problem does this PR solve? Related PR: #67904 Problem Summary: Cached Paimon tables and system wrappers could use a different schema than statement binding, and branch reads could observe later commits. Carry the exact schema generation and positive or empty branch data fence through binding, native planning and JNI serialization. Use explicit NTZ literals in the timestamp precision regression. ### Release note Paimon statement reads retain consistent schema and data generations. ### Check List (For Author) - Test: Paimon package build (570 passed, 1 skipped); FE MVCC unit tests (68 passed); FE Checkstyle. External regression updated, not run locally. - Behavior changed: Yes, schema and data fences remain consistent through reads. - Does this need documentation: No. --- .../connector/paimon/PaimonCatalogOps.java | 15 +- .../paimon/PaimonConnectorMetadata.java | 78 +++--- .../connector/paimon/PaimonScanParams.java | 17 +- .../paimon/PaimonScanPlanProvider.java | 23 +- .../PaimonConnectorMetadataMvccTest.java | 5 +- .../paimon/PaimonStatementSchemaTest.java | 252 ++++++++++++++++++ .../mvcc/PluginDrivenMvccExternalTable.java | 20 +- .../PluginDrivenMvccExternalTableTest.java | 18 ++ ...imon_schema_only_snapshot_precision.groovy | 5 +- 9 files changed, 372 insertions(+), 61 deletions(-) create mode 100644 fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java index 0198d7ed5e5781..253cb85306c917 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonCatalogOps.java @@ -205,17 +205,28 @@ public long schemaId() { * offline without faking a concrete paimon {@code TableSchema}. */ final class PaimonSchemaSnapshot { + private final long schemaId; private final List fields; private final List partitionKeys; private final List primaryKeys; public PaimonSchemaSnapshot(List fields, List partitionKeys, List primaryKeys) { + this(-1L, fields, partitionKeys, primaryKeys); + } + + public PaimonSchemaSnapshot(long schemaId, List fields, List partitionKeys, + List primaryKeys) { + this.schemaId = schemaId; this.fields = fields; this.partitionKeys = partitionKeys; this.primaryKeys = primaryKeys; } + public long schemaId() { + return schemaId; + } + /** The schema's fields ({@code tableSchema.fields()}). */ public List fields() { return fields; @@ -368,7 +379,7 @@ public PaimonSchemaSnapshot schemaAt(Table table, long schemaId) { // (legacy PaimonExternalTable.initSchema(schemaId) reads the same accessors). TableSchema tableSchema = ((DataTable) table).schemaManager().schema(schemaId); return new PaimonSchemaSnapshot( - tableSchema.fields(), tableSchema.partitionKeys(), tableSchema.primaryKeys()); + tableSchema.id(), tableSchema.fields(), tableSchema.partitionKeys(), tableSchema.primaryKeys()); } @Override @@ -381,7 +392,7 @@ public Optional latestSchema(Table table) { return Optional.empty(); } return ((DataTable) table).schemaManager().latest() - .map(s -> new PaimonSchemaSnapshot(s.fields(), s.partitionKeys(), s.primaryKeys())); + .map(s -> new PaimonSchemaSnapshot(s.id(), s.fields(), s.partitionKeys(), s.primaryKeys())); } @Override diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java index 9d353024dc0451..fb3df5611eef6c 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java @@ -100,6 +100,9 @@ public class PaimonConnectorMetadata implements ConnectorMetadata { // existing direct-construction tests compile unchanged; production goes through the 5-arg ctor. private final PaimonLatestSnapshotCache latestSnapshotCache; + // Metadata is statement-scoped: aliases sharing a data fence must also share one schema generation. + private final Map statementSchemaIds = new java.util.concurrent.ConcurrentHashMap<>(); + // PERF-06: cross-query DERIVED partition-view cache A (generic ConnectorMetadataCache), injected by the // owning PaimonConnector; null = no cross-query derived layer (the convenience/test ctors used by ~15 // existing direct-construction tests pass null). Layered ABOVE the raw remote catalogOps.listPartitions @@ -365,8 +368,12 @@ private Table resolveSystemTableAt(ConnectorSession session, PaimonTableHandle pinned = snapshot == null ? paimonHandle : (PaimonTableHandle) applySnapshot(session, paimonHandle, snapshot); - Table table = resolveTable(pinned); Map scanOptions = pinned.getScanOptions(); + if (PaimonScanParams.preservesBoundSchema(scanOptions)) { + // Schema-derived wrappers must be built over the bound source before exposing their fields. + return new PaimonScanPlanProvider(catalogProperties, catalogOps, context).resolveScanTable(pinned); + } + Table table = resolveTable(pinned); if (scanOptions != null && !scanOptions.isEmpty() && PaimonScanParams.isOptionsPin(scanOptions)) { return PaimonScanParams.applyOptions(table, scanOptions); } @@ -577,7 +584,14 @@ public Optional beginQuerySnapshot( Identifier identifier = Identifier.create(paimonHandle.getDatabaseName(), paimonHandle.getTableName()); long id = latestSnapshotCache.getOrLoad(identifier, () -> catalogOps.latestSnapshotId(resolveTable(paimonHandle)).orElse(-1L)); - return Optional.of(ConnectorMvccSnapshot.builder().snapshotId(id).build()); + return Optional.of(ConnectorMvccSnapshot.builder().snapshotId(id) + .schemaId(statementSchemaId(paimonHandle, resolveTable(paimonHandle))).build()); + } + + private long statementSchemaId(PaimonTableHandle handle, Table table) { + return statementSchemaIds.computeIfAbsent(handle, + ignored -> catalogOps.latestSchema(table).map(PaimonCatalogOps.PaimonSchemaSnapshot::schemaId) + .orElse(-1L)); } @Override @@ -619,11 +633,10 @@ public boolean usesStatementSnapshotForOptions( * loads the branch as its OWN table (independent schema/snapshots, via the 3-arg branch * Identifier through {@link PaimonTableHandle#withBranch}) and pins its LATEST snapshot — * branches have NO in-branch time-travel (legacy {@code PaimonExternalTable} reads the - * branch's {@code latestSnapshot()} only). Its schema id remains {@code -1} so binding uses - * the branch's current schema, which can advance without a new data snapshot. The branch - * identity is carried to + * branch's {@code latestSnapshot()} only). The current schema id is captured independently + * because it can advance without a data snapshot. The branch identity is carried to * {@link #applySnapshot} via an internal sentinel ({@code CoreOptions.BRANCH} key, NOT a - * scan-copy option); no {@code scan.snapshot-id} is pinned (the branch reads its own latest). + * scan-copy option), together with the resolved data fence. * An empty branch also pins {@code snapshotId=-1}; both empty and non-empty branches bind * against the current branch schema. * @@ -735,13 +748,12 @@ public Optional resolveTimeTravel( Table branchTable = resolveTable(paimonHandle.withBranch(branchName)); long snapshotId = catalogOps.latestSnapshotId(branchTable).orElse(-1L); // A schema-only ALTER advances the branch schema without creating a data snapshot. - // Keep the data fence but bind the branch's current schema instead of that snapshot's old schema. - long schemaId = -1L; + // Bind the branch's exact current schema independently of its latest data snapshot. + long schemaId = statementSchemaId(paimonHandle.withBranch(branchName), branchTable); // Carry the branch identity to applySnapshot via an internal sentinel // (CoreOptions.BRANCH key). Branch is a handle-IDENTITY change, not a scan-copy // option: applySnapshot reads this sentinel and routes it to handle.withBranch (it is - // never threaded into Table.copy). No scan.snapshot-id is pinned (the branch table - // natively reads its own latest). + // never threaded into Table.copy). The data fence is applied after changing identity. return Optional.of(ConnectorMvccSnapshot.builder() .snapshotId(snapshotId) .schemaId(schemaId) @@ -787,9 +799,9 @@ public Optional resolveTimeTravel( : pinnedSnapshotId(table, resolved); // The statement fence pins data visibility, not schema time travel. Planning-only // aliases must retain the latest-schema projection used by the plain relation. - long schemaId = usesStatementFence || pinnedId < 0 - ? -1L - : catalogOps.snapshotSchemaId(table, pinnedId).orElse(-1L); + long schemaId = usesStatementFence + ? statementSchemaId(paimonHandle, table) + : pinnedId < 0 ? -1L : catalogOps.snapshotSchemaId(table, pinnedId).orElse(-1L); // resolved is never empty for a startup selector; for a selector-free @options (e.g. only // scan.manifest-parallelism) it is the user map verbatim, which applySnapshot still // threads -- those keys tune HOW the scan runs, not WHICH version it reads. @@ -933,38 +945,32 @@ public ConnectorTableHandle applySnapshot(ConnectorSession session, return paimonHandle; } PaimonScanParams.validateSystemTableOptions(snapshot.getProperties()); - return paimonHandle.withScanOptions(snapshot.getProperties()); + return paimonHandle.withScanOptions(PaimonScanParams.withBoundSchema( + snapshot.getProperties(), snapshot.getSchemaId())); } if (snapshot != null) { String branch = snapshot.getProperties().get(CoreOptions.BRANCH.key()); if (branch != null) { - // Branch time-travel is a handle-identity change (a different table load), not a scan - // option: route to withBranch (which clears the transient base Table so resolveTable - // reloads the branch). The branch reads its own latest, so no scan.snapshot-id is - // pinned. Detected BEFORE the generic properties path so the branch sentinel never - // becomes a scan-copy option. - return paimonHandle.withBranch(branch); + // Branch identity and data visibility are independent: switching tables must not + // discard the resolved positive or empty fence when a branch commits during planning. + return paimonHandle.withBranch(branch).withScanOptions(PaimonScanParams.withBoundSchema( + PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), snapshot.getSnapshotId()), + snapshot.getSchemaId())); } if (!snapshot.getProperties().isEmpty()) { // Explicit time-travel: the connector already resolved the exact scan options // (scan.snapshot-id OR scan.tag-name etc.) in resolveTimeTravel — thread them verbatim. - return paimonHandle.withScanOptions(snapshot.getProperties()); + return paimonHandle.withScanOptions(PaimonScanParams.withBoundSchema( + snapshot.getProperties(), snapshot.getSchemaId())); } } if (snapshot == null) { return paimonHandle; } - if (snapshot.getSnapshotId() < 0) { - // Empty latest is still a statement-scoped state. Carry only Doris' internal marker; - // Paimon's scan.snapshot-id=-1 would address a non-existent snapshot file. - return paimonHandle.withScanOptions( - PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), -1L)); - } - // The latest statement snapshot fences data only. Its schema was bound from the current table, - // which may be newer after a schema-only ALTER that created no data snapshot. - Map scanOptions = PaimonScanParams.pinOptionsToSnapshot( - Collections.emptyMap(), snapshot.getSnapshotId()); - return paimonHandle.withScanOptions(scanOptions); + // The latest statement fence owns both axes, even if no data snapshot exists yet. + return paimonHandle.withScanOptions(PaimonScanParams.withBoundSchema( + PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), snapshot.getSnapshotId()), + snapshot.getSchemaId())); } /** @@ -1355,11 +1361,9 @@ private List cachedPartitions(PaimonTableHandle paimonHa * and a new snapshot (data change, once the entry expires or REFRESH invalidates it) naturally mints a new key. * *

schemaId: pinned {@code -1} ("unversioned" for that axis, matching - * {@link ConnectorTableKey}'s documented convention). Unlike iceberg, paimon's {@link PaimonTableHandle} - * carries no schemaId — {@code applySnapshot} threads only {@code scanOptions} (an opaque properties map; - * see its javadoc) onto the handle, and {@link #beginQuerySnapshot} (the common latest-pin path) never - * resolves a schemaId either (its {@code ConnectorMvccSnapshot} keeps the builder default {@code -1}). This - * is not a loss for THIS view: {@link #collectPartitions} derives its output from {@code partitionKeys} + * {@link ConnectorTableKey}'s documented convention). Statement-fenced handles bypass this cache. + * Unversioned partition views do not need a schema generation: {@link #collectPartitions} derives + * its output from {@code partitionKeys} * (fixed at handle-build time) and paimon's raw partition specs, and paimon partition columns are immutable * post-creation, so schema evolution (e.g. ADD COLUMN) does not change what this method computes. */ diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java index 45b3ea75f9ac00..eda74d6a11392c 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java @@ -61,6 +61,7 @@ public final class PaimonScanParams { private static final String PINNED_FILE_CREATION_TIME = "doris.internal.paimon.file-creation-time-millis"; private static final String PINNED_EMPTY_SCAN = "doris.internal.paimon.empty-scan"; + private static final String BOUND_SCHEMA_ID = "doris.internal.paimon.bound-schema-id"; private static final String PRESERVE_BOUND_SCHEMA = "doris.internal.paimon.preserve-bound-schema"; /** @@ -196,8 +197,12 @@ public static FileStoreTable applyOptionsWithoutTimeTravel( .filter(key -> !tableOptions.containsKey(key)) .forEach(key -> isolatedOptions.put(key, null)); } - // The statement fence already selected the schema generation. Preserve that generation - // while carrying only the resolved read selector and execution options into this copy. + String schemaId = options.get(BOUND_SCHEMA_ID); + if (schemaId != null && table.schema().id() != Long.parseLong(schemaId)) { + // A cached table can predate binding, and latest can advance again after binding. + // Copy the exact schema while retaining catalog options, decorators and branch identity. + table = table.copy(table.schemaManager().schema(Long.parseLong(schemaId)).copy(table.options())); + } FileStoreTable effectiveTable = (FileStoreTable) PaimonReaderOptions.runtimeSafeTable( table.copyWithoutTimeTravel(isolatedOptions)); PaimonReaderOptions.validateEffectiveTable(effectiveTable); @@ -386,6 +391,14 @@ public static Map pinOptionsToSnapshot( return pinned; } + public static Map withBoundSchema(Map options, long schemaId) { + Map bound = new HashMap<>(options); + if (schemaId >= 0 && preservesBoundSchema(options)) { + bound.put(BOUND_SCHEMA_ID, Long.toString(schemaId)); + } + return bound; + } + public static boolean preservesBoundSchema(Map options) { return options != null && Boolean.parseBoolean(options.get(PRESERVE_BOUND_SCHEMA)); } diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java index 75bb1ce7797462..8128f30d04d5cd 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java @@ -345,7 +345,9 @@ Table resolveScanTable(PaimonTableHandle paimonHandle) { Table table = resolveTable(paimonHandle); Map scanOptions = paimonHandle.getScanOptions(); Table finalTable = table; - if (scanOptions != null && !scanOptions.isEmpty()) { + if (scanOptions != null && !scanOptions.isEmpty() + && !(paimonHandle.isSystemTable() && PaimonScanParams.preservesBoundSchema(scanOptions))) { + // Statement-fenced system wrappers are rebuilt below from their schema-bound source. if (table instanceof FileStoreTable && PaimonScanParams.preservesBoundSchema(scanOptions)) { // A statement fence owns data visibility, not schema time travel. Reusing Table.copy @@ -1229,7 +1231,8 @@ Table tableForBackend(PaimonTableHandle handle, Table scanTable) { if (optionsAppliedToSource) { // Fallback snapshot translation consults each branch catalog, so options must be // resolved while both loaders are still present and only then made BE-safe. - preparedDataTable = (FileStoreTable) PaimonScanParams.applyOptions( + // Rebuild the backend wrapper with the same schema provenance used during binding and planning. + preparedDataTable = (FileStoreTable) PaimonReaderOptions.runtimeSafeSystemSource( preparedDataTable, scanOptions); } FileStoreTable baseForBackend = dropCatalogLoader(preparedDataTable); @@ -1417,9 +1420,11 @@ static FileStoreTable dropCatalogLoader(FileStoreTable dataTable) { } private static FileStoreTable rebuildWithoutCatalogLoader(FileStoreTable branch) { + // The factory evaluates scan.snapshot-id and can rewind a schema-only ALTER. Restore + // the already-bound schema after removing the loader so JNI reads the planned field ids. return FileStoreTableFactory.createWithoutFallbackBranch( branch.fileIO(), branch.location(), branch.schema(), new Options(), - CatalogEnvironment.empty()); + CatalogEnvironment.empty()).copy(branch.schema()); } /** @@ -1446,14 +1451,10 @@ private Table resolveSchemaDictTable(Table table, PaimonTableHandle handle) { } if (table instanceof ReadOptimizedTable) { FileStoreTable pinnedSource = handle.getSysBaseTable(); - if (pinnedSource != null) { - // $ro reads the field ids of its embedded source; a catalog reload here can observe - // schema generation B while the wrapper still plans generation A's files. Relation scan - // options must also select this source, or historical splits get the latest dictionary. - return reapplyScanParams( - pinnedSource, pinnedSource, false, handle.getScanOptions()); - } - return reloadBaseTable(handle); + // A reloaded source still needs the bound schema and data selector; otherwise the + // native dictionary can disagree with the wrapper after either cache or handle reload. + return PaimonReaderOptions.runtimeSafeSystemSource( + pinnedSource == null ? reloadBaseTable(handle) : pinnedSource, handle.getScanOptions()); } return null; } diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java index ac9cd639e417c6..b20ee4cd36fe2e 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonConnectorMetadataMvccTest.java @@ -1082,8 +1082,11 @@ public void applySnapshotWithBranchSentinelRoutesToWithBranch() { // assertions red. Assertions.assertEquals("b1", pinned.getBranchName(), "the branch sentinel must route to withBranch (handle identity), not a scan option"); - Assertions.assertTrue(pinned.getScanOptions().isEmpty(), + Assertions.assertFalse(pinned.getScanOptions().containsKey(CoreOptions.BRANCH.key()), "a branch pin must NOT thread the sentinel as a scan-copy option"); + Assertions.assertEquals("7", pinned.getScanOptions().get(CoreOptions.SCAN_SNAPSHOT_ID.key()), + "a branch pin must retain its resolved data fence after switching identity"); + Assertions.assertTrue(PaimonScanParams.preservesBoundSchema(pinned.getScanOptions())); Assertions.assertNull(pinned.getPaimonTable(), "withBranch must clear the transient base Table so the branch reloads"); } diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java new file mode 100644 index 00000000000000..20ab7ff109cc09 --- /dev/null +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java @@ -0,0 +1,252 @@ +// 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. + +package org.apache.doris.connector.paimon; + +import org.apache.doris.connector.spi.ConnectorColumn; +import org.apache.doris.connector.spi.ConnectorSession; +import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; +import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; +import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec; +import org.apache.doris.connector.spi.scan.ConnectorScanRequest; +import org.apache.doris.thrift.TFileScanRangeParams; + +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.FileSystemCatalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.InstantiationUtil; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +public class PaimonStatementSchemaTest { + @Test + public void warmTableKeepsExactStatementSchema(@TempDir Path warehouse) throws Exception { + checkSchemaMutation(warehouse, false, false); + } + + @Test + public void warmOptionsTableKeepsExactStatementSchema(@TempDir Path warehouse) throws Exception { + checkSchemaMutation(warehouse, true, false); + } + + @Test + public void systemOptionsKeepsExactStatementSchema(@TempDir Path warehouse) throws Exception { + checkSchemaMutation(warehouse, true, true); + } + + private void checkSchemaMutation(Path warehouse, boolean options, boolean system) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .column("old_name", DataTypes.INT()).option("file.format", "parquet").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + append(warm, GenericRow.of(1, 10)); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("old_name", "bound_name")), false); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata(ops, + PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext()); + PaimonScanPlanProvider provider = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), ops); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(warm); + ConnectorMvccSnapshot snapshot = metadata.beginQuerySnapshot(null, handle).get(); + if (options) { + snapshot = metadata.resolveTimeTravel(null, handle, ConnectorTimeTravelSpec.options( + Collections.singletonMap("scan.plan-sort-partition", "true"), + snapshot.getSnapshotId())).get(); + } + if (system) { + handle = (PaimonTableHandle) metadata.getSysTableHandle(null, handle, "ro").get(); + } + Assertions.assertTrue(metadata.getTableSchema(null, handle, snapshot).getColumns().stream() + .map(ConnectorColumn::getName).collect(Collectors.toList()).contains("bound_name")); + // The second ALTER must not change a statement that already bound the first rename. + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("bound_name", "later_name")), false); + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + Table scan = provider.resolveScanTable(pinned); + Assertions.assertEquals(Arrays.asList("id", "bound_name"), scan.rowType().getFieldNames()); + List columns = Arrays.asList(new PaimonColumnHandle("id", 0), + new PaimonColumnHandle("bound_name", 1)); + Assertions.assertFalse(provider.planScan(session(), + ConnectorScanRequest.builder(pinned, columns).build()).isEmpty()); + Map properties = provider.getScanNodeProperties(session(), pinned, columns, + Optional.empty()); + TFileScanRangeParams params = new TFileScanRangeParams(); + PaimonScanPlanProvider.applySchemaEvolutionParam(params, properties.get("paimon.schema_evolution")); + Assertions.assertEquals("bound_name", params.getHistorySchemaInfo().get(0) + .getRootField().getFields().get(1).getFieldPtr().getName()); + Table backend = InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(provider.tableForBackend(pinned, scan)), + getClass().getClassLoader()); + Assertions.assertEquals(scan.rowType(), backend.rowType()); + Assertions.assertEquals(Collections.singletonList(1), readIds(backend)); + } + } + + @Test + public void branchCommitAfterResolutionIsInvisible(@TempDir Path warehouse) throws Exception { + checkBranchMutation(warehouse, false); + } + + @Test + public void emptyBranchRemainsEmptyAfterFirstCommit(@TempDir Path warehouse) throws Exception { + checkBranchMutation(warehouse, true); + } + + private void checkBranchMutation(Path warehouse, boolean empty) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .option("file.format", "parquet").build(), false); + FileStoreTable base = (FileStoreTable) catalog.getTable(id); + base.branchManager().createBranch("dev"); + Identifier branchId = new Identifier("db", "t", "dev"); + FileStoreTable branch = (FileStoreTable) catalog.getTable(branchId); + if (!empty) { + append(branch, GenericRow.of(1)); + } + catalog.alterTable(branchId, Collections.singletonList( + SchemaChange.addColumn("bound_name", DataTypes.INT())), false); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata(ops, + PaimonCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext()); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(base); + ConnectorMvccSnapshot snapshot = metadata.resolveTimeTravel(null, handle, + ConnectorTimeTravelSpec.branch("dev")).get(); + catalog.alterTable(branchId, Collections.singletonList( + SchemaChange.renameColumn("bound_name", "later_name")), false); + append((FileStoreTable) catalog.getTable(branchId), GenericRow.of(2, 20)); + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + PaimonScanPlanProvider provider = new PaimonScanPlanProvider( + PaimonCatalogProperties.of(Collections.emptyMap()), ops); + Assertions.assertEquals(Arrays.asList("id", "bound_name"), + provider.resolveScanTable(pinned).rowType().getFieldNames()); + if (empty) { + Assertions.assertTrue(provider.planScan(session(), ConnectorScanRequest.builder(pinned, + Collections.singletonList(new PaimonColumnHandle("id", 0))).build()).isEmpty()); + } else { + Table scan = provider.resolveScanTable(pinned); + Assertions.assertEquals(Collections.singletonList(1), readIds(scan)); + Table backend = InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(provider.tableForBackend(pinned, scan)), + getClass().getClassLoader()); + Assertions.assertEquals(Collections.singletonList(1), readIds(backend)); + } + } + } + + private static void append(FileStoreTable table, GenericRow row) throws Exception { + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = builder.newWrite(); BatchTableCommit commit = builder.newCommit()) { + write.write(row); + commit.commit(write.prepareCommit()); + } + } + + private static List readIds(Table table) throws Exception { + List ids = new ArrayList<>(); + for (Split split : table.newReadBuilder().newScan().plan().splits()) { + try (RecordReader reader = table.newReadBuilder().newRead().createReader(split)) { + reader.forEachRemaining(row -> ids.add(row.getInt(0))); + } + } + Collections.sort(ids); + return ids; + } + + private static ConnectorSession session() { + return new ConnectorSession() { + @Override + public String getQueryId() { + return "q"; + } + + @Override + public String getUser() { + return "u"; + } + + @Override + public String getTimeZone() { + return "UTC"; + } + + @Override + public String getLocale() { + return "en_US"; + } + + @Override + public long getCatalogId() { + return 0; + } + + @Override + public String getCatalogName() { + return "c"; + } + + @Override + public T getProperty(String name, Class type) { + return null; + } + + @Override + public Map getCatalogProperties() { + return Collections.emptyMap(); + } + + @Override + public Map getSessionProperties() { + return Collections.emptyMap(); + } + }; + } + +} diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java index d4223f9f7c60d2..c18bc709b35379 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java @@ -172,6 +172,14 @@ private PluginDrivenMvccSnapshot materializeLatest( // legacy listPartitions/LIST/timestamp path below (byte-unchanged; the no-op applySnapshot for the // latest pin is side-effect-free for both paimon and iceberg). ConnectorTableHandle pinnedHandle = metadata.applySnapshot(session, handle, connectorSnapshot); + PluginDrivenSchemaCacheValue pinnedSchema = null; + if (connectorSnapshot.getSchemaId() >= 0) { + // Latest data and schema can advance independently. Keep the connector's exact schema + // on the statement pin so analysis cannot fall back to a different cached generation. + ConnectorTableSchema atSchema = metadata.getTableSchema(session, pinnedHandle, connectorSnapshot); + pinnedSchema = toSchemaCacheValue(metadata, session, + db != null ? db.getRemoteName() : "", getRemoteName(), atSchema); + } // Partition counts feed COUNT(*) pushdown and SQL block rules, so even the initial latest // materialization must enumerate the same pinned generation that the data scan reads. ConnectorTableHandle partitionHandle = pinnedHandle; @@ -182,7 +190,7 @@ private PluginDrivenMvccSnapshot materializeLatest( if (connectorSnapshot.getSnapshotId() < 0) { // A negative query-begin pin is the connector's resolved-empty generation. Falling through to // a live LIST read would mix partitions committed after the data scan's empty boundary. - return buildFromRangeView(connectorSnapshot, view); + return buildFromRangeView(connectorSnapshot, view, pinnedSchema); } // A non-RANGE (UNPARTITIONED) view is the connector's "not RANGE / not MTMV-range-eligible" verdict // (iceberg's range view only covers single time-transform specs). If the table nonetheless declares @@ -199,16 +207,16 @@ private PluginDrivenMvccSnapshot materializeLatest( Map listLastModifiedMillis = Maps.newHashMap(); listLatestPartitions(metadata, session, partitionHandle, listItems, listLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, listItems, listLastModifiedMillis, - null, PartitionType.UNPARTITIONED, false, 0L); + pinnedSchema, PartitionType.UNPARTITIONED, false, 0L); } - return buildFromRangeView(connectorSnapshot, view); + return buildFromRangeView(connectorSnapshot, view, pinnedSchema); } Map nameToPartitionItem = Maps.newHashMap(); Map nameToLastModifiedMillis = Maps.newHashMap(); listLatestPartitions(metadata, session, partitionHandle, nameToPartitionItem, nameToLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, nameToPartitionItem, - nameToLastModifiedMillis); + nameToLastModifiedMillis, pinnedSchema); } /** @@ -221,7 +229,7 @@ private PluginDrivenMvccSnapshot materializeLatest( * per-partition log-and-skip). */ private PluginDrivenMvccSnapshot buildFromRangeView(ConnectorMvccSnapshot connectorSnapshot, - ConnectorMvccPartitionView view) { + ConnectorMvccPartitionView view, PluginDrivenSchemaCacheValue pinnedSchema) { PartitionType partitionType = view.getStyle() == ConnectorMvccPartitionView.Style.RANGE ? PartitionType.RANGE : PartitionType.UNPARTITIONED; boolean snapshotIdFreshness = @@ -244,7 +252,7 @@ private PluginDrivenMvccSnapshot buildFromRangeView(ConnectorMvccSnapshot connec } } return new PluginDrivenMvccSnapshot(connectorSnapshot, nameToPartitionItem, nameToFreshnessValue, - null, partitionType, snapshotIdFreshness, view.getNewestUpdateMonotonicMarker(), + pinnedSchema, partitionType, snapshotIdFreshness, view.getNewestUpdateMonotonicMarker(), view.getNewestUpdateWallClockMillis()); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java index ba82c074d541a6..cbce74013065bf 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java @@ -567,6 +567,24 @@ public void testLoadSnapshotEmptyMaterializes() { "the latest pin must carry the materialized partition view"); } + @Test + public void testLatestSnapshotCarriesConnectorBoundSchema() { + Fixture f = Fixture.timeTravel(); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder() + .snapshotId(PINNED_SNAPSHOT_ID).schemaId(Fixture.TT_SCHEMA_ID).build(); + Mockito.when(f.metadata.beginQuerySnapshot(f.session, f.handle)) + .thenReturn(Optional.of(snapshot)); + Mockito.when(f.metadata.applySnapshot(f.session, f.handle, snapshot)) + .thenReturn(f.pinnedHandle); + + PluginDrivenMvccSnapshot pin = (PluginDrivenMvccSnapshot) + f.table.loadSnapshot(Optional.empty(), Optional.empty()); + + Assertions.assertNotNull(pin.getPinnedSchema()); + Assertions.assertEquals("v1", pin.getPinnedSchema().getSchema().get(0).getName(), + "a connector's latest schema generation must survive the data fence materialization"); + } + @Test public void testInitialLatestPartitionAccountingUsesPinnedHandle() { Fixture f = Fixture.partitioned(); diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy index 4b5627a76b7b05..388c05b4dbdc01 100644 --- a/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy @@ -54,17 +54,18 @@ suite("test_paimon_schema_only_snapshot_precision", "p0,external,paimon") { """ try { + // Explicit NTZ makes this exercise predicate pushdown instead of LTZ residual filtering. spark_paimon_multi """ create database if not exists paimon.${dbName}; drop table if exists paimon.${dbName}.${tableName}; create table paimon.${dbName}.${tableName} ( id int, old_name string, - event_time timestamp + event_time timestamp_ntz ) using paimon tblproperties ('file.format'='parquet'); insert into paimon.${dbName}.${tableName} - values (1, 'base', timestamp '2024-01-01 00:00:00.123456'); + values (1, 'base', timestamp_ntz '2024-01-01 00:00:00.123456'); """ String dataSnapshotId = latestSnapshotId() spark_paimon_multi """ From 75ceac12b856aca84f572e2337b52870837950b7 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 13 Sep 2026 16:59:12 +0800 Subject: [PATCH 04/11] [fix](fe) Preserve schema provenance and safe timestamp predicates ### What problem does this PR solve? Related PR: #67904 Problem Summary: Restoring a bound Paimon schema could retain field-referencing options from a different generation or overwrite a fallback branch's schema and identity. Exact nanosecond timestamp pushdown could reject values that compare equal after Doris truncation. Iceberg cached schema pins could expose live partition names and specs. Preserve each schema's options and fallback provenance, retain privilege checks, keep high-precision comparisons residual, and pin Iceberg partition specs with their schema. ### Release note External scans retain consistent schema and partition metadata. Paimon nanosecond timestamps are compared at Doris precision without unsafe source pruning. ### Check List (For Author) - Test: 1,963 connector unit tests passed, 6 skipped; one existing Iceberg test failure reproduced before the changes and excluded. FE Checkstyle and plugin packages passed. Generated and verified all five regression baselines through the regression harness. - Behavior changed: Yes, preserve schema provenance and avoid false timestamp pruning. - Does this need documentation: No. --- .../iceberg/IcebergConnectorMetadata.java | 45 ++++++---- .../iceberg/IcebergLatestSnapshotCache.java | 11 ++- .../IcebergConnectorMetadataMvccTest.java | 22 +++++ .../paimon/PaimonPredicateConverter.java | 6 ++ .../connector/paimon/PaimonScanParams.java | 32 ++++++- .../paimon/PaimonTableDecorators.java | 24 +++++ .../paimon/PaimonPredicateConverterTest.java | 14 +++ .../paimon/PaimonStatementSchemaTest.java | 90 +++++++++++++++++++ ..._paimon_schema_only_snapshot_precision.out | 16 ++++ ...imon_schema_only_snapshot_precision.groovy | 20 ++--- 10 files changed, 246 insertions(+), 34 deletions(-) create mode 100644 regression-test/data/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.out diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java index 1caeae8fce7f38..b09baa5934ec36 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java @@ -113,6 +113,7 @@ public class IcebergConnectorMetadata implements ConnectorMetadata { // Internal sentinel property carrying a tag/branch ref name from resolveTimeTravel to applySnapshot (the // typed ConnectorMvccSnapshot has snapshotId/schemaId carriers but no ref field). NOT a BE scan option. static final String REF_PROPERTY = "iceberg.scan.ref"; + private static final String PARTITION_SPEC_ID_PROPERTY = "iceberg.partition.spec.id"; private static final String EMPTY_PARTITION_STYLE_PROPERTY = "iceberg.empty.partition.style"; // Iceberg v3 row-lineage hidden columns. Local literal copies of the Doris-side constants — the @@ -499,17 +500,24 @@ public ConnectorTableSchema getTableSchema( schema = table.schema(); } } - return buildTableSchema(iceHandle.getTableName(), table, schema, true); + String specId = snapshot.getProperties().get(PARTITION_SPEC_ID_PROPERTY); + PartitionSpec spec = specId == null ? table.spec() : table.specs().get(Integer.parseInt(specId)); + return buildTableSchema(iceHandle.getTableName(), table, schema, spec, true); } /** * Assembles the {@link ConnectorTableSchema} for {@code table} from {@code schema} (the latest schema, or a - * historical schema for a time-travel read). The {@code iceberg.format-version} / {@code location} / - * {@code iceberg.partition-spec} properties are table-level (not schema-versioned). Factored out so the - * latest and at-snapshot paths share ONE assembly. + * historical schema for a time-travel read). Pinned reads also supply the partition spec from their + * metadata generation; table properties and location still come from the loaded table. Factored out + * so the latest and at-snapshot paths share one assembly. */ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Schema schema, boolean appendDataFileMetadataColumns) { + return buildTableSchema(tableName, table, schema, table.spec(), appendDataFileMetadataColumns); + } + + private ConnectorTableSchema buildTableSchema(String tableName, Table table, Schema schema, + PartitionSpec spec, boolean appendDataFileMetadataColumns) { List columns = parseSchema(schema); // Iceberg file metadata columns are always available for data tables, but are hidden from @@ -552,7 +560,7 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch if (table.location() != null) { tableProps.put(ConnectorTableSchema.SHOW_LOCATION_KEY, table.location()); } - String partitionClause = buildShowPartitionClause(table); + String partitionClause = buildShowPartitionClause(schema, spec); if (!partitionClause.isEmpty()) { tableProps.put(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY, partitionClause); } @@ -560,14 +568,10 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch if (!sortClause.isEmpty()) { tableProps.put(ConnectorTableSchema.SHOW_SORT_CLAUSE_KEY, sortClause); } - if (!table.spec().isUnpartitioned()) { - // Generic FE partition-column contract: post-cutover, PluginDrivenExternalTable derives the - // table's partition columns SOLELY from a "partition_columns" CSV property (toSchemaCacheValue), - // the same key MaxCompute/paimon emit. Mirror legacy IcebergUtils.loadTableSchemaCacheValue: - // walk the CURRENT spec, resolve each partition field's SOURCE column name (NO identity filter), - // case-preserved to match parseSchema's case-preserved column names (#65094 read-path - // alignment; fromRemoteColumnName is identity for iceberg, so the FE consumer looks the names up - // case-sensitively). + if (!spec.isUnpartitioned()) { + // A cached latest pin can outlive schema-only renames and spec evolution while REST + // credentials require a fresh Table. Resolve partition source IDs in the pinned schema + // and spec so FE never receives historical columns paired with live partition names. // DEDUPED per source column (LinkedHashSet, first-occurrence order): this CSV becomes a SET of // partition COLUMNS on the FE side, not a list of spec FIELDS. fe-core maps each name to one scan // Slot (PruneFileScanPartition) and OneListPartitionEvaluator collects Slot -> literal into an @@ -577,8 +581,8 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch // deduped column sequence, so the two stay index-aligned (the arity checkState in // PluginDrivenMvccExternalTable.toListPartitionItem). Set partitionColumns = new LinkedHashSet<>(); - for (PartitionField field : table.spec().fields()) { - Types.NestedField source = table.schema().findField(field.sourceId()); + for (PartitionField field : spec.fields()) { + Types.NestedField source = schema.findField(field.sourceId()); if (source != null) { partitionColumns.add(source.name()); } @@ -601,14 +605,13 @@ private ConnectorTableSchema buildTableSchema(String tableName, Table table, Sch * {@code bucket[N]}/{@code truncate[W]}/{@code year}/{@code month}/{@code day}/{@code hour} -> the * matching Doris partition function. Returns "" for an unpartitioned table or no renderable field. */ - private String buildShowPartitionClause(Table table) { - PartitionSpec spec = table.spec(); + private String buildShowPartitionClause(Schema schema, PartitionSpec spec) { if (spec == null || spec.isUnpartitioned()) { return ""; } List fields = new ArrayList<>(); for (PartitionField field : spec.fields()) { - String colName = table.schema().findColumnName(field.sourceId()); + String colName = schema.findColumnName(field.sourceId()); if (colName == null) { continue; } @@ -2131,6 +2134,9 @@ public Optional beginQuerySnapshot( : loadLatestSnapshotPin(session, iceHandle); ConnectorMvccSnapshot.Builder snapshot = ConnectorMvccSnapshot.builder() .snapshotId(pin.snapshotId).schemaId(pin.schemaId); + if (pin.specId >= 0) { + snapshot.property(PARTITION_SPEC_ID_PROPERTY, Integer.toString(pin.specId)); + } if (pin.snapshotId < 0) { snapshot.property(EMPTY_PARTITION_STYLE_PROPERTY, pin.emptyPartitionStyle.name()); } @@ -2177,7 +2183,8 @@ private IcebergLatestSnapshotCache.CachedSnapshot latestSnapshotPin(Table table) ? ConnectorMvccPartitionView.Style.RANGE : ConnectorMvccPartitionView.Style.UNPARTITIONED; return new IcebergLatestSnapshotCache.CachedSnapshot( - current == null ? -1L : current.snapshotId(), table.schema().schemaId(), emptyPartitionStyle); + current == null ? -1L : current.snapshotId(), table.schema().schemaId(), + table.spec().specId(), emptyPartitionStyle); } /** diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java index a20ab5fd2ecf4d..c9550219fea517 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergLatestSnapshotCache.java @@ -37,7 +37,7 @@ * query-begin pin ({@link IcebergConnectorMetadata#beginQuerySnapshot}) reads the SAME snapshot until the * entry expires or is invalidated by {@code REFRESH TABLE}/{@code REFRESH CATALOG}. * - *

Value carries snapshotId, schemaId and the resolved-empty partition style. + *

Value carries snapshotId, schemaId, specId and the resolved-empty partition style. * {@code beginQuerySnapshot} pins the snapshot id and the LATEST schema id * ({@code table.schema().schemaId()} — not {@code currentSnapshot().schemaId()}, mirroring legacy * {@code IcebergUtils.getLatestIcebergSnapshot}). A schema-only {@code ALTER} bumps the latest schema id @@ -55,10 +55,11 @@ */ final class IcebergLatestSnapshotCache { - /** Immutable atomic pin for the latest snapshot/schema and its resolved-empty partition style. */ + /** Immutable atomic pin for the latest snapshot/schema/spec and its resolved-empty partition style. */ static final class CachedSnapshot { final long snapshotId; final long schemaId; + final int specId; final ConnectorMvccPartitionView.Style emptyPartitionStyle; CachedSnapshot(long snapshotId, long schemaId) { @@ -67,8 +68,14 @@ static final class CachedSnapshot { CachedSnapshot(long snapshotId, long schemaId, ConnectorMvccPartitionView.Style emptyPartitionStyle) { + this(snapshotId, schemaId, -1, emptyPartitionStyle); + } + + CachedSnapshot(long snapshotId, long schemaId, int specId, + ConnectorMvccPartitionView.Style emptyPartitionStyle) { this.snapshotId = snapshotId; this.schemaId = schemaId; + this.specId = specId; this.emptyPartitionStyle = emptyPartitionStyle; } } diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java index b0dd3baaecaedf..3f1e80939b3875 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java @@ -171,6 +171,28 @@ public void beginQuerySnapshotEnabledCachePinsStableAndLoadsOnce() { Assertions.assertEquals(1, loads, "an enabled cache must load the table at most once within the TTL"); } + @Test + public void latestCacheHitKeepsPartitionSchemaAfterLiveRename() { + Table table = dayPartitionedTable(); + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = table; + IcebergLatestSnapshotCache cache = new IcebergLatestSnapshotCache(100, 1000); + IcebergConnectorMetadata firstQuery = new IcebergConnectorMetadata(ops, + IcebergCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot first = firstQuery.beginQuerySnapshot(null, handle()).get(); + table.updateSchema().renameColumn("ts", "renamed_ts").commit(); + table.updateSpec().addField("id").commit(); + IcebergConnectorMetadata nextQuery = new IcebergConnectorMetadata(ops, + IcebergCatalogProperties.of(Collections.emptyMap()), new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot cached = nextQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(first.getSchemaId(), cached.getSchemaId()); + ConnectorTableSchema schema = nextQuery.getTableSchema(null, handle(), cached); + Assertions.assertTrue(columnNames(schema).contains("ts")); + Assertions.assertEquals("ts", schema.getProperties().get(ConnectorTableSchema.PARTITION_COLUMNS_KEY)); + Assertions.assertEquals("PARTITION BY LIST (DAY(`ts`)) ()", + schema.getProperties().get(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY)); + } + @Test public void beginQuerySnapshotDisabledCacheLoadsEveryCall() { Fixture f = fixture(); diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java index c03c9675645b81..b959a88a12a027 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java @@ -38,6 +38,7 @@ import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.RowType; +import org.apache.paimon.types.TimestampType; import java.math.BigDecimal; import java.time.LocalDate; @@ -328,6 +329,11 @@ private Object convertLiteralValue(ConnectorLiteral literal, DataType paimonType } return null; case TIMESTAMP_WITHOUT_TIME_ZONE: + // Doris truncates source nanoseconds to DATETIMEV2(6). Exact source comparisons + // would reject rows that become equal after that truncation, so keep them residual. + if (((TimestampType) paimonType).getPrecision() > 6) { + return null; + } // Preserve the complete wall-clock value: narrowing it to epoch milliseconds can // make Paimon prune every file matching a non-millisecond-aligned predicate. if (value instanceof LocalDateTime) { diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java index eda74d6a11392c..f3f6276139f7c0 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java @@ -24,6 +24,9 @@ import org.apache.paimon.Snapshot; import org.apache.paimon.options.ConfigOption; import org.apache.paimon.options.FallbackKey; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; import org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner; @@ -33,6 +36,7 @@ import java.util.HashMap; import java.util.Locale; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -199,9 +203,7 @@ public static FileStoreTable applyOptionsWithoutTimeTravel( } String schemaId = options.get(BOUND_SCHEMA_ID); if (schemaId != null && table.schema().id() != Long.parseLong(schemaId)) { - // A cached table can predate binding, and latest can advance again after binding. - // Copy the exact schema while retaining catalog options, decorators and branch identity. - table = table.copy(table.schemaManager().schema(Long.parseLong(schemaId)).copy(table.options())); + table = restoreBoundSchema(table, Long.parseLong(schemaId)); } FileStoreTable effectiveTable = (FileStoreTable) PaimonReaderOptions.runtimeSafeTable( table.copyWithoutTimeTravel(isolatedOptions)); @@ -209,6 +211,30 @@ public static FileStoreTable applyOptionsWithoutTimeTravel( return effectiveTable; } + private static FileStoreTable restoreBoundSchema(FileStoreTable table, long schemaId) { + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + // Schema IDs are branch-local. Broadcasting the main schema through copy(TableSchema) + // overwrites the fallback's provenance and can even reset its branch to main. + return new FallbackReadFileStoreTable(restoreBoundSchema(pair.wrapped(), schemaId), pair.fallback()); + } + if (table instanceof DelegatedFileStoreTable) { + FileStoreTable wrapped = ((DelegatedFileStoreTable) table).wrapped(); + return PaimonTableDecorators.replaceWrapped(table, restoreBoundSchema(wrapped, schemaId)); + } + TableSchema bound = table.schemaManager().schema(schemaId); + Map persisted = table.schemaManager().schema(table.schema().id()).options(); + Map merged = new HashMap<>(bound.options()); + // Field-referencing options evolve with the schema (e.g. bucket-key and sequence.field + // on rename). Only replay the catalog/runtime delta, never another generation's options. + table.options().forEach((key, value) -> { + if (!Objects.equals(persisted.get(key), value)) { + merged.put(key, value); + } + }); + return table.copy(bound.copy(merged)); + } + /** * Paimon's inherited read-state family: startup mode, startup position and incremental range, plus * every fallback key. Shared with {@link PaimonIncrementalScanParams#applyResetsIfIncremental}, which diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java index fbebd7b1989a90..b762ff35e4fc03 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTableDecorators.java @@ -17,10 +17,15 @@ package org.apache.doris.connector.paimon; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.privilege.PrivilegeChecker; +import org.apache.paimon.privilege.PrivilegedFileStoreTable; import org.apache.paimon.table.DelegatedFileStoreTable; import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; +import java.lang.reflect.Field; + /** * The one place that knows how Paimon stacks {@link DelegatedFileStoreTable} decorators on a loaded * table, and how far they may be peeled off. @@ -30,6 +35,25 @@ final class PaimonTableDecorators { private PaimonTableDecorators() { } + static FileStoreTable replaceWrapped(FileStoreTable original, FileStoreTable replacement) { + if (!(original instanceof PrivilegedFileStoreTable)) { + throw new IllegalArgumentException("Unsupported Paimon planning table delegate: " + + original.getClass().getName()); + } + try { + // Paimon exposes no delegate-replacement API. Retain the original checker and identity + // so rebuilding a fallback pair does not discard authorization on subsequent reads. + Field checker = PrivilegedFileStoreTable.class.getDeclaredField("privilegeChecker"); + Field identifier = PrivilegedFileStoreTable.class.getDeclaredField("identifier"); + checker.setAccessible(true); + identifier.setAccessible(true); + return PrivilegedFileStoreTable.wrap(replacement, + (PrivilegeChecker) checker.get(original), (Identifier) identifier.get(original)); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException("Failed to preserve Paimon privilege delegate", e); + } + } + /** * Peel the decorators Paimon may have stacked on top of the table, down to the fallback-branch * pair - the one layer that must stay on top, because that is what dispatches a read to the diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java index 4dde877319eac0..1b0763cfd2d311 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonPredicateConverterTest.java @@ -100,6 +100,20 @@ public void ntzPushPreservesMicroseconds() { "an NTZ predicate literal must retain precision below one millisecond"); } + @Test + public void nanosecondTimestampComparisonsRemainResidual() { + RowType rowType = RowType.builder().field("ts", DataTypes.TIMESTAMP(9)).build(); + LocalDateTime stored = LocalDateTime.of(2024, 1, 1, 0, 0, 0, 123_456_789); + LocalDateTime visible = stored.withNano(123_456_000); + Assertions.assertNotEquals(Timestamp.fromLocalDateTime(stored), Timestamp.fromLocalDateTime(visible)); + for (ConnectorComparison.Operator operator : ConnectorComparison.Operator.values()) { + ConnectorComparison comparison = new ConnectorComparison(operator, + new ConnectorColumnRef("ts", ANY), new ConnectorLiteral(ANY, visible)); + Assertions.assertTrue(new PaimonPredicateConverter(rowType).convert(comparison).isEmpty(), + "sub-microsecond source values must be compared after Doris truncation: " + operator); + } + } + @Test public void ltzNotPushed() { RowType rowType = RowType.builder() diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java index 20ab7ff109cc09..e35a0092874936 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java @@ -31,9 +31,13 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.privilege.PrivilegeChecker; +import org.apache.paimon.privilege.PrivilegedFileStoreTable; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; import org.apache.paimon.table.sink.BatchTableCommit; @@ -46,6 +50,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.lang.reflect.Proxy; import java.nio.file.Path; import java.util.ArrayList; import java.util.Arrays; @@ -53,6 +58,8 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; public class PaimonStatementSchemaTest { @@ -71,6 +78,89 @@ public void systemOptionsKeepsExactStatementSchema(@TempDir Path warehouse) thro checkSchemaMutation(warehouse, true, true); } + @Test + public void warmKeyRenameUsesBoundSchemaOptions(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("old_key", DataTypes.INT().notNull()).primaryKey("id", "old_key") + .option("bucket", "1").option("bucket-key", "old_key") + .option("sequence.field", "old_key").option("file.format", "parquet").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + append(warm, GenericRow.of(1, 10)); + warm = warm.copyWithoutTimeTravel(Collections.singletonMap("read.batch-size", "64")); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("old_key", "bound_key")), false); + long boundId = ((FileStoreTable) catalog.getTable(id)).schema().id(); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("bound_key", "later_key")), false); + for (FileStoreTable loaded : Arrays.asList(warm, (FileStoreTable) catalog.getTable(id))) { + FileStoreTable pinned = PaimonScanParams.applyOptionsWithoutTimeTravel(loaded, + PaimonScanParams.withBoundSchema( + PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), 1), boundId)); + Assertions.assertEquals("bound_key", pinned.options().get("bucket-key")); + Assertions.assertEquals("bound_key", pinned.options().get("sequence.field")); + Assertions.assertEquals(Collections.singletonList(1), readIds(pinned)); + if (loaded == warm) { + Assertions.assertEquals("64", pinned.options().get("read.batch-size")); + } + } + } + } + + @Test + public void stalePrivilegedFallbackKeepsBranchProvenance(@TempDir Path warehouse) throws Exception { + try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()) + .column("value", DataTypes.INT()).partitionKeys("id") + .option("file.format", "parquet").option("scan.manifest.parallelism", "1").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + warm.createBranch("backup"); + FileStoreTable fallback = warm.switchToBranch("backup"); + append(fallback, GenericRow.of(2, 20)); + append(warm, GenericRow.of(1, 10)); + catalog.alterTable(id, Collections.singletonList(SchemaChange.setOption("read.batch-size", "32")), + false); + FileStoreTable latest = (FileStoreTable) catalog.getTable(id); + AtomicInteger selectChecks = new AtomicInteger(); + AtomicBoolean denied = new AtomicBoolean(); + PrivilegeChecker checker = (PrivilegeChecker) Proxy.newProxyInstance( + PrivilegeChecker.class.getClassLoader(), new Class[] {PrivilegeChecker.class}, + (proxy, method, args) -> { + if (method.getName().equals("assertCanSelect")) { + selectChecks.incrementAndGet(); + if (denied.get()) { + throw new SecurityException("Select denied"); + } + } + return null; + }); + FileStoreTable privileged = PrivilegedFileStoreTable.wrap( + new FallbackReadFileStoreTable(warm, fallback), checker, id); + FileStoreTable pinned = PaimonScanParams.applyOptionsWithoutTimeTravel(privileged, + PaimonScanParams.withBoundSchema( + PaimonScanParams.pinOptionsToSnapshot(Collections.emptyMap(), 1), latest.schema().id())); + Assertions.assertInstanceOf(PrivilegedFileStoreTable.class, pinned); + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) + ((DelegatedFileStoreTable) pinned).wrapped(); + Assertions.assertEquals("backup", pair.fallback().coreOptions().branch()); + Assertions.assertEquals(fallback.schema().id(), pair.fallback().schema().id()); + Assertions.assertEquals(latest.schema().id(), pair.wrapped().schema().id()); + List ids = readIds(pinned); + Collections.sort(ids); + Assertions.assertEquals(Arrays.asList(1, 2), ids); + Assertions.assertTrue(selectChecks.get() >= 2); + denied.set(true); + Assertions.assertThrows(SecurityException.class, pinned::newScan); + Assertions.assertThrows(SecurityException.class, pinned::newRead); + } + } + private void checkSchemaMutation(Path warehouse, boolean options, boolean system) throws Exception { try (Catalog catalog = new FileSystemCatalog(LocalFileIO.create(), new org.apache.paimon.fs.Path(warehouse.toUri()))) { diff --git a/regression-test/data/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.out b/regression-test/data/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.out new file mode 100644 index 00000000000000..9233f0f07c5937 --- /dev/null +++ b/regression-test/data/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.out @@ -0,0 +1,16 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !plain_schema -- +1 base + +-- !options_schema -- +1 base + +-- !branch_schema -- +1 base + +-- !native_precision -- +1 + +-- !jni_precision -- +1 + diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy index 388c05b4dbdc01..3da6a189f095c2 100644 --- a/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy @@ -92,32 +92,32 @@ suite("test_paimon_schema_only_snapshot_precision", "p0,external,paimon") { sql """use ${dbName}""" sql """refresh table ${tableName}""" - assertEquals([[1, "base"]], sql(""" + order_qt_plain_schema """ select id, current_name from ${tableName} order by id - """)) - assertEquals([[1, "base"]], sql(""" + """ + order_qt_options_schema """ select id, current_name from ${tableName}@options('scan.plan-sort-partition'='true') order by id - """)) - assertEquals([[1, "base"]], sql(""" + """ + order_qt_branch_schema """ select id, branch_name from ${tableName}@branch(${branchName}) order by id - """)) + """ // Sub-millisecond precision must survive FE predicate conversion or Paimon's file // statistics can reject the only matching file before either reader sees it. sql """set force_jni_scanner=false""" - assertEquals([[1]], sql(""" + order_qt_native_precision """ select id from ${tableName} where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) - """)) + """ sql """set force_jni_scanner=true""" - assertEquals([[1]], sql(""" + order_qt_jni_precision """ select id from ${tableName} where event_time = cast('2024-01-01 00:00:00.123456' as datetime(6)) - """)) + """ } finally { sql """set force_jni_scanner=false""" sql """drop catalog if exists ${catalogName}""" From 56cfe11da78b7f4a1cad21fdbd4cd6b1c7d673d0 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 13 Sep 2026 18:07:01 +0800 Subject: [PATCH 05/11] [fix](fe) Authenticate bound schema reads and preserve partition types ### What problem does this PR solve? Related PR: #67904 Problem Summary: Statement schema capture, materialization and restoration can read Paimon schema files after table resolution leaves the authenticated connector scope. Run those reads with the connector authentication and classloader context. The generic FE bridge also used ambient partition columns while constructing a new pinned schema; pass the pinned columns through RANGE/LIST builders so type or arity changes cannot corrupt partition items. ### Release note Keep statement schema reads authenticated and construct partition items with the bound schema's column types. ### Check List (For Author) - Test: Paimon module package, 576 passed and one skipped; FE MVCC suite, 71 passed; FE Checkstyle. Six new tests fail before the fixes and pass afterward. The unrelated IvmNormalizeMTMVJoinTest compile failure was temporarily excluded and the configuration restored. - Behavior changed: Yes, preserve authenticated schema access and partition schema consistency. - Does this need documentation: No. --- .../paimon/PaimonConnectorMetadata.java | 31 +++++- .../paimon/PaimonScanPlanProvider.java | 27 +++++- .../paimon/PaimonStatementSchemaTest.java | 96 +++++++++++++++++++ .../mvcc/PluginDrivenMvccExternalTable.java | 26 ++--- .../PluginDrivenMvccExternalTableTest.java | 60 +++++++++++- 5 files changed, 221 insertions(+), 19 deletions(-) diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java index fb3df5611eef6c..b0549139c07341 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java @@ -65,6 +65,7 @@ import java.util.Optional; import java.util.OptionalLong; import java.util.Set; +import java.util.function.Supplier; /** * {@link ConnectorMetadata} implementation for Paimon. @@ -254,7 +255,8 @@ public ConnectorTableSchema getTableSchema( // tables (isSystemTable()) always keep their synthetic rowType() (no schema-version history; some // are not DataTable). Sharing buildTableSchema with the at-snapshot path keeps the two from drifting. if (!paimonHandle.isSystemTable()) { - Optional latest = catalogOps.latestSchema(table); + Optional latest = + readSchemaAuthenticated(() -> catalogOps.latestSchema(table)); if (latest.isPresent()) { PaimonCatalogOps.PaimonSchemaSnapshot schema = latest.get(); return buildTableSchema( @@ -317,7 +319,8 @@ public ConnectorTableSchema getTableSchema( // carries branchName in equals/hashCode) so a branch@schemaId and a base@same-schemaId cannot // collide in this long-lived memo. resolveTable runs ONCE, outside the loader. PaimonCatalogOps.PaimonSchemaSnapshot schema = - schemaAtMemo.getOrLoad(pinned, schemaId, () -> catalogOps.schemaAt(table, schemaId)); + schemaAtMemo.getOrLoad(pinned, schemaId, + () -> readSchemaAuthenticated(() -> catalogOps.schemaAt(table, schemaId))); return buildTableSchema( paimonHandle.getTableName(), table, @@ -588,9 +591,21 @@ public Optional beginQuerySnapshot( .schemaId(statementSchemaId(paimonHandle, resolveTable(paimonHandle))).build()); } + private T readSchemaAuthenticated(Supplier read) { + // Cached tables do not cache schema files: exact/latest schema reads still need plugin UGI and TCCL. + try { + return context.executeAuthenticated(read::get); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException("Failed to read Paimon schema", e); + } + } + private long statementSchemaId(PaimonTableHandle handle, Table table) { return statementSchemaIds.computeIfAbsent(handle, - ignored -> catalogOps.latestSchema(table).map(PaimonCatalogOps.PaimonSchemaSnapshot::schemaId) + ignored -> readSchemaAuthenticated(() -> catalogOps.latestSchema(table)) + .map(PaimonCatalogOps.PaimonSchemaSnapshot::schemaId) .orElse(-1L)); } @@ -1191,7 +1206,8 @@ public Map getColumnHandles( // (aborts the BE). latestSchema() is empty for a non-DataTable/schema-less backend -> fall back to // rowType(). System tables keep their synthetic rowType() (no schema-version history). if (!paimonHandle.isSystemTable()) { - Optional latest = catalogOps.latestSchema(table); + Optional latest = + readSchemaAuthenticated(() -> catalogOps.latestSchema(table)); if (latest.isPresent()) { return buildColumnHandles(latest.get().fields(), true); } @@ -1243,7 +1259,8 @@ public Map getColumnHandles( // per-catalog and long-lived, so keying on the base handle would let a branch@schemaId poison a // later base@same-schemaId read (each has its own independently-evolved schema-). PaimonCatalogOps.PaimonSchemaSnapshot schema = - schemaAtMemo.getOrLoad(pinned, schemaId, () -> catalogOps.schemaAt(table, schemaId)); + schemaAtMemo.getOrLoad(pinned, schemaId, + () -> readSchemaAuthenticated(() -> catalogOps.schemaAt(table, schemaId))); return buildColumnHandles(schema.fields(), true); } @@ -1608,6 +1625,10 @@ private Table runtimeSafeSystemTable( return systemTable; } Table dataTable = PaimonTableResolver.resolveSystemSource(catalogOps, handle, context); + if (PaimonScanParams.preservesBoundSchema(scanOptions)) { + return readSchemaAuthenticated(() -> PaimonReaderOptions.runtimeSafeSystemTable( + handle.getSysTableName(), systemTable, dataTable, scanOptions)); + } return PaimonReaderOptions.runtimeSafeSystemTable( handle.getSysTableName(), systemTable, dataTable, scanOptions); } diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java index 8128f30d04d5cd..011cf9d3a57d04 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java @@ -114,6 +114,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Supplier; import java.util.stream.Collectors; /** @@ -343,6 +344,24 @@ Table resolveTable(PaimonTableHandle paimonHandle) { */ Table resolveScanTable(PaimonTableHandle paimonHandle) { Table table = resolveTable(paimonHandle); + return withBoundSchemaAuthentication(paimonHandle, () -> applyScanOptions(paimonHandle, table)); + } + + private T withBoundSchemaAuthentication(PaimonTableHandle handle, Supplier action) { + if (context == null || !PaimonScanParams.preservesBoundSchema(handle.getScanOptions())) { + return action.get(); + } + // Restoring a bound schema can read FileIO after table resolution has left the authenticated scope. + try { + return context.executeAuthenticated(action::get); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException("Failed to restore Paimon statement schema", e); + } + } + + private Table applyScanOptions(PaimonTableHandle paimonHandle, Table table) { Map scanOptions = paimonHandle.getScanOptions(); Table finalTable = table; if (scanOptions != null && !scanOptions.isEmpty() @@ -1203,6 +1222,10 @@ OptionalInt backendManifestParallelism(PaimonTableHandle handle, Table scanTable */ // Package-private for direct unit testing (PaimonBackendBoundTableTest). Table tableForBackend(PaimonTableHandle handle, Table scanTable) { + return withBoundSchemaAuthentication(handle, () -> buildBackendTable(handle, scanTable)); + } + + private Table buildBackendTable(PaimonTableHandle handle, Table scanTable) { if (scanTable instanceof FileStoreTable) { // resolveScanTable's copy(...) merged the relation's dynamic options into the schema, // and the rebuild below goes through that schema, so this branch needs no re-application. @@ -1453,8 +1476,8 @@ private Table resolveSchemaDictTable(Table table, PaimonTableHandle handle) { FileStoreTable pinnedSource = handle.getSysBaseTable(); // A reloaded source still needs the bound schema and data selector; otherwise the // native dictionary can disagree with the wrapper after either cache or handle reload. - return PaimonReaderOptions.runtimeSafeSystemSource( - pinnedSource == null ? reloadBaseTable(handle) : pinnedSource, handle.getScanOptions()); + return withBoundSchemaAuthentication(handle, () -> PaimonReaderOptions.runtimeSafeSystemSource( + pinnedSource == null ? reloadBaseTable(handle) : pinnedSource, handle.getScanOptions())); } return null; } diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java index e35a0092874936..e7973f928aca06 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonStatementSchemaTest.java @@ -18,7 +18,9 @@ package org.apache.doris.connector.paimon; import org.apache.doris.connector.spi.ConnectorColumn; +import org.apache.doris.connector.spi.ConnectorContext; import org.apache.doris.connector.spi.ConnectorSession; +import org.apache.doris.connector.spi.ForwardingConnectorContext; import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; import org.apache.doris.connector.spi.mvcc.ConnectorMvccSnapshot; import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec; @@ -30,6 +32,7 @@ import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.privilege.PrivilegeChecker; import org.apache.paimon.privilege.PrivilegedFileStoreTable; @@ -50,6 +53,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Proxy; import java.nio.file.Path; import java.util.ArrayList; @@ -58,11 +62,103 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; public class PaimonStatementSchemaTest { + @Test + public void latestCacheHitCapturesSchemaInsideAuth(@TempDir Path warehouse) throws Exception { + checkAuthenticatedSchemaRead(warehouse, "capture"); + } + + @Test + public void pinnedSchemaMaterializesInsideAuth(@TempDir Path warehouse) throws Exception { + checkAuthenticatedSchemaRead(warehouse, "materialize"); + } + + @Test + public void scanRestoresSchemaInsideAuth(@TempDir Path warehouse) throws Exception { + checkAuthenticatedSchemaRead(warehouse, "restore"); + } + + private void checkAuthenticatedSchemaRead(Path warehouse, String operation) throws Exception { + AtomicBoolean enforceScope = new AtomicBoolean(); + ThreadLocal authenticated = ThreadLocal.withInitial(() -> false); + AtomicInteger reads = new AtomicInteger(); + ClassLoader pluginLoader = new ClassLoader(getClass().getClassLoader()) {}; + ClassLoader callerLoader = Thread.currentThread().getContextClassLoader(); + FileIO local = LocalFileIO.create(); + FileIO guarded = (FileIO) Proxy.newProxyInstance(getClass().getClassLoader(), + new Class[] {FileIO.class}, (proxy, method, args) -> { + if (enforceScope.get() && args != null && args.length > 0 + && args[0] instanceof org.apache.paimon.fs.Path + && args[0].toString().contains("/schema")) { + Assertions.assertTrue(authenticated.get(), "schema FileIO must run inside auth"); + Assertions.assertSame(pluginLoader, Thread.currentThread().getContextClassLoader()); + reads.incrementAndGet(); + } + try { + return method.invoke(local, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + }); + ConnectorContext context = new TcclPinningConnectorContext( + new ForwardingConnectorContext(new RecordingConnectorContext()) { + @Override + public T executeAuthenticated(Callable task) throws Exception { + boolean previous = authenticated.get(); + authenticated.set(true); + try { + return task.call(); + } finally { + authenticated.set(previous); + } + } + }, pluginLoader, () -> null); + try (Catalog catalog = new FileSystemCatalog(guarded, + new org.apache.paimon.fs.Path(warehouse.toUri()))) { + catalog.createDatabase("db", false); + Identifier id = Identifier.create("db", "t"); + catalog.createTable(id, Schema.newBuilder().column("old_name", DataTypes.INT()) + .option("scan.manifest.parallelism", "1").build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + PaimonTableHandle handle = new PaimonTableHandle("db", "t", + Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(warm); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonLatestSnapshotCache cache = new PaimonLatestSnapshotCache(100, 1000); + PaimonCatalogProperties props = PaimonCatalogProperties.of(Collections.emptyMap()); + new PaimonConnectorMetadata(ops, props, context, new PaimonSchemaAtMemo(1000), cache) + .beginQuerySnapshot(null, handle); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("old_name", "bound_name")), false); + long schemaId = ((FileStoreTable) catalog.getTable(id)).schema().id(); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata( + ops, props, context, new PaimonSchemaAtMemo(1000), cache); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder().snapshotId(-1L) + .schemaId(schemaId).build(); + enforceScope.set(true); + if (operation.equals("capture")) { + Assertions.assertEquals(schemaId, metadata.beginQuerySnapshot(null, handle).get().getSchemaId()); + } else if (operation.equals("materialize")) { + Assertions.assertEquals("bound_name", + metadata.getTableSchema(null, handle, snapshot).getColumns().get(0).getName()); + } else { + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + Assertions.assertEquals(Collections.singletonList("bound_name"), + new PaimonScanPlanProvider(props, ops, context).resolveScanTable(pinned) + .rowType().getFieldNames()); + } + Assertions.assertTrue(reads.get() > 0, "the assertion must exercise real schema-file IO"); + Assertions.assertFalse(authenticated.get()); + Assertions.assertSame(callerLoader, Thread.currentThread().getContextClassLoader()); + enforceScope.set(false); + } + } + @Test public void warmTableKeepsExactStatementSchema(@TempDir Path warehouse) throws Exception { checkSchemaMutation(warehouse, false, false); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java index c18bc709b35379..9c4360832a328a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java @@ -180,6 +180,9 @@ private PluginDrivenMvccSnapshot materializeLatest( pinnedSchema = toSchemaCacheValue(metadata, session, db != null ? db.getRemoteName() : "", getRemoteName(), atSchema); } + // This pin is not in StatementContext yet; ambient schema lookup can see a later generation. + List partitionColumns = pinnedSchema == null + ? getPartitionColumns() : pinnedSchema.getPartitionColumns(); // Partition counts feed COUNT(*) pushdown and SQL block rules, so even the initial latest // materialization must enumerate the same pinned generation that the data scan reads. ConnectorTableHandle partitionHandle = pinnedHandle; @@ -190,7 +193,7 @@ private PluginDrivenMvccSnapshot materializeLatest( if (connectorSnapshot.getSnapshotId() < 0) { // A negative query-begin pin is the connector's resolved-empty generation. Falling through to // a live LIST read would mix partitions committed after the data scan's empty boundary. - return buildFromRangeView(connectorSnapshot, view, pinnedSchema); + return buildFromRangeView(connectorSnapshot, view, pinnedSchema, partitionColumns); } // A non-RANGE (UNPARTITIONED) view is the connector's "not RANGE / not MTMV-range-eligible" verdict // (iceberg's range view only covers single time-transform specs). If the table nonetheless declares @@ -202,19 +205,21 @@ private PluginDrivenMvccSnapshot materializeLatest( // A RANGE view (range items) and a genuinely unpartitioned table (no partition columns) are // unaffected. Freshness matches the legacy LIST path (timestamps / 0), harmless here because the // UNPARTITIONED verdict keeps this table out of the snapshot-id MTMV path. - if (view.getStyle() != ConnectorMvccPartitionView.Style.RANGE && !getPartitionColumns().isEmpty()) { + if (view.getStyle() != ConnectorMvccPartitionView.Style.RANGE && !partitionColumns.isEmpty()) { Map listItems = Maps.newHashMap(); Map listLastModifiedMillis = Maps.newHashMap(); - listLatestPartitions(metadata, session, partitionHandle, listItems, listLastModifiedMillis); + listLatestPartitions(metadata, session, partitionHandle, partitionColumns, + listItems, listLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, listItems, listLastModifiedMillis, pinnedSchema, PartitionType.UNPARTITIONED, false, 0L); } - return buildFromRangeView(connectorSnapshot, view, pinnedSchema); + return buildFromRangeView(connectorSnapshot, view, pinnedSchema, partitionColumns); } Map nameToPartitionItem = Maps.newHashMap(); Map nameToLastModifiedMillis = Maps.newHashMap(); - listLatestPartitions(metadata, session, partitionHandle, nameToPartitionItem, nameToLastModifiedMillis); + listLatestPartitions(metadata, session, partitionHandle, partitionColumns, + nameToPartitionItem, nameToLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, nameToPartitionItem, nameToLastModifiedMillis, pinnedSchema); } @@ -229,7 +234,7 @@ private PluginDrivenMvccSnapshot materializeLatest( * per-partition log-and-skip). */ private PluginDrivenMvccSnapshot buildFromRangeView(ConnectorMvccSnapshot connectorSnapshot, - ConnectorMvccPartitionView view, PluginDrivenSchemaCacheValue pinnedSchema) { + ConnectorMvccPartitionView view, PluginDrivenSchemaCacheValue pinnedSchema, List partitionColumns) { PartitionType partitionType = view.getStyle() == ConnectorMvccPartitionView.Style.RANGE ? PartitionType.RANGE : PartitionType.UNPARTITIONED; boolean snapshotIdFreshness = @@ -237,7 +242,6 @@ private PluginDrivenMvccSnapshot buildFromRangeView(ConnectorMvccSnapshot connec Map nameToPartitionItem = Maps.newHashMap(); Map nameToFreshnessValue = Maps.newHashMap(); if (view.getStyle() == ConnectorMvccPartitionView.Style.RANGE) { - List partitionColumns = getPartitionColumns(); for (ConnectorMvccPartition partition : view.getPartitions()) { try { nameToPartitionItem.put(partition.getName(), @@ -291,9 +295,8 @@ private static List toPartitionValues(List bound) { * rather than failing the whole query. */ private void listLatestPartitions(ConnectorMetadata metadata, ConnectorSession session, - ConnectorTableHandle handle, Map nameToPartitionItem, - Map nameToLastModifiedMillis) { - List partitionColumns = getPartitionColumns(); + ConnectorTableHandle handle, List partitionColumns, + Map nameToPartitionItem, Map nameToLastModifiedMillis) { List types = partitionColumns.stream().map(Column::getType).collect(Collectors.toList()); List parts = metadata.listPartitions(session, handle, Optional.empty()); for (ConnectorPartitionInfo part : parts) { @@ -469,7 +472,8 @@ private MvccSnapshot loadSnapshotInternal( // normal-read materializeLatest path — NOT a snapshot-pinned handle. Map nameToPartitionItem = Maps.newHashMap(); Map nameToLastModifiedMillis = Maps.newHashMap(); - listLatestPartitions(metadata, session, handle, nameToPartitionItem, nameToLastModifiedMillis); + listLatestPartitions(metadata, session, handle, getPartitionColumns(), + nameToPartitionItem, nameToLastModifiedMillis); return new PluginDrivenMvccSnapshot(connectorSnapshot, nameToPartitionItem, nameToLastModifiedMillis, null); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java index cbce74013065bf..815e23825a2cdd 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTableTest.java @@ -585,6 +585,64 @@ public void testLatestSnapshotCarriesConnectorBoundSchema() { "a connector's latest schema generation must survive the data fence materialization"); } + @Test + public void testLatestRangeUsesPinnedPartitionTypeAndArity() { + Fixture f = pinnedPartitionFixture(); + Mockito.when(f.metadata.getMvccPartitionView(f.session, f.pinnedHandle)) + .thenReturn(Optional.of(rangeView(rangePart("p1", "10", "20", FRESH_555)))); + PluginDrivenMvccSnapshot pin = (PluginDrivenMvccSnapshot) + f.table.loadSnapshot(Optional.empty(), Optional.empty()); + RangePartitionItem item = (RangePartitionItem) pin.getNameToPartitionItem().get("p1"); + Assertions.assertEquals(1, item.getItems().lowerEndpoint().getKeys().size()); + Assertions.assertEquals(Type.INT, item.getItems().lowerEndpoint().getKeys().get(0).getType()); + Assertions.assertEquals("10", item.getItems().lowerEndpoint().getKeys().get(0).getStringValue()); + } + + @Test + public void testLatestListUsesPinnedPartitionTypeAndArity() { + checkLatestPinnedList(false); + } + + @Test + public void testLatestUnpartitionedViewUsesPinnedListColumns() { + checkLatestPinnedList(true); + } + + private void checkLatestPinnedList(boolean unpartitionedView) { + Fixture f = pinnedPartitionFixture(); + if (unpartitionedView) { + // The ambient schema has no partition columns, but this pin still owns a LIST partition. + f.latestCacheValue.getPartitionColumns().clear(); + Mockito.when(f.metadata.getMvccPartitionView(f.session, f.pinnedHandle)) + .thenReturn(Optional.of(ConnectorMvccPartitionView.unpartitioned())); + } + Mockito.when(f.metadata.listPartitions(Mockito.eq(f.session), Mockito.eq(f.pinnedHandle), Mockito.any())) + .thenReturn(Collections.singletonList(cpi("key=10", TS_2024_01_01))); + PluginDrivenMvccSnapshot pin = (PluginDrivenMvccSnapshot) + f.table.loadSnapshot(Optional.empty(), Optional.empty()); + Assertions.assertEquals(1, pin.getNameToPartitionItem().size()); + PartitionKey key = ((ListPartitionItem) pin.getNameToPartitionItem().get("key=10")).getItems().get(0); + Assertions.assertEquals(1, key.getKeys().size()); + Assertions.assertEquals(Type.INT, key.getKeys().get(0).getType()); + Assertions.assertEquals("10", key.getKeys().get(0).getStringValue()); + } + + private Fixture pinnedPartitionFixture() { + Fixture f = Fixture.timeTravel(); + // The local pin predates a partition type/arity change in the ambient latest schema. + f.latestCacheValue.getPartitionColumns().clear(); + f.latestCacheValue.getPartitionColumns().add(new Column("key", Type.DATEV2)); + f.latestCacheValue.getPartitionColumns().add(new Column("added", Type.INT)); + ConnectorMvccSnapshot snapshot = ConnectorMvccSnapshot.builder() + .snapshotId(PINNED_SNAPSHOT_ID).schemaId(Fixture.TT_SCHEMA_ID).build(); + Mockito.when(f.metadata.beginQuerySnapshot(f.session, f.handle)).thenReturn(Optional.of(snapshot)); + ConnectorTableSchema schema = new ConnectorTableSchema("REMOTE_TBL", + Collections.singletonList(new ConnectorColumn("key", ConnectorType.of("INT"), "", true, null)), + "", Collections.singletonMap(ConnectorTableSchema.PARTITION_COLUMNS_KEY, "key")); + Mockito.when(f.metadata.getTableSchema(f.session, f.pinnedHandle, snapshot)).thenReturn(schema); + return f; + } + @Test public void testInitialLatestPartitionAccountingUsesPinnedHandle() { Fixture f = Fixture.partitioned(); @@ -1532,7 +1590,7 @@ private static Fixture build(List partitions, boolean ti // string-key path) — the LATEST schema. List schema = Collections.singletonList(new Column("dt", partitionColType)); PluginDrivenSchemaCacheValue latestCacheValue = new PluginDrivenSchemaCacheValue( - schema, schema, Collections.singletonList("dt")); + schema, new ArrayList<>(schema), Collections.singletonList("dt")); ConnectorMvccSnapshot resolvedSnapshot = ConnectorMvccSnapshot.builder() .snapshotId(7L).schemaId(TT_SCHEMA_ID).build(); From 1cf784fb4a972f2ec242919db51c05c4cfa121c0 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Sun, 13 Sep 2026 19:04:44 +0800 Subject: [PATCH 06/11] [fix](fe) Preserve Iceberg synthetic columns and empty-table schema pins ### What problem does this PR solve? Related PR: #67904 Problem Summary: Eagerly retained latest schemas made the MVCC guard reject request-scoped Iceberg row IDs, breaking hidden-column reads and DML. Exempt connector SYNTHESIZED columns on schema misses while retaining physical and GENERATED column checks. Empty tables also bypassed cached schema IDs after schema-only changes. Resolve schema and column handles through the same pinned schema lookup before and after the first append. ### Release note Preserve Iceberg hidden-column reads and DML with pinned schemas, and keep empty-table schema generations consistent across schema-only changes. ### Check List (For Author) - Test: 234 FE scan/MVCC and 251 Iceberg metadata/MVCC/scan-provider tests passed; regression tests reproduced both bugs before fixes; FE build and Checkstyle passed; original hidden-row-ID regression suite passed locally after the fix. - Behavior changed: Yes, honor cached schema IDs and synthesized column semantics. - Does this need documentation: No. --- .../iceberg/IcebergConnectorMetadata.java | 35 ++++------- .../IcebergConnectorMetadataMvccTest.java | 60 ++++++++++++++++++ .../datasource/scan/PluginDrivenScanNode.java | 13 ++-- ...uginDrivenScanNodeMvccSchemaGuardTest.java | 63 +++++++++++++++---- 4 files changed, 133 insertions(+), 38 deletions(-) diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java index b09baa5934ec36..c93df68e5ef486 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java @@ -465,8 +465,8 @@ public ConnectorTableSchema getTableSchema( /** * Returns the schema AS OF {@code snapshot.getSchemaId()} (the pinned schema version, for time-travel reads * under schema evolution), or the LATEST schema when there is no pinned schema id (null snapshot or - * {@code schemaId < 0}). Mirrors legacy {@code IcebergUtils.getSchema}: {@code table.schemas().get(schemaId)} - * when the id is set and a current snapshot exists, else {@code table.schema()}. Shares + * {@code schemaId < 0}). Resolves {@code table.schemas().get(schemaId)} even before the first append, + * since schema-only changes do not create data snapshots. Shares * {@link #buildTableSchema} with the latest path so the two cannot drift. */ @Override @@ -484,27 +484,21 @@ public ConnectorTableSchema getTableSchema( return getTableSchema(session, handle); } Table table = loadTable(session, iceHandle); - Schema schema; - if (table.currentSnapshot() == null) { - // Empty table: legacy getSchema falls back to the latest schema (NEWEST_SCHEMA_ID path). - schema = table.schema(); - } else { - schema = table.schemas().get((int) snapshot.getSchemaId()); - if (schema == null) { - // Defensive: a pinned id absent from table.schemas() (legacy would NPE) -> latest. - // INVARIANT: this SLOT-schema fallback MUST stay identical to the DICT-schema fallback in - // IcebergScanPlanProvider.pinnedSchema (same getSchemaId() lookup + same silent -> table.schema()). - // If the two diverge, the field-id dict names and the BE scan-slot names resolve DIFFERENT - // schemas -> BE children.at() std::out_of_range-SIGABRT on a schema-evolved time-travel read - // (reverify #65185 L16). Do not harden ONE side to throw without the other. - schema = table.schema(); - } - } + Schema schema = resolvePinnedSchema(table, snapshot); String specId = snapshot.getProperties().get(PARTITION_SPEC_ID_PROPERTY); PartitionSpec spec = specId == null ? table.spec() : table.specs().get(Integer.parseInt(specId)); return buildTableSchema(iceHandle.getTableName(), table, schema, spec, true); } + private static Schema resolvePinnedSchema(Table table, ConnectorMvccSnapshot snapshot) { + // An empty table can evolve its schema while a cached snapshot-less pin remains unchanged. + // Slots and handles must honor that schema ID both before and after the first append. + Schema schema = table.schemas().get((int) snapshot.getSchemaId()); + // Keep the missing-ID fallback aligned with IcebergScanPlanProvider.pinnedSchema so the + // reader's field-ID dictionary and FE slots cannot resolve different schema generations. + return schema == null ? table.schema() : schema; + } + /** * Assembles the {@link ConnectorTableSchema} for {@code table} from {@code schema} (the latest schema, or a * historical schema for a time-travel read). Pinned reads also supply the partition spec from their @@ -768,10 +762,7 @@ public Map getColumnHandles( return getColumnHandles(session, handle); } Table table = loadTable(session, iceHandle); - Schema schema = table.currentSnapshot() == null - ? table.schema() : table.schemas().get((int) snapshot.getSchemaId()); - // Keep the handle-schema fallback identical to getTableSchema so slots and handles cannot diverge. - return buildColumnHandles(schema == null ? table.schema() : schema, true); + return buildColumnHandles(resolvePinnedSchema(table, snapshot), true); } @Override diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java index 3f1e80939b3875..81edd7d3555366 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java @@ -193,6 +193,66 @@ public void latestCacheHitKeepsPartitionSchemaAfterLiveRename() { schema.getProperties().get(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY)); } + @Test + public void emptyTableCacheHitKeepsSchemaAndHandlesThroughFirstAppend() { + InMemoryCatalog catalog = new InMemoryCatalog(); + catalog.initialize("test", Collections.emptyMap()); + catalog.createNamespace(Namespace.of("db1")); + PartitionSpec spec = PartitionSpec.builderFor(PARTITIONED_SCHEMA).day("ts").build(); + Table table = catalog.createTable(TableIdentifier.of("db1", "t1"), PARTITIONED_SCHEMA, spec); + RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps(); + ops.table = table; + IcebergLatestSnapshotCache cache = new IcebergLatestSnapshotCache(100, 1000); + IcebergCatalogProperties properties = IcebergCatalogProperties.of(Collections.emptyMap()); + IcebergConnectorMetadata firstQuery = new IcebergConnectorMetadata( + ops, properties, new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot first = firstQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(-1L, first.getSnapshotId()); + Assertions.assertTrue(first.getSchemaId() >= 0); + table.updateSchema().renameColumn("ts", "renamed_ts").commit(); + Assertions.assertNull(table.currentSnapshot()); + + // Keep only the latest-pin cache warm; each query reloads live metadata as vended catalogs do. + for (int stage = 0; stage < 2; stage++) { + IcebergConnectorMetadata nextQuery = new IcebergConnectorMetadata( + ops, properties, new RecordingConnectorContext(), cache); + ConnectorMvccSnapshot cached = nextQuery.beginQuerySnapshot(null, handle()).get(); + Assertions.assertEquals(first.getSnapshotId(), cached.getSnapshotId()); + Assertions.assertEquals(first.getSchemaId(), cached.getSchemaId()); + ConnectorTableSchema schema = nextQuery.getTableSchema(null, handle(), cached); + Assertions.assertAll( + () -> Assertions.assertTrue(columnNames(schema).contains("ts")), + () -> Assertions.assertFalse(columnNames(schema).contains("renamed_ts")), + () -> Assertions.assertEquals("ts", + schema.getProperties().get(ConnectorTableSchema.PARTITION_COLUMNS_KEY)), + () -> Assertions.assertEquals("PARTITION BY LIST (DAY(`ts`)) ()", + schema.getProperties().get(ConnectorTableSchema.SHOW_PARTITION_CLAUSE_KEY)), + () -> Assertions.assertTrue(nextQuery.getColumnHandles(null, handle(), cached).containsKey("ts")), + () -> Assertions.assertFalse(nextQuery.getColumnHandles(null, handle(), cached) + .containsKey("renamed_ts"))); + if (stage == 0) { + table.newAppend().appendFile(DataFiles.builder(spec) + .withPath("s3://bucket/db1/t1/first.parquet").withFileSizeInBytes(100).withRecordCount(1) + .withPartitionPath("ts_day=1970-04-11").withFormat(FileFormat.PARQUET).build()).commit(); + } + } + } + + @Test + public void emptyTableMissingPinnedSchemaFallsBackForSchemaAndHandles() { + InMemoryCatalog catalog = new InMemoryCatalog(); + catalog.initialize("test", Collections.emptyMap()); + catalog.createNamespace(Namespace.of("db1")); + Table table = catalog.createTable( + TableIdentifier.of("db1", "t1"), SCHEMA_V0, PartitionSpec.unpartitioned()); + IcebergConnectorMetadata metadata = metadataFor(table, new RecordingIcebergCatalogOps()); + ConnectorMvccSnapshot missing = ConnectorMvccSnapshot.builder().snapshotId(-1).schemaId(999).build(); + Assertions.assertEquals(columnNames(metadata.getTableSchema(null, handle())), + columnNames(metadata.getTableSchema(null, handle(), missing))); + Assertions.assertEquals(metadata.getColumnHandles(null, handle()).keySet(), + metadata.getColumnHandles(null, handle(), missing).keySet()); + } + @Test public void beginQuerySnapshotDisabledCacheLoadsEveryCall() { Fixture f = fixture(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java index 3229689f0175d4..9bb9e73b8a6de0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java @@ -1278,7 +1278,7 @@ private void pinMvccSnapshot() throws UserException { // loaded, checks this reference's own pin), so it catches the latest-masked / @incr / MTMV-refresh // cases the version-blind analysis-time binding cannot. Per user decision 2026-07-13: throw for now; // the per-reference version-aware schema-binding refactor is tracked as D-MVCC-VERSION-SCHEMA. A - // latest / @incr / hive scan carries a null pinnedSchema -> no-op; so does a sys-table scan on a + // reference without a pinned schema is a no-op; so is a sys-table scan on a // connector that rejects sys-table time travel (resolveSysTableSnapshotPin returns empty). A sys-table // scan on a connector that DOES honor it (iceberg) is excluded inside the guard — see there. if (snapshot.isPresent() && snapshot.get() instanceof PluginDrivenMvccSnapshot) { @@ -1307,12 +1307,13 @@ private void pinMvccSnapshot() throws UserException { * (BE matches iceberg columns by id, so a rename that keeps the id is fine, a renumber / added column is * caught); {@code uniqueId < 0} (paimon has no top-level field-id) matches by name. Reader-synthesized * row-id columns ({@link Column#GLOBAL_ROWID_COL}) are skipped — they are not table columns and are - * absent from every pinned schema by construction.

+ * absent from every pinned schema by construction. Connector-synthesized columns are likewise exempt + * on a schema miss; their classification comes from the scan provider, not connector-specific names.

* *

Two no-ops, both because the guard's precondition — {@code boundColumns} and {@code pinnedSchema} * describe the SAME table — does not hold: *