github-actions[bot] commented on code in PR #66176:
URL: https://github.com/apache/doris/pull/66176#discussion_r3668358964
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExecuteCommand.java:
##########
@@ -77,6 +79,29 @@ public void run(ConnectContext ctx, StmtExecutor executor)
throws Exception {
}
PrepareCommand prepareCommand = preparedStmtCtx.command;
LogicalPlan logicalPlan = prepareCommand.getLogicalPlan();
+ LogicalPlan relationRoot = logicalPlan;
+ if (logicalPlan instanceof InsertIntoTableCommand) {
+ relationRoot = ((InsertIntoTableCommand)
logicalPlan).getLogicalQuery();
+ } else if (logicalPlan instanceof InsertOverwriteTableCommand) {
+ relationRoot = ((InsertOverwriteTableCommand)
logicalPlan).getLogicalQuery();
+ } else if (logicalPlan instanceof UpdateCommand) {
+ relationRoot = ((UpdateCommand) logicalPlan).getLogicalQuery();
+ } else if (logicalPlan instanceof Command) {
Review Comment:
[P1] Reset every relation-bearing prepared command
This generic branch includes relation-bearing DML that retains source plans:
`DeleteFromUsingCommand.logicalQuery` is executed through an insert command,
and top-level `MergeIntoCommand` reuses its `source` for both OLAP and Iceberg
targets. Either source can contain a Paimon `@options` relation, but neither is
traversed here, so repeated EXECUTEs keep the snapshot resolved by the first
execution. Please introduce a common relation-root contract (or explicitly
cover both commands) and test repeated DELETE USING and MERGE after advancing
the Paimon source.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java:
##########
@@ -106,6 +106,32 @@ public Table getPaimonTable(Optional<MvccSnapshot>
snapshot) {
}
}
+ public Table getPaimonTable(TableScanParams scanParams) {
+ if (scanParams != null && scanParams.isOptions()) {
+ Map<String, String> options = scanParams.getMapParams();
+ Table statementTable =
getPaimonTable(MvccUtil.getSnapshotFromContext(this));
+ Table resolutionTable =
PaimonScanParams.usesStatementSnapshot(options)
+ ? statementTable
+ : getBasePaimonTable();
+ Map<String, String> resolvedOptions =
scanParams.getOrResolveMapParams(
+ relationOptions ->
PaimonScanParams.resolveOptions(resolutionTable, relationOptions));
+ // Startup options are normalized to an immutable snapshot before
schema binding. The
+ // scan phase reuses that exact resolution instead of consulting a
mutable tag or clock.
+ Table table = PaimonScanParams.selectsSchema(resolvedOptions)
Review Comment:
[P1] Preserve current schema when pinning latest state
For selector-free `default`/`latest`/`latest-full`/`full`, resolution starts
from the statement table, which is intentionally a pinned snapshot followed by
`copyWithLatestSchema()`. File-creation-time resolution similarly pins the
latest snapshot while native Paimon keeps the current row type. Both paths then
carry a `scan.snapshot-id` into this condition, rebuild from the base table,
and time-travel to the data snapshot's older schema. After a schema-only ADD
COLUMN, these OPTIONS forms can reject a column native latest/file-creation
reads expose as null. Please preserve latest-schema provenance for both pinned
states and add schema-only evolution regressions.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java:
##########
@@ -0,0 +1,391 @@
+// 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 org.apache.doris.analysis.TableScanParams;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.FallbackKey;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Validation and application rules for relation-scoped Paimon scan parameters.
+ */
+public final class PaimonScanParams {
+ private static final String PINNED_FILE_CREATION_TIME =
+ "doris.internal.paimon.file-creation-time-millis";
+ private static final String PINNED_EMPTY_SCAN =
"doris.internal.paimon.empty-scan";
+
+ private static final Set<String> QUERY_OPTION_KEYS = ImmutableSet.of(
+ CoreOptions.SCAN_MODE.key(),
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_VERSION.key(),
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
+ CoreOptions.SCAN_PLAN_SORT_PARTITION.key());
+
+ private static final Set<String> STARTUP_POSITION_KEYS = ImmutableSet.of(
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_VERSION.key());
+
+ private static final Set<String> INHERITED_READ_STATE_KEYS =
inheritedReadStateKeys();
+
+ // FilesScan enumerates the latest partitions before applying its
range-aware per-partition scan,
+ // so it cannot safely read a range when a partition in that range has
since been dropped.
+ private static final Set<String> INCREMENTAL_SYSTEM_TABLES =
ImmutableSet.of(
+ "audit_log", "binlog", "partitions", "ro", "row_tracking");
+
+ private static final Set<String> PAIMON_READER_SYSTEM_TABLES =
ImmutableSet.of(
+ "audit_log", "binlog", "row_tracking");
+
+ private static final Set<String> OPTIONS_SYSTEM_TABLES = ImmutableSet.of(
+ // A system table may advertise OPTIONS only when every
row-producing stage observes
+ // the selected snapshot; files and buckets still consult latest
metadata internally.
+ "audit_log", "binlog", "manifests", "partitions", "ro",
+ "row_tracking", "table_indexes");
+
+ private PaimonScanParams() {
+ }
+
+ public static void validateOptions(Map<String, String> options) {
+ if (options.containsKey(CoreOptions.SCAN_FALLBACK_BRANCH.key())) {
+ throw new IllegalArgumentException("Paimon query option '"
+ + CoreOptions.SCAN_FALLBACK_BRANCH.key()
+ + "' is not supported because it requires rebuilding the
table through the catalog factory.");
+ }
+
+ Set<String> unsupported = options.keySet().stream()
+ .filter(key -> !QUERY_OPTION_KEYS.contains(key))
+ .collect(Collectors.toSet());
+ if (!unsupported.isEmpty()) {
+ throw new IllegalArgumentException("Unsupported Paimon query
option(s): " + unsupported);
+ }
+
+ String scanMode = options.get(CoreOptions.SCAN_MODE.key());
+ if ("from-creation-timestamp".equalsIgnoreCase(scanMode)
+ && options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()) ==
null) {
+ // Paimon 1.3.1 does not validate this newer mode, but its
starting scanner
+ // requires the creation timestamp and otherwise fails after
analysis.
+ throw new IllegalArgumentException("Paimon scan mode
'from-creation-timestamp' requires query option '"
+ + CoreOptions.SCAN_CREATION_TIME_MILLIS.key() + "'.");
+ }
+
+ long positionCount =
options.keySet().stream().filter(STARTUP_POSITION_KEYS::contains).count();
+ if (positionCount > 1) {
+ throw new IllegalArgumentException(
+ "Only one Paimon startup position can be specified: " +
STARTUP_POSITION_KEYS);
+ }
+
+ if (options.containsKey(CoreOptions.SCAN_MODE.key()) && positionCount
== 1) {
+ String position = options.keySet().stream()
+ .filter(STARTUP_POSITION_KEYS::contains)
+ .findFirst()
+ .get();
+ String mode = scanMode.toLowerCase(Locale.ROOT);
+ if (!isCompatibleStartupMode(position, mode)) {
+ throw new IllegalArgumentException("Paimon scan mode '" + mode
+ + "' is incompatible with startup position '" +
position + "'.");
+ }
+ }
+ }
+
+ public static Table applyOptions(Table table, Map<String, String> options)
{
+ Map<String, String> tableOptions = userOptions(options);
+ validateOptions(tableOptions);
+ Map<String, String> isolatedOptions = new HashMap<>(tableOptions);
+ if (hasStartupOptions(tableOptions)) {
+ // Startup mode, position, and range form one inherited state
family in Paimon. Clear
+ // absent members so one relation cannot accidentally reuse
another relation's read state.
+ INHERITED_READ_STATE_KEYS.stream()
+ .filter(key -> !tableOptions.containsKey(key))
+ .forEach(key -> isolatedOptions.put(key, null));
+ }
+ return table.copy(isolatedOptions);
+ }
+
+ private static Set<String> inheritedReadStateKeys() {
+ ImmutableSet.Builder<String> keys = ImmutableSet.builder();
+ for (ConfigOption<?> option : Arrays.asList(
+ CoreOptions.SCAN_TIMESTAMP,
+ CoreOptions.SCAN_TIMESTAMP_MILLIS,
+ CoreOptions.SCAN_WATERMARK,
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS,
+ CoreOptions.SCAN_CREATION_TIME_MILLIS,
+ CoreOptions.SCAN_SNAPSHOT_ID,
+ CoreOptions.SCAN_TAG_NAME,
+ CoreOptions.SCAN_VERSION,
+ CoreOptions.SCAN_MODE,
+ CoreOptions.SCAN_BOUNDED_WATERMARK,
+ CoreOptions.INCREMENTAL_BETWEEN,
+ CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP,
+ CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE,
+ CoreOptions.INCREMENTAL_TO_AUTO_TAG)) {
+ keys.add(option.key());
+ for (FallbackKey fallbackKey : option.fallbackKeys()) {
+ keys.add(fallbackKey.getKey());
+ }
+ }
+ return keys.build();
+ }
+
+ public static boolean hasStartupOptions(Map<String, String> options) {
+ return options.containsKey(CoreOptions.SCAN_MODE.key())
+ ||
options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains);
+ }
+
+ public static boolean selectsSchema(Map<String, String> options) {
+ return hasStartupOptions(options);
+ }
+
+ public static boolean usesStatementSnapshot(Map<String, String> options) {
+ if
(options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains)) {
+ return false;
+ }
+ String mode = options.get(CoreOptions.SCAN_MODE.key());
+ return mode == null
+ || "default".equalsIgnoreCase(mode)
+ || "latest".equalsIgnoreCase(mode)
+ || "latest-full".equalsIgnoreCase(mode)
+ || "full".equalsIgnoreCase(mode);
+ }
+
+ public static Map<String, String> resolveOptions(Table table, Map<String,
String> options) {
+ validateOptions(options);
+ if (!hasStartupOptions(options)) {
+ return options;
+ }
+ if (usesStatementSnapshot(options)) {
+ String pinnedSnapshotId =
table.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key());
+ if (pinnedSnapshotId != null) {
+ return resolvedSnapshotOptions(options, pinnedSnapshotId);
+ }
+ }
+ if (!(table instanceof FileStoreTable)) {
+ throw new IllegalArgumentException("Paimon startup options require
a file-store data table.");
+ }
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ if
(options.containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())) {
+ return resolveFileCreationTime(
+ options,
+ fileStoreTable,
+
Long.parseLong(options.get(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())));
+ }
+ if (options.containsKey(CoreOptions.SCAN_CREATION_TIME_MILLIS.key())) {
+ long creationTime =
Long.parseLong(options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()));
+ Long previousSnapshotId = TimeTravelUtil.earlierThanTimeMills(
+ fileStoreTable.snapshotManager(),
+ fileStoreTable.changelogManager(),
+ creationTime,
+ fileStoreTable.coreOptions().changelogLifecycleDecoupled(),
+ true);
+ if (previousSnapshotId != null
+ &&
fileStoreTable.snapshotManager().snapshotExists(previousSnapshotId + 1)) {
+ return resolvedSnapshotOptions(options,
String.valueOf(previousSnapshotId + 1));
+ }
+ return resolveFileCreationTime(options, fileStoreTable,
creationTime);
+ }
+
+ Table selectedTable = applyOptions(table, options);
+ if
("compacted-full".equalsIgnoreCase(options.get(CoreOptions.SCAN_MODE.key()))) {
+ Long snapshotId = compactedFullSnapshotId((FileStoreTable)
selectedTable);
+ return snapshotId == null
+ ? resolvedEmptyOptions(options)
+ : resolvedSnapshotOptions(options,
String.valueOf(snapshotId));
+ }
+ Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest((FileStoreTable)
selectedTable);
+ if (snapshot == null) {
+ return resolvedEmptyOptions(options);
+ }
+ String tagName = selectedTagName(options, (FileStoreTable)
selectedTable);
+ if (tagName != null) {
+ // A tag owns a retained Snapshot copy even after the ordinary
snapshot file expires.
+ // Keep the canonical tag selector so planning reads that retained
metadata path.
+ return resolvedTagOptions(options, tagName);
Review Comment:
[P1] Pin the selected tag state, not only its mutable name
This keeps `scan.tag-name` so expired ordinary snapshots remain readable,
but both schema binding and scan initialization later call `applyOptions()` on
fresh base tables. Each Paimon `Table.copy` resolves the tag name again, so a
concurrent tag replacement can make one statement bind the old tag schema and
execute the new tag snapshot. Please retain an immutable selected tagged
table/snapshot that both phases reuse, and test replacing a tag between output
binding and split initialization.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -559,8 +597,17 @@ boolean shouldForceJniForSystemTable() {
}
PaimonSysExternalTable paimonSysExternalTable =
(PaimonSysExternalTable) externalTable;
String sysTableType = paimonSysExternalTable.getSysTableType();
- return PAIMON_BINLOG_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType)
- ||
PAIMON_AUDIT_LOG_SYSTEM_TABLE_TYPE.equalsIgnoreCase(sysTableType);
+ return PaimonScanParams.requiresPaimonReader(sysTableType);
Review Comment:
[P1] Make the V1 scanner honor required JNI routing
This forces a logical split and FE sets `reader_type=PAIMON_JNI`, but when
FileScannerV2 is disabled the V1 `FileScanner` ignores that field and chooses
`PaimonCppReader` whenever `enable_paimon_cpp_reader=true`. The C++ reader does
not consume the serialized processed table or the required
row-tracking/audit/binlog transformations, so these queries can execute on the
wrong table semantics. Please make V1 honor `PAIMON_JNI` before the session
override and add V1-disabled routing coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -610,11 +657,37 @@ public List<org.apache.paimon.table.source.Split>
getPaimonSplitFromAPI() throws
long startTime = System.currentTimeMillis();
try {
Table paimonTable = getProcessedTable();
+ Map<String, String> resolvedOptions = scanParams == null
+ ? Collections.emptyMap()
+ :
scanParams.getResolvedMapParams().orElse(Collections.emptyMap());
+ if (PaimonScanParams.isPinnedEmptyScan(resolvedOptions)) {
+ return Collections.emptyList();
+ }
+ Optional<Long> fileCreationTime =
PaimonScanParams.getPinnedFileCreationTime(resolvedOptions);
+ if (fileCreationTime.isPresent()) {
+ if (!(paimonTable instanceof FileStoreTable)) {
+ throw new UserException("Paimon file-creation OPTIONS
require a data table.");
+ }
+ FileStoreTable fileStoreTable = (FileStoreTable) paimonTable;
+ SnapshotReader snapshotReader =
fileStoreTable.newSnapshotReader()
Review Comment:
[P1] Preserve Paimon query authorization on the direct snapshot path
Normal `DataTableBatchScan.plan()` calls Paimon's `authQuery()` with the
projected field names; for REST catalogs with `query-auth.enabled`, that
reaches `Catalog.authTableQuery()` and can deny the table or columns. This
file-creation path instead constructs `SnapshotReader` directly and returns its
splits, bypassing that authorization hook entirely. A user denied by the normal
projected query can therefore plan the same files through
`@options('scan.file-creation-time-millis'=...)`. Please run the standard
table/column authorization before this direct read, or retain it through a scan
API that supports the pinned manifest filter, with an equivalent-denial REST
test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java:
##########
@@ -0,0 +1,391 @@
+// 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 org.apache.doris.analysis.TableScanParams;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.FallbackKey;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Validation and application rules for relation-scoped Paimon scan parameters.
+ */
+public final class PaimonScanParams {
+ private static final String PINNED_FILE_CREATION_TIME =
+ "doris.internal.paimon.file-creation-time-millis";
+ private static final String PINNED_EMPTY_SCAN =
"doris.internal.paimon.empty-scan";
+
+ private static final Set<String> QUERY_OPTION_KEYS = ImmutableSet.of(
+ CoreOptions.SCAN_MODE.key(),
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_VERSION.key(),
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
+ CoreOptions.SCAN_PLAN_SORT_PARTITION.key());
+
+ private static final Set<String> STARTUP_POSITION_KEYS = ImmutableSet.of(
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_VERSION.key());
+
+ private static final Set<String> INHERITED_READ_STATE_KEYS =
inheritedReadStateKeys();
+
+ // FilesScan enumerates the latest partitions before applying its
range-aware per-partition scan,
+ // so it cannot safely read a range when a partition in that range has
since been dropped.
+ private static final Set<String> INCREMENTAL_SYSTEM_TABLES =
ImmutableSet.of(
+ "audit_log", "binlog", "partitions", "ro", "row_tracking");
+
+ private static final Set<String> PAIMON_READER_SYSTEM_TABLES =
ImmutableSet.of(
+ "audit_log", "binlog", "row_tracking");
+
+ private static final Set<String> OPTIONS_SYSTEM_TABLES = ImmutableSet.of(
+ // A system table may advertise OPTIONS only when every
row-producing stage observes
+ // the selected snapshot; files and buckets still consult latest
metadata internally.
+ "audit_log", "binlog", "manifests", "partitions", "ro",
+ "row_tracking", "table_indexes");
+
+ private PaimonScanParams() {
+ }
+
+ public static void validateOptions(Map<String, String> options) {
+ if (options.containsKey(CoreOptions.SCAN_FALLBACK_BRANCH.key())) {
+ throw new IllegalArgumentException("Paimon query option '"
+ + CoreOptions.SCAN_FALLBACK_BRANCH.key()
+ + "' is not supported because it requires rebuilding the
table through the catalog factory.");
+ }
+
+ Set<String> unsupported = options.keySet().stream()
+ .filter(key -> !QUERY_OPTION_KEYS.contains(key))
+ .collect(Collectors.toSet());
+ if (!unsupported.isEmpty()) {
+ throw new IllegalArgumentException("Unsupported Paimon query
option(s): " + unsupported);
+ }
+
+ String scanMode = options.get(CoreOptions.SCAN_MODE.key());
+ if ("from-creation-timestamp".equalsIgnoreCase(scanMode)
+ && options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()) ==
null) {
+ // Paimon 1.3.1 does not validate this newer mode, but its
starting scanner
+ // requires the creation timestamp and otherwise fails after
analysis.
+ throw new IllegalArgumentException("Paimon scan mode
'from-creation-timestamp' requires query option '"
+ + CoreOptions.SCAN_CREATION_TIME_MILLIS.key() + "'.");
+ }
+
+ long positionCount =
options.keySet().stream().filter(STARTUP_POSITION_KEYS::contains).count();
+ if (positionCount > 1) {
+ throw new IllegalArgumentException(
+ "Only one Paimon startup position can be specified: " +
STARTUP_POSITION_KEYS);
+ }
+
+ if (options.containsKey(CoreOptions.SCAN_MODE.key()) && positionCount
== 1) {
+ String position = options.keySet().stream()
+ .filter(STARTUP_POSITION_KEYS::contains)
+ .findFirst()
+ .get();
+ String mode = scanMode.toLowerCase(Locale.ROOT);
+ if (!isCompatibleStartupMode(position, mode)) {
+ throw new IllegalArgumentException("Paimon scan mode '" + mode
+ + "' is incompatible with startup position '" +
position + "'.");
+ }
+ }
+ }
+
+ public static Table applyOptions(Table table, Map<String, String> options)
{
+ Map<String, String> tableOptions = userOptions(options);
+ validateOptions(tableOptions);
+ Map<String, String> isolatedOptions = new HashMap<>(tableOptions);
+ if (hasStartupOptions(tableOptions)) {
+ // Startup mode, position, and range form one inherited state
family in Paimon. Clear
+ // absent members so one relation cannot accidentally reuse
another relation's read state.
+ INHERITED_READ_STATE_KEYS.stream()
+ .filter(key -> !tableOptions.containsKey(key))
+ .forEach(key -> isolatedOptions.put(key, null));
+ }
+ return table.copy(isolatedOptions);
+ }
+
+ private static Set<String> inheritedReadStateKeys() {
+ ImmutableSet.Builder<String> keys = ImmutableSet.builder();
+ for (ConfigOption<?> option : Arrays.asList(
+ CoreOptions.SCAN_TIMESTAMP,
+ CoreOptions.SCAN_TIMESTAMP_MILLIS,
+ CoreOptions.SCAN_WATERMARK,
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS,
+ CoreOptions.SCAN_CREATION_TIME_MILLIS,
+ CoreOptions.SCAN_SNAPSHOT_ID,
+ CoreOptions.SCAN_TAG_NAME,
+ CoreOptions.SCAN_VERSION,
+ CoreOptions.SCAN_MODE,
+ CoreOptions.SCAN_BOUNDED_WATERMARK,
+ CoreOptions.INCREMENTAL_BETWEEN,
+ CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP,
+ CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE,
+ CoreOptions.INCREMENTAL_TO_AUTO_TAG)) {
+ keys.add(option.key());
+ for (FallbackKey fallbackKey : option.fallbackKeys()) {
+ keys.add(fallbackKey.getKey());
+ }
+ }
+ return keys.build();
+ }
+
+ public static boolean hasStartupOptions(Map<String, String> options) {
+ return options.containsKey(CoreOptions.SCAN_MODE.key())
+ ||
options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains);
+ }
+
+ public static boolean selectsSchema(Map<String, String> options) {
+ return hasStartupOptions(options);
+ }
+
+ public static boolean usesStatementSnapshot(Map<String, String> options) {
+ if
(options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains)) {
+ return false;
+ }
+ String mode = options.get(CoreOptions.SCAN_MODE.key());
+ return mode == null
+ || "default".equalsIgnoreCase(mode)
+ || "latest".equalsIgnoreCase(mode)
+ || "latest-full".equalsIgnoreCase(mode)
+ || "full".equalsIgnoreCase(mode);
+ }
+
+ public static Map<String, String> resolveOptions(Table table, Map<String,
String> options) {
+ validateOptions(options);
+ if (!hasStartupOptions(options)) {
+ return options;
+ }
+ if (usesStatementSnapshot(options)) {
+ String pinnedSnapshotId =
table.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key());
+ if (pinnedSnapshotId != null) {
+ return resolvedSnapshotOptions(options, pinnedSnapshotId);
+ }
+ }
+ if (!(table instanceof FileStoreTable)) {
+ throw new IllegalArgumentException("Paimon startup options require
a file-store data table.");
+ }
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ if
(options.containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())) {
+ return resolveFileCreationTime(
+ options,
+ fileStoreTable,
+
Long.parseLong(options.get(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())));
+ }
+ if (options.containsKey(CoreOptions.SCAN_CREATION_TIME_MILLIS.key())) {
+ long creationTime =
Long.parseLong(options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()));
+ Long previousSnapshotId = TimeTravelUtil.earlierThanTimeMills(
+ fileStoreTable.snapshotManager(),
+ fileStoreTable.changelogManager(),
+ creationTime,
+ fileStoreTable.coreOptions().changelogLifecycleDecoupled(),
+ true);
+ if (previousSnapshotId != null
+ &&
fileStoreTable.snapshotManager().snapshotExists(previousSnapshotId + 1)) {
+ return resolvedSnapshotOptions(options,
String.valueOf(previousSnapshotId + 1));
+ }
+ return resolveFileCreationTime(options, fileStoreTable,
creationTime);
+ }
+
+ Table selectedTable = applyOptions(table, options);
+ if
("compacted-full".equalsIgnoreCase(options.get(CoreOptions.SCAN_MODE.key()))) {
+ Long snapshotId = compactedFullSnapshotId((FileStoreTable)
selectedTable);
+ return snapshotId == null
+ ? resolvedEmptyOptions(options)
+ : resolvedSnapshotOptions(options,
String.valueOf(snapshotId));
+ }
+ Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest((FileStoreTable)
selectedTable);
+ if (snapshot == null) {
+ return resolvedEmptyOptions(options);
+ }
+ String tagName = selectedTagName(options, (FileStoreTable)
selectedTable);
+ if (tagName != null) {
+ // A tag owns a retained Snapshot copy even after the ordinary
snapshot file expires.
+ // Keep the canonical tag selector so planning reads that retained
metadata path.
+ return resolvedTagOptions(options, tagName);
+ }
+ return resolvedSnapshotOptions(options, String.valueOf(snapshot.id()));
+ }
+
+ private static String selectedTagName(Map<String, String> options,
FileStoreTable selectedTable) {
+ String tagName = options.get(CoreOptions.SCAN_TAG_NAME.key());
+ if (tagName != null) {
+ return tagName;
+ }
+ // Paimon normalizes a tag-valued scan.version while copying the
selected table.
+ return selectedTable.options().get(CoreOptions.SCAN_TAG_NAME.key());
+ }
+
+ private static Long compactedFullSnapshotId(FileStoreTable table) {
+ CoreOptions coreOptions = table.coreOptions();
+ int deltaCommits = coreOptions.toConfiguration()
+ .getOptional(CoreOptions.FULL_COMPACTION_DELTA_COMMITS)
+ .orElse(1);
+ if (coreOptions.changelogProducer() ==
CoreOptions.ChangelogProducer.FULL_COMPACTION
+ ||
coreOptions.toConfiguration().contains(CoreOptions.FULL_COMPACTION_DELTA_COMMITS))
{
+ return table.snapshotManager().pickOrLatest(snapshot ->
+ snapshot.commitKind() == Snapshot.CommitKind.COMPACT
+ &&
FullCompactedStartingScanner.isFullCompactedIdentifier(
+ snapshot.commitIdentifier(),
deltaCommits));
+ }
+ // COMPACTED_FULL means the newest compact snapshot, which can be
older than latest.
+ return table.snapshotManager().pickOrLatest(
+ snapshot -> snapshot.commitKind() ==
Snapshot.CommitKind.COMPACT);
+ }
+
+ private static Map<String, String> resolveFileCreationTime(
+ Map<String, String> options, FileStoreTable table, long
creationTime) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ Optional<Snapshot> latestSnapshot = table.latestSnapshot();
+ if (latestSnapshot.isPresent()) {
+ resolved.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ String.valueOf(latestSnapshot.get().id()));
+ } else {
+ resolved.put(PINNED_EMPTY_SCAN, Boolean.TRUE.toString());
+ }
+ // Paimon's file-creation scanner consults latest lazily. Keep its
filter as internal
+ // metadata while replacing the live latest lookup with the snapshot
fixed above.
+ resolved.put(PINNED_FILE_CREATION_TIME, String.valueOf(creationTime));
+ return resolved;
+ }
+
+ private static Map<String, String> resolvedSnapshotOptions(
+ Map<String, String> options, String snapshotId) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ resolved.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId);
+ return resolved;
+ }
+
+ private static Map<String, String> resolvedTagOptions(Map<String, String>
options, String tagName) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ resolved.put(CoreOptions.SCAN_TAG_NAME.key(), tagName);
+ return resolved;
+ }
+
+ private static Map<String, String> resolvedEmptyOptions(Map<String,
String> options) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ // An empty table is also a statement state. Remember it explicitly so
a commit between
+ // binding and split planning cannot turn the relation into a
non-empty scan.
+ resolved.put(PINNED_EMPTY_SCAN, Boolean.TRUE.toString());
+ return resolved;
+ }
+
+ private static Map<String, String> userOptions(Map<String, String>
options) {
+ return options.entrySet().stream()
+ .filter(entry ->
!entry.getKey().startsWith("doris.internal.paimon."))
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
+ }
+
+ public static Optional<Long> getPinnedFileCreationTime(Map<String, String>
options) {
+ return
Optional.ofNullable(options.get(PINNED_FILE_CREATION_TIME)).map(Long::parseLong);
+ }
+
+ public static boolean isPinnedEmptyScan(Map<String, String> options) {
+ return Boolean.parseBoolean(options.get(PINNED_EMPTY_SCAN));
+ }
+
+ public static Map<String, String> isolateIncrementalRead(Map<String,
String> incrementalOptions) {
+ Map<String, String> isolatedOptions = new HashMap<>();
+ INHERITED_READ_STATE_KEYS.forEach(key -> isolatedOptions.put(key,
null));
+ isolatedOptions.putAll(incrementalOptions);
+ return isolatedOptions;
+ }
+
+ private static boolean isCompatibleStartupMode(String position, String
mode) {
+ if ("default".equals(mode)) {
+ return true;
+ }
+ if (CoreOptions.SCAN_TIMESTAMP.key().equals(position)
+ || CoreOptions.SCAN_TIMESTAMP_MILLIS.key().equals(position)) {
+ return "from-timestamp".equals(mode);
+ }
+ if (CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key().equals(position))
{
+ return "from-file-creation-time".equals(mode);
+ }
+ if (CoreOptions.SCAN_CREATION_TIME_MILLIS.key().equals(position)) {
+ return "from-creation-timestamp".equals(mode);
+ }
+ return "from-snapshot".equals(mode)
Review Comment:
[P2] Canonicalize numeric scan.version before mode validation
A digit-only `scan.version` is Paimon's alias for `scan.snapshot-id`; Paimon
1.3.1 converts it during `Table.copy()` before validating that
`from-snapshot-full` has a snapshot ID. This pre-validation looks only at the
raw key, so Doris rejects a combination the bundled dependency accepts and
implements. Please canonicalize the version first (or classify a digit-only
value as snapshot ID here) and add a positive full-snapshot test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonSysExternalTable.java:
##########
@@ -133,6 +134,29 @@ public Table getSysPaimonTable() {
return paimonSysTable;
}
+ public Table getSysPaimonTable(TableScanParams scanParams) {
+ Table table = getSysPaimonTable();
+ if (scanParams == null || !scanParams.isOptions()) {
+ return table;
+ }
+ Map<String, String> resolvedOptions = scanParams.getOrResolveMapParams(
+ options ->
PaimonScanParams.resolveOptions(sourceTable.getBasePaimonTable(), options));
Review Comment:
[P1] Resolve system-table latest OPTIONS from the statement snapshot
Selector-free system-table OPTIONS are classified as latest statement reads,
but this resolver uses `sourceTable.getBasePaimonTable()` directly. Meta-table
binding returns before the normal source-table `loadSnapshots()` path, whereas
data-table OPTIONS resolve from the statement-pinned table. An external commit
between binding data and system aliases can thus make one statement observe
different source commits. Please load/reuse the source table's MVCC snapshot
here and add a mixed data/system-alias consistency test.
##########
regression-test/suites/external_table_p0/PAIMON_ICEBERG_READ_WRITE_P0_COVERAGE.md:
##########
@@ -0,0 +1,102 @@
+<!--
+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.
+-->
+
+# Paimon / Iceberg Read and Write P0 Coverage
+
+## Scope
+
+This matrix compares the supported Doris surface with the format capabilities
documented by
+[Apache Paimon](https://paimon.apache.org/docs/1.0/), its
+[ecosystem matrix](https://paimon.apache.org/docs/master/ecosystem/), and
Iceberg's
+[write](https://iceberg.apache.org/docs/latest/spark-writes/) and
+[evolution](https://iceberg.apache.org/docs/latest/evolution/) documentation.
A format feature is a Doris P0 contract
+only when Doris supports it. Unsupported write or format boundaries require
deterministic negative
+coverage and no metadata or data mutation.
+
+Doris currently documents Paimon data access as read-only. Master also
supports Paimon catalog,
+database and table metadata creation/deletion, but it has no Paimon data sink.
Iceberg supports read,
+DDL, INSERT, INSERT OVERWRITE, CTAS and, for compatible V2/V3 tables, DELETE,
UPDATE and MERGE INTO.
+
+## Audit result
+
+The master inventory before this change contained 41 Paimon suites and 145
Iceberg suites, including
+35 Iceberg write suites and four dedicated Iceberg DML suites. Iceberg P0 is
complete for the Doris-
+supported read/write surface: its existing matrices cover file-format
versions, delete encodings,
+evolution, time travel, DDL, distributed writes, row-level DML, failure
atomicity and unsupported
+boundaries.
+
+Paimon P0 was incomplete. No suite selected a `merge-engine`, and write
rejection was not checked
+across all DML shapes with pre/post data and snapshot invariants. PM01-PM03
below close those gaps.
+PM04 exposes one remaining product defect as an opt-in negative regression:
failed Paimon CTAS leaves
+target metadata. Streaming writes, overwrite, delete/update and merge listed
as unsupported by the
+Paimon ecosystem matrix are boundary tests rather than positive Doris P0
contracts.
+
+## Risks
+
+| ID | Risk | Source | Impact | Priority |
+| --- | --- | --- | --- | --- |
+| R01 | Append and primary-key tables are planned with the wrong merge
semantics | Black box: Paimon table models | Silent wrong results | P0 |
+| R02 | Deduplicate, partial-update, aggregation and first-row produce the
same result for duplicate keys | White box: LSM sorted-run merge | Silent wrong
results | P0 |
+| R03 | Fixed and dynamic bucket tables expose duplicate or stale
cross-partition keys | Paimon data distribution | Silent wrong results | P0 |
+| R04 | Automatic and forced-JNI routing disagree on Paimon merged rows, or
native raw-file reads disagree where conversion is supported | Doris split
routing | Query correctness | P0 |
+| R05 | A rejected Paimon write creates a snapshot, changes data or leaves a
CTAS table | Doris read-only boundary | Data or metadata mutation | P0 |
+| R06 | Iceberg schema or partition evolution binds old field/spec IDs during
current or historical reads | Iceberg evolution | Silent wrong results | P0 |
+| R07 | Position/equality deletes or V3 deletion vectors are applied to the
wrong file or snapshot | Iceberg row-level deletes | Deleted data visible or
live data lost | P0 |
+| R08 | Iceberg writes lose rows, route them to the wrong transform, or
publish partial failed commits | Doris distributed Iceberg sink | Data loss or
corruption | P0 |
+| R09 | Snapshot, tag or branch reads/writes leak schema or data across
references | Both formats | Historical data corruption | P0 |
+| R10 | Unsupported format, DML mode or CTAS mutates state before rejection |
Capability boundary | Partial commits or orphan metadata | P0 |
+
+## Feature matrix
+
+| Format | Capability | P0 status | Main suites |
+| --- | --- | --- | --- |
+| Paimon | Append table, partitioned table, primitive and nested types |
Covered | `test_paimon_catalog`, `test_paimon_partition_table`,
`test_paimon_full_schema_change` |
+| Paimon | Primary-key deduplicate, partial-update, aggregation and first-row
| Covered | `test_paimon_merge_engine_matrix` |
+| Paimon | Fixed bucket, dynamic bucket and cross-partition update | Covered |
`test_paimon_merge_engine_matrix`, `test_paimon_partition_pk_delete_refs` |
+| Paimon | Parquet/ORC and mixed-format reads; JNI/native parity | Covered |
`test_paimon_merge_engine_matrix`, `paimon_tb_mix_format`,
`test_paimon_cpp_reader` |
Review Comment:
[P2] Require evidence that the parity runs use different readers
The cited `test_paimon_cpp_reader` toggles `enable_paimon_cpp_reader`, but
that flag is consumed only by legacy V1 while FileScannerV2 defaults on. The
other cited suites compare forced JNI with automatic routing without asserting
that the automatic run contains any native split, so all comparisons can
exercise JNI and still pass. Please add EXPLAIN assertions for JNI versus
`paimonNativeReadSplits`/NATIVE (as `test_paimon_partition_pk_delete_refs`
does), or cite and extend a suite that proves both routes before marking parity
covered.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/ExternalTablePreloadInfo.java:
##########
@@ -50,6 +50,7 @@ public boolean hasNonLatestRelation() {
}
public boolean shouldPreloadLatestSnapshot() {
- return hasLatestOnlyRelation && !hasNonLatestRelation;
+ // A historical alias has independent scan state and must not cancel
the latest alias warmup.
+ return hasLatestOnlyRelation;
Review Comment:
[P1] Preserve the historical alias's snapshot when preloading latest
When one statement references this table through both a latest alias and a
historical alias, returning true here makes `PreloadExternalMetadata` install
the latest snapshot in `StatementContext.snapshots`. That map is keyed only by
table identity, and `loadSnapshots()` never replaces an existing entry, so the
later explicit snapshot/tag load is ignored. The historical alias can
consequently bind latest schema/partitions (or fail after a historical
rename/drop). Please either skip this preload when any non-latest alias exists
or key the cached snapshot by relation/selector, and extend the mixed-alias
test through historical binding rather than only asserting that latest preload
ran.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java:
##########
@@ -121,6 +141,91 @@ public Map<String, String> getCatalogOptionsMap() {
}
}
+ public Map<String, String> getTableOptionsMap() {
+ return tableOptionsMap;
+ }
+
+ /**
+ * Returns Catalog table options which are not explicitly configured by
the Paimon table.
+ *
+ * <p>The comparison is based on Paimon {@link ConfigOption}s so canonical
and fallback keys
+ * follow the same precedence rule.
+ */
+ public Map<String, String> getTableOptionsForCopy(Map<String, String>
currentTableOptions) {
+ if (tableOptionsMap.isEmpty() || currentTableOptions.isEmpty()) {
+ return tableOptionsMap;
+ }
+
+ Options existingOptions = new Options(currentTableOptions);
+ Map<String, String> optionsForCopy = new LinkedHashMap<>();
+ tableOptionsMap.forEach((key, value) -> {
+ ConfigOption<?> option = SUPPORTED_TABLE_OPTIONS.find(key);
+ if (!existingOptions.contains(option)) {
+ optionsForCopy.put(key, value);
+ }
+ });
+ return Collections.unmodifiableMap(optionsForCopy);
+ }
+
+ public static boolean isTableOptionProperty(String key) {
+ return key.toLowerCase(Locale.ROOT).startsWith(TABLE_OPTION_PREFIX);
+ }
+
+ private Map<String, String> extractTableOptions() {
+ Map<String, String> tableOptions = new LinkedHashMap<>();
+ origProps.forEach((key, value) -> {
+ if (isTableOptionProperty(key)) {
+ String tableOptionKey =
key.substring(TABLE_OPTION_PREFIX.length());
+ if (StringUtils.isBlank(tableOptionKey)) {
+ throw new IllegalArgumentException(
+ "Paimon table option name must not be empty after
prefix " + TABLE_OPTION_PREFIX);
+ }
+ validateTableOption(tableOptionKey, value);
+ tableOptions.put(tableOptionKey, value);
+ }
+ });
+ return Collections.unmodifiableMap(tableOptions);
+ }
+
+ private void validateTableOption(String key, String value) {
+ ConfigOption<?> option = SUPPORTED_TABLE_OPTIONS.find(key);
+ if (option == null) {
+ throw new IllegalArgumentException("Unsupported Paimon table
option '" + key
+ + "' for the bundled Paimon version");
+ }
+
+ try {
+ new Options(Collections.singletonMap(key, value)).get(option);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Invalid value for Paimon table
option '" + key + "': "
+ + e.getMessage(), e);
+ }
+ }
+
+ private static final class SupportedTableOptions {
+ /** Canonical and fallback option names which support direct lookup. */
+ private final Map<String, ConfigOption<?>> exactOptions;
+
+ private SupportedTableOptions(Map<String, ConfigOption<?>>
exactOptions) {
+ this.exactOptions = exactOptions;
+ }
+
+ private static SupportedTableOptions build() {
+ Map<String, ConfigOption<?>> exactOptions = new HashMap<>();
+ for (ConfigOption<?> option : CoreOptions.getOptions()) {
Review Comment:
[P1] Restrict catalog defaults to options safe across every internal copy
This allowlist includes every reflected `CoreOptions` entry, while
validation only parses its type. Canonical immutable keys such as
`merge-engine` fail on the later `table.copy()`, while fallback aliases can
evade that raw-key guard: `deduplicate.ignore-delete=true` is still resolved by
Paimon and can suppress retracts, changing visible PK rows. Even copy-legal
startup defaults are unsafe: `scan.mode=latest` later conflicts with the
projection loader's snapshot ID, while a catalog snapshot ID is silently
replaced by live latest. Please canonicalize names and define a copy- and
lifecycle-safe allowlist that excludes immutable aliases and read-state
selectors, tested through real nonempty-table initialization.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java:
##########
@@ -53,13 +59,25 @@ public abstract class AbstractPaimonProperties extends
MetastoreProperties {
public abstract String getPaimonCatalogType();
private static final String USER_PROPERTY_PREFIX = "paimon.";
+ private static final String DORIS_JNI_PROPERTY_PREFIX = "paimon.jni.";
+ /** The suffix after this prefix is passed to Paimon as a dynamic table
option. */
+ public static final String TABLE_OPTION_PREFIX = "paimon.table-option.";
+ private static final SupportedTableOptions SUPPORTED_TABLE_OPTIONS =
SupportedTableOptions.build();
+
+ private Map<String, String> tableOptionsMap = Collections.emptyMap();
protected AbstractPaimonProperties(Map<String, String> props) {
super(Type.PAIMON, props);
}
public abstract Catalog initializeCatalog(String catalogName,
List<StorageProperties> storagePropertiesList);
+ @Override
+ public void initNormalizeAndCheckProps() {
+ super.initNormalizeAndCheckProps();
+ tableOptionsMap = extractTableOptions();
Review Comment:
[P1] Roll back catalog properties on these validation failures
`ALTER CATALOG` first applies the new map through `tryModifyCatalogProps()`
and only then calls `checkProperties()`. The rollback in
`CatalogMgr.replayAlterCatalogProps` catches `DdlException`, but this new
validation throws `IllegalArgumentException` (and the metastore property path
does not translate it). A rejected value such as a non-integer
`read.batch-size` therefore remains in the running catalog even though no edit
log was written. Please translate these failures or roll back on every
validation exception, and assert the post-failure property map in an ALTER test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -95,15 +100,15 @@ public class PaimonScanNode extends FileQueryScanNode {
private static final String DORIS_END_TIMESTAMP = "endTimestamp";
private static final String DORIS_INCREMENTAL_BETWEEN_SCAN_MODE =
"incrementalBetweenScanMode";
private static final String PAIMON_PROPERTY_PREFIX = "paimon.";
- private static final String DORIS_ENABLE_JNI_IO_MANAGER =
"doris.enable_jni_io_manager";
- private static final String DORIS_JNI_IO_MANAGER_TMP_DIR =
"doris.jni_io_manager.tmp_dir";
- private static final String DORIS_JNI_IO_MANAGER_IMPL_CLASS =
"doris.jni_io_manager.impl_class";
+ private static final String DORIS_ENABLE_FILE_READER_ASYNC =
"jni.enable_file_reader_async";
+ private static final String DORIS_ENABLE_JNI_IO_MANAGER =
"jni.enable_jni_io_manager";
+ private static final String DORIS_JNI_IO_MANAGER_TMP_DIR =
"jni.io_manager.tmp_dir";
+ private static final String DORIS_JNI_IO_MANAGER_IMPL_CLASS =
"jni.io_manager.impl_class";
private static final List<String> BACKEND_PAIMON_OPTIONS = Arrays.asList(
DORIS_ENABLE_JNI_IO_MANAGER,
DORIS_JNI_IO_MANAGER_TMP_DIR,
- DORIS_JNI_IO_MANAGER_IMPL_CLASS);
- private static final String PAIMON_BINLOG_SYSTEM_TABLE_TYPE = "binlog";
- private static final String PAIMON_AUDIT_LOG_SYSTEM_TABLE_TYPE =
"audit_log";
+ DORIS_JNI_IO_MANAGER_IMPL_CLASS,
+ DORIS_ENABLE_FILE_READER_ASYNC);
Review Comment:
[P2] Make the forwarded async-reader option effective
This adds `jni.enable_file_reader_async` to the backend options, but neither
the Java scanner nor another consumer reads
`paimon.jni.enable_file_reader_async`; reader construction still uses Paimon's
unrelated `file-reader-async-threshold`. Toggling the catalog boolean therefore
has no effect. Please map it into the actual reader configuration before
`newReadBuilder()`, with true/false construction tests, or remove the
ineffective advertised option.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonScanParams.java:
##########
@@ -0,0 +1,391 @@
+// 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 org.apache.doris.analysis.TableScanParams;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.FallbackKey;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.snapshot.FullCompactedStartingScanner;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Validation and application rules for relation-scoped Paimon scan parameters.
+ */
+public final class PaimonScanParams {
+ private static final String PINNED_FILE_CREATION_TIME =
+ "doris.internal.paimon.file-creation-time-millis";
+ private static final String PINNED_EMPTY_SCAN =
"doris.internal.paimon.empty-scan";
+
+ private static final Set<String> QUERY_OPTION_KEYS = ImmutableSet.of(
+ CoreOptions.SCAN_MODE.key(),
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_VERSION.key(),
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
+ CoreOptions.SCAN_PLAN_SORT_PARTITION.key());
+
+ private static final Set<String> STARTUP_POSITION_KEYS = ImmutableSet.of(
+ CoreOptions.SCAN_TIMESTAMP.key(),
+ CoreOptions.SCAN_TIMESTAMP_MILLIS.key(),
+ CoreOptions.SCAN_WATERMARK.key(),
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_CREATION_TIME_MILLIS.key(),
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ CoreOptions.SCAN_TAG_NAME.key(),
+ CoreOptions.SCAN_VERSION.key());
+
+ private static final Set<String> INHERITED_READ_STATE_KEYS =
inheritedReadStateKeys();
+
+ // FilesScan enumerates the latest partitions before applying its
range-aware per-partition scan,
+ // so it cannot safely read a range when a partition in that range has
since been dropped.
+ private static final Set<String> INCREMENTAL_SYSTEM_TABLES =
ImmutableSet.of(
+ "audit_log", "binlog", "partitions", "ro", "row_tracking");
+
+ private static final Set<String> PAIMON_READER_SYSTEM_TABLES =
ImmutableSet.of(
+ "audit_log", "binlog", "row_tracking");
+
+ private static final Set<String> OPTIONS_SYSTEM_TABLES = ImmutableSet.of(
+ // A system table may advertise OPTIONS only when every
row-producing stage observes
+ // the selected snapshot; files and buckets still consult latest
metadata internally.
+ "audit_log", "binlog", "manifests", "partitions", "ro",
+ "row_tracking", "table_indexes");
+
+ private PaimonScanParams() {
+ }
+
+ public static void validateOptions(Map<String, String> options) {
+ if (options.containsKey(CoreOptions.SCAN_FALLBACK_BRANCH.key())) {
+ throw new IllegalArgumentException("Paimon query option '"
+ + CoreOptions.SCAN_FALLBACK_BRANCH.key()
+ + "' is not supported because it requires rebuilding the
table through the catalog factory.");
+ }
+
+ Set<String> unsupported = options.keySet().stream()
+ .filter(key -> !QUERY_OPTION_KEYS.contains(key))
+ .collect(Collectors.toSet());
+ if (!unsupported.isEmpty()) {
+ throw new IllegalArgumentException("Unsupported Paimon query
option(s): " + unsupported);
+ }
+
+ String scanMode = options.get(CoreOptions.SCAN_MODE.key());
+ if ("from-creation-timestamp".equalsIgnoreCase(scanMode)
+ && options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()) ==
null) {
+ // Paimon 1.3.1 does not validate this newer mode, but its
starting scanner
+ // requires the creation timestamp and otherwise fails after
analysis.
+ throw new IllegalArgumentException("Paimon scan mode
'from-creation-timestamp' requires query option '"
+ + CoreOptions.SCAN_CREATION_TIME_MILLIS.key() + "'.");
+ }
+
+ long positionCount =
options.keySet().stream().filter(STARTUP_POSITION_KEYS::contains).count();
+ if (positionCount > 1) {
+ throw new IllegalArgumentException(
+ "Only one Paimon startup position can be specified: " +
STARTUP_POSITION_KEYS);
+ }
+
+ if (options.containsKey(CoreOptions.SCAN_MODE.key()) && positionCount
== 1) {
+ String position = options.keySet().stream()
+ .filter(STARTUP_POSITION_KEYS::contains)
+ .findFirst()
+ .get();
+ String mode = scanMode.toLowerCase(Locale.ROOT);
+ if (!isCompatibleStartupMode(position, mode)) {
+ throw new IllegalArgumentException("Paimon scan mode '" + mode
+ + "' is incompatible with startup position '" +
position + "'.");
+ }
+ }
+ }
+
+ public static Table applyOptions(Table table, Map<String, String> options)
{
+ Map<String, String> tableOptions = userOptions(options);
+ validateOptions(tableOptions);
+ Map<String, String> isolatedOptions = new HashMap<>(tableOptions);
+ if (hasStartupOptions(tableOptions)) {
+ // Startup mode, position, and range form one inherited state
family in Paimon. Clear
+ // absent members so one relation cannot accidentally reuse
another relation's read state.
+ INHERITED_READ_STATE_KEYS.stream()
+ .filter(key -> !tableOptions.containsKey(key))
+ .forEach(key -> isolatedOptions.put(key, null));
+ }
+ return table.copy(isolatedOptions);
+ }
+
+ private static Set<String> inheritedReadStateKeys() {
+ ImmutableSet.Builder<String> keys = ImmutableSet.builder();
+ for (ConfigOption<?> option : Arrays.asList(
+ CoreOptions.SCAN_TIMESTAMP,
+ CoreOptions.SCAN_TIMESTAMP_MILLIS,
+ CoreOptions.SCAN_WATERMARK,
+ CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS,
+ CoreOptions.SCAN_CREATION_TIME_MILLIS,
+ CoreOptions.SCAN_SNAPSHOT_ID,
+ CoreOptions.SCAN_TAG_NAME,
+ CoreOptions.SCAN_VERSION,
+ CoreOptions.SCAN_MODE,
+ CoreOptions.SCAN_BOUNDED_WATERMARK,
+ CoreOptions.INCREMENTAL_BETWEEN,
+ CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP,
+ CoreOptions.INCREMENTAL_BETWEEN_SCAN_MODE,
+ CoreOptions.INCREMENTAL_TO_AUTO_TAG)) {
+ keys.add(option.key());
+ for (FallbackKey fallbackKey : option.fallbackKeys()) {
+ keys.add(fallbackKey.getKey());
+ }
+ }
+ return keys.build();
+ }
+
+ public static boolean hasStartupOptions(Map<String, String> options) {
+ return options.containsKey(CoreOptions.SCAN_MODE.key())
+ ||
options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains);
+ }
+
+ public static boolean selectsSchema(Map<String, String> options) {
+ return hasStartupOptions(options);
+ }
+
+ public static boolean usesStatementSnapshot(Map<String, String> options) {
+ if
(options.keySet().stream().anyMatch(STARTUP_POSITION_KEYS::contains)) {
+ return false;
+ }
+ String mode = options.get(CoreOptions.SCAN_MODE.key());
+ return mode == null
+ || "default".equalsIgnoreCase(mode)
+ || "latest".equalsIgnoreCase(mode)
+ || "latest-full".equalsIgnoreCase(mode)
+ || "full".equalsIgnoreCase(mode);
+ }
+
+ public static Map<String, String> resolveOptions(Table table, Map<String,
String> options) {
+ validateOptions(options);
+ if (!hasStartupOptions(options)) {
+ return options;
+ }
+ if (usesStatementSnapshot(options)) {
+ String pinnedSnapshotId =
table.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key());
+ if (pinnedSnapshotId != null) {
+ return resolvedSnapshotOptions(options, pinnedSnapshotId);
+ }
+ }
+ if (!(table instanceof FileStoreTable)) {
+ throw new IllegalArgumentException("Paimon startup options require
a file-store data table.");
+ }
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ if
(options.containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())) {
+ return resolveFileCreationTime(
+ options,
+ fileStoreTable,
+
Long.parseLong(options.get(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key())));
+ }
+ if (options.containsKey(CoreOptions.SCAN_CREATION_TIME_MILLIS.key())) {
+ long creationTime =
Long.parseLong(options.get(CoreOptions.SCAN_CREATION_TIME_MILLIS.key()));
+ Long previousSnapshotId = TimeTravelUtil.earlierThanTimeMills(
+ fileStoreTable.snapshotManager(),
+ fileStoreTable.changelogManager(),
+ creationTime,
+ fileStoreTable.coreOptions().changelogLifecycleDecoupled(),
+ true);
+ if (previousSnapshotId != null
+ &&
fileStoreTable.snapshotManager().snapshotExists(previousSnapshotId + 1)) {
+ return resolvedSnapshotOptions(options,
String.valueOf(previousSnapshotId + 1));
+ }
+ return resolveFileCreationTime(options, fileStoreTable,
creationTime);
+ }
+
+ Table selectedTable = applyOptions(table, options);
+ if
("compacted-full".equalsIgnoreCase(options.get(CoreOptions.SCAN_MODE.key()))) {
+ Long snapshotId = compactedFullSnapshotId((FileStoreTable)
selectedTable);
+ return snapshotId == null
+ ? resolvedEmptyOptions(options)
+ : resolvedSnapshotOptions(options,
String.valueOf(snapshotId));
+ }
+ Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest((FileStoreTable)
selectedTable);
+ if (snapshot == null) {
+ return resolvedEmptyOptions(options);
+ }
+ String tagName = selectedTagName(options, (FileStoreTable)
selectedTable);
+ if (tagName != null) {
+ // A tag owns a retained Snapshot copy even after the ordinary
snapshot file expires.
+ // Keep the canonical tag selector so planning reads that retained
metadata path.
+ return resolvedTagOptions(options, tagName);
+ }
+ return resolvedSnapshotOptions(options, String.valueOf(snapshot.id()));
+ }
+
+ private static String selectedTagName(Map<String, String> options,
FileStoreTable selectedTable) {
+ String tagName = options.get(CoreOptions.SCAN_TAG_NAME.key());
+ if (tagName != null) {
+ return tagName;
+ }
+ // Paimon normalizes a tag-valued scan.version while copying the
selected table.
+ return selectedTable.options().get(CoreOptions.SCAN_TAG_NAME.key());
+ }
+
+ private static Long compactedFullSnapshotId(FileStoreTable table) {
+ CoreOptions coreOptions = table.coreOptions();
+ int deltaCommits = coreOptions.toConfiguration()
+ .getOptional(CoreOptions.FULL_COMPACTION_DELTA_COMMITS)
+ .orElse(1);
+ if (coreOptions.changelogProducer() ==
CoreOptions.ChangelogProducer.FULL_COMPACTION
+ ||
coreOptions.toConfiguration().contains(CoreOptions.FULL_COMPACTION_DELTA_COMMITS))
{
+ return table.snapshotManager().pickOrLatest(snapshot ->
+ snapshot.commitKind() == Snapshot.CommitKind.COMPACT
+ &&
FullCompactedStartingScanner.isFullCompactedIdentifier(
+ snapshot.commitIdentifier(),
deltaCommits));
+ }
+ // COMPACTED_FULL means the newest compact snapshot, which can be
older than latest.
+ return table.snapshotManager().pickOrLatest(
+ snapshot -> snapshot.commitKind() ==
Snapshot.CommitKind.COMPACT);
+ }
+
+ private static Map<String, String> resolveFileCreationTime(
+ Map<String, String> options, FileStoreTable table, long
creationTime) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ Optional<Snapshot> latestSnapshot = table.latestSnapshot();
+ if (latestSnapshot.isPresent()) {
+ resolved.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ String.valueOf(latestSnapshot.get().id()));
+ } else {
+ resolved.put(PINNED_EMPTY_SCAN, Boolean.TRUE.toString());
+ }
+ // Paimon's file-creation scanner consults latest lazily. Keep its
filter as internal
+ // metadata while replacing the live latest lookup with the snapshot
fixed above.
+ resolved.put(PINNED_FILE_CREATION_TIME, String.valueOf(creationTime));
+ return resolved;
+ }
+
+ private static Map<String, String> resolvedSnapshotOptions(
+ Map<String, String> options, String snapshotId) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ resolved.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId);
+ return resolved;
+ }
+
+ private static Map<String, String> resolvedTagOptions(Map<String, String>
options, String tagName) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ resolved.put(CoreOptions.SCAN_TAG_NAME.key(), tagName);
+ return resolved;
+ }
+
+ private static Map<String, String> resolvedEmptyOptions(Map<String,
String> options) {
+ Map<String, String> resolved = new HashMap<>(options);
+ INHERITED_READ_STATE_KEYS.forEach(resolved::remove);
+ // An empty table is also a statement state. Remember it explicitly so
a commit between
+ // binding and split planning cannot turn the relation into a
non-empty scan.
+ resolved.put(PINNED_EMPTY_SCAN, Boolean.TRUE.toString());
+ return resolved;
+ }
+
+ private static Map<String, String> userOptions(Map<String, String>
options) {
+ return options.entrySet().stream()
+ .filter(entry ->
!entry.getKey().startsWith("doris.internal.paimon."))
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
+ }
+
+ public static Optional<Long> getPinnedFileCreationTime(Map<String, String>
options) {
+ return
Optional.ofNullable(options.get(PINNED_FILE_CREATION_TIME)).map(Long::parseLong);
+ }
+
+ public static boolean isPinnedEmptyScan(Map<String, String> options) {
+ return Boolean.parseBoolean(options.get(PINNED_EMPTY_SCAN));
+ }
+
+ public static Map<String, String> isolateIncrementalRead(Map<String,
String> incrementalOptions) {
+ Map<String, String> isolatedOptions = new HashMap<>();
+ INHERITED_READ_STATE_KEYS.forEach(key -> isolatedOptions.put(key,
null));
+ isolatedOptions.putAll(incrementalOptions);
+ return isolatedOptions;
+ }
+
+ private static boolean isCompatibleStartupMode(String position, String
mode) {
+ if ("default".equals(mode)) {
+ return true;
+ }
+ if (CoreOptions.SCAN_TIMESTAMP.key().equals(position)
+ || CoreOptions.SCAN_TIMESTAMP_MILLIS.key().equals(position)) {
+ return "from-timestamp".equals(mode);
+ }
+ if (CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key().equals(position))
{
+ return "from-file-creation-time".equals(mode);
+ }
+ if (CoreOptions.SCAN_CREATION_TIME_MILLIS.key().equals(position)) {
+ return "from-creation-timestamp".equals(mode);
+ }
+ return "from-snapshot".equals(mode)
+ || (CoreOptions.SCAN_SNAPSHOT_ID.key().equals(position)
+ && "from-snapshot-full".equals(mode));
+ }
+
+ public static boolean supportsIncrementalRead(String systemTableType) {
+ return systemTableType != null &&
INCREMENTAL_SYSTEM_TABLES.contains(systemTableType.toLowerCase());
+ }
+
+ public static boolean supportsOptions(String systemTableType) {
+ return systemTableType != null &&
OPTIONS_SYSTEM_TABLES.contains(systemTableType.toLowerCase());
+ }
+
+ public static boolean requiresPaimonReader(String systemTableType) {
+ // Range support and reader requirements are independent capabilities.
File-backed system
+ // tables can honor INCR while still using Doris' native reader.
Unknown mocked or legacy
+ // system-table types retain the previous behavior and do not force
the Paimon reader.
+ return systemTableType != null &&
PAIMON_READER_SYSTEM_TABLES.contains(systemTableType.toLowerCase());
+ }
+
+ public static void validateSystemTable(String systemTableType,
TableScanParams scanParams) {
+ if (scanParams == null) {
+ return;
+ }
+ if (scanParams.incrementalRead() &&
!supportsIncrementalRead(systemTableType)) {
+ throw new IllegalArgumentException(
+ "Paimon system table '" + systemTableType + "' does not
support INCR scan params.");
+ }
+ if (scanParams.isOptions() && !supportsOptions(systemTableType)) {
+ throw new IllegalArgumentException(
+ "Paimon system table '" + systemTableType + "' does not
support OPTIONS scan params.");
+ }
+ if (scanParams.isOptions()
+ &&
scanParams.getMapParams().containsKey(CoreOptions.SCAN_FILE_CREATION_TIME_MILLIS.key()))
{
Review Comment:
[P2] Make system-table creation-time support independent of retention
This rejects `scan.file-creation-time-millis` but still admits
`scan.creation-time-millis` for every OPTIONS-capable system table. The latter
resolves to a snapshot only when `previousSnapshotId + 1` is retained; at a
boundary or after expiration it falls back to the same internal file-creation
predicate, which `PaimonSysExternalTable` then rejects. Identical SQL can
therefore switch from success to failure as retention changes. Please reject
both forms up front or implement the filter for these wrappers, and test a
supported system table at the retention boundary.
##########
regression-test/data/external_table_p0/paimon/paimon_system_table.out:
##########
@@ -65,3 +65,12 @@ delta_record_count bigint Yes true \N NONE
changelog_record_count bigint Yes true \N NONE
watermark bigint Yes true \N NONE
+-- !files_at_first_snapshot --
Review Comment:
[P2] Remove or execute these orphan expected-output blocks
The suite now executes only `order_qt_files_without_options`; there are no
`files_at_first_snapshot` or `files_at_latest_snapshot` query tags anywhere in
`paimon_system_table.groovy`. These two added blocks therefore are not produced
by the suite and leave the checked-in output out of correspondence. Please
remove them, or restore real tagged queries only for a system-table selector
that remains supported.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractPaimonProperties.java:
##########
@@ -121,6 +141,91 @@ public Map<String, String> getCatalogOptionsMap() {
}
}
+ public Map<String, String> getTableOptionsMap() {
+ return tableOptionsMap;
+ }
+
+ /**
+ * Returns Catalog table options which are not explicitly configured by
the Paimon table.
+ *
+ * <p>The comparison is based on Paimon {@link ConfigOption}s so canonical
and fallback keys
+ * follow the same precedence rule.
+ */
+ public Map<String, String> getTableOptionsForCopy(Map<String, String>
currentTableOptions) {
+ if (tableOptionsMap.isEmpty() || currentTableOptions.isEmpty()) {
+ return tableOptionsMap;
+ }
+
+ Options existingOptions = new Options(currentTableOptions);
+ Map<String, String> optionsForCopy = new LinkedHashMap<>();
+ tableOptionsMap.forEach((key, value) -> {
+ ConfigOption<?> option = SUPPORTED_TABLE_OPTIONS.find(key);
+ if (!existingOptions.contains(option)) {
+ optionsForCopy.put(key, value);
+ }
+ });
+ return Collections.unmodifiableMap(optionsForCopy);
+ }
+
+ public static boolean isTableOptionProperty(String key) {
+ return key.toLowerCase(Locale.ROOT).startsWith(TABLE_OPTION_PREFIX);
+ }
+
+ private Map<String, String> extractTableOptions() {
+ Map<String, String> tableOptions = new LinkedHashMap<>();
+ origProps.forEach((key, value) -> {
+ if (isTableOptionProperty(key)) {
+ String tableOptionKey =
key.substring(TABLE_OPTION_PREFIX.length());
+ if (StringUtils.isBlank(tableOptionKey)) {
+ throw new IllegalArgumentException(
+ "Paimon table option name must not be empty after
prefix " + TABLE_OPTION_PREFIX);
+ }
+ validateTableOption(tableOptionKey, value);
+ tableOptions.put(tableOptionKey, value);
+ }
+ });
+ return Collections.unmodifiableMap(tableOptions);
+ }
+
+ private void validateTableOption(String key, String value) {
+ ConfigOption<?> option = SUPPORTED_TABLE_OPTIONS.find(key);
+ if (option == null) {
+ throw new IllegalArgumentException("Unsupported Paimon table
option '" + key
+ + "' for the bundled Paimon version");
+ }
+
+ try {
+ new Options(Collections.singletonMap(key, value)).get(option);
Review Comment:
[P1] Reject a zero Paimon read batch size
Type parsing accepts `paimon.table-option.read.batch-size=0`. In Paimon
1.3.1 the Parquet reader then advances by `min(batchSize, remaining)`, returns
a non-null zero-row batch, and never increments its row position.
`PaimonJniScanner` exhausts that empty iterator and immediately calls
`readBatch()` again, so a nonempty scan spins without reaching EOF. Please
require a strictly positive value and add zero-value CREATE/ALTER plus scanner
forward-progress coverage.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -95,15 +100,15 @@ public class PaimonScanNode extends FileQueryScanNode {
private static final String DORIS_END_TIMESTAMP = "endTimestamp";
private static final String DORIS_INCREMENTAL_BETWEEN_SCAN_MODE =
"incrementalBetweenScanMode";
private static final String PAIMON_PROPERTY_PREFIX = "paimon.";
- private static final String DORIS_ENABLE_JNI_IO_MANAGER =
"doris.enable_jni_io_manager";
- private static final String DORIS_JNI_IO_MANAGER_TMP_DIR =
"doris.jni_io_manager.tmp_dir";
- private static final String DORIS_JNI_IO_MANAGER_IMPL_CLASS =
"doris.jni_io_manager.impl_class";
+ private static final String DORIS_ENABLE_FILE_READER_ASYNC =
"jni.enable_file_reader_async";
+ private static final String DORIS_ENABLE_JNI_IO_MANAGER =
"jni.enable_jni_io_manager";
Review Comment:
[P1] Preserve the existing IO-manager property namespace during upgrades
These are persisted catalog properties and FE/BE wire keys, but the patch
replaces `paimon.doris.*` everywhere instead of accepting it as a legacy alias.
Existing catalogs lose their configured IO manager after an FE upgrade, and
either direction of a rolling FE/BE upgrade sends keys the other side does not
consume, silently disabling spill management. Please accept both namespaces for
a compatibility window (new key wins) and cover a persisted old catalog plus
mixed-version wire contracts.
--
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]
