Gabriel39 commented on code in PR #66297:
URL: https://github.com/apache/doris/pull/66297#discussion_r3690587611
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java:
##########
@@ -248,7 +263,11 @@ public BaseAnalysisTask createAnalysisTask(AnalysisInfo
info) {
public long fetchRowCount() {
makeSureInitialized();
long rowCount = 0;
- List<Split> splits =
getBasePaimonTable().newReadBuilder().newScan().plan().splits();
+ Table effectiveTable = getBasePaimonTable();
+ // Statistics and row-count cache planning run before ScanNode and
must not reach an
+ // unsafe manifest executor, even when the foreground relation later
supplies an override.
+ PaimonReaderOptions.validateEffectiveTable(effectiveTable);
Review Comment:
Valid. Manifest-planning consumers now build runtime-capped execution
copies, including hidden system sources.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -958,8 +961,61 @@ public Optional<MvccSnapshot> getSnapshot(TableIf tableIf)
{
if (!(tableIf instanceof MvccTable)) {
return Optional.empty();
}
- MvccTableInfo mvccTableInfo = new MvccTableInfo(tableIf);
- return Optional.ofNullable(snapshots.get(mvccTableInfo));
+ MvccTableInfo defaultKey = new MvccTableInfo(tableIf);
+ MvccSnapshot defaultSnapshot = snapshots.get(defaultKey);
+ if (defaultSnapshot != null) {
+ return Optional.of(defaultSnapshot);
+ }
+ MvccSnapshot only = null;
+ for (Map.Entry<MvccTableInfo, MvccSnapshot> entry :
snapshots.entrySet()) {
+ if (defaultKey.isSameTable(entry.getKey())) {
+ if (only != null) {
+ return Optional.empty();
Review Comment:
Valid. Table-only descriptor consumers use a safe pinned projection when
multiple relation projections exist.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java:
##########
@@ -967,8 +980,74 @@ public Optional<MvccSnapshot> getSnapshot(TableIf tableIf)
{
if (!(tableIf instanceof MvccTable)) {
return Optional.empty();
}
- MvccTableInfo mvccTableInfo = new MvccTableInfo(tableIf);
- return Optional.ofNullable(snapshots.get(mvccTableInfo));
+ MvccTableInfo defaultKey = new MvccTableInfo(tableIf);
+ MvccSnapshot defaultSnapshot = snapshots.get(defaultKey);
+ if (defaultSnapshot != null) {
+ return Optional.of(defaultSnapshot);
+ }
+ MvccSnapshot only = null;
+ for (Map.Entry<MvccTableInfo, MvccSnapshot> entry :
snapshots.entrySet()) {
+ if (defaultKey.isSameTable(entry.getKey())) {
+ if (only != null) {
+ return Optional.empty();
+ }
+ only = entry.getValue();
+ }
+ }
+ return Optional.ofNullable(only);
+ }
+
+ public Optional<MvccSnapshot> getSnapshot(TableIf tableIf,
Optional<TableSnapshot> tableSnapshot,
+ Optional<TableScanParams> scanParams) {
+ if (!(tableIf instanceof MvccTable)) {
+ return Optional.empty();
+ }
+ return Optional.ofNullable(snapshots.get(
+ new MvccTableInfo(tableIf, versionKeyOf(tableSnapshot,
scanParams))));
+ }
+
+ /**
+ * Return a validated statement projection for metadata consumers without
relation identity.
+ */
+ public Optional<MvccSnapshot> getSnapshotForTableMetadata(TableIf tableIf)
{
+ Optional<MvccSnapshot> unambiguous = getSnapshot(tableIf);
+ if (unambiguous.isPresent() || !(tableIf instanceof MvccTable)) {
+ return unambiguous;
+ }
+ // Descriptor serialization has no relation key. Reuse a validated
statement projection
+ // instead of reopening a neutral handle after multiple OPTIONS
aliases were bound.
+ return Optional.ofNullable(tableMetadataSnapshots.get(new
MvccTableInfo(tableIf)));
+ }
+
+ private static String versionKeyOf(Optional<TableSnapshot> tableSnapshot,
+ Optional<TableScanParams> scanParams) {
+ // Limit the backport to relation-scoped OPTIONS: branch-4.1's older
Iceberg/Paimon
+ // time-travel paths still resolve their handles outside this map,
while OPTIONS needs an
+ // exact content key shared by analysis and scan planning.
+ if (scanParams != null && scanParams.isPresent() &&
scanParams.get().isOptions()) {
+ TableScanParams params = scanParams.get();
+ StringBuilder key = new StringBuilder("p");
+ appendVersionKeyPart(key, params.getParamType());
+ Map<String, String> sortedParams = new
TreeMap<>(params.getMapParams());
Review Comment:
Valid. Latest snapshot identity is shared independently from
relation-specific planning projections.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonLatestSnapshotProjectionLoader.java:
##########
@@ -69,6 +71,44 @@ public PaimonSnapshotCacheValue load(NameMapping
nameMapping, Table paimonTable)
}
}
+ public PaimonSnapshotCacheValue loadFence(NameMapping nameMapping, Table
paimonTable) {
+ try {
+ // A statement fence needs version/schema identity only;
enumerating partitions here
+ // can fail before relation-level options have replaced an unsafe
physical setting.
+ return new PaimonSnapshotCacheValue(
+ PaimonPartitionInfo.EMPTY,
resolveLatestSnapshot(paimonTable));
+ } catch (Exception e) {
+ throw new CacheException("failed to load paimon snapshot fence
%s.%s.%s: %s",
+ e, nameMapping.getCtlId(), nameMapping.getLocalDbName(),
nameMapping.getLocalTblName(),
+ e.getMessage());
+ }
+ }
+
+ public PaimonSnapshotCacheValue loadAtFence(
+ NameMapping nameMapping, Table paimonTable, PaimonSnapshot fence) {
+ try {
+ FileStoreTable latestSchemaTable = ((FileStoreTable)
paimonTable).copyWithLatestSchema();
Review Comment:
Valid. Commit d6d387df114 carries the exact fenced table generation into
later aliases instead of reopening current metadata.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -571,17 +670,137 @@ static Optional<Long> parseDataSizeBytes(String value) {
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
+ table = applyBackendManifestParallelism(table,
+ params.get(PAIMON_OPTION_PREFIX +
DORIS_MANIFEST_PARALLELISM_CAP),
+ Runtime.getRuntime().availableProcessors());
+ table = applyDefaultReadBatchSize(table, batchSize);
+ paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
+ }
+ }
+
+ static Table applyDefaultReadBatchSize(Table table, int dorisBatchSize) {
+ validateSerializedReaderOptions(table);
+ if (hasReadBatchSize(table)) {
+ // Doris' output block size and Paimon's reader batch are
independent controls; an
+ // explicitly validated value on any hidden reader must survive
transport unchanged.
+ return table;
+ }
// The serialized table may pin an older data snapshot while carrying
the latest schema
// after a schema change. Applying a normal copy would time travel to
that snapshot's
// schema again and make renamed or newly added columns disappear.
Map<String, String> readOptions = Collections.singletonMap(
- CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(batchSize));
- table = table instanceof FileStoreTable
+ CoreOptions.READ_BATCH_SIZE.key(),
String.valueOf(dorisBatchSize));
+ return table instanceof FileStoreTable
? ((FileStoreTable) table).copyWithoutTimeTravel(readOptions)
: table.copy(readOptions);
- paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
- if (LOG.isDebugEnabled()) {
- LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
+ }
+
+ static Table applyBackendManifestParallelism(
+ Table table, String feParallelismCap, int localCapacity) {
+ List<Integer> configuredValues = new ArrayList<>();
+ collectManifestParallelism(table, configuredValues);
+ int requestedBound = localCapacity;
+ if (feParallelismCap != null) {
+ requestedBound =
Math.min(parsePositiveManifestParallelism(feParallelismCap), localCapacity);
+ }
+ final int safeBound = requestedBound;
+ // The FE cap is a requested bound, not proof that every serialized
wrapper carries it;
+ // a later table rebuild can expose the original physical value to
this BE.
+ if (configuredValues.isEmpty()
Review Comment:
Valid. Commit d6d387df114 transports the exact hidden system source and
rebuilds a real PartitionsTable under the smaller BE cap.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -571,17 +670,137 @@ static Optional<Long> parseDataSizeBytes(String value) {
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
+ table = applyBackendManifestParallelism(table,
+ params.get(PAIMON_OPTION_PREFIX +
DORIS_MANIFEST_PARALLELISM_CAP),
+ Runtime.getRuntime().availableProcessors());
+ table = applyDefaultReadBatchSize(table, batchSize);
+ paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
+ }
+ }
+
+ static Table applyDefaultReadBatchSize(Table table, int dorisBatchSize) {
+ validateSerializedReaderOptions(table);
+ if (hasReadBatchSize(table)) {
+ // Doris' output block size and Paimon's reader batch are
independent controls; an
+ // explicitly validated value on any hidden reader must survive
transport unchanged.
+ return table;
+ }
// The serialized table may pin an older data snapshot while carrying
the latest schema
// after a schema change. Applying a normal copy would time travel to
that snapshot's
// schema again and make renamed or newly added columns disappear.
Map<String, String> readOptions = Collections.singletonMap(
- CoreOptions.READ_BATCH_SIZE.key(), String.valueOf(batchSize));
- table = table instanceof FileStoreTable
+ CoreOptions.READ_BATCH_SIZE.key(),
String.valueOf(dorisBatchSize));
+ return table instanceof FileStoreTable
? ((FileStoreTable) table).copyWithoutTimeTravel(readOptions)
: table.copy(readOptions);
- paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
- if (LOG.isDebugEnabled()) {
- LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
+ }
+
+ static Table applyBackendManifestParallelism(
+ Table table, String feParallelismCap, int localCapacity) {
+ List<Integer> configuredValues = new ArrayList<>();
+ collectManifestParallelism(table, configuredValues);
+ int requestedBound = localCapacity;
+ if (feParallelismCap != null) {
+ requestedBound =
Math.min(parsePositiveManifestParallelism(feParallelismCap), localCapacity);
+ }
+ final int safeBound = requestedBound;
+ // The FE cap is a requested bound, not proof that every serialized
wrapper carries it;
+ // a later table rebuild can expose the original physical value to
this BE.
+ if (configuredValues.isEmpty()
+ || configuredValues.stream().noneMatch(value -> value >
safeBound)) {
+ return table;
+ }
+ int safeParallelism = Math.min(
+
configuredValues.stream().mapToInt(Integer::intValue).min().getAsInt(),
+ safeBound);
+ Map<String, String> cap = Collections.singletonMap(
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
String.valueOf(safeParallelism));
+ // Preserve the FE-selected schema while lowering only the BE-local
execution bound.
+ return table instanceof FileStoreTable
+ ? ((FileStoreTable) table).copyWithoutTimeTravel(cap)
+ : table.copy(cap);
+ }
+
+ private static int parsePositiveManifestParallelism(String value) {
+ try {
+ int parsed = Integer.parseInt(value);
+ if (parsed < 1) {
+ throw new IllegalArgumentException("Paimon manifest
parallelism cap must be positive.");
+ }
+ return parsed;
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException("Paimon manifest parallelism
cap must be an integer.", e);
+ }
+ }
+
+ private static void collectManifestParallelism(Table table, List<Integer>
values) {
+ String configured =
table.options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key());
+ if (configured != null) {
+ values.add(parsePositiveManifestParallelism(configured));
+ }
+ if (table instanceof FallbackReadFileStoreTable) {
+ collectManifestParallelism(((FallbackReadFileStoreTable)
table).fallback(), values);
+ }
+ if (table instanceof DelegatedFileStoreTable) {
+ collectManifestParallelism(((DelegatedFileStoreTable)
table).wrapped(), values);
+ }
+ }
+
+ private static boolean hasReadBatchSize(Table table) {
+ if (table.options().containsKey(CoreOptions.READ_BATCH_SIZE.key())) {
+ return true;
+ }
+ if (table instanceof FallbackReadFileStoreTable
+ && hasReadBatchSize(((FallbackReadFileStoreTable)
table).fallback())) {
+ return true;
+ }
+ return table instanceof DelegatedFileStoreTable
+ && hasReadBatchSize(((DelegatedFileStoreTable)
table).wrapped());
+ }
+
+ private static void validateSerializedReaderOptions(Table table) {
+
validateSerializedReadBatchSize(table.options().get(CoreOptions.READ_BATCH_SIZE.key()));
+
validateSerializedAsyncThreshold(table.options().get(CoreOptions.FILE_READER_ASYNC_THRESHOLD.key()));
+ if (table instanceof FallbackReadFileStoreTable) {
+ validateSerializedReaderOptions(((FallbackReadFileStoreTable)
table).fallback());
+ }
+ if (table instanceof DelegatedFileStoreTable) {
+ validateSerializedReaderOptions(((DelegatedFileStoreTable)
table).wrapped());
+ }
+ }
+
+ private static void validateSerializedAsyncThreshold(String value) {
+ if (value == null) {
+ return;
+ }
+ Optional<Long> bytes = parseDataSizeBytes(value);
Review Comment:
Valid. Commit d6d387df114 uses Paimon MemorySize parsing and covers long IEC
units plus boundaries.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalTable.java:
##########
@@ -191,6 +193,19 @@ private PaimonSnapshotCacheValue
getPaimonSnapshotCacheValue(Optional<TableSnaps
"Failed to get Paimon snapshot: " + (e.getMessage() ==
null ? "unknown cause" : e.getMessage()),
e);
}
+ } else if (scanParams.isPresent() && scanParams.get().isOptions()) {
+ Table baseTable = getBasePaimonTable();
+ Map<String, String> resolvedOptions =
scanParams.get().getOrResolveMapParams(
+ options -> PaimonScanParams.resolveOptions(baseTable,
options));
+ Table effectiveTable = PaimonScanParams.applyOptions(baseTable,
resolvedOptions);
+ if (PaimonScanParams.hasOnlyReaderOptions(resolvedOptions)) {
Review Comment:
Valid. Commit d6d387df114 classifies reader-only tuning from raw user
options before injecting the statement snapshot selector.
--
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]
