From 0043b71832c6d308d8e2be1251e6c0a45e225a37 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 23 Oct 2026 20:33:29 +0800 Subject: [PATCH 1/2] [fix](paimon) Preserve schema options across renamed-field reads --- .../connector/paimon/PaimonCatalogOps.java | 2 +- .../paimon/PaimonConnectorMetadata.java | 38 +- .../connector/paimon/PaimonReaderOptions.java | 10 +- .../connector/paimon/PaimonScanParams.java | 68 ++- .../paimon/PaimonScanPlanProvider.java | 31 +- .../paimon/PaimonTableDecorators.java | 17 + .../PaimonHistoricalSchemaOptionsTest.java | 386 ++++++++++++++++++ .../test_paimon_renamed_field_time_travel.out | 101 +++++ ...st_paimon_renamed_field_time_travel.groovy | 185 +++++++++ 9 files changed, 803 insertions(+), 35 deletions(-) create mode 100644 fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java create mode 100644 regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out create mode 100644 regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy 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 48a69bf114ff54..bd605d696913a5 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 @@ -312,7 +312,7 @@ public Table getTable(Identifier identifier) throws Catalog.TableNotExistExcepti // so this must not invalidate: REFRESH, Doris writes and DDL invalidate through CatalogMetaCache. Table table = catalog.getTable(identifier); Map optionsForCopy = PaimonTableOptions.forCopy(tableOptions); - return optionsForCopy.isEmpty() ? table : table.copy(optionsForCopy); + return optionsForCopy.isEmpty() ? table : PaimonScanParams.copyWithHistoricalSchema(table, optionsForCopy); } @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 79ecda41202d36..c9de0bba92184d 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 @@ -420,15 +420,11 @@ private Table resolveSystemTableAt(ConnectorSession session, ? paimonHandle : (PaimonTableHandle) applySnapshot(session, paimonHandle, snapshot); Map scanOptions = pinned.getScanOptions(); - if (PaimonScanParams.preservesBoundSchema(scanOptions)) { + if (PaimonScanParams.preservesBoundSchema(scanOptions) || PaimonScanParams.isOptionsPin(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); - } - return table; + return resolveTable(pinned); } /** @@ -870,7 +866,8 @@ private Optional resolveTimeTravelAuthenticated( resolved = PaimonScanParams.pinOptionsToSnapshot( spec.getOptions(), spec.getLatestSnapshotFence().getAsLong()); } else { - resolved = PaimonScanParams.resolveOptions(table, spec.getOptions()); + resolved = PaimonScanParams.resolveOptions(table, spec.getOptions(), + PaimonTableOptions.extractCompatible(catalogProperties.getRaw())); } String pinnedTag = resolved.get(CoreOptions.SCAN_TAG_NAME.key()); if (pinnedTag != null) { @@ -1755,6 +1752,13 @@ private ConnectorTableKey partitionViewCacheKey(PaimonTableHandle paimonHandle) * the partition columns and escapes path-special characters in the name via the Paimon SDK. */ private List collectPartitions(PaimonTableHandle paimonHandle) { + if (paimonHandle.getScanOptions().isEmpty()) { + return collectPartitionsAuthenticated(paimonHandle); + } + return readMetadataAuthenticated(() -> collectPartitionsAuthenticated(paimonHandle)); + } + + private List collectPartitionsAuthenticated(PaimonTableHandle paimonHandle) { if (PaimonScanParams.isPinnedEmptyScan(paimonHandle.getScanOptions())) { // Do not reopen latest metadata after the statement fenced an empty table. return Collections.emptyList(); @@ -1767,7 +1771,8 @@ private List collectPartitions(PaimonTableHandle paimonH } Table resolvedTable = resolveTable(paimonHandle); - Map scanOptions = paimonHandle.getScanOptions(); + Map scanOptions = PaimonScanParams.withCatalogOptions( + paimonHandle.getScanOptions(), PaimonTableOptions.extractCompatible(catalogProperties.getRaw())); boolean optionsPin = PaimonScanParams.isOptionsPin(scanOptions); Table table; if (optionsPin) { @@ -1778,8 +1783,9 @@ private List collectPartitions(PaimonTableHandle paimonH // partitions so EXPLAIN and block-rule accounting describe the same version as the scan. Table partitionTable = snapshotId == null ? resolvedTable - : resolvedTable.copy(Collections.singletonMap( - CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId)); + : PaimonScanParams.copyWithHistoricalSchema(resolvedTable, PaimonScanParams.withCatalogOptions( + Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId), + PaimonTableOptions.extractCompatible(catalogProperties.getRaw()))); // Partition projection never opens a data reader, so reader-only settings must not // invalidate metadata that a later relation-scoped override can make safe. // Metadata planning can also touch Paimon's global manifest executor, so apply the @@ -1949,6 +1955,11 @@ public Optional getTableStatistics( if (snapshot == null) { return getTableStatistics(session, handle); } + return readMetadataAuthenticated(() -> getTableStatisticsAtSnapshot(session, handle, snapshot)); + } + + private Optional getTableStatisticsAtSnapshot( + ConnectorSession session, ConnectorTableHandle handle, ConnectorMvccSnapshot snapshot) { long rowCount; try { PaimonTableHandle pinned = (PaimonTableHandle) applySnapshot(session, handle, snapshot); @@ -1957,17 +1968,18 @@ public Optional getTableStatistics( // first commit even though execution is still required to scan zero rows. return Optional.empty(); } - Map scanOptions = pinned.getScanOptions(); + Map scanOptions = PaimonScanParams.withCatalogOptions( + pinned.getScanOptions(), PaimonTableOptions.extractCompatible(catalogProperties.getRaw())); // applyOptions removes this Doris-only marker, but execution still filters files by it. // This also covers scan.creation-time-millis when it resolves to a file-creation scan. if (PaimonScanParams.getPinnedFileCreationTime(scanOptions).isPresent()) { return Optional.empty(); } Table table = resolveTable(pinned); - if (scanOptions != null && !scanOptions.isEmpty()) { + if (scanOptions != null && !scanOptions.isEmpty() && !pinned.isSystemTable()) { table = PaimonScanParams.isOptionsPin(scanOptions) ? PaimonScanParams.applyOptions(table, scanOptions) - : table.copy(scanOptions); + : PaimonScanParams.copyWithHistoricalSchema(table, scanOptions); } table = PaimonReaderOptions.runtimeSafeTable(table); table = runtimeSafeSystemTable(pinned, table, scanOptions); diff --git a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java index 8907dc38ed0df5..053eac694db6d1 100644 --- a/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java +++ b/fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java @@ -189,6 +189,14 @@ public static Table runtimeSafeSystemTable( } return rebuilt; } + // Non-file sources cannot be rebuilt through SystemTableLoader; retain their wrapper's + // own copy semantics after independently validating the source's effective options. + if (scanOptions != null && !scanOptions.isEmpty()) { + systemTable = PaimonScanParams.isOptionsPin(scanOptions) + ? PaimonScanParams.applyOptions(systemTable, scanOptions) + : PaimonScanParams.copyWithHistoricalSchema(systemTable, + PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions)); + } return runtimeSafeTable(systemTable); } @@ -207,7 +215,7 @@ public static Table runtimeSafeSystemSource(Table sourceTable, Map options) { } // Apply the relation override first, then normalize each physical planning branch. A // single combined copy would flatten independent fallback values into one sibling's cap. - Table effectiveTable = PaimonReaderOptions.runtimeSafeTable(table.copy(isolatedOptions)); + Table effectiveTable = PaimonReaderOptions.runtimeSafeTable( + copyWithHistoricalSchema(table, isolatedOptions, options)); // Validate after every copy so relation options participate in the documented // relation > catalog > physical precedence before the effective value is judged. PaimonReaderOptions.validateEffectiveTable(effectiveTable); return effectiveTable; } + static Table copyWithHistoricalSchema(Table table, Map options) { + return copyWithHistoricalSchema(table, userOptions(options), options); + } + + private static Table copyWithHistoricalSchema( + Table table, Map dynamicOptions, Map provenance) { + if (!(table instanceof FileStoreTable) || table instanceof ChainGroupReadTable + || PaimonTableDecorators.isChainTable(table)) { + // Chain groups have their own option filtering and are not ordinary fallback readers. + return table.copy(dynamicOptions); + } + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + // Each branch has independent snapshot IDs. Keep the SDK's timestamp translation + // before restoring either schema; copying one historical schema onto the pair is invalid. + return PaimonTableDecorators.rebuildFallbackPair(pair, + (FileStoreTable) copyWithHistoricalSchema(pair.wrapped(), dynamicOptions, provenance), + (FileStoreTable) copyWithHistoricalSchema(pair.other(), + PaimonTableDecorators.otherOptions(pair, dynamicOptions), provenance)); + } + FileStoreTable dataTable = (FileStoreTable) table; + Map effectiveOptions = new HashMap<>(table.options()); + dynamicOptions.forEach((key, value) -> { + if (value == null) { + effectiveOptions.remove(key); + } else { + effectiveOptions.put(key, value); + } + }); + Optional snapshot; + try { + snapshot = TimeTravelUtil.tryTravelToSnapshot(Options.fromMap(effectiveOptions), + dataTable.snapshotManager(), dataTable.tagManager()); + } catch (Exception e) { + // Match the SDK's schema-selection fallback (for example a timestamp on an empty table). + // Its normal copy below still owns option validation and the eventual scan error. + return table.copy(dynamicOptions); + } + if (snapshot.isPresent() && snapshot.get().schemaId() != dataTable.schema().id()) { + // Paimon overlays current options onto historical columns. Restore the matching + // sequence/aggregation options first, or renamed fields fail before scan planning. + dataTable = restoreBoundSchema(dataTable, snapshot.get().schemaId(), provenance, ""); + } + return dataTable.copy(dynamicOptions); + } + public static FileStoreTable applyOptionsWithoutTimeTravel( FileStoreTable table, Map options) { Map tableOptions = userOptions(options); @@ -353,6 +402,11 @@ public static boolean usesStatementSnapshot(Map options) { } public static Map resolveOptions(Table table, Map options) { + return resolveOptions(table, options, java.util.Collections.emptyMap()); + } + + static Map resolveOptions( + Table table, Map options, Map catalogOptions) { validateOptions(options); if (!hasStartupOptions(options)) { return options; @@ -388,7 +442,7 @@ public static Map resolveOptions(Table table, Map resolvedEmptyOptions(Map opti } private static Map userOptions(Map options) { - return options.entrySet().stream() - .filter(entry -> !entry.getKey().startsWith(INTERNAL_PREFIX)) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + Map result = new HashMap<>(); + options.forEach((key, value) -> { + if (!key.startsWith(INTERNAL_PREFIX)) { + result.put(key, value); + } + }); + return result; } /** 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 6c5493765e13f0..548907b52493d6 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 @@ -361,10 +361,11 @@ Table resolveScanTable(PaimonTableHandle paimonHandle) { } private T withBoundSchemaAuthentication(PaimonTableHandle handle, Supplier action) { - if (context == null || !PaimonScanParams.preservesBoundSchema(handle.getScanOptions())) { + if (context == null || (handle.getScanOptions().isEmpty() + && PaimonTableOptions.extractCompatible(catalogProps.getRaw()).isEmpty())) { return action.get(); } - // Restoring a bound schema can read FileIO after table resolution has left the authenticated scope. + // Historical and statement schema restoration read FileIO after table resolution leaves its scope. try { return context.executeAuthenticated(action::get); } catch (RuntimeException e) { @@ -376,20 +377,17 @@ private T withBoundSchemaAuthentication(PaimonTableHandle handle, Supplier effectiveScanOptions(PaimonTableHandle handle) { Map options = handle.getScanOptions(); - if (PaimonScanParams.preservesBoundSchema(options)) { - // Replayed catalogs already ignored legacy options during binding; scans must use the same policy. - return PaimonScanParams.withCatalogOptions(options, - PaimonTableOptions.extractCompatible(catalogProps.getRaw())); - } - return options; + // An explicit catalog value may equal latest's value but differ from the historical schema. + return PaimonScanParams.withCatalogOptions(options, + PaimonTableOptions.extractCompatible(catalogProps.getRaw())); } private Table applyScanOptions(PaimonTableHandle paimonHandle, Table table) { Map scanOptions = effectiveScanOptions(paimonHandle); Table finalTable = table; - if (scanOptions != null && !scanOptions.isEmpty() - && !(paimonHandle.isSystemTable() && PaimonScanParams.preservesBoundSchema(scanOptions))) { - // Statement-fenced system wrappers are rebuilt below from their schema-bound source. + if (scanOptions != null && !scanOptions.isEmpty() && !paimonHandle.isSystemTable()) { + // System wrappers are rebuilt below from their schema-bound source. Copying the wrapper + // first would time-travel inside the SDK before historical field options can be restored. if (table instanceof FileStoreTable && PaimonScanParams.preservesBoundSchema(scanOptions)) { // A statement fence owns data visibility, not schema time travel. Reusing Table.copy @@ -406,7 +404,8 @@ private Table applyScanOptions(PaimonTableHandle paimonHandle, Table table) { // scan.snapshot-id/scan.mode here (the single Table.copy chokepoint shared by both the // native/JNI scan path and the JNI serialized-table path) so a stale persisted pin on the // base table cannot hijack incremental-between. Non-incremental pins pass through unchanged. - finalTable = table.copy(PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions)); + finalTable = PaimonScanParams.copyWithHistoricalSchema(table, + PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions)); } } finalTable = PaimonReaderOptions.runtimeSafeTable(finalTable); @@ -1799,10 +1798,12 @@ private static FileStoreTable dropCatalogLoaderOfBranches(FileStoreTable table) } 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. + // The factory invokes copy(Map) before returning, so selectors can rewind a schema-only + // ALTER and fail field-option validation before the bound schema can be restored. + Map factoryOptions = new HashMap<>(branch.schema().options()); + PaimonScanParams.inheritedReadStateKeys().forEach(factoryOptions::remove); return FileStoreTableFactory.createWithoutFallbackBranch( - branch.fileIO(), branch.location(), branch.schema(), new Options(), + branch.fileIO(), branch.location(), branch.schema().copy(factoryOptions), new Options(), CatalogEnvironment.empty()).copy(branch.schema()); } 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 48e2e11ad94a25..f6f17e34b3cb54 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 @@ -23,6 +23,8 @@ import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; +import java.util.Map; + /** * The one place that knows how Paimon stacks {@link DelegatedFileStoreTable} decorators on a loaded * table, and how far they may be peeled off. @@ -45,6 +47,21 @@ static FileStoreTable rebuildFallbackPair(FallbackReadFileStoreTable pair, FileS return new FallbackReadFileStoreTable(wrapped, other, PaimonReaderOptions.isWrappedFirst(pair)); } + static Map otherOptions(FallbackReadFileStoreTable pair, Map options) { + return new FallbackOptions(pair).rewriteOtherOptions(options); + } + + private static final class FallbackOptions extends FallbackReadFileStoreTable { + private FallbackOptions(FallbackReadFileStoreTable pair) { + super(pair.wrapped(), pair.other(), PaimonReaderOptions.isWrappedFirst(pair)); + } + + @Override + public Map rewriteOtherOptions(Map options) { + return super.rewriteOtherOptions(options); + } + } + /** * Whether {@code table} reads as a chain table: a fallback pair whose other branch is a * {@link ChainGroupReadTable}. A branch of a chain table read on its own is a plain table. diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java new file mode 100644 index 00000000000000..d0069936b67447 --- /dev/null +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java @@ -0,0 +1,386 @@ +// 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.mvcc.ConnectorMvccSnapshot; +import org.apache.doris.connector.spi.mvcc.ConnectorTimeTravelSpec; + +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.options.Options; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.CatalogEnvironment; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +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.table.system.SystemTableLoader; +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.HashMap; +import java.util.List; +import java.util.Map; + +public class PaimonHistoricalSchemaOptionsTest { + @Test + public void snapshotUsesHistoricalSequenceFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, false, "snapshot"); + } + + @Test + public void timestampUsesHistoricalSequenceFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, false, "timestamp"); + } + + @Test + public void tagUsesHistoricalSequenceFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, false, "tag"); + } + + @Test + public void optionsUsesHistoricalSequenceFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, false, "options"); + } + + @Test + public void currentReadUsesRenamedSequenceFieldFromWarmTable(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, false, "current"); + } + + @Test + public void snapshotUsesHistoricalAggregationFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, true, "snapshot"); + } + + @Test + public void timestampUsesHistoricalAggregationFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, true, "timestamp"); + } + + @Test + public void tagUsesHistoricalAggregationFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, true, "tag"); + } + + @Test + public void optionsUsesHistoricalAggregationFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, true, "options"); + } + + @Test + public void currentReadUsesRenamedAggregationFieldFromWarmTable(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, true, "current"); + } + + @Test + public void systemOptionsUsesHistoricalSequenceFieldAfterRename(@TempDir Path warehouse) throws Exception { + checkRenamedField(warehouse, false, "system"); + } + + @Test + public void historicalSchemaKeepsExplicitCatalogReaderOption(@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", "reader_history"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("value", DataTypes.INT().notNull()).primaryKey("id", "value").partitionKeys("id") + .option("bucket", "1").option("sequence.field", "value") + .option("read.batch-size", "64").option("file.format", "parquet").build(), false); + FileStoreTable original = (FileStoreTable) catalog.getTable(id); + append(original, GenericRow.of(1, 10)); + catalog.alterTable(id, Arrays.asList(SchemaChange.renameColumn("value", "renamed_value"), + SchemaChange.setOption("read.batch-size", "32")), false); + Map properties = Collections.singletonMap("paimon.table-option.read.batch-size", "32"); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps( + catalog, Collections.singletonMap("read.batch-size", "32")); + PaimonCatalogProperties catalogProperties = PaimonCatalogProperties.of(properties); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata( + ops, catalogProperties, new RecordingConnectorContext()); + PaimonTableHandle handle = (PaimonTableHandle) metadata.getTableHandle(null, "db", "reader_history").get(); + ConnectorMvccSnapshot snapshot = metadata.resolveTimeTravel(null, handle, + ConnectorTimeTravelSpec.snapshotId("1")).get(); + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + Assertions.assertEquals(Collections.singletonList("id=1"), metadata.listPartitionNames(null, pinned)); + Assertions.assertTrue(metadata.getTableStatistics(null, handle, snapshot).isPresent()); + Table scan = new PaimonScanPlanProvider(catalogProperties, ops).resolveScanTable(pinned); + // Equal current/catalog values still carry an explicit override of the historical default. + Assertions.assertEquals("32", scan.options().get("read.batch-size")); + Assertions.assertEquals(Collections.singletonList("1:10"), readRows(scan)); + } + } + + @Test + public void fallbackBranchesRestoreIndependentHistoricalOptions(@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", "fallback_history"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("value", DataTypes.INT()).primaryKey("id") + .option("bucket", "1").option("sequence.field", "value") + .option("file.format", "parquet").build(), false); + FileStoreTable original = (FileStoreTable) catalog.getTable(id); + original.createBranch("backup"); + Identifier backupId = new Identifier("db", "fallback_history", "backup"); + FileStoreTable backup = (FileStoreTable) catalog.getTable(backupId); + append(backup, GenericRow.of(2, 8)); + append(original, GenericRow.of(1, 10)); + catalog.alterTable(backupId, Collections.singletonList( + SchemaChange.renameColumn("value", "renamed_value")), false); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("value", "renamed_value")), false); + FileStoreTable latest = (FileStoreTable) catalog.getTable(id); + FileStoreTable latestBackup = (FileStoreTable) catalog.getTable(backupId); + append(latestBackup, GenericRow.of(2, 18)); + append(latest, GenericRow.of(1, 20)); + FallbackReadFileStoreTable pair = new FallbackReadFileStoreTable(latest, latestBackup, false); + Map options = new HashMap<>(); + options.put("scan.snapshot-id", "1"); + options.put("branch", "main"); + FallbackReadFileStoreTable historical = (FallbackReadFileStoreTable) + PaimonScanParams.copyWithHistoricalSchema(pair, options); + Assertions.assertEquals("value", historical.wrapped().options().get("sequence.field")); + Assertions.assertEquals("value", historical.other().options().get("sequence.field")); + Assertions.assertEquals("backup", historical.other().coreOptions().branch()); + Assertions.assertSame(latest.catalogEnvironment(), historical.wrapped().catalogEnvironment()); + Assertions.assertSame(latestBackup.catalogEnvironment(), historical.other().catalogEnvironment()); + Assertions.assertEquals(Collections.singletonList("1:10"), readRows(historical.wrapped())); + Assertions.assertEquals(Collections.singletonList("2:8"), readRows(historical.other())); + } + } + + @Test + public void timestampOnEmptyTableKeepsSdkBehavior(@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", "empty_history"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT()).build(), false); + Table table = catalog.getTable(id); + Map options = Collections.singletonMap("scan.timestamp-millis", "1"); + Table selected = PaimonScanParams.copyWithHistoricalSchema(table, options); + Assertions.assertEquals(table.copy(options).rowType(), selected.rowType()); + IllegalArgumentException expected = Assertions.assertThrows( + IllegalArgumentException.class, () -> readRows(table.copy(options))); + IllegalArgumentException actual = Assertions.assertThrows( + IllegalArgumentException.class, () -> readRows(selected)); + Assertions.assertEquals(expected.getMessage(), actual.getMessage()); + } + } + + @Test + public void catalogBackedSchemaOnlyRenameKeepsStatementSchemaOnBackend(@TempDir Path warehouse) + throws Exception { + checkCatalogBackedTransport(warehouse, true, false); + } + + @Test + public void catalogBackedHistoricalReadKeepsHistoricalSchemaOnBackend(@TempDir Path warehouse) + throws Exception { + checkCatalogBackedTransport(warehouse, false, false); + } + + @Test + public void catalogBackedAggregationRenameKeepsStatementSchemaOnBackend(@TempDir Path warehouse) + throws Exception { + checkCatalogBackedTransport(warehouse, true, true); + } + + @Test + public void catalogBackedAggregationHistoryKeepsHistoricalSchemaOnBackend(@TempDir Path warehouse) + throws Exception { + checkCatalogBackedTransport(warehouse, false, true); + } + + private void checkCatalogBackedTransport(Path warehouse, boolean current, boolean aggregate) 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", "transport_history"); + Schema.Builder schema = Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("value", DataTypes.INT()).primaryKey("id") + .option("bucket", "1").option("file.format", "parquet"); + if (aggregate) { + schema.option("merge-engine", "aggregation").option("fields.value.aggregate-function", "sum"); + } else { + schema.option("sequence.field", "value"); + } + catalog.createTable(id, schema.build(), false); + FileStoreTable original = (FileStoreTable) catalog.getTable(id); + append(original, GenericRow.of(1, 10)); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("value", "renamed_value")), false); + FileStoreTable latest = (FileStoreTable) catalog.getTable(id); + CatalogEnvironment environment = new CatalogEnvironment(id, null, () -> catalog, + null, null, null, false, false); + latest = FileStoreTableFactory.createWithoutFallbackBranch(latest.fileIO(), latest.location(), + latest.schema(), new Options(), environment); + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonCatalogProperties properties = PaimonCatalogProperties.of(Collections.emptyMap()); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata( + ops, properties, new RecordingConnectorContext()); + PaimonTableHandle handle = new PaimonTableHandle("db", "transport_history", + Collections.emptyList(), Collections.emptyList()); + handle.setPaimonTable(latest); + ConnectorMvccSnapshot snapshot = current ? metadata.beginQuerySnapshot(null, handle).get() + : metadata.resolveTimeTravel(null, handle, ConnectorTimeTravelSpec.snapshotId("1")).get(); + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + PaimonScanPlanProvider provider = new PaimonScanPlanProvider(properties, ops); + FileStoreTable scan = (FileStoreTable) provider.resolveScanTable(pinned); + String expectedName = current ? "renamed_value" : "value"; + Assertions.assertNotNull(scan.catalogEnvironment().catalogLoader()); + Assertions.assertEquals(Arrays.asList("id", expectedName), scan.rowType().getFieldNames()); + // A filesystem-only fixture skips loader removal; serialize the catalog-backed path too. + FileStoreTable backend = InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(provider.tableForBackend(pinned, scan)), + getClass().getClassLoader()); + Assertions.assertNull(backend.catalogEnvironment().catalogLoader()); + Assertions.assertEquals(scan.schema(), backend.schema()); + Assertions.assertEquals("1", backend.options().get("scan.snapshot-id")); + if (aggregate) { + Assertions.assertEquals("sum", backend.options().get("fields." + expectedName + ".aggregate-function")); + } else { + Assertions.assertEquals(expectedName, backend.options().get("sequence.field")); + } + Assertions.assertEquals(Collections.singletonList("1:10"), readRows(backend)); + } + } + + private void checkRenamedField(Path warehouse, boolean aggregate, String selector) 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", "field_history"); + Schema.Builder schema = Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("value", DataTypes.INT()).primaryKey("id") + .option("bucket", "1").option("file.format", "parquet"); + if (aggregate) { + schema.option("merge-engine", "aggregation") + .option("fields.value.aggregate-function", "sum"); + } else { + schema.option("sequence.field", "value"); + } + catalog.createTable(id, schema.build(), false); + FileStoreTable warm = (FileStoreTable) catalog.getTable(id); + append(warm, GenericRow.of(1, 10)); + // A second value proves that the configured merge rule remains effective during reads. + append(warm, GenericRow.of(1, 5)); + long snapshotId = warm.snapshotManager().latestSnapshotId(); + long timestamp = warm.snapshotManager().snapshot(snapshotId).timeMillis(); + warm.createTag("before_rename", snapshotId); + // Read-optimized views can omit un-compacted files; preserve their real SDK baseline. + List systemBaseline = "system".equals(selector) + ? readRows(SystemTableLoader.load("ro", warm)) : Collections.emptyList(); + catalog.alterTable(id, Collections.singletonList( + SchemaChange.renameColumn("value", "renamed_value")), false); + FileStoreTable latest = (FileStoreTable) catalog.getTable(id); + append(latest, GenericRow.of(1, 20)); + String historicalRow = aggregate ? "1:15" : "1:10"; + String latestRow = aggregate ? "1:35" : "1:20"; + Assertions.assertEquals(Collections.singletonList(latestRow), readRows(latest)); + + PaimonCatalogOps ops = new PaimonCatalogOps.CatalogBackedPaimonCatalogOps(catalog); + PaimonCatalogProperties properties = PaimonCatalogProperties.of(Collections.emptyMap()); + PaimonConnectorMetadata metadata = new PaimonConnectorMetadata( + ops, properties, new RecordingConnectorContext()); + PaimonTableHandle handle = new PaimonTableHandle("db", "field_history", + Collections.emptyList(), Collections.emptyList()); + boolean current = "current".equals(selector); + handle.setPaimonTable(current ? warm : latest); + // Current reads rebind a warm table; explicit selectors must restore historical field options. + ConnectorMvccSnapshot snapshot = current + ? metadata.beginQuerySnapshot(null, handle).get() + : metadata.resolveTimeTravel(null, handle, selector(selector, snapshotId, timestamp)).get(); + if ("system".equals(selector)) { + handle = (PaimonTableHandle) metadata.getSysTableHandle(null, handle, "ro").get(); + } + Assertions.assertEquals(current ? "renamed_value" : "value", + metadata.getTableSchema(null, handle, snapshot).getColumns().get(1).getName()); + PaimonTableHandle pinned = (PaimonTableHandle) metadata.applySnapshot(null, handle, snapshot); + PaimonScanPlanProvider provider = new PaimonScanPlanProvider(properties, ops); + Table scan = provider.resolveScanTable(pinned); + String expectedName = current ? "renamed_value" : "value"; + String expectedRow = current ? latestRow : historicalRow; + Assertions.assertEquals(Arrays.asList("id", expectedName), scan.rowType().getFieldNames()); + if (aggregate) { + Assertions.assertEquals("sum", scan.options().get("fields." + expectedName + ".aggregate-function")); + } else { + Assertions.assertEquals(expectedName, scan.options().get("sequence.field")); + } + List expectedRows = "system".equals(selector) + ? systemBaseline : Collections.singletonList(expectedRow); + Assertions.assertEquals(expectedRows, readRows(scan)); + Assertions.assertEquals(expectedRows, readRows(provider.tableForBackend(pinned, scan))); + } + } + + private static ConnectorTimeTravelSpec selector(String selector, long snapshotId, long timestamp) { + switch (selector) { + case "snapshot": + return ConnectorTimeTravelSpec.snapshotId(Long.toString(snapshotId)); + case "timestamp": + return ConnectorTimeTravelSpec.timestamp(Long.toString(timestamp), true); + case "tag": + return ConnectorTimeTravelSpec.tag("before_rename"); + case "system": + case "options": + return ConnectorTimeTravelSpec.options( + Collections.singletonMap("scan.snapshot-id", Long.toString(snapshotId))); + default: + throw new IllegalArgumentException("Unexpected test selector: " + selector); + } + } + + 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 readRows(Table table) throws Exception { + List rows = new ArrayList<>(); + for (Split split : table.newReadBuilder().newScan().plan().splits()) { + try (RecordReader reader = table.newReadBuilder().newRead().createReader(split)) { + reader.forEachRemaining(row -> rows.add(row.getInt(0) + ":" + row.getInt(1))); + } + } + Collections.sort(rows); + return rows; + } +} diff --git a/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out b/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out new file mode 100644 index 00000000000000..ac3aab3611ec26 --- /dev/null +++ b/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out @@ -0,0 +1,101 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !sequence_schema_only_false -- +1 10 base +2 20 retained + +-- !sequence_schema_only_true -- +1 10 base +2 20 retained + +-- !aggregation_schema_only_false -- +1 10 +2 20 + +-- !aggregation_schema_only_true -- +1 10 +2 20 + +-- !sequence_latest_false -- +1 30 updated +2 20 retained +3 15 new + +-- !sequence_old_snapshot_false -- +1 10 base +2 20 retained + +-- !sequence_new_snapshot_false -- +1 30 updated +2 20 retained +3 15 new + +-- !sequence_options_false -- +1 10 base +2 20 retained + +-- !sequence_tag_false -- +1 10 base +2 20 retained + +-- !aggregation_latest_false -- +1 22 +2 27 + +-- !aggregation_old_snapshot_false -- +1 10 +2 20 + +-- !aggregation_intermediate_snapshot_false -- +1 15 +2 23 + +-- !aggregation_options_false -- +1 10 +2 20 + +-- !aggregation_tag_false -- +1 10 +2 20 + +-- !sequence_latest_true -- +1 30 updated +2 20 retained +3 15 new + +-- !sequence_old_snapshot_true -- +1 10 base +2 20 retained + +-- !sequence_new_snapshot_true -- +1 30 updated +2 20 retained +3 15 new + +-- !sequence_options_true -- +1 10 base +2 20 retained + +-- !sequence_tag_true -- +1 10 base +2 20 retained + +-- !aggregation_latest_true -- +1 22 +2 27 + +-- !aggregation_old_snapshot_true -- +1 10 +2 20 + +-- !aggregation_intermediate_snapshot_true -- +1 15 +2 23 + +-- !aggregation_options_true -- +1 10 +2 20 + +-- !aggregation_tag_true -- +1 10 +2 20 + diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy new file mode 100644 index 00000000000000..8b961554be6a98 --- /dev/null +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy @@ -0,0 +1,185 @@ +// 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_renamed_field_time_travel", "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_renamed_field_time_travel" + String dbName = "paimon_renamed_field_time_travel_db" + + def latestSnapshotId = { String tableName -> + def 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' + ) + """ + + try { + spark_paimon_multi """ + CREATE DATABASE IF NOT EXISTS paimon.${dbName}; + DROP TABLE IF EXISTS paimon.${dbName}.sequence_timeline; + CREATE TABLE paimon.${dbName}.sequence_timeline ( + id INT, old_sequence BIGINT, payload STRING + ) USING paimon + TBLPROPERTIES ( + 'primary-key'='id', + 'bucket'='1', + 'file.format'='parquet', + 'sequence.field'='old_sequence' + ); + INSERT INTO paimon.${dbName}.sequence_timeline + VALUES (1, 10, 'base'), (2, 20, 'retained'); + """ + String sequenceBase = latestSnapshotId("sequence_timeline") + spark_paimon_multi """ + CALL paimon.sys.create_tag( + table => '${dbName}.sequence_timeline', tag => 'sequence_base' + ); + ALTER TABLE paimon.${dbName}.sequence_timeline + RENAME COLUMN old_sequence TO current_sequence; + """ + + sql """switch ${catalogName}""" + sql """use ${dbName}""" + sql """refresh table sequence_timeline""" + // A schema-only rename updates sequence.field without creating a data snapshot. + // Latest reads must keep those two pieces of metadata from the same schema. + assertEquals(sequenceBase, latestSnapshotId("sequence_timeline")) + [false, true].each { forceJni -> + sql """set force_jni_scanner=${forceJni}""" + "order_qt_sequence_schema_only_${forceJni}" """ + SELECT id, current_sequence, payload FROM sequence_timeline ORDER BY id + """ + } + + spark_paimon """ + INSERT INTO paimon.${dbName}.sequence_timeline VALUES + (1, 30, 'updated'), (2, 5, 'stale'), (3, 15, 'new') + """ + String sequenceCurrent = latestSnapshotId("sequence_timeline") + sql """refresh table sequence_timeline""" + + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.aggregation_timeline; + CREATE TABLE paimon.${dbName}.aggregation_timeline ( + id INT, original_total BIGINT + ) USING paimon + TBLPROPERTIES ( + 'primary-key'='id', + 'bucket'='1', + 'file.format'='parquet', + 'merge-engine'='aggregation', + 'fields.original_total.aggregate-function'='sum' + ); + INSERT INTO paimon.${dbName}.aggregation_timeline VALUES (1, 10), (2, 20); + """ + String aggregationBase = latestSnapshotId("aggregation_timeline") + spark_paimon_multi """ + CALL paimon.sys.create_tag( + table => '${dbName}.aggregation_timeline', tag => 'aggregation_base' + ); + ALTER TABLE paimon.${dbName}.aggregation_timeline + RENAME COLUMN original_total TO intermediate_total; + """ + sql """refresh catalog ${catalogName}""" + assertEquals(aggregationBase, latestSnapshotId("aggregation_timeline")) + [false, true].each { forceJni -> + sql """set force_jni_scanner=${forceJni}""" + "order_qt_aggregation_schema_only_${forceJni}" """ + SELECT id, intermediate_total FROM aggregation_timeline ORDER BY id + """ + } + spark_paimon """ + INSERT INTO paimon.${dbName}.aggregation_timeline VALUES (1, 5), (2, 3) + """ + String aggregationIntermediate = latestSnapshotId("aggregation_timeline") + spark_paimon_multi """ + ALTER TABLE paimon.${dbName}.aggregation_timeline + RENAME COLUMN intermediate_total TO current_total; + INSERT INTO paimon.${dbName}.aggregation_timeline VALUES (1, 7), (2, 4); + """ + sql """refresh catalog ${catalogName}""" + + // Historical fields must use their historical sequence/aggregation options; + // latest options reference names that do not exist in the old schema. + [false, true].each { forceJni -> + sql """set force_jni_scanner=${forceJni}""" + "order_qt_sequence_latest_${forceJni}" """ + SELECT id, current_sequence, payload FROM sequence_timeline ORDER BY id + """ + "order_qt_sequence_old_snapshot_${forceJni}" """ + SELECT id, old_sequence, payload + FROM sequence_timeline FOR VERSION AS OF ${sequenceBase} ORDER BY id + """ + "order_qt_sequence_new_snapshot_${forceJni}" """ + SELECT id, current_sequence, payload + FROM sequence_timeline FOR VERSION AS OF ${sequenceCurrent} ORDER BY id + """ + "order_qt_sequence_options_${forceJni}" """ + SELECT id, old_sequence, payload + FROM sequence_timeline@options('scan.snapshot-id'='${sequenceBase}') + ORDER BY id + """ + "order_qt_sequence_tag_${forceJni}" """ + SELECT id, old_sequence, payload + FROM sequence_timeline FOR VERSION AS OF 'sequence_base' ORDER BY id + """ + "order_qt_aggregation_latest_${forceJni}" """ + SELECT id, current_total FROM aggregation_timeline ORDER BY id + """ + "order_qt_aggregation_old_snapshot_${forceJni}" """ + SELECT id, original_total + FROM aggregation_timeline FOR VERSION AS OF ${aggregationBase} ORDER BY id + """ + "order_qt_aggregation_intermediate_snapshot_${forceJni}" """ + SELECT id, intermediate_total + FROM aggregation_timeline FOR VERSION AS OF ${aggregationIntermediate} ORDER BY id + """ + "order_qt_aggregation_options_${forceJni}" """ + SELECT id, original_total + FROM aggregation_timeline@options('scan.snapshot-id'='${aggregationBase}') + ORDER BY id + """ + "order_qt_aggregation_tag_${forceJni}" """ + SELECT id, original_total + FROM aggregation_timeline@tag(aggregation_base) ORDER BY id + """ + } + } finally { + sql """set force_jni_scanner=false""" + } +} From 61e929f76396937c87f49ec132aa46fb157a6abb Mon Sep 17 00:00:00 2001 From: Gabriel Date: Fri, 23 Oct 2026 21:16:11 +0800 Subject: [PATCH 2/2] [fix](paimon) Preserve effective read state across schema restoration ### What problem does this PR solve? Historical schema options could revive a removed startup mode or obsolete primary-branch policy. The former rejects valid snapshot reads; the latter changes selected rows when manifest parallelism normalization rebuilds a fallback reader. Preserve effective selectors and branch policy, including absent keys, while restoring historical field options. ### Release note Fix historical Paimon reads after removing startup options or changing primary/fallback branch priority. ### Check List (For Author) - Test: Full Paimon connector install and Checkstyle; real-table unit tests for removed scan mode and overlapping branches at capacities 64/128; cluster regression with both scanner settings. - Behavior changed: Yes. Historical schema restoration retains effective read state and branch policy. - Does this need documentation: No. --- .../connector/paimon/PaimonScanParams.java | 18 +++++ .../PaimonHistoricalSchemaOptionsTest.java | 66 +++++++++++++++++++ .../test_paimon_renamed_field_time_travel.out | 6 ++ ...st_paimon_renamed_field_time_travel.groovy | 22 +++++++ 4 files changed, 112 insertions(+) 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 a145def09dd155..a05adb64f03a2d 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 @@ -105,6 +105,13 @@ public final class PaimonScanParams { private static final Set INHERITED_READ_STATE_KEYS = buildInheritedReadStateKeys(); + private static final Set SCHEMA_INDEPENDENT_READ_KEYS = ImmutableSet.builder() + .addAll(INHERITED_READ_STATE_KEYS) + .add(CoreOptions.BRANCH.key()) + .add(CoreOptions.SCAN_PRIMARY_BRANCH.key()) + .add(CoreOptions.SCAN_FALLBACK_BRANCH.key()) + .build(); + private static final Set TIME_TRAVEL_SELECTOR_KEYS = ImmutableSet.of( CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), CoreOptions.SCAN_WATERMARK.key(), @@ -331,6 +338,17 @@ private static FileStoreTable restoreBoundSchema( merged.put(key, value); } }); + // Read selectors and branch topology belong to the resolved table, not its historical schema. + // Preserve absent keys too: reviving an old mode can invalidate a snapshot selector, while + // reviving an old branch policy can change fallback priority when reader caps rebuild the pair. + SCHEMA_INDEPENDENT_READ_KEYS.forEach(key -> { + String value = table.options().get(key); + if (value == null) { + merged.remove(key); + } else { + merged.put(key, value); + } + }); // Explicit catalog overrides can equal old physical values; equality is not provenance. options.forEach((key, value) -> { if (key.startsWith(INTERNAL_PREFIX + "catalog-option.")) { diff --git a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java index d0069936b67447..3ab8f855e38a5d 100644 --- a/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java +++ b/fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonHistoricalSchemaOptionsTest.java @@ -55,6 +55,72 @@ import java.util.Map; public class PaimonHistoricalSchemaOptionsTest { + @Test + public void snapshotReadDoesNotRestoreRemovedStartupMode(@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", "removed_startup_mode"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("value", DataTypes.INT()).primaryKey("id") + .option("bucket", "1").option("file.format", "parquet") + .option("scan.mode", "latest-full").build(), false); + append((FileStoreTable) catalog.getTable(id), GenericRow.of(1, 10)); + catalog.alterTable(id, Collections.singletonList(SchemaChange.removeOption("scan.mode")), false); + Table latest = catalog.getTable(id); + Assertions.assertFalse(latest.options().containsKey("scan.mode")); + Map selector = Collections.singletonMap("scan.snapshot-id", "1"); + // Removing a startup mode must remain effective when a historical schema is restored. + Assertions.assertEquals(Collections.singletonList("1:10"), readRows(latest.copy(selector))); + Table historical = PaimonScanParams.copyWithHistoricalSchema(latest, selector); + Assertions.assertEquals(Collections.singletonList("1:10"), readRows(historical)); + } + } + + @Test + public void historicalFallbackPriorityIsStableAt64(@TempDir Path warehouse) throws Exception { + checkHistoricalFallbackPriority(warehouse, 64); + } + + @Test + public void historicalFallbackPriorityIsStableAt128(@TempDir Path warehouse) throws Exception { + checkHistoricalFallbackPriority(warehouse, 128); + } + + private void checkHistoricalFallbackPriority(Path warehouse, int capacity) 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", "branch_priority_history"); + catalog.createTable(id, Schema.newBuilder().column("id", DataTypes.INT().notNull()) + .column("value", DataTypes.INT().notNull()).primaryKey("id", "value").partitionKeys("id") + .option("bucket", "1").option("file.format", "parquet").build(), false); + FileStoreTable original = (FileStoreTable) catalog.getTable(id); + original.createBranch("backup"); + Identifier backupId = new Identifier("db", "branch_priority_history", "backup"); + append((FileStoreTable) catalog.getTable(backupId), GenericRow.of(1, 20)); + catalog.alterTable(id, Arrays.asList(SchemaChange.setOption("scan.primary-branch", "backup"), + SchemaChange.setOption("scan.manifest.parallelism", "128")), false); + FallbackReadFileStoreTable primary = (FallbackReadFileStoreTable) catalog.getTable(id); + append(primary.wrapped(), GenericRow.of(1, 10)); + Assertions.assertEquals(Collections.singletonList("1:20"), readRows(primary)); + catalog.alterTable(id, Arrays.asList(SchemaChange.removeOption("scan.primary-branch"), + SchemaChange.setOption("scan.fallback-branch", "backup")), false); + FallbackReadFileStoreTable latest = (FallbackReadFileStoreTable) catalog.getTable(id); + Assertions.assertEquals(Collections.singletonList("1:10"), readRows(latest)); + Table historical = PaimonScanParams.copyWithHistoricalSchema(latest, + Collections.singletonMap("scan.snapshot-id", "1")); + Assertions.assertEquals("backup", historical.options().get("scan.fallback-branch")); + Assertions.assertFalse(historical.options().containsKey("scan.primary-branch")); + // The same overlapping partition must choose the same branch on different FE capacities. + Table normalized = PaimonReaderOptions.runtimeSafeTable(historical, capacity); + Assertions.assertEquals(Integer.toString(capacity), normalized.options().get("scan.manifest.parallelism")); + Assertions.assertEquals("backup", normalized.options().get("scan.fallback-branch")); + Assertions.assertFalse(normalized.options().containsKey("scan.primary-branch")); + Assertions.assertEquals(Collections.singletonList("1:10"), readRows(normalized)); + } + } + @Test public void snapshotUsesHistoricalSequenceFieldAfterRename(@TempDir Path warehouse) throws Exception { checkRenamedField(warehouse, false, "snapshot"); diff --git a/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out b/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out index ac3aab3611ec26..881f364cfa0465 100644 --- a/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out +++ b/regression-test/data/external_table_p0/paimon/test_paimon_renamed_field_time_travel.out @@ -57,6 +57,9 @@ 1 10 2 20 +-- !removed_scan_mode_false -- +1 10 + -- !sequence_latest_true -- 1 30 updated 2 20 retained @@ -99,3 +102,6 @@ 1 10 2 20 +-- !removed_scan_mode_true -- +1 10 + diff --git a/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy b/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy index 8b961554be6a98..5151cc45934829 100644 --- a/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy +++ b/regression-test/suites/external_table_p0/paimon/test_paimon_renamed_field_time_travel.groovy @@ -134,6 +134,23 @@ suite("test_paimon_renamed_field_time_travel", "p0,external,paimon") { """ sql """refresh catalog ${catalogName}""" + spark_paimon_multi """ + DROP TABLE IF EXISTS paimon.${dbName}.scan_state_history; + CREATE TABLE paimon.${dbName}.scan_state_history ( + id INT, value BIGINT + ) USING paimon + TBLPROPERTIES ( + 'primary-key'='id', 'bucket'='1', 'file.format'='parquet', + 'scan.mode'='latest-full' + ); + INSERT INTO paimon.${dbName}.scan_state_history VALUES (1, 10); + """ + String scanStateSnapshot = latestSnapshotId("scan_state_history") + spark_paimon """ + ALTER TABLE paimon.${dbName}.scan_state_history UNSET TBLPROPERTIES ('scan.mode') + """ + sql """refresh catalog ${catalogName}""" + // Historical fields must use their historical sequence/aggregation options; // latest options reference names that do not exist in the old schema. [false, true].each { forceJni -> @@ -178,6 +195,11 @@ suite("test_paimon_renamed_field_time_travel", "p0,external,paimon") { SELECT id, original_total FROM aggregation_timeline@tag(aggregation_base) ORDER BY id """ + // Historical schema restoration must not revive a removed startup mode. + "order_qt_removed_scan_mode_${forceJni}" """ + SELECT id, value FROM scan_state_history + FOR VERSION AS OF ${scanStateSnapshot} ORDER BY id + """ } } finally { sql """set force_jni_scanner=false"""