github-actions[bot] commented on code in PR #66247:
URL: https://github.com/apache/doris/pull/66247#discussion_r3690119438
##########
fe/fe-connector/fe-connector-api/src/main/java/org/apache/doris/connector/api/ConnectorMetadata.java:
##########
@@ -70,6 +71,12 @@ default Optional<ConnectorMvccSnapshot> beginQuerySnapshot(
return Optional.empty();
}
+ /** Whether relation-level options need the statement's latest snapshot as
their version fence. */
+ default boolean usesStatementSnapshotForOptions(
Review Comment:
[P1] Regenerate the ConnectorMetadata surface baseline
This adds a public `usesStatementSnapshotForOptions(...)` method, but the
recorded `connector-metadata-methods.txt` surface still lacks its rendered
signature. `ConnectorMetadataSurfaceTest` reflects every public method and
exact-compares that set with this resource, so the `fe-connector-api` suite
fails unconditionally on this head. Regenerate and commit the baseline in the
same change, then run the connector API surface test.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java:
##########
@@ -0,0 +1,331 @@
+// 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.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.system.SystemTableLoader;
+
+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(
+ String systemTableType, Table systemTable, Table sourceTable,
Map<String, String> scanOptions) {
+ Table effectiveSource = runtimeSafeSystemSource(sourceTable,
scanOptions);
+ validateEffectiveTable(effectiveSource);
+ OptionalInt parallelism =
runtimeSafeManifestParallelism(effectiveSource);
+ if (!parallelism.isPresent()) {
+ return systemTable;
+ }
+ Map<String, String> cap = Collections.singletonMap(
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
+ String.valueOf(parallelism.getAsInt()));
+ if (effectiveSource instanceof FileStoreTable) {
+ // Copying a system wrapper replays inherited time-travel options
and can rewind a
+ // schema-only ALTER; cap the source without time travel, then
rebuild the same wrapper.
+ FileStoreTable cappedSource = ((FileStoreTable)
effectiveSource).copyWithoutTimeTravel(cap);
+ Table rebuilt = SystemTableLoader.load(systemTableType,
cappedSource);
Review Comment:
[P1] Preserve the fallback pair when rebuilding a capped `$ro`
`getSysTableHandle` deliberately peels `PrivilegedFileStoreTable` before
building `$ro`, because `ReadOptimizedTable` selects `FallbackReadScan` only
when `FallbackReadFileStoreTable` is its immediate child. Here
`effectiveSource` comes from the decorated `systemTableSource`,
`copyWithoutTimeTravel` preserves that decorator, and this rebuild produces
`ReadOptimized(Privileged(FallbackRead(...)))`; `$ro` then scans only the main
branch and silently omits fallback-only rows. Any configured manifest value,
even `1`, reaches this path. Peel non-fallback delegates from the capped source
before rebuilding, and cover a privilege-wrapped fallback `$ro` scan/statistics
case.
##########
fe/fe-connector/fe-connector-paimon/src/test/java/org/apache/doris/connector/paimon/PaimonScanParamsTest.java:
##########
@@ -72,6 +72,69 @@ public void testValidateKnownScanOptions() {
"scan.plan-sort-partition", "true"));
}
+ @Test
+ public void testValidateRelationScopedReaderOptions() {
+ PaimonScanParams.validateOptions(ImmutableMap.of(
+ "read.batch-size", "4096",
+ "file-reader-async-threshold", "16 MB",
+ "file-index.read.enabled", "false",
+ "source.split.target-size", "64 MB",
+ "source.split.open-file-cost", "1 MB",
+ "scan.manifest.parallelism", "1",
+ "scan.plan-sort-partition", "true"));
+
+ for (Map<String, String> options : new Map[] {
+ ImmutableMap.of("read.batch-size", "0"),
+ ImmutableMap.of("read.batch-size", "-1"),
+ ImmutableMap.of("read.batch-size", "65537"),
+ ImmutableMap.of("file-reader-async-threshold", "512 KB"),
+ ImmutableMap.of("file-reader-async-threshold", "2 GB")
+ }) {
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> PaimonScanParams.validateOptions(options));
+ }
+ }
+
+ @Test
+ public void testPlanningOptionsDoNotReuseMetadataProjection() {
+
Assertions.assertTrue(PaimonScanParams.hasOnlyReaderOptions(ImmutableMap.of(
+ "file-index.read.enabled", "false",
+ "source.split.target-size", "64 MB")));
+ Assertions.assertFalse(PaimonScanParams.hasOnlyReaderOptions(
+ ImmutableMap.of("scan.manifest.parallelism", "1")));
+ Assertions.assertFalse(PaimonScanParams.hasOnlyReaderOptions(
+ ImmutableMap.of("scan.plan-sort-partition", "true")));
+ }
+
+ @Test
+ public void testManifestParallelismCannotMutateGlobalPoolCapacity() {
+ int availableProcessors = Runtime.getRuntime().availableProcessors();
+ PaimonScanParams.validateOptions(ImmutableMap.of(
Review Comment:
[P1] Keep this test valid above 256 processors
This success case passes the raw `availableProcessors()` value, but
production validation first enforces the intentional hardware-independent
`1..256` limit and only then checks the local CPU bound. A JVM exposing 257 or
more processors therefore throws on the supposedly valid call, and the Maven
test configuration does not clamp `ActiveProcessorCount`; adjacent tests
already clamp or skip this boundary. Use `min(availableProcessors,
MAX_MANIFEST_PARALLELISM)` for the valid case, guard the local `+1` assertion
when applicable, and retain an explicit `MAX_MANIFEST_PARALLELISM + 1`
rejection.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java:
##########
@@ -749,7 +767,10 @@ public Optional<ConnectorMvccSnapshot> resolveTimeTravel(
.properties(PaimonScanParams.markAsOptions(resolved))
.build());
}
- long pinnedId = pinnedSnapshotId(table, resolved);
+ long pinnedId = spec.getLatestSnapshotFence().isPresent()
Review Comment:
[P1] Keep latest-schema semantics for planning-only OPTIONS
When this branch reuses statement fence `S`, the code below derives
`schemaId` from snapshot `S`. Paimon schema-only ALTERs advance
`schemaManager.latest()` without creating a new data snapshot, which is why
`beginQuerySnapshot` deliberately leaves `schemaId=-1`. A plain `t` therefore
binds the current schema, while `t@options('scan.manifest.parallelism'='1')`
binds `S`'s historical schema even though the option is planning-only and both
aliases share the same version. Preserve latest-schema (`schemaId=-1`)
semantics for fence-derived options, and add a schema-only ALTER alias test.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonReaderOptions.java:
##########
@@ -0,0 +1,331 @@
+// 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.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.system.SystemTableLoader;
+
+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(
+ String systemTableType, Table systemTable, Table sourceTable,
Map<String, String> scanOptions) {
+ Table effectiveSource = runtimeSafeSystemSource(sourceTable,
scanOptions);
+ validateEffectiveTable(effectiveSource);
+ OptionalInt parallelism =
runtimeSafeManifestParallelism(effectiveSource);
+ if (!parallelism.isPresent()) {
+ return systemTable;
+ }
+ Map<String, String> cap = Collections.singletonMap(
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
+ String.valueOf(parallelism.getAsInt()));
+ if (effectiveSource instanceof FileStoreTable) {
+ // Copying a system wrapper replays inherited time-travel options
and can rewind a
+ // schema-only ALTER; cap the source without time travel, then
rebuild the same wrapper.
+ FileStoreTable cappedSource = ((FileStoreTable)
effectiveSource).copyWithoutTimeTravel(cap);
+ Table rebuilt = SystemTableLoader.load(systemTableType,
cappedSource);
+ if (rebuilt == null) {
+ throw new IllegalArgumentException("Unsupported Paimon system
table '"
+ + systemTableType + "'");
+ }
+ return rebuilt;
+ }
+ return systemTable.copy(cap);
+ }
+
+ public static Table runtimeSafeSystemSource(Table sourceTable, Map<String,
String> scanOptions) {
+ return PaimonScanParams.isOptionsPin(scanOptions)
+ ? PaimonScanParams.applyOptions(sourceTable, scanOptions)
+ : runtimeSafeTable(sourceTable);
Review Comment:
[P1] Reapply `@incr` before rebuilding the system wrapper
`resolveScanTable` first applies the reset-aware incremental range to the
system wrapper, but this helper rebuilds from the original source and reapplies
scan options only for OPTIONS-marked maps. For `@incr`, the configured manifest
value therefore makes `SystemTableLoader.load` replace the correctly ranged
wrapper with an unbounded/latest one; this affects supported incremental system
tables during FE planning and statistics, before any later BE reapply can help.
Apply the non-OPTIONS incremental map to the exact source before
capping/rebuilding, and test a real incremental system wrapper with manifest
parallelism configured.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java:
##########
@@ -336,6 +350,16 @@ private static Map<String, String> resolvedSnapshotOptions(
return resolved;
}
+ public static Map<String, String> pinOptionsToSnapshot(
+ Map<String, String> options, long snapshotId) {
+ // Statement-fence pinning removes inherited selectors, so validate
the raw map first;
+ // otherwise an unsupported inherited key can disappear before the
common validation path.
+ validateOptions(options);
+ return snapshotId < 0
Review Comment:
[P1] Freeze an empty latest fence for plain aliases too
This maps fence id `-1` to the internal pinned-empty marker only for an
OPTIONS projection. A plain alias sharing the same fence goes through
`applySnapshot(-1)`, which returns the base handle unchanged and leaves its
scan live. If the table receives its first commit after fence capture, the
OPTIONS alias remains empty while plain `t` can read the new rows. Represent an
empty latest fence with scan semantics honored by both paths, and test the
first-commit race in both alias orders.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java:
##########
@@ -347,15 +355,71 @@ 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 MvccSnapshot loadLatestSnapshotFence() {
+ makeSureInitialized();
+ PluginDrivenExternalCatalog pluginCatalog =
(PluginDrivenExternalCatalog) catalog;
+ Connector connector = pluginCatalog.getConnector();
+ if (connector == null) {
+ return new PluginDrivenMvccSnapshot(emptySnapshot(),
+ Collections.emptyMap(), Collections.emptyMap());
+ }
+ ConnectorSession session = pluginCatalog.buildConnectorSession();
+ ConnectorMetadata metadata = PluginDrivenMetadata.get(session,
connector);
+ Optional<ConnectorTableHandle> handle =
resolveConnectorTableHandle(session, metadata);
+ ConnectorMvccSnapshot fence = handle.isPresent()
+ ? metadata.beginQuerySnapshot(session,
handle.get()).orElseGet(this::emptySnapshot)
+ : emptySnapshot();
+ // A fence carries version identity only; raw partitions may be
invalid for relation options.
+ return new PluginDrivenMvccSnapshot(fence, Collections.emptyMap(),
Collections.emptyMap());
+ }
+
+ @Override
+ public boolean requiresLatestSnapshotFence(
+ Optional<TableSnapshot> tableSnapshot, Optional<TableScanParams>
scanParams) {
+ if (tableSnapshot.isPresent() || !scanParams.isPresent() ||
!scanParams.get().isOptions()) {
+ return false;
+ }
+ makeSureInitialized();
+ PluginDrivenExternalCatalog pluginCatalog =
(PluginDrivenExternalCatalog) catalog;
+ Connector connector = pluginCatalog.getConnector();
+ if (connector == null) {
+ return false;
+ }
+ ConnectorSession session = pluginCatalog.buildConnectorSession();
+ ConnectorMetadata metadata = PluginDrivenMetadata.get(session,
connector);
+ Optional<ConnectorTableHandle> handle =
resolveConnectorTableHandle(session, metadata);
+ return handle.isPresent() && metadata.usesStatementSnapshotForOptions(
+ session, handle.get(), scanParams.get().getMapParams());
+ }
+
+ @Override
+ public MvccSnapshot loadSnapshot(
Review Comment:
[P1] Thread the statement fence into system-table OPTIONS
This fence-aware overload is never reached by
`PluginDrivenSysExternalTable`: `BindRelation.handleMetaTable` returns before
`StatementContext.loadSnapshots`, and `resolveScanPin` calls the old
two-argument source loader. Consequently `t` joined with
`t$partitions@options('scan.manifest.parallelism'='1')` can pin `S` for the
plain alias and resolve live `S+1` for the system view after an intervening
commit. The system table's memo only prevents drift within that one relation.
Route its source pin through the statement-scoped fence registry (in both
relation orders) and add a commit-between-resolutions test.
--
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]
