github-actions[bot] commented on code in PR #66247:
URL: https://github.com/apache/doris/pull/66247#discussion_r3697830053
##########
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:
[P2] Hydrate initial partition accounting from the pinned handle
For an ordinary `materializeLatest()` call, `existingFence` is empty, so
this chooses the live base handle even after `beginQuerySnapshot` captured
snapshot S. Paimon later replans from the pushed predicate, so the mismatch
does not drop rows, but this live listing initializes
`selectedPartitionNum`/`totalPartitionNum`. Normal scans replace the selected
count with Paimon's actual scanned-partition count; `COUNT(*)` pushdown
deliberately keeps the Nereids count. If S+1 removes partition `p` between
snapshot capture and this listing, a COUNT(*) scan pinned to S can therefore
report fewer S+1 partitions and evade a `sql_block_rule` `partition_num` limit.
The existing positive-fence thread covers options-first hydration; this is the
initial plain/latest and preload path. Hydrate from `pinnedHandle` (or publish
no partition view when enumeration cannot honor the pin), and add a
partition-removal race that asserts COUNT(*) partition accounting and
block-rule enforcement.
##########
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:
[P1] Backstop the newly bounded split target on the BE
The FE now rejects `source.split.target-size < 1` because zero defeats
Paimon's bin packing and expands planning to one split per data file, but this
rolling-upgrade guard checks only batch size and async threshold. An older FE
can still serialize `source.split.target-size=0 B`; for a deferred `$files`
reader, planning happens after deserialization and the zero value reaches
`newReadBuilder()` unchanged. The existing old-FE threads cover async bounds
and hidden batch readers, not this newly allowlisted option. Mirror the
positive split-target check here across the deserialized/hidden system source
and add an old-FE deferred `$files` fixture with a zero target.
##########
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:
[P1] Resolve the fallback pin before dropping its catalog loader
This statement-fence path turns selector-free tuning such as
`t$ro@options('read.batch-size'='4096')` into `scan.snapshot-id=S`. Later,
`tableForBackend` removes both branch catalog loaders and `reapplyScanParams`
copies that pin onto the catalog-less fallback pair; Paimon's fallback-id
rewrite then searches the newest snapshot files rather than the version-managed
fallback catalog pointer. After an unpublished commit or rollback, `$ro` can
therefore return fallback rows from a generation the catalog does not expose.
This is distinct from the decorator thread: the fallback pair is retained here,
but it selects the wrong generation. Resolve both branches' catalog-visible
pins while their loaders are still present (or avoid replaying this synthetic
id on the catalog-less pair), and cover reader-only OPTIONS on a
version-managed fallback `$ro` with a newer fallback snapshot file beyond the
catalog pointer.
--
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]
