Gabriel39 commented on code in PR #66247:
URL: https://github.com/apache/doris/pull/66247#discussion_r3698073332
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -523,64 +627,146 @@ 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));
+ validateSerializedReaderOptions(table);
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);
+ int requestedBound = localCapacity;
Review Comment:
Fixed in 5e3567ff85. The old-FE backstop now always includes the stable 256
ceiling, including hidden fallback children, with null-cap/high-capacity
regression coverage.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -316,20 +322,48 @@ Table resolveTable(PaimonTableHandle paimonHandle) {
Table resolveScanTable(PaimonTableHandle paimonHandle) {
Table table = resolveTable(paimonHandle);
Map<String, String> scanOptions = paimonHandle.getScanOptions();
+ Table finalTable = table;
if (scanOptions != null && !scanOptions.isEmpty()) {
if (PaimonScanParams.isOptionsPin(scanOptions)) {
// An @options pin owns the whole scan-startup state:
applyOptions strips the internal
// markers and nulls out the absent members of paimon's
inherited read-state family, so a
// scan.mode / tag persisted on the base table cannot leak
into this relation's read.
- return PaimonScanParams.applyOptions(table, scanOptions);
+ finalTable = PaimonScanParams.applyOptions(table, scanOptions);
+ } else {
+ // FIX-INCR-SCAN-RESET: for an @incr read, reapply legacy's
null reset of
+ // scan.snapshot-id/scan.mode here (the single Table.copy
chokepoint shared by both the
+ // native/JNI scan path and the JNI serialized-table path) so
a stale persisted pin on the
+ // base table cannot hijack incremental-between.
Non-incremental pins pass through unchanged.
+ finalTable =
table.copy(PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
+ }
+ }
+ finalTable = PaimonReaderOptions.runtimeSafeTable(finalTable);
+ finalTable = runtimeSafeSystemTable(paimonHandle, finalTable,
scanOptions);
+ // This is the last common boundary before planning and serialization.
Normalize and
+ // validate only after relation > catalog > physical precedence is
established.
+ PaimonReaderOptions.validateEffectiveTable(finalTable);
+ return finalTable;
+ }
+
+ private Table runtimeSafeSystemTable(
+ PaimonTableHandle handle, Table systemTable, Map<String, String>
scanOptions) {
+ if (!handle.isSystemTable()) {
+ return systemTable;
+ }
+ try {
+ Table dataTable = handle.getSystemTableSource();
+ if (dataTable == null) {
+ dataTable = handle.getSysBaseTable();
+ }
+ if (dataTable == null) {
+ dataTable = catalogOps.getTable(
Review Comment:
Fixed. Transient-free system-source reloads now use the connector
authentication context in both scan planning and statistics, with coverage for
the fallback path.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java:
##########
@@ -731,8 +741,17 @@ public Optional<ConnectorMvccSnapshot> resolveTimeTravel(
// must not be re-evaluated later, or split planning would
read a different version than
// the one whose schema was bound. Resolution runs against the
LATEST table, because the
// options themselves are what selects the version.
- Map<String, String> resolved =
- PaimonScanParams.resolveOptions(table,
spec.getOptions());
+ boolean usesStatementFence =
spec.getLatestSnapshotFence().isPresent()
+ &&
PaimonScanParams.usesStatementSnapshot(spec.getOptions());
+ Map<String, String> resolved;
+ if (usesStatementFence) {
+ // Planning-only aliases own different table projections,
not different
+ // versions. Reuse the statement fence even if latest
advances between binds.
+ resolved = PaimonScanParams.pinOptionsToSnapshot(
Review Comment:
Fixed in 5e3567ff85. Both fallback branches resolve their catalog-visible
snapshot pins before catalog loaders are removed, and the version-managed
fallback read is covered.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java:
##########
@@ -157,15 +162,18 @@ private PluginDrivenMvccSnapshot materializeLatest() {
ConnectorTableHandle handle = handleOpt.get();
// An empty (no-snapshot) connector still pins: fall back to a
snapshot id of -1.
- ConnectorMvccSnapshot connectorSnapshot =
- metadata.beginQuerySnapshot(session,
handle).orElseGet(this::emptySnapshot);
+ ConnectorMvccSnapshot connectorSnapshot = existingFence.orElseGet(
+ () -> metadata.beginQuerySnapshot(session,
handle).orElseGet(this::emptySnapshot));
// Range-view path (e.g. iceberg): thread the query's pin onto the
handle FIRST (applySnapshot), so
// the partition/freshness enumeration stays consistent with the
data-scan pin, then ask the connector
// for its range-aware view. A connector without a range view returns
empty -> fall through to the
// legacy listPartitions/LIST/timestamp path below (byte-unchanged;
the no-op applySnapshot for the
// latest pin is side-effect-free for both paimon and iceberg).
ConnectorTableHandle pinnedHandle = metadata.applySnapshot(session,
handle, connectorSnapshot);
+ // Fence hydration must list the pinned version; the ordinary latest
path keeps its legacy
+ // base-handle partition semantics for connectors whose applySnapshot
is scan-only.
+ ConnectorTableHandle partitionHandle = existingFence.isPresent() ?
pinnedHandle : handle;
Review Comment:
Fixed in 5e3567ff85. Initial partition accounting now enumerates the pinned
handle, including the empty/latest fence path, so COUNT pushdown and block-rule
accounting use the scanned snapshot.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -523,64 +628,156 @@ 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));
+ validateSerializedReaderOptions(table);
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()) {
+ if (systemSource == null && !(table instanceof FileStoreTable)) {
+ // Legacy FEs serialize only the system wrapper, whose public
options hide the
+ // source planner. Wrapper copy is the only compatible way to
enforce the BE cap.
+ return table.copy(Collections.singletonMap(
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
String.valueOf(safeBound)));
+ }
+ return table;
+ }
+ if (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));
+ // File-store copies must retain the FE-selected schema while only
lowering an execution
+ // bound; ordinary copy can re-resolve time travel and undo schema
pinning.
+ if (systemSource != null && systemTableType != null) {
+ FileStoreTable cappedSource =
systemSource.copyWithoutTimeTravel(cap);
+ // Read-only wrappers hide the data table's option map. Rebuild
from the transported
+ // exact source so a smaller BE can lower that hidden planner
without rewinding schema.
+ Table rebuilt = SystemTableLoader.load(systemTableType,
cappedSource);
+ if (rebuilt == null) {
+ throw new IllegalArgumentException(
+ "Unsupported Paimon system table '" + systemTableType
+ "'");
+ }
+ return rebuilt;
+ }
+ 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 void validateSerializedReaderOptions(Table table) {
Review Comment:
Fixed in 5e3567ff85. The BE validates a positive split target through hidden
serialized system sources; the legacy deferred files wrapper regression is
included.
--
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]
