Gabriel39 commented on code in PR #66297: URL: https://github.com/apache/doris/pull/66297#discussion_r3690582107
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonReaderOptions.java: ########## @@ -0,0 +1,210 @@ +// 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 com.google.common.collect.ImmutableSet; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** Validation shared by catalog-scoped and relation-scoped Paimon reader tuning. */ +public final class PaimonReaderOptions { + public static final String TABLE_OPTION_PREFIX = "paimon.table-option."; + public static final int MIN_READ_BATCH_SIZE = 1; + public static final int MAX_READ_BATCH_SIZE = 65536; + public static final long MIN_ASYNC_THRESHOLD_BYTES = 1024L * 1024L; + public static final long MAX_ASYNC_THRESHOLD_BYTES = 1024L * 1024L * 1024L; + + // Keep this list to batch-read controls consumed by Doris' Paimon scan path. Context selectors, + // streaming-source settings, storage layout, and write options are unsafe after schema binding. + private static final Set<String> SUPPORTED_OPTIONS = ImmutableSet.of( + CoreOptions.READ_BATCH_SIZE.key(), Review Comment: Valid. The current head preserves validated read.batch-size through scanner initialization and retains the old-sender guard. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java: ########## @@ -190,6 +190,19 @@ private PaimonSnapshotCacheValue getPaimonSnapshotCacheValue(Optional<TableSnaps "Failed to get Paimon snapshot: " + (e.getMessage() == null ? "unknown cause" : e.getMessage()), e); } + } else if (scanParams.isPresent() && scanParams.get().isOptions()) { Review Comment: Valid. The current head keeps option-bearing projections relation-scoped and covers both alias orders. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonPartitionInfoLoader.java: ########## @@ -44,6 +45,7 @@ public PaimonPartitionInfo load(NameMapping nameMapping, Table snapshotTable, Li // maps null and blank strings to the same partition.default-name there, so those // specs cannot be used as logical partition identities. PartitionEntry retains the // typed BinaryRow and is also bound to the data snapshot represented by this table. Review Comment: Valid. The current head avoids neutral pre-lock preload before relation options are available. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonReaderOptions.java: ########## @@ -0,0 +1,210 @@ +// 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 com.google.common.collect.ImmutableSet; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.DelegatedFileStoreTable; +import org.apache.paimon.table.FallbackReadFileStoreTable; +import org.apache.paimon.table.Table; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Set; + +/** Validation shared by catalog-scoped and relation-scoped Paimon reader tuning. */ +public final class PaimonReaderOptions { + public static final String TABLE_OPTION_PREFIX = "paimon.table-option."; + public static final int MIN_READ_BATCH_SIZE = 1; + public static final int MAX_READ_BATCH_SIZE = 65536; + public static final long MIN_ASYNC_THRESHOLD_BYTES = 1024L * 1024L; + public static final long MAX_ASYNC_THRESHOLD_BYTES = 1024L * 1024L * 1024L; + + // Keep this list to batch-read controls consumed by Doris' Paimon scan path. Context selectors, + // streaming-source settings, storage layout, and write options are unsafe after schema binding. + private static final Set<String> SUPPORTED_OPTIONS = ImmutableSet.of( + CoreOptions.READ_BATCH_SIZE.key(), + CoreOptions.FILE_READER_ASYNC_THRESHOLD.key(), + CoreOptions.FILE_INDEX_READ_ENABLED.key(), + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), + CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key(), + CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), + CoreOptions.SCAN_PLAN_SORT_PARTITION.key()); + + // These settings do not alter the selected snapshot or manifest projection, so relation-local + // copies can reuse the memoized partition projection while still planning splits from the copy. + private static final Set<String> METADATA_NEUTRAL_OPTIONS = ImmutableSet.of( + CoreOptions.READ_BATCH_SIZE.key(), + CoreOptions.FILE_READER_ASYNC_THRESHOLD.key(), + CoreOptions.FILE_INDEX_READ_ENABLED.key(), + CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key(), + CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key()); + + private PaimonReaderOptions() { + } + + public static Set<String> supportedOptions() { + return SUPPORTED_OPTIONS; + } + + public static Set<String> metadataNeutralOptions() { + return METADATA_NEUTRAL_OPTIONS; + } + + public static void validate(String key, String value) { + if (!SUPPORTED_OPTIONS.contains(key)) { + throw new IllegalArgumentException("Unsupported Paimon dynamic reader option '" + key + + "'. Supported options are " + SUPPORTED_OPTIONS); + } + + if (CoreOptions.READ_BATCH_SIZE.key().equals(key)) { + int batchSize = parse(key, value, CoreOptions.READ_BATCH_SIZE); + // A zero batch can make Paimon's vectorized reader report success without + // advancing input; the upper bound also prevents one relation from over-allocating. + requireRange(key, batchSize, MIN_READ_BATCH_SIZE, MAX_READ_BATCH_SIZE); + } else if (CoreOptions.FILE_READER_ASYNC_THRESHOLD.key().equals(key)) { + MemorySize threshold = parse(key, value, CoreOptions.FILE_READER_ASYNC_THRESHOLD); + // Bound the trigger on both sides so a query cannot fan out tiny async reads or + // silently disable asynchronous reading with an effectively infinite threshold. + requireRange(key, threshold.getBytes(), + MIN_ASYNC_THRESHOLD_BYTES, MAX_ASYNC_THRESHOLD_BYTES); + } else if (CoreOptions.SOURCE_SPLIT_TARGET_SIZE.key().equals(key)) { + MemorySize targetSize = parse(key, value, CoreOptions.SOURCE_SPLIT_TARGET_SIZE); + // A split-size option represents byte capacity; non-positive values silently defeat + // Paimon's bin packing and turn every data file into a separate Doris scan range. + requireRange(key, targetSize.getBytes(), 1, Long.MAX_VALUE); + } else if (CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST.key().equals(key)) { + parse(key, value, CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST); + } else if (CoreOptions.SCAN_MANIFEST_PARALLELISM.key().equals(key)) { + validateManifestParallelism(value); + } else if (CoreOptions.FILE_INDEX_READ_ENABLED.key().equals(key)) { + parse(key, value, CoreOptions.FILE_INDEX_READ_ENABLED); + } else { + parse(key, value, CoreOptions.SCAN_PLAN_SORT_PARTITION); + } + } + + public static void validateCatalogProperties(Map<String, String> properties) { + properties.forEach((key, value) -> { + if (!key.toLowerCase(Locale.ROOT).startsWith(TABLE_OPTION_PREFIX)) { + return; + } + String optionKey = key.substring(TABLE_OPTION_PREFIX.length()); + if (optionKey.isEmpty()) { + throw new IllegalArgumentException( + "Paimon table option name must not be empty after prefix " + TABLE_OPTION_PREFIX); + } + validate(optionKey, value); + }); + } + + public static Map<String, String> compatibleCatalogOptions(Map<String, String> properties) { + Map<String, String> compatibleOptions = new LinkedHashMap<>(); + properties.forEach((key, value) -> { + if (!key.toLowerCase(Locale.ROOT).startsWith(TABLE_OPTION_PREFIX)) { + return; + } + String optionKey = key.substring(TABLE_OPTION_PREFIX.length()); + try { + validate(optionKey, value); + compatibleOptions.put(optionKey, value); + } catch (IllegalArgumentException ignored) { + // Images written before the reader-only allowlist may contain arbitrary Paimon + // options. Keep the catalog loadable, but never apply an unsafe legacy option. + } + }); + return Collections.unmodifiableMap(compatibleOptions); + } + + public static void validateEffectiveTableOptions(Map<String, String> options) { + SUPPORTED_OPTIONS.stream() + .filter(options::containsKey) + .forEach(key -> validate(key, options.get(key))); + } + + public static void validateEffectiveTable(Table table) { + validateEffectiveTableOptions(table.options()); + if (table instanceof FallbackReadFileStoreTable) { + // The fallback scan plans its private child independently, so the visible main options + // cannot prove that every manifest executor input is safe. + validateEffectiveTable(((FallbackReadFileStoreTable) table).fallback()); + } + if (table instanceof DelegatedFileStoreTable) { + // Privilege and other supported delegates can hide a fallback planner behind their + // own visible main options, so traverse the complete planning-handle chain. + validateEffectiveTable(((DelegatedFileStoreTable) table).wrapped()); + } + } + + public static void validateEffectivePlanningTable(Table table) { + // Partition projection never opens a data reader. Revalidating batch/async values here + // would reject a raw handle even when the final relation copy safely overrides them. + validateIfPresent(table.options(), CoreOptions.SCAN_MANIFEST_PARALLELISM.key()); + validateIfPresent(table.options(), CoreOptions.SCAN_PLAN_SORT_PARTITION.key()); + if (table instanceof FallbackReadFileStoreTable) { + validateEffectivePlanningTable(((FallbackReadFileStoreTable) table).fallback()); + } + if (table instanceof DelegatedFileStoreTable) { + validateEffectivePlanningTable(((DelegatedFileStoreTable) table).wrapped()); + } + } + + private static void validateIfPresent(Map<String, String> options, String key) { + if (options.containsKey(key)) { + validate(key, options.get(key)); + } + } + + private static void validateManifestParallelism(String value) { + if (value == null) { + return; + } + int parallelism; + try { + parallelism = Integer.parseInt(value); + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Invalid value for Paimon option '" + + CoreOptions.SCAN_MANIFEST_PARALLELISM.key() + "': " + value, e); + } + int maximum = Runtime.getRuntime().availableProcessors(); Review Comment: Valid. Persisted validation uses the stable 1..256 contract, while execution applies the local CPU cap. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java: ########## @@ -190,6 +190,19 @@ private PaimonSnapshotCacheValue getPaimonSnapshotCacheValue(Optional<TableSnaps "Failed to get Paimon snapshot: " + (e.getMessage() == null ? "unknown cause" : e.getMessage()), e); } + } else if (scanParams.isPresent() && scanParams.get().isOptions()) { + Table baseTable = getBasePaimonTable(); + Map<String, String> resolvedOptions = scanParams.get().getOrResolveMapParams( + options -> PaimonScanParams.resolveOptions(baseTable, options)); + Table effectiveTable = PaimonScanParams.applyOptions(baseTable, resolvedOptions); + if (PaimonScanParams.hasOnlyReaderOptions(resolvedOptions)) { + // Reader tuning cannot change snapshot metadata. Reuse the memoized projection so + // a per-query batch-size change does not enumerate every partition again. + return PaimonUtils.getLatestSnapshotCacheValue(this); + } + // The shared latest cache was built from the catalog-scoped handle. Relation options + // need their own projection so partition enumeration uses the final safe table copy. + return PaimonUtils.loadSnapshotProjection(this, effectiveTable); Review Comment: Valid. Resolved startup selectors are isolated before the statement snapshot is pinned. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java: ########## @@ -139,9 +140,12 @@ public Table getPaimonTable(NameMapping nameMapping, String branch, String query } return executionAuthenticator.execute(() -> { Table table = catalog.getTable(identifier); - Map<String, String> tableOptions = - paimonProperties.getTableOptionsForCopy(table.options()); - return tableOptions.isEmpty() ? table : table.copy(tableOptions); + Map<String, String> tableOptions = paimonProperties.getTableOptionsForCopy(); + // Relation options are applied after this cached handle is returned. Defer final + // validation so a safe relation value can override an unsafe physical value. + Map<String, String> runtimeOptions = + PaimonReaderOptions.runtimeSafeCopyOptions(table, tableOptions); Review Comment: Valid. The base catalog cache remains neutral; normalization happens after relation and catalog options are composed. -- 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]
