github-actions[bot] commented on code in PR #65984: URL: https://github.com/apache/doris/pull/65984#discussion_r3654135777
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java: ########## @@ -0,0 +1,368 @@ +// 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 = ImmutableSet.of( + CoreOptions.SCAN_MODE.key(), + 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(), + CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), + CoreOptions.SCAN_PLAN_SORT_PARTITION.key()); + + 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(); + + // FilesScan enumerates the latest partitions before applying its range-aware per-partition scan, + // so it cannot safely read a range when a partition in that range has since been dropped. + private static final Set<String> INCREMENTAL_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "partitions", "ro", "row_tracking"); Review Comment: [P1] Do not push COUNT(*) past incremental binlog packing Adding `binlog` to the INCR allowlist exposes a count path with different row semantics. Paimon's incremental delta splits are streaming, so `BinlogRead` uses `PackChangelogReader` to combine an `UPDATE_BEFORE`/`UPDATE_AFTER` pair into one logical `+U` binlog row. `PaimonScanNode`, however, handles `COUNT(*)` before `forceJniForSystemTable` and sums `DataSplit.mergedRowCount()`, which counts the physical changelog records. A raw-convertible update pair can therefore return one row from SELECT but `COUNT(*) = 2`. Please disable this pushdown for incremental binlog (or obtain the packed logical count) and add a paired-update regression over the same INCR range. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java: ########## @@ -0,0 +1,368 @@ +// 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 = ImmutableSet.of( + CoreOptions.SCAN_MODE.key(), + 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(), + CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), + CoreOptions.SCAN_PLAN_SORT_PARTITION.key()); + + 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(); + + // FilesScan enumerates the latest partitions before applying its range-aware per-partition scan, + // so it cannot safely read a range when a partition in that range has since been dropped. + private static final Set<String> INCREMENTAL_SYSTEM_TABLES = ImmutableSet.of( + "audit_log", "binlog", "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); + } + + String scanMode = options.get(CoreOptions.SCAN_MODE.key()); + if ("from-creation-timestamp".equalsIgnoreCase(scanMode) + && options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()) == null) { + // Paimon 1.3.1 does not validate this newer mode, but its starting scanner + // requires the creation timestamp and otherwise fails after analysis. + throw new IllegalArgumentException("Paimon scan mode 'from-creation-timestamp' requires query option '" + + CoreOptions.SCAN_CREATION_TIME_MILLIS.key() + "'."); + } + + 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 = scanMode.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); + if (snapshot == null) { + return resolvedEmptyOptions(options); + } + return resolvedSnapshotOptions(options, String.valueOf(snapshot.id())); Review Comment: [P1] Preserve retained tag metadata when pinning the relation Resolving a tag to `snapshot.id()` is not equivalent after snapshot expiration. Paimon's tag scanner reads the `Snapshot` copy retained in the tag directory, while the later `scan.snapshot-id` path reopens `/snapshot/snapshot-<id>` and fails once that ordinary snapshot file has expired. Thus an unchanged retained tag remains readable through normal Paimon time travel but fails through `@options('scan.tag-name'=...)` (and tag-valued `scan.version`). Please carry the resolved tag `Snapshot` through binding/planning instead of reducing it to an ID, and add OPTIONS tag/version reads after the suite's existing `expire_snapshots` call. -- 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]
