github-actions[bot] commented on code in PR #66297:
URL: https://github.com/apache/doris/pull/66297#discussion_r3689955962
##########
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:
[P1] Do not treat an empty visible option map as proof that the transported
cap is unnecessary. Paimon 1.3.1 ReadonlyTable.options() is always empty, and
PartitionsTable keeps the manifest-planning FileStoreTable private while
copy(dynamicOptions) forwards options to it. That is exactly why the FE sends
doris.scan.manifest.parallelism-cap for system wrappers. On an 8-core FE and
4-core BE, this branch sees no configured values and returns the serialized
wrapper unchanged, leaving the hidden planner at 8 instead of lowering it to 4.
This is distinct from the existing FE raw-handle thread: the FE copy was
already safe locally; the smaller BE loses its own cap after transport. Apply
the non-null transported cap through table.copy(...) even when options() is
empty, and add a serialized PartitionsTable/heterogeneous-cap test.
##########
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:
[P1] Carry the fenced schema/table generation into later aliases, not only
the snapshot ID. The statement fence captures (S, schema A), but both follow-on
paths reopen current table metadata instead of preserving that generation. In
options-first/plain-later, loadAtFence -> copyWithLatestSchema() produces a
later plain relation whose snapshot and tuple slots advertise A while its table
handle has schema B; after a rename or drop, PaimonScanNode fails with "Paimon
scan schema does not contain all bound Doris columns." In
plain-first/options-later, the later OPTIONS projection instead binds B while
the earlier plain relation stays on A, so the statement's visible schemas
depend on alias order. Rows are still pinned to S, so this is distinct from the
existing S/S+1 thread. Build every relation projection from the fenced
table/schema generation and test both alias orders across a schema-only ALTER.
##########
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:
[P1] Reuse Paimon's size grammar here instead of maintaining a different
one. FE validation delegates this option to Paimon 1.3.1 MemorySize, which
accepts kibibytes/mebibytes/gibibytes/tebibytes (not kib/mib/gib/tib). Thus
file-reader-async-threshold='16 mebibytes' passes CREATE/ALTER/relation
validation and is within 1 MB-1 GB, but this parser returns empty and
validateSerializedAsyncThreshold aborts both V1 and V2 scans during
initialization. Conversely, this parser accepts unit spellings Paimon rejects.
Parse with MemorySize/Options or mirror its exact grammar, and add transport
tests for the long IEC spellings and boundary values.
##########
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:
[P2] Classify reader-only tuning before adding the statement snapshot
selector. In production, these OPTIONS relations first pass through the fenced
overload, which rewrites resolvedOptions with pinOptionsToSnapshot(...) and
adds scan.snapshot-id. hasOnlyReaderOptions then sees that non-reader key, so
the first occurrence of each distinct reader-only tuning map takes the uncached
projection loader and enumerates every partition instead of reusing the
memoized latest projection; identical aliases merely reuse that result. The new
cache-reuse test calls loadSnapshot directly and therefore misses the fenced
path. Base the neutrality check on the original user options (or ignore the
fence-injected selector) and cover StatementContext.loadSnapshots for a
partitioned table.
--
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]
