Gabriel39 commented on code in PR #66297: URL: https://github.com/apache/doris/pull/66297#discussion_r3698361507
########## 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: Fixed by replacing the scalar/map-level cap with recursive per-leaf normalization after relation options are applied. Missing values are materialized independently when the local processor default exceeds the stable 256 ceiling, so an absent sibling cannot escape a safe explicit branch. The 512-capacity tests cover both mixed directions and the all-absent case. ########## 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: Fixed on the BE backstop as well. It now traverses delegated tables and hidden old-FE system-table sources recursively, preserves each fallback branch independently, and unwraps non-routing delegates before rebuilding a system wrapper even when no numeric cap changes. Scanner tests cover an asymmetric fallback and a safe fallback hidden behind a privilege delegate. -- 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]
