github-actions[bot] commented on code in PR #68791:
URL: https://github.com/apache/doris/pull/68791#discussion_r4244942275
##########
fe/fe-connector/fe-connector-fluss/src/main/java/org/apache/doris/connector/fluss/FlussLakeRange.java:
##########
@@ -178,6 +178,12 @@ public Optional<String> getPath() {
return inner.getPath();
}
+ @Override
+ public String getSplitIdentity() {
+ return inner.getSplitIdentity() + "\u0001"
Review Comment:
[P2] Keep a lake file's hash identity stable as the log tail grows.
`bindTailToLakeSplit` wraps the same Paimon file with a tail ending at the
current `bucketState.stop`; each new log write can change `tail.encode()` while
the file path and byte range stay unchanged. On groups where ring selection is
used, hashing this value can remap the unchanged native lake file even with
spread=1, so Fluss union reads miss the opted-in file cache and warm another
BE. Use the inner file identity and byte extent for file-backed lake ranges;
the tail is read metadata, not a new cache key for that file.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java:
##########
@@ -252,7 +269,15 @@ public Multimap<Backend, Split>
computeScanRangeAssignment(List<Split> splits) t
Optional<Backend> chosenNode = candidateNodes.stream()
.min(Comparator.comparingLong(ownerNode ->
assignedWeightPerBackend.get(ownerNode)));
- if (chosenNode.isPresent()) {
+ // A host hint is a local preference for remotely readable
splits. In spread mode, do not
+ // let a repeated hint monopolize one backend while other
eligible backends are idle.
+ boolean localNodeIsWithinGlobalMinimum =
chosenNode.isPresent();
+ if (localNodeIsWithinGlobalMinimum && isSpreadEnabled()) {
+ long minimumAssignedWeight =
Collections.min(assignedWeightPerBackend.values());
Review Comment:
[P2] Avoid scanning every backend for each hinted split.
`Collections.min(assignedWeightPerBackend.values())` runs inside the
remote-split loop whenever a preferred host matches. With S such splits and B
eligible BEs, even spread=2 adds O(S×B) map visits during planning, including
under `SplitAssignment.assignLock` for lazy batches. Large Trino/HDFS scans can
pay this cost despite a two-node hash bound. Track the current minimum as
weights change, or use a priority structure, so the locality check stays cheap.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/scan/FederationBackendPolicy.java:
##########
@@ -280,12 +305,20 @@ 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 {
+ int candidateCount = isSpreadEnabled()
+ ? Math.min(consistentHashSpreadNum,
backends.size())
+ :
Config.split_assigner_min_consistent_hash_candidate_num;
+ candidateNodes = candidateCount >= backends.size()
? backends
Review Comment:
[P1] Preserve ring order when spread is disabled. With file cache or the
hash selector enabled, the default spread value 1 and default two hash
candidates make a two-BE group enter this shortcut for every split. It returns
the same backend list for every file; `chooseNodeForSplit` resolves the initial
zero-weight tie to its last member, so independent single-file scans of
different paths all land on one BE and lose the original file-cache affinity.
Global redistribution does not move a single standard-weight split because its
variance threshold is 1. The previous path used `consistentHash.getNode(split,
2)`, whose order depends on the split key. Restrict this shortcut to enabled
spread (or preserve ring order for legacy mode).
##########
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() {
Review Comment:
[P1] Bump the connector plugin API major for this new SPI method. The parent
pom still stamps both FE and connector plugins as 13.0, and `ApiVersionGate`
accepts any matching major. A plugin built against this method can therefore be
loaded by an older major-13 FE whose `ConnectorScanRange` lacks it; if that
plugin calls the hook, scan planning fails with `NoSuchMethodError`. Advance
the major with the pinned test value and record `ConnectorScanRange` in the
surface baseline so this API change is gated.
--
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]
