github-actions[bot] commented on code in PR #68791:
URL: https://github.com/apache/doris/pull/68791#discussion_r4254802456
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java:
##########
@@ -280,12 +310,23 @@ public Multimap<Backend, Split>
computeScanRangeAssignment(List<Split> splits) t
}
case RANDOM: {
randomCandidates.reset();
- candidateNodes =
selectNodes(Config.split_assigner_min_random_candidate_num, randomCandidates);
+ candidateNodes = consistentHashSpreadNum == 0 ?
backends
+ :
selectNodes(Config.split_assigner_min_random_candidate_num, randomCandidates);
break;
}
case CONSISTENT_HASHING: {
- candidateNodes = consistentHash.getNode(split,
-
Config.split_assigner_min_consistent_hash_candidate_num);
+ if (consistentHashSpreadNum == 0) {
+ candidateNodes = backends;
+ } else {
+ if (isSpreadEnabled()) {
+ int candidateCount =
Math.min(consistentHashSpreadNum, backends.size());
+ candidateNodes = candidateCount >=
backends.size() ? backends
+ : consistentHash.getNode(split,
candidateCount);
Review Comment:
[P2] Include the MaxCompute read session in bounded-spread identity
MaxCompute partition batches can still produce identical hash keys for
different work. With `mc.split_strategy=row_count`, `mc.split_row_count=1000`,
`num_partitions_in_batch_mode=1`, and 100 selected partitions of 1000 rows,
`PluginDrivenScanNode` creates a separate read session for each partition. Each
session's row loop emits `(/row_offset, 0, 1000)`, while `MaxComputeScanRange`
inherits the path-only identity and omits its session ID. With hashing enabled,
spread=2 and 10 eligible BEs, this branch gives all 100 ranges the same two
candidates; disabled redistribution leaves eight BEs unable to receive them.
Include the read-session/work-unit identity through the connector hook and
cover equal offsets from different sessions. This is a separate MaxCompute
instance from the existing Paimon/Trino/Remote Doris thread: the path is
nonempty but synthetic, and coordinates are session-relative.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/scan/ConnectorScanRange.java:
##########
@@ -46,6 +46,14 @@ default Optional<String> getPath() {
return Optional.empty();
}
+ /**
+ * Returns a stable identity used to assign connector ranges to
external-scan backends. Connectors whose JNI
+ * ranges do not have a path should override this with their serialized
split identity.
+ */
+ default String getSplitIdentity() {
+ return getPath().orElse("");
Review Comment:
[P2] Give Iceberg metadata tasks their own scheduling identity
The path default also collapses Iceberg JNI metadata ranges.
`IcebergScanPlanProvider.doPlanSystemTableScan` wraps every task with
`/dummyPath`, and `IcebergScanRange` leaves start=0 and length=-1 without
overriding this hook. For a `$files` query over many unmerged manifests,
[Iceberg 1.11.0 plans one task per
manifest](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/BaseFilesTable.java);
distinct serialized tasks therefore all hash as `(/dummyPath, 0, -1)`. With
hashing enabled, spread=2 and 10 eligible BEs, all metadata work is confined to
the same two BEs because redistribution is disabled. Use the immutable
serialized task identity for Iceberg's JNI metadata arm, preserve actual file
identity for native arms, and add same-dummy-path/different-task coverage. This
Iceberg metadata-table instance is distinct from the existing
Paimon/Trino/Remote Doris report.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java:
##########
@@ -239,6 +261,7 @@ public Multimap<Backend, Split>
computeScanRangeAssignment(List<Split> splits) t
for (List<Backend> backendList : backendMap.values()) {
backends.addAll(backendList);
}
+ rebuildAssignedWeightCounts();
Review Comment:
[P2] Keep the maintained weight histogram across streamed batches
This unconditional rebuild reconstructs an already-current histogram for
every batch. `PluginDrivenScanNode.startStreamingSplit` calls `addToQueue` once
per individual file, so a million-file scan executes this rebuild a million
times under `SplitAssignment.assignLock`. Each call scans all B backend weights
and merges them into a cleared TreeMap, adding B tree operations and recreating
its entries even for hostless S3 ranges or spread=1, where the histogram is
never queried. All production weight writes already go through
`setAssignedWeight`/`addAssignedWeight`, which keep the counts synchronized;
the exposed map's production consumer only reads it for the profile. Retain the
incremental counts and handle test-only injected weights separately. This is a
distinct new trigger from the earlier hinted-split `Collections.min` issue: the
rebuild runs per singleton streamed batch, including unhinted and
disabled-spread scans.
--
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]
