github-actions[bot] commented on code in PR #66297:
URL: https://github.com/apache/doris/pull/66297#discussion_r3695921480
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/metacache/paimon/PaimonLatestSnapshotProjectionLoader.java:
##########
@@ -79,8 +124,10 @@ private PaimonSnapshot resolveLatestSnapshot(Table
paimonTable) {
// Pin the data snapshot for MVCC while retaining the latest table
schema. A normal
// copy applies time travel and falls back to the snapshot's
schema, which can be stale
// immediately after a schema change that has not produced a new
data snapshot.
+ // The lightweight fence must not validate physical planning
options before relation
+ // overrides are composed; validation belongs to
loadEffectiveAtFence's final copy.
snapshotTable = latestSchemaTable.copyWithoutTimeTravel(
-
Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(),
String.valueOf(latestSnapshotId)));
+ PaimonScanParams.isolateSnapshotRead(latestSnapshotId));
Review Comment:
[P1] Keep the full latest projection CPU-capped while leaving only the
lightweight fence neutral. Both load() and loadFence() call
resolveLatestSnapshot(), and this copy no longer applies
runtimeSafeCopyOptions(). On a 4-core FE with a cluster-valid physical or
catalog scan.manifest.parallelism=8 and a partitioned table, the ordinary
latest cache now sends 8 into PaimonPartitionInfoLoader, whose runtime
validation rejects it before partition enumeration; the unpartitioned path
reaches the later ScanNode cap, so the current tests miss this. Split
lightweight fence construction from full projection normalization (or
parameterize this helper), and cover a partitioned ordinary latest scan on a
smaller FE.
##########
fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -519,69 +622,173 @@ static Optional<Long> parseDataSizeBytes(String value) {
if (value == null || value.trim().isEmpty()) {
return Optional.empty();
}
- String normalized = value.trim().toLowerCase(Locale.ROOT).replace("_",
"").replace(" ", "");
- int unitStart = 0;
- while (unitStart < normalized.length()
- && (Character.isDigit(normalized.charAt(unitStart)) ||
normalized.charAt(unitStart) == '.')) {
- unitStart++;
- }
- if (unitStart == 0) {
- return Optional.empty();
- }
try {
- double number = Double.parseDouble(normalized.substring(0,
unitStart));
- String unit = normalized.substring(unitStart);
- long multiplier;
- switch (unit) {
- case "":
- case "b":
- case "byte":
- case "bytes":
- multiplier = 1L;
- break;
- case "k":
- case "kb":
- case "kib":
- multiplier = 1024L;
- break;
- case "m":
- case "mb":
- case "mib":
- multiplier = 1024L * 1024L;
- break;
- case "g":
- case "gb":
- case "gib":
- multiplier = 1024L * 1024L * 1024L;
- break;
- case "t":
- case "tb":
- case "tib":
- multiplier = 1024L * 1024L * 1024L * 1024L;
- break;
- default:
- return Optional.empty();
- }
- return Optional.of((long) (number * multiplier));
- } catch (NumberFormatException e) {
+ // Keep the BE guard's accepted grammar identical to the Paimon
option parser that will
+ // consume this value; accepting a superset lets invalid
serialized options reach scans.
+ return Optional.of(MemorySize.parse(value).getBytes());
+ } catch (IllegalArgumentException e) {
return Optional.empty();
}
}
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
+ String encodedSystemSource = params.get(PAIMON_OPTION_PREFIX +
DORIS_SERIALIZED_SYSTEM_SOURCE);
+ FileStoreTable systemSource = encodedSystemSource == null
+ ? null : PaimonUtils.deserialize(encodedSystemSource);
+ table = applyBackendManifestParallelism(table,
+ params.get(PAIMON_OPTION_PREFIX +
DORIS_MANIFEST_PARALLELISM_CAP),
+ Runtime.getRuntime().availableProcessors(), systemSource,
+ params.get(PAIMON_OPTION_PREFIX + DORIS_SYSTEM_TABLE_TYPE));
+ 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) {
+ return applyBackendManifestParallelism(
+ table, feParallelismCap, localCapacity, null, null);
+ }
+
+ static Table applyBackendManifestParallelism(
+ Table table, String feParallelismCap, int localCapacity,
+ FileStoreTable systemSource, String systemTableType) {
+ Table planningTable = systemSource == null ? table : systemSource;
+ List<Integer> configuredValues = new ArrayList<>();
+ collectManifestParallelism(planningTable, configuredValues);
+ // Old FEs do not send a cap, so the BE must still preserve the
hardware-independent
+ // ceiling that prevents one scan from growing Paimon's JVM-global
executor beyond 256.
+ int requestedBound = Math.min(localCapacity, MAX_MANIFEST_PARALLELISM);
+ if (feParallelismCap != null) {
+ requestedBound =
Math.min(parsePositiveManifestParallelism(feParallelismCap), requestedBound);
+ }
+ 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] Preserve the manifest cap for old-FE system-table payloads too. During
a rolling upgrade, an old FE sends serialized_table but none of the new
cap/source/type keys, so systemSource and feParallelismCap are null. For a
serialized PartitionsTable, options() is empty while its private storeTable
still owns scan.manifest.parallelism; configuredValues is therefore empty and
this return bypasses both min(localCapacity, 256) and the hidden planner. An
8-core old FE can thus send a wrapper pinned at 8 to a 4-core new BE, and
pre-PR values above 256 also escape the backstop. When source metadata is
absent, apply the computed legacy backstop through the wrapper copy (or version
and reject/reschedule the scan), and test an old-sender PartitionsTable with
none of the new keys.
--
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]
