github-actions[bot] commented on code in PR #66297:
URL: https://github.com/apache/doris/pull/66297#discussion_r3682287070
##########
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:
[P2] Keep this option-bearing projection relation-scoped.
`StatementContext.loadSnapshots` has only one `MvccTableInfo` slot per
catalog/db/table, so if the first alias uses
`@options('scan.manifest.parallelism'='1')`, this snapshot retains `1` and a
later unannotated alias reuses it through `PaimonSource`; reversing alias order
instead rejects the raw physical `0`. That makes relation isolation and
validation traversal-order-dependent. Please keep the shared latest snapshot
neutral or key option projections by relation/scan parameters, and add a
two-alias order test.
##########
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:
[P2] Do not advertise `read.batch-size` until its value reaches the reader.
After this effective table is serialized, `PaimonJniScanner.initTable()`
unconditionally copies it with `CoreOptions.READ_BATCH_SIZE = batchSize` from
the Doris runtime, so physical, catalog, and relation values (including the new
`1` versus `8192` isolation case) are all replaced immediately before
`newReadBuilder()`. Please either preserve the validated effective option and
keep the Doris output-block cap separate, or remove this key from the
allowlist; add a scanner-level assertion for the value seen by the opened
reader.
##########
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:
[P1] Preserve the relation override during pre-lock preload. Selector-free
Paimon `@options` are registered as a latest relation, but for a mixed
internal/external query `PreloadExternalMetadata` calls `loadSnapshots(table,
empty, empty)` before `BindRelation`. A partitioned physical table with
`scan.manifest.parallelism=0` therefore reaches this validator and fails before
its safe relation override of `1` is applied. Please retain the scan
params/effective handle in preload, or skip latest-snapshot preload for these
relations, and cover the mixed-table preload path.
##########
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:
[P1] Do not make persisted catalog validity depend on this FE's CPU count. A
value accepted on an 8-core leader is replayed without validation, then
`compatibleCatalogOptions()` re-runs this check on a 4-core follower and
silently drops it. After failover the catalog therefore changes behavior, and
an override that masked a physical `scan.manifest.parallelism=0` disappears so
affected queries fail. Please use a deterministic cluster-stable
bound/normalization, or fail explicitly during reconstruction, and cover
heterogeneous-CPU replay/failover.
--
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]
