github-actions[bot] commented on code in PR #65984: URL: https://github.com/apache/doris/pull/65984#discussion_r3650617547
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java: ########## @@ -0,0 +1,323 @@ +// 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.datasource.paimon; + +import org.apache.doris.analysis.TableScanParams; + +import com.google.common.collect.ImmutableSet; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.FallbackKey; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.source.snapshot.TimeTravelUtil; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Validation and application rules for relation-scoped Paimon scan parameters. + */ +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 Set<String> QUERY_OPTION_KEYS = CoreOptions.getOptions().stream() + .map(option -> option.key()) + .filter(key -> key.startsWith("scan.")) + .collect(Collectors.toSet()); + + private static final Set<String> STARTUP_POSITION_KEYS = ImmutableSet.of( + CoreOptions.SCAN_TIMESTAMP.key(), + CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), + CoreOptions.SCAN_WATERMARK.key(), + CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), + CoreOptions.SCAN_CREATION_TIME_MILLIS.key(), + CoreOptions.SCAN_SNAPSHOT_ID.key(), + CoreOptions.SCAN_TAG_NAME.key(), + CoreOptions.SCAN_VERSION.key()); + + private static final Set<String> INHERITED_READ_STATE_KEYS = inheritedReadStateKeys(); + + private static final Set<String> INCREMENTAL_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "files", "partitions", "ro", "row_tracking"); + + private static final Set<String> PAIMON_READER_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "row_tracking"); + + private static final Set<String> OPTIONS_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "buckets", "files", "manifests", "partitions", "ro", + "row_tracking", "table_indexes"); + + private PaimonScanParams() { + } + + public static void validateOptions(Map<String, String> options) { + if (options.containsKey(CoreOptions.SCAN_FALLBACK_BRANCH.key())) { + throw new IllegalArgumentException("Paimon query option '" + + CoreOptions.SCAN_FALLBACK_BRANCH.key() + + "' is not supported because it requires rebuilding the table through the catalog factory."); + } + + Set<String> unsupported = options.keySet().stream() + .filter(key -> !QUERY_OPTION_KEYS.contains(key)) + .collect(Collectors.toSet()); + if (!unsupported.isEmpty()) { + throw new IllegalArgumentException("Unsupported Paimon query option(s): " + unsupported); + } + + long positionCount = options.keySet().stream().filter(STARTUP_POSITION_KEYS::contains).count(); + if (positionCount > 1) { + throw new IllegalArgumentException( + "Only one Paimon startup position can be specified: " + STARTUP_POSITION_KEYS); + } + + if (options.containsKey(CoreOptions.SCAN_MODE.key()) && positionCount == 1) { + String position = options.keySet().stream() + .filter(STARTUP_POSITION_KEYS::contains) + .findFirst() + .get(); + String mode = options.get(CoreOptions.SCAN_MODE.key()).toLowerCase(Locale.ROOT); + if (!isCompatibleStartupMode(position, mode)) { + throw new IllegalArgumentException("Paimon scan mode '" + mode + + "' is incompatible with startup position '" + position + "'."); + } + } + } + + public static Table applyOptions(Table table, Map<String, String> options) { + Map<String, String> tableOptions = userOptions(options); + validateOptions(tableOptions); + Map<String, String> isolatedOptions = new HashMap<>(tableOptions); + if (hasStartupOptions(tableOptions)) { + // Startup mode, position, and range form one inherited state family in Paimon. Clear + // absent members so one relation cannot accidentally reuse another relation's read state. + INHERITED_READ_STATE_KEYS.stream() + .filter(key -> !tableOptions.containsKey(key)) + .forEach(key -> isolatedOptions.put(key, null)); + } + return table.copy(isolatedOptions); + } + + private static Set<String> inheritedReadStateKeys() { + ImmutableSet.Builder<String> keys = ImmutableSet.builder(); + for (ConfigOption<?> option : Arrays.asList( + CoreOptions.SCAN_TIMESTAMP, + CoreOptions.SCAN_TIMESTAMP_MILLIS, + CoreOptions.SCAN_WATERMARK, + CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS, + CoreOptions.SCAN_CREATION_TIME_MILLIS, + CoreOptions.SCAN_SNAPSHOT_ID, + CoreOptions.SCAN_TAG_NAME, + CoreOptions.SCAN_VERSION, + CoreOptions.SCAN_MODE, + CoreOptions.SCAN_BOUNDED_WATERMARK, + CoreOptions.INCREMENTAL_BETWEEN, + CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP, + CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE, + CoreOptions.INCREMENTAL_TO_AUTO_TAG)) { + keys.add(option.key()); + for (FallbackKey fallbackKey : option.fallbackKeys()) { + keys.add(fallbackKey.getKey()); + } + } + return keys.build(); + } + + public static boolean hasStartupOptions(Map<String, String> options) { + return options.containsKey(CoreOptions.SCAN_MODE.key()) + || options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains); + } + + public static boolean selectsSchema(Map<String, String> options) { + return hasStartupOptions(options); + } + + public static boolean usesStatementSnapshot(Map<String, String> options) { + if (options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains)) { + return false; + } + String mode = options.get(CoreOptions.SCAN_MODE.key()); + return mode == null + || "default".equalsIgnoreCase(mode) + || "latest".equalsIgnoreCase(mode) + || "latest-full".equalsIgnoreCase(mode) + || "full".equalsIgnoreCase(mode); + } + + public static Map<String, String> resolveOptions(Table table, Map<String, String> options) { + validateOptions(options); + if (!hasStartupOptions(options)) { + return options; + } + if (usesStatementSnapshot(options)) { + String pinnedSnapshotId = table.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key()); + if (pinnedSnapshotId != null) { + return resolvedSnapshotOptions(options, pinnedSnapshotId); + } + } + if (!(table instanceof FileStoreTable)) { + throw new IllegalArgumentException("Paimon startup options require a file-store data table."); + } + FileStoreTable fileStoreTable = (FileStoreTable) table; + if (options.containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())) { + return resolveFileCreationTime( + options, + fileStoreTable, + Long.parseLong(options.get(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key()))); + } + if (options.containsKey(CoreOptions.SCAN_CREATION_TIME_MILLIS.key())) { + long creationTime = Long.parseLong(options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key())); + Long previousSnapshotId = TimeTravelUtil.earlierThanTimeMills( + fileStoreTable.snapshotManager(), + fileStoreTable.changelogManager(), + creationTime, + fileStoreTable.coreOptions().changelogLifecycleDecoupled(), + true); + if (previousSnapshotId != null + && fileStoreTable.snapshotManager().snapshotExists(previousSnapshotId + 1)) { + return resolvedSnapshotOptions(options, String.valueOf(previousSnapshotId + 1)); + } + return resolveFileCreationTime(options, fileStoreTable, creationTime); + } + + Table selectedTable = applyOptions(table, options); + Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest((FileStoreTable) selectedTable); Review Comment: [P1] Preserve compacted-full startup semantics `@options('scan.mode'='compacted-full')` reaches `TimeTravelUtil.tryTravelOrLatest`, but that Paimon 1.3.1 helper ignores `scan.mode` and therefore returns the latest snapshot. Paimon's normal batch path instead dispatches this mode to `CompactedStartingScanner`/`FullCompactedStartingScanner`, which can select an older compaction snapshot. `resolvedSnapshotOptions` then removes the mode, so planning cannot recover the intended result. Please resolve this mode through its actual starting scanner or reject it for relation OPTIONS, and add a compaction-followed-by-append test that distinguishes compacted from latest. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java: ########## @@ -0,0 +1,323 @@ +// 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.datasource.paimon; + +import org.apache.doris.analysis.TableScanParams; + +import com.google.common.collect.ImmutableSet; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.Snapshot; +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.FallbackKey; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.table.source.snapshot.TimeTravelUtil; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * Validation and application rules for relation-scoped Paimon scan parameters. + */ +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 Set<String> QUERY_OPTION_KEYS = CoreOptions.getOptions().stream() + .map(option -> option.key()) + .filter(key -> key.startsWith("scan.")) + .collect(Collectors.toSet()); + + private static final Set<String> STARTUP_POSITION_KEYS = ImmutableSet.of( + CoreOptions.SCAN_TIMESTAMP.key(), + CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), + CoreOptions.SCAN_WATERMARK.key(), + CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(), + CoreOptions.SCAN_CREATION_TIME_MILLIS.key(), + CoreOptions.SCAN_SNAPSHOT_ID.key(), + CoreOptions.SCAN_TAG_NAME.key(), + CoreOptions.SCAN_VERSION.key()); + + private static final Set<String> INHERITED_READ_STATE_KEYS = inheritedReadStateKeys(); + + private static final Set<String> INCREMENTAL_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "files", "partitions", "ro", "row_tracking"); + + private static final Set<String> PAIMON_READER_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "row_tracking"); + + private static final Set<String> OPTIONS_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "buckets", "files", "manifests", "partitions", "ro", Review Comment: [P1] Accept OPTIONS only when the whole system scan is snapshot-aware Two entries here can silently return the wrong historical metadata in Paimon 1.3.1. `$buckets` calls `newSnapshotReader().bucketEntries()` directly; that reader creates `store().newScan()` without a starting scanner or `withSnapshot`, so the copied `scan.snapshot-id` selects a schema but the bucket entries still come from latest. `$files` replans each emitted split at the selected snapshot, but `FilesScan` first enumerates partitions through the same latest `SnapshotReader`, so a partition present only in the requested snapshot is never emitted. The new bucket assertions only check non-null results and the file fixture has no old-only partition, so both substitutions pass. Please reject these forms or explicitly pin their enumeration/readers, with snapshot-distinguishing bucket and dropped-partition file tests. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java: ########## @@ -607,11 +637,36 @@ public List<org.apache.paimon.table.source.Split> getPaimonSplitFromAPI() throws long startTime = System.currentTimeMillis(); try { Table paimonTable = getProcessedTable(); + Map<String, String> resolvedOptions = scanParams == null + ? Collections.emptyMap() + : scanParams.getResolvedMapParams().orElse(Collections.emptyMap()); + if (PaimonScanParams.isPinnedEmptyScan(resolvedOptions)) { + return Collections.emptyList(); + } + Optional<Long> fileCreationTime = PaimonScanParams.getPinnedFileCreationTime(resolvedOptions); + if (fileCreationTime.isPresent()) { + if (!(paimonTable instanceof FileStoreTable)) { + throw new UserException("Paimon file-creation OPTIONS require a data table."); + } + FileStoreTable fileStoreTable = (FileStoreTable) paimonTable; + SnapshotReader snapshotReader = fileStoreTable.newSnapshotReader() Review Comment: [P1] Preserve DataTableBatchScan filters here Building a raw `SnapshotReader` bypasses correctness configuration that Paimon 1.3.1 installs in `DataTableBatchScan` before its starting scanner runs. For primary-key tables with deletion vectors or `FIRST_ROW` in batch mode `NONE`, the normal path applies `level > 0` plus value filtering; for postponed-bucket tables it applies `onlyReadRealBuckets()`. This creation-time path can therefore plan level-0 or `bucket=-1` files that an ordinary batch scan excludes, changing or duplicating returned rows. Please preserve the normal batch-scan configuration while pinning the snapshot, and add result tests for skip-level-0 and postponed-bucket tables. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java: ########## @@ -502,7 +505,11 @@ public void registerExternalTableForPreload(TableIf table, Optional<TableSnapsho } ExternalTablePreloadInfo preloadInfo = externalTablePreloadInfos.computeIfAbsent(table.getId(), id -> new ExternalTablePreloadInfo((ExternalTable) table)); - if (tableSnapshot.isPresent() || scanParams.isPresent()) { + boolean selectorFreePaimonOptions = scanParams.isPresent() + && scanParams.get().isOptions() + && (table instanceof PaimonExternalTable || table instanceof PaimonSysExternalTable) + && PaimonScanParams.usesStatementSnapshot(scanParams.get().getMapParams()); + if (tableSnapshot.isPresent() || (scanParams.isPresent() && !selectorFreePaimonOptions)) { Review Comment: [P2] Let mixed aliases preload the latest state This correctly marks a selector-free OPTIONS alias as latest-equivalent, but preload state is aggregated by table ID and `shouldPreloadLatestSnapshot()` requires `hasLatestOnlyRelation && !hasNonLatestRelation`. If the same Paimon table also appears as `@options('scan.snapshot-id'='1')`, the historical alias therefore cancels snapshot, schema, and partition warmup needed by the latest/selector-free alias. `BindRelation` then performs that alias's first Paimon load after unrelated internal-table locks are acquired. Please preload latest state whenever any latest-equivalent relation exists and track historical resolution separately; add a mixed latest + historical alias lock-order test. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
