github-actions[bot] commented on code in PR #66247:
URL: https://github.com/apache/doris/pull/66247#discussion_r3689309870
##########
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:
[P1] Validate the async-reader threshold in the old-FE backstop too.
`initTable()` names this a serialized-reader-options guard, but it recursively
checks only `read.batch-size`; an older FE can still send
`file-reader-async-threshold=0 B` or `2 GB` on the visible or fallback table,
and the new BE passes it unchanged to `newReadBuilder()`, bypassing the 1 MiB-1
GiB resource bound introduced by this PR. Please parse and range-check this
option on every traversed child and extend the serialized-table compatibility
tests.
##########
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:
[P1] Do not treat the advertised FE cap as proof that this serialized
wrapper already carries it. `tableForBackend` rebuilds a system table from
`handle.getSysBaseTable()`, discarding the capped hidden source returned by
`resolveScanTable`; for example, physical 200 on a 32-core FE can serialize as
200 plus an out-of-band cap of 32. A 64-core BE takes this early return (`32 <=
64`) and deferred `$files`/`$partitions` planning still uses 200, resizing
Paimon's JVM-global executor. Please inspect the deserialized children and
apply `min(feCap, localCapacity)` whenever any actual value exceeds it, and add
a real system-wrapper test where the FE cap is lower than BE capacity. The
existing schema-fence thread covers the smaller-BE copy branch, not this
no-copy branch.
##########
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:
[P1] Preserve the FE system wrapper's schema generation while applying the
cap. This runs before planning/serialization and `systemTable.copy(cap)`
delegates to the hidden `FileStoreTable.copy`; Paimon then re-evaluates any
inherited `scan.snapshot-id` and swaps in that snapshot's schema. After a
schema-only ALTER, merely having `scan.manifest.parallelism` set can therefore
rewind the wrapper after its current-schema columns were bound (and this copy
happens even when the configured value is already within the local CPU limit).
Please rebuild the same wrapper over a hidden source capped with
`copyWithoutTimeTravel`, and cover a real system table with a retained snapshot
selector plus a newer schema. The existing BE-side thread at r3688450282 covers
the later deserialization copy, not this FE planning/statistics path.
##########
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:
[P1] Do not materialize the raw latest projection for every `@options`
relation. Returning true here makes `StatementContext` call
`loadSnapshot(empty, empty)` first; on a partitioned table that lists/validates
the unannotated base handle, so physical `scan.manifest.parallelism=0` fails
before `@options('scan.manifest.parallelism'='1')` can apply—the exact success
case added to this PR's P0 suite, now even with preload disabled. It is also
unnecessary for explicit snapshot/tag/timestamp selectors because Paimon later
reports `usesStatementSnapshot=false`. Please obtain a lightweight version-only
fence or let the connector decide before raw partition planning, and cover
direct binding of the physical-zero/safe-override case plus an explicit
selector.
##########
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:
[P1] Check the manifest before instantiating a classpath provider.
`ServiceLoader.forEach` creates each provider before this lambda runs, so an
API-1 implementation can execute its constructor—or fail linkage/construction
with an uncaught `ServiceConfigurationError` and abort discovery—before
`rejectionReasonForClass` gets a chance to skip it. That leaves the classpath
gate from r3684457168 fail-closed only for harmless providers like the test
fixture. Please enumerate provider types/descriptors first, apply the jar-major
gate, and instantiate only accepted providers; add an incompatible provider
whose constructor records or throws to prove it is never run.
--
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]
