Gabriel39 commented on code in PR #66247: URL: https://github.com/apache/doris/pull/66247#discussion_r3689762790
########## fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java: ########## @@ -0,0 +1,319 @@ +// 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.connector.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; + } + + OptionalInt safeParallelism = runtimeSafeManifestParallelism(table); + if (!safeParallelism.isPresent()) { + return safeOptions; + } + List<Integer> configuredValues = new ArrayList<>(); + collectManifestParallelism(table, configuredValues); + int localCapacity = Runtime.getRuntime().availableProcessors(); + if (configuredValues.stream().anyMatch(value -> value > localCapacity)) { + // Keep persisted semantics stable across heterogeneous FEs, but cap the execution copy + // conservatively across every nested planner hidden by a wrapper. + safeOptions.put(key, String.valueOf(safeParallelism.getAsInt())); + } + return safeOptions; + } + + public static OptionalInt runtimeSafeManifestParallelism(Table table) { + List<Integer> configuredValues = new ArrayList<>(); + collectManifestParallelism(table, configuredValues); + if (configuredValues.isEmpty()) { + return OptionalInt.empty(); + } + int localCapacity = Runtime.getRuntime().availableProcessors(); + return OptionalInt.of(Math.min( + configuredValues.stream().mapToInt(Integer::intValue).min().getAsInt(), + localCapacity)); + } + + public static Table runtimeSafeTable(Table table) { + Map<String, String> runtimeOptions = runtimeSafeCopyOptions(table, Collections.emptyMap()); + // Catalog handles stay hardware-neutral; every local planning consumer receives its own + // capped copy before it can resize Paimon's JVM-wide manifest executor. + return runtimeOptions.isEmpty() ? table : table.copy(runtimeOptions); + } + + public static Table runtimeSafeSystemTable( + Table systemTable, Table sourceTable, Map<String, String> scanOptions) { + Table effectiveSource = runtimeSafeSystemSource(sourceTable, scanOptions); + validateEffectiveTable(effectiveSource); + OptionalInt parallelism = runtimeSafeManifestParallelism(effectiveSource); + if (!parallelism.isPresent()) { + return systemTable; + } + // Paimon system wrappers keep the real FileStoreTable in a private field and expose empty + // outer options, so copy the cap onto the wrapper that will actually perform planning. + return systemTable.copy(Collections.singletonMap( Review Comment: Fixed in 068ee111cd0. Runtime capping now applies copyWithoutTimeTravel to the exact FileStoreTable source and rebuilds the same system wrapper, preserving the FE-bound schema generation. PaimonBackendBoundTableTest covers a real schema-only ALTER with a retained snapshot selector. ########## fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java: ########## @@ -575,12 +676,98 @@ static Optional<Long> parseDataSizeBytes(String value) { private void initTable() { Preconditions.checkState(params.containsKey("serialized_table")); table = PaimonUtils.deserialize(params.get("serialized_table")); + table = applyBackendManifestParallelism(table, + params.get(PAIMON_OPTION_PREFIX + DORIS_MANIFEST_PARALLELISM_CAP), + Runtime.getRuntime().availableProcessors()); + validateSerializedReaderOptions(table); paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType()); if (LOG.isDebugEnabled()) { LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames); } } + static Table applyBackendManifestParallelism( + Table table, String feParallelismCap, int localCapacity) { + int safeParallelism; + if (feParallelismCap != null) { + safeParallelism = parsePositiveManifestParallelism(feParallelismCap); + if (safeParallelism <= localCapacity) { + return table; + } + safeParallelism = localCapacity; + } else { + List<Integer> configuredValues = new ArrayList<>(); + collectManifestParallelism(table, configuredValues); + if (configuredValues.isEmpty() + || configuredValues.stream().noneMatch(value -> value > localCapacity)) { + return table; + } + safeParallelism = Math.min( + configuredValues.stream().mapToInt(Integer::intValue).min().getAsInt(), + localCapacity); + } + Map<String, String> cap = Collections.singletonMap( + CoreOptions.SCAN_MANIFEST_PARALLELISM.key(), String.valueOf(safeParallelism)); + // File-store copies must retain the FE-selected schema while only lowering an execution + // bound; ordinary copy can re-resolve time travel and undo schema pinning. + return table instanceof FileStoreTable + ? ((FileStoreTable) table).copyWithoutTimeTravel(cap) + : table.copy(cap); + } + + private static int parsePositiveManifestParallelism(String value) { + try { + int parsed = Integer.parseInt(value); + if (parsed < 1) { + throw new IllegalArgumentException("Paimon manifest parallelism cap must be positive."); + } + return parsed; + } catch (NumberFormatException e) { + throw new IllegalArgumentException("Paimon manifest parallelism cap must be an integer.", e); + } + } + + private static void collectManifestParallelism(Table table, List<Integer> values) { + String configured = table.options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key()); + if (configured != null) { + values.add(parsePositiveManifestParallelism(configured)); + } + if (table instanceof FallbackReadFileStoreTable) { + collectManifestParallelism(((FallbackReadFileStoreTable) table).fallback(), values); + } + if (table instanceof DelegatedFileStoreTable) { + collectManifestParallelism(((DelegatedFileStoreTable) table).wrapped(), values); + } + } + + private static void validateSerializedReaderOptions(Table table) { + validateSerializedReadBatchSize(table.options().get(CoreOptions.READ_BATCH_SIZE.key())); Review Comment: Fixed in 068ee111cd0. The old-FE backstop now parses and range-checks file-reader-async-threshold recursively across visible, fallback, and delegated readers. Compatibility tests cover 0 B and 2 GB values. ########## fe/fe-core/src/main/java/org/apache/doris/connector/ConnectorPluginManager.java: ########## @@ -104,8 +104,20 @@ public class ConnectorPluginManager { /** Called at FE startup to load built-in providers from classpath. */ public void loadBuiltins() { - ServiceLoader.load(ConnectorProvider.class) + loadBuiltins(ConnectorProvider.class.getClassLoader()); + } + + /** Classloader seam used to verify embedded/classpath providers with their own defining jars. */ + void loadBuiltins(ClassLoader classLoader) { + ServiceLoader.load(ConnectorProvider.class, classLoader) Review Comment: Fixed in 068ee111cd0. Built-in discovery now inspects ServiceLoader.Provider.type(), applies the manifest/API gate, and calls get() only for accepted providers; construction errors are isolated. The new test proves an incompatible provider constructor is never invoked. ########## fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java: ########## @@ -575,12 +676,98 @@ static Optional<Long> parseDataSizeBytes(String value) { private void initTable() { Preconditions.checkState(params.containsKey("serialized_table")); table = PaimonUtils.deserialize(params.get("serialized_table")); + table = applyBackendManifestParallelism(table, + params.get(PAIMON_OPTION_PREFIX + DORIS_MANIFEST_PARALLELISM_CAP), + Runtime.getRuntime().availableProcessors()); + validateSerializedReaderOptions(table); paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType()); if (LOG.isDebugEnabled()) { LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames); } } + static Table applyBackendManifestParallelism( + Table table, String feParallelismCap, int localCapacity) { + int safeParallelism; + if (feParallelismCap != null) { + safeParallelism = parsePositiveManifestParallelism(feParallelismCap); + if (safeParallelism <= localCapacity) { Review Comment: Fixed in 068ee111cd0. The BE now inspects the actual serialized wrapper children and normalizes any value above min(FE cap, local capacity); the advertised cap is no longer treated as proof. The test covers physical 200, FE cap 32, and BE capacity 64. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java: ########## @@ -347,6 +347,27 @@ private static ListPartitionItem toListPartitionItem(String partitionName, List< @Override public MvccSnapshot loadSnapshot(Optional<TableSnapshot> tableSnapshot, Optional<TableScanParams> scanParams) { + return loadSnapshotInternal(tableSnapshot, scanParams, Optional.empty()); + } + + @Override + public boolean requiresLatestSnapshotFence( + Optional<TableSnapshot> tableSnapshot, Optional<TableScanParams> scanParams) { + return !tableSnapshot.isPresent() && scanParams.isPresent() && scanParams.get().isOptions(); Review Comment: Fixed in 068ee111cd0. Selector-free option projections now acquire a lightweight version-only fence and hydrate the exact version only when needed, so the raw partition projection is not materialized. Explicit snapshot/tag/timestamp selectors skip this fence. Tests cover the physical-zero/safe-override path, alias ordering, and explicit selectors. -- 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]
