Copilot commented on code in PR #16744:
URL: https://github.com/apache/pinot/pull/16744#discussion_r2318724950
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/adaptiveserverselector/ServerSelectionContext.java:
##########
@@ -52,15 +55,22 @@ public class ServerSelectionContext {
*
* @param queryOptions map of query options that may contain server
selection preferences
*/
- public ServerSelectionContext(Map<String, String> queryOptions) {
+ public ServerSelectionContext(Map<String, String> queryOptions,
InstanceSelectorConfig instanceSelectorConfig) {
_queryOptions = queryOptions == null ? Collections.emptyMap() :
queryOptions;
+ _config = instanceSelectorConfig;
_orderedPreferredPools =
QueryOptionsUtils.getOrderedPreferredPools(_queryOptions);
+ Boolean isUseFixedReplica =
QueryOptionsUtils.isUseFixedReplica(_queryOptions);
+ _isUseFixedReplica = isUseFixedReplica != null ? isUseFixedReplica :
instanceSelectorConfig.isUseFixedReplica();
}
public Map<String, String> getQueryOptions() {
return _queryOptions;
}
+ public Boolean isUseFixedReplica() {
Review Comment:
The method returns `Boolean` but the field `_isUseFixedReplica` is a
primitive `boolean`. This creates an unnecessary boxing operation and
inconsistent return type. Change the return type to `boolean` to match the
field type and avoid boxing.
```suggestion
public boolean isUseFixedReplica() {
```
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java:
##########
@@ -85,42 +86,52 @@ Pair<Map<String, String>, Map<String, String>>
select(List<String> segments, int
for (int idx = 0; idx < serverRankList.size(); idx++) {
serverRankMap.put(serverRankList.get(idx), idx);
}
- return selectServersUsingAdaptiveServerSelector(segments, requestId,
segmentStates, serverRankMap, ctx);
+ return selectServers(segments, requestId, segmentStates, serverRankMap,
ctx);
} else {
// Adaptive Server Selection is NOT enabled.
- return selectServersUsingRoundRobin(segments, requestId, segmentStates,
ctx);
+ return selectServers(segments, requestId, segmentStates, null, ctx);
}
}
- private Pair<Map<String, String>, Map<String, String>>
selectServersUsingRoundRobin(List<String> segments,
- int requestId, SegmentStates segmentStates, ServerSelectionContext ctx) {
+ private Pair<Map<String, String>, Map<String, String>>
selectServers(List<String> segments, int requestId,
+ SegmentStates segmentStates, @Nullable Map<String, Integer>
serverRankMap, ServerSelectionContext ctx) {
+
Map<String, String> segmentToSelectedInstanceMap = new
HashMap<>(HashUtil.getHashMapCapacity(segments.size()));
// No need to adjust this map per total segment numbers, as optional
segments should be empty most of the time.
Map<String, String> optionalSegmentToInstanceMap = new HashMap<>();
Map<Integer, Integer> poolToSegmentCount = new HashMap<>();
- boolean useFixedReplica = isUseFixedReplica(ctx.getQueryOptions());
+ boolean useFixedReplica = ctx.isUseFixedReplica();
Integer numReplicaGroupsToQuery =
QueryOptionsUtils.getNumReplicaGroupsToQuery(ctx.getQueryOptions());
int numReplicaGroups = numReplicaGroupsToQuery != null ?
numReplicaGroupsToQuery : 1;
int replicaOffset = 0;
for (String segment : segments) {
List<SegmentInstanceCandidate> candidates =
segmentStates.getCandidates(segment);
- // NOTE: candidates can be null when there is no enabled instances for
the segment, or the instance selector has
+ // NOTE: candidates can be null when there are no enabled instances for
the segment, or the instance selector has
// not been updated (we update all components for routing in sequence)
if (candidates == null) {
continue;
}
- // Round robin selection.
- int numCandidates = candidates.size();
- int instanceIdx;
+ // Round-robin selection (default behavior)
+ int numCandidates = candidates.size();
+ int instanceIdx = (requestId + replicaOffset) % numCandidates;
+ SegmentInstanceCandidate selectedInstance = candidates.get(instanceIdx);
if (useFixedReplica) {
- // candidates array is always sorted
- instanceIdx = (_tableNameHashForFixedReplicaRouting + replicaOffset) %
numCandidates;
- } else {
- instanceIdx = (requestId + replicaOffset) % numCandidates;
+ // Adaptive Server Selection cannot be used with fixed replica routing.
+ // The candidates array is always sorted
+ selectedInstance =
candidates.get((_tableNameHashForFixedReplicaRouting + replicaOffset) %
numCandidates);
+ } else if (MapUtils.isNotEmpty(serverRankMap)) {
+ // Adaptive Server Selection is enabled.
+ // Use the instance with the best rank if all servers have stats
populated, else use the round-robin selected
+ // instance
+ selectedInstance = candidates.stream()
+ .anyMatch(candidate ->
!serverRankMap.containsKey(candidate.getInstance()))
+ ? selectedInstance
+ : candidates.stream()
+ .min(Comparator.comparingInt(candidate ->
serverRankMap.get(candidate.getInstance())))
+ .orElse(selectedInstance);
Review Comment:
The logic for fixed replica routing is incorrectly placed inside the general
selection block. When `useFixedReplica` is true, the code should skip the
adaptive server selection logic entirely, but the current structure allows both
fixed replica and adaptive selection to potentially execute. Move the fixed
replica check to be mutually exclusive with the adaptive selection block.
--
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]
