Gabriel39 · Oct 23, 2026 · Oct 23, 2026 · github-actions · Oct 23, 2026
@@ -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<String, String> optionsForCopy = PaimonTableOptions.forCopy(tableOptions);
-            return optionsForCopy.isEmpty() ? table : table.copy(optionsForCopy);
+            return optionsForCopy.isEmpty() ? table : PaimonScanParams.copyWithHistoricalSchema(table, optionsForCopy);
         }
 
         @Override

@@ -420,15 +420,11 @@ private Table resolveSystemTableAt(ConnectorSession session,
                 ? paimonHandle
                 : (PaimonTableHandle) applySnapshot(session, paimonHandle, snapshot);
         Map<String, String> 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<ConnectorMvccSnapshot> 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<ConnectorPartitionInfo> collectPartitions(PaimonTableHandle paimonHandle) {
+        if (paimonHandle.getScanOptions().isEmpty()) {
+            return collectPartitionsAuthenticated(paimonHandle);
+        }
+        return readMetadataAuthenticated(() -> collectPartitionsAuthenticated(paimonHandle));
+    }
+
+    private List<ConnectorPartitionInfo> 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<ConnectorPartitionInfo> collectPartitions(PaimonTableHandle paimonH
         }
 
         Table resolvedTable = resolveTable(paimonHandle);
-        Map<String, String> scanOptions = paimonHandle.getScanOptions();
+        Map<String, String> scanOptions = PaimonScanParams.withCatalogOptions(
+                paimonHandle.getScanOptions(), PaimonTableOptions.extractCompatible(catalogProperties.getRaw()));
         boolean optionsPin = PaimonScanParams.isOptionsPin(scanOptions);
         Table table;
         if (optionsPin) {
@@ -1778,8 +1783,9 @@ private List<ConnectorPartitionInfo> 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<ConnectorTableStatistics> getTableStatistics(
         if (snapshot == null) {
             return getTableStatistics(session, handle);
         }
+        return readMetadataAuthenticated(() -> getTableStatisticsAtSnapshot(session, handle, snapshot));
+    }
+
+    private Optional<ConnectorTableStatistics> getTableStatisticsAtSnapshot(
+            ConnectorSession session, ConnectorTableHandle handle, ConnectorMvccSnapshot snapshot) {
         long rowCount;
         try {
             PaimonTableHandle pinned = (PaimonTableHandle) applySnapshot(session, handle, snapshot);
@@ -1957,17 +1968,18 @@ public Optional<ConnectorTableStatistics> getTableStatistics(
                 // first commit even though execution is still required to scan zero rows.
                 return Optional.empty();
             }
-            Map<String, String> scanOptions = pinned.getScanOptions();
+            Map<String, String> 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);

@@ -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<String, Strin
         if (scanOptions != null && !scanOptions.isEmpty()) {
             // Incremental ranges are relation state too; rebuilding a capped system wrapper from the
             // undecorated source must not silently turn @incr back into a latest scan.
-            effectiveSource = sourceTable.copy(
+            effectiveSource = PaimonScanParams.copyWithHistoricalSchema(sourceTable,
                     PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
         }
         return runtimeSafeTable(effectiveSource);

@@ -24,7 +24,9 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.options.ConfigOption;
 import org.apache.paimon.options.FallbackKey;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.ChainGroupReadTable;
 import org.apache.paimon.table.FallbackReadFileStoreTable;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
@@ -103,6 +105,13 @@ public final class PaimonScanParams {
 
     private static final Set<String> INHERITED_READ_STATE_KEYS = buildInheritedReadStateKeys();
 
+    private static final Set<String> SCHEMA_INDEPENDENT_READ_KEYS = ImmutableSet.<String>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<String> TIME_TRAVEL_SELECTOR_KEYS = ImmutableSet.of(
             CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
             CoreOptions.SCAN_WATERMARK.key(),
@@ -204,13 +213,60 @@ public static Table applyOptions(Table table, Map<String, String> 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<String, String> options) {
+        return copyWithHistoricalSchema(table, userOptions(options), options);
+    }
+
+    private static Table copyWithHistoricalSchema(
+            Table table, Map<String, String> dynamicOptions, Map<String, String> 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<String, String> effectiveOptions = new HashMap<>(table.options());
+        dynamicOptions.forEach((key, value) -> {
+            if (value == null) {
+                effectiveOptions.remove(key);
+            } else {
+                effectiveOptions.put(key, value);
+            }
+        });
+        Optional<Snapshot> 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<String, String> options) {
         Map<String, String> tableOptions = userOptions(options);
@@ -282,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.")) {
@@ -353,6 +420,11 @@ public static boolean usesStatementSnapshot(Map<String, String> options) {
     }
 
     public static Map<String, String> resolveOptions(Table table, Map<String, String> options) {
+        return resolveOptions(table, options, java.util.Collections.emptyMap());
+    }
+
+    static Map<String, String> resolveOptions(
+            Table table, Map<String, String> options, Map<String, String> catalogOptions) {
         validateOptions(options);
         if (!hasStartupOptions(options)) {
             return options;
@@ -388,7 +460,7 @@ public static Map<String, String> resolveOptions(Table table, Map<String, String
             return resolveFileCreationTime(options, fileStoreTable, creationTime);
         }
 
-        Table selectedTable = applyOptions(table, options);
+        Table selectedTable = applyOptions(table, withCatalogOptions(options, catalogOptions));
         if ("compacted-full".equalsIgnoreCase(options.get(CoreOptions.SCAN_MODE.key()))) {
             Long snapshotId = compactedFullSnapshotId((FileStoreTable) selectedTable);
             return snapshotId == null
@@ -508,9 +580,13 @@ private static Map<String, String> resolvedEmptyOptions(Map<String, String> opti
     }
 
     private static Map<String, String> userOptions(Map<String, String> options) {
-        return options.entrySet().stream()
-                .filter(entry -> !entry.getKey().startsWith(INTERNAL_PREFIX))
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        Map<String, String> result = new HashMap<>();
+        options.forEach((key, value) -> {
+            if (!key.startsWith(INTERNAL_PREFIX)) {
+                result.put(key, value);
+            }
+        });
+        return result;
     }
 
     /**

@@ -361,10 +361,11 @@ Table resolveScanTable(PaimonTableHandle paimonHandle) {
     }
 
     private <T> T withBoundSchemaAuthentication(PaimonTableHandle handle, Supplier<T> 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> T withBoundSchemaAuthentication(PaimonTableHandle handle, Supplier<T
 
     private Map<String, String> effectiveScanOptions(PaimonTableHandle handle) {
         Map<String, String> 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<String, String> 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<String, String> 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());
     }