github-actions[bot] commented on code in PR #66297: URL: https://github.com/apache/doris/pull/66297#discussion_r3698169404
########## fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonReaderOptions.java: ########## @@ -0,0 +1,312 @@ +// 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.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.OptionalInt; +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; + // Keep catalog replay deterministic while bounding a single option's JVM-wide thread impact. + public static final int MAX_MANIFEST_PARALLELISM = 256; + 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 validateReaderOptions(Map<String, String> options) { + SUPPORTED_OPTIONS.stream() + .filter(options::containsKey) + .forEach(key -> validate(key, options.get(key))); + } + + public static void validateEffectiveTableOptions(Map<String, String> options) { + validateReaderOptions(options); + validateIfPresentForRuntime(options, CoreOptions.SCAN_MANIFEST_PARALLELISM.key()); + } + + public static Map<String, String> runtimeSafeCopyOptions(Table table, Map<String, String> copyOptions) { + Map<String, String> safeOptions = new LinkedHashMap<>(copyOptions); + String key = CoreOptions.SCAN_MANIFEST_PARALLELISM.key(); + if (safeOptions.containsKey(key)) { + String configured = safeOptions.get(key); + if (configured == null) { + return safeOptions; + } + validateManifestParallelism(configured); + int requested = Integer.parseInt(configured); + int localCapacity = Runtime.getRuntime().availableProcessors(); + if (requested > localCapacity) { + safeOptions.put(key, String.valueOf(localCapacity)); + } + return safeOptions; + } + + List<Integer> configuredValues = new ArrayList<>(); + collectManifestParallelism(table, configuredValues); + if (configuredValues.isEmpty()) { Review Comment: [P1] Materialize the stable ceiling for every absent FE planning branch. The all-absent case returns here, but mixed fallback trees also escape: on a >256-core FE, Fallback(main=1, fallback=unset) makes `configuredValues` nonempty, no collected value exceeds `localCapacity`, and the fallback still inherits `availableProcessors()`. FE partition/split, row-count, and metadata-TVF planning runs before the BE cap. This is distinct from the fixed BE-absence thread; the direct-fallback thread covers explicit sibling preferences, not an absent child exceeding 256. Normalize absent children independently to `min(localCapacity, 256)` without overwriting a lower explicit sibling, and test all-absent plus both mixed explicit/absent orders at capacity 512. ########## fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java: ########## @@ -519,69 +622,212 @@ static Optional<Long> parseDataSizeBytes(String value) { if (value == null || value.trim().isEmpty()) { return Optional.empty(); } - String normalized = value.trim().toLowerCase(Locale.ROOT).replace("_", "").replace(" ", ""); - int unitStart = 0; - while (unitStart < normalized.length() - && (Character.isDigit(normalized.charAt(unitStart)) || normalized.charAt(unitStart) == '.')) { - unitStart++; - } - if (unitStart == 0) { - return Optional.empty(); - } try { - double number = Double.parseDouble(normalized.substring(0, unitStart)); - String unit = normalized.substring(unitStart); - long multiplier; - switch (unit) { - case "": - case "b": - case "byte": - case "bytes": - multiplier = 1L; - break; - case "k": - case "kb": - case "kib": - multiplier = 1024L; - break; - case "m": - case "mb": - case "mib": - multiplier = 1024L * 1024L; - break; - case "g": - case "gb": - case "gib": - multiplier = 1024L * 1024L * 1024L; - break; - case "t": - case "tb": - case "tib": - multiplier = 1024L * 1024L * 1024L * 1024L; - break; - default: - return Optional.empty(); - } - return Optional.of((long) (number * multiplier)); - } catch (NumberFormatException e) { + // Keep the BE guard's accepted grammar identical to the Paimon option parser that will + // consume this value; accepting a superset lets invalid serialized options reach scans. + return Optional.of(MemorySize.parse(value).getBytes()); + } catch (IllegalArgumentException e) { return Optional.empty(); } } private void initTable() { Preconditions.checkState(params.containsKey("serialized_table")); table = PaimonUtils.deserialize(params.get("serialized_table")); + String encodedSystemSource = params.get(PAIMON_OPTION_PREFIX + DORIS_SERIALIZED_SYSTEM_SOURCE); + FileStoreTable systemSource = encodedSystemSource == null + ? null : PaimonUtils.deserialize(encodedSystemSource); + table = applyBackendManifestParallelism(table, + params.get(PAIMON_OPTION_PREFIX + DORIS_MANIFEST_PARALLELISM_CAP), + Runtime.getRuntime().availableProcessors(), systemSource, + params.get(PAIMON_OPTION_PREFIX + DORIS_SYSTEM_TABLE_TYPE)); + table = applyDefaultReadBatchSize(table, batchSize); + paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType()); + if (LOG.isDebugEnabled()) { + LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames); + } + } + + static Table applyDefaultReadBatchSize(Table table, int dorisBatchSize) { + validateSerializedReaderOptions(table); + if (hasReadBatchSize(table)) { + // Doris' output block size and Paimon's reader batch are independent controls; an + // explicitly validated value on any hidden reader must survive transport unchanged. + return table; + } // The serialized table may pin an older data snapshot while carrying the latest schema // after a schema change. Applying a normal copy would time travel to that snapshot's // schema again and make renamed or newly added columns disappear. Map<String, String> readOptions = Collections.singletonMap( - CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(batchSize)); - table = table instanceof FileStoreTable + CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(dorisBatchSize)); + return table instanceof FileStoreTable ? ((FileStoreTable) table).copyWithoutTimeTravel(readOptions) : table.copy(readOptions); - paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType()); - if (LOG.isDebugEnabled()) { - LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames); + } + + static Table applyBackendManifestParallelism( + Table table, String feParallelismCap, int localCapacity) { + return applyBackendManifestParallelism( + table, feParallelismCap, localCapacity, null, null); + } + + static Table applyBackendManifestParallelism( + Table table, String feParallelismCap, int localCapacity, + FileStoreTable systemSource, String systemTableType) { + // Old FEs do not send a cap, so the BE must still preserve the hardware-independent + // ceiling that prevents one scan from growing Paimon's JVM-global executor beyond 256. + int requestedBound = Math.min(localCapacity, MAX_MANIFEST_PARALLELISM); + if (feParallelismCap != null) { + requestedBound = Math.min(parsePositiveManifestParallelism(feParallelismCap), requestedBound); + } + final int safeBound = requestedBound; + if (systemSource != null && systemTableType != null) { + FileStoreTable cappedSource = + (FileStoreTable) applyManifestParallelismBound(systemSource, safeBound); + // Read-only wrappers hide the data table's option map. Rebuild from the transported + // exact source so a smaller BE can lower that hidden planner without rewinding schema. + Table rebuilt = SystemTableLoader.load(systemTableType, cappedSource); + if (rebuilt == null) { + throw new IllegalArgumentException( + "Unsupported Paimon system table '" + systemTableType + "'"); + } + return rebuilt; + } + return applyManifestParallelismBound(table, safeBound); + } + + private static Table applyManifestParallelismBound(Table table, int safeBound) { + if (table instanceof FallbackReadFileStoreTable) { + FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable) table; + FileStoreTable main = applyManifestParallelismBound(pair.wrapped(), safeBound); + FileStoreTable fallback = applyManifestParallelismBound(pair.fallback(), safeBound); + if (main == pair.wrapped() && fallback == pair.fallback()) { + return table; + } + // Each branch owns an independent planner setting; a smaller sibling is not an + // execution ceiling and must never throttle the other branch. + return new FallbackReadFileStoreTable(main, fallback); + } + + String configured = table.options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()); Review Comment: [P1] Traverse delegated wrappers before trusting the visible manifest setting. `PrivilegedFileStoreTable` can wrap `FallbackReadFileStoreTable` (the FE test constructs this exact shape), so with main=1, fallback=128, and a 64-core BE this code sees the delegate's visible 1 and returns the whole wrapper; the hidden fallback planner still opens at 128. This is distinct from the existing direct-fallback thread: the outer delegate prevents lines 700-709 from ever running. Recurse/rebuild delegated children before this early return (or reject an unrebuildable wrapper), and cover a serialized privilege delegate around heterogeneous fallback branches. -- 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]
