github-actions[bot] commented on code in PR #65984: URL: https://github.com/apache/doris/pull/65984#discussion_r3651741321
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java: ########## @@ -0,0 +1,349 @@ +// 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.FullCompactedStartingScanner; +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"); Review Comment: [P1] Do not advertise incremental files until enumeration is range-aware Paimon 1.3.1 plans `$files` in two stages: `FilesScan.innerPlan()` first calls `fileStoreTable.newSnapshotReader().partitions()` with no starting scanner or pinned snapshot, and only each resulting `FilesSplit` is later replanned through the option-bearing `storeTable.newScan()`. If partition `p` contributes a file in `(start,end]` and is dropped after `end`, the first stage enumerates current/latest state, emits no split for `p`, and the incremental scan silently omits that file. The new fixture is unpartitioned, so its count cannot catch this. Please remove `files` from this allowlist until the initial enumeration is derived from the requested range/end snapshot, or pin that stage too; cover a range partition that is absent from latest. ########## fe/fe-core/src/main/java/org/apache/doris/analysis/TableScanParams.java: ########## @@ -88,6 +96,30 @@ public Map<String, String> getMapParams() { return mapParams; } + /** + * Resolve dynamic parameters once for the lifetime of this relation. + */ + public Map<String, String> getOrResolveMapParams( + Function<Map<String, String>, Map<String, String>> resolver) { + Map<String, String> resolved = resolvedMapParams; + if (resolved == null) { + synchronized (this) { + resolved = resolvedMapParams; + if (resolved == null) { Review Comment: [P1] Keep resolved OPTIONS state execution-scoped `PREPARE` analyzes the retained inner plan to produce result metadata, so a relation such as `p@options('scan.mode'='latest')` reaches this assignment during PREPARE and caches that snapshot ID. `PreparedStatementContext` then keeps the same `PrepareCommand`, and `ExecuteCommand` replans `prepareCommand.getLogicalPlan()` without recreating or clearing its `TableScanParams`. A commit after PREPARE therefore remains invisible to the first and every later EXECUTE; mutable tags, time-relative selectors, and the pinned-empty marker leak across executions in the same way. Please keep this resolution in per-execution state (or recreate the relation scan state for each EXECUTE), and add a prepared-query regression with commits before the first EXECUTE and between two EXECUTEs. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java: ########## @@ -0,0 +1,349 @@ +// 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.FullCompactedStartingScanner; +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( + // A system table may advertise OPTIONS only when every row-producing stage observes + // the selected snapshot; files and buckets still consult latest metadata internally. + "audit_log", "binlog", "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); + if ("compacted-full".equalsIgnoreCase(options.get(CoreOptions.SCAN_MODE.key()))) { + Long snapshotId = compactedFullSnapshotId((FileStoreTable) selectedTable); + return snapshotId == null + ? resolvedEmptyOptions(options) + : resolvedSnapshotOptions(options, String.valueOf(snapshotId)); + } + Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest((FileStoreTable) selectedTable); Review Comment: [P2] Require a time for from-creation-timestamp `@options('scan.mode'='from-creation-timestamp')` reaches this fallback with no `scan.creation-time-millis`. Paimon 1.3.1's copy-time `SchemaValidation` happens not to validate this newer startup mode, and `TimeTravelUtil.tryTravelOrLatest()` ignores the mode itself, so Doris pins latest and removes the requested mode instead of reporting the missing position. Paimon's real batch starting scanner requires/unboxes that creation time. Please require `scan.creation-time-millis` for this mode and add a negative query test for the mode-only form. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java: ########## @@ -0,0 +1,349 @@ +// 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.FullCompactedStartingScanner; +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.")) Review Comment: [P2] Reject scan options that the batch path cannot consume Deriving support from every canonical `scan.*` key also admits streaming/enumerator-only settings. In Paimon 1.3.1, `scan.bounded.watermark` is read only by `DataTableStreamScan.createBoundedChecker()`, and `scan.max-splits-per-task` controls a source enumerator; Doris creates a batch `TableScan`, which consumes neither. Both OPTIONS therefore succeed while returning the same ordinary batch rows. Please use an allowlist for settings honored by Doris's actual batch planning/read path and reject these no-op keys, with negative coverage for at least `scan.bounded.watermark`. -- 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]
