gortiz commented on code in PR #18947:
URL: https://github.com/apache/pinot/pull/18947#discussion_r3896511336
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/StrictReplicaGroupInstanceSelector.java:
##########
@@ -18,44 +18,52 @@
*/
package org.apache.pinot.broker.routing.instanceselector;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
+import
org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+
/// Instance selector for strict replica-group routing strategy.
///
-/// ```
-/// The strict replica-group routing strategy always routes the query to the
instances within the same replica-group.
-/// (Note that the replica-group information is derived from the ideal state
of the table, where the instances are
-/// sorted alphabetically in the instance state map, so the replica-groups in
the instance selector might not match the
-/// replica-groups in the instance partitions). The goal of this algorithm is
to ensure that segments from the same
-/// partition are never served from multiple different instances. The
instances in a replica-group should have all the
-/// online segments (segments with ONLINE/CONSUMING instances in the ideal
state and selected by the pre-selector)
-/// available (ONLINE/CONSUMING in the external view) in order to serve
queries. If any segment is unavailable in the
-/// replica-group, we mark the whole replica-group down and not serve queries
with this replica-group.
-///
-/// The selection algorithm is the same as {@link
ReplicaGroupInstanceSelector}, and will always evenly distribute the
-/// traffic to all replica-groups that have all online segments available.
+/// The strict replica-group routing strategy always routes same-partition
segments to the same instance. During
+/// routing state construction, [#updateSegmentMapsForUpsertTable(IdealState,
ExternalView, Set, Map)] removes from
+/// every segment in a partition any replica that is unavailable for any old
segment in that partition. Consequently,
+/// all same-partition segments have identical, ordered candidate identities.
///
-/// The algorithm relies on the mirror segment assignment from replica-group
segment assignment strategy. With mirror
-/// segment assignment, any server in one replica-group will always have a
corresponding server in other replica-groups
-/// that have the same segments assigned. For example, if S1 is a server in
replica-group 1, and it has mirror server S2
-/// in replica-group 2 and S3 in replica-group 3. All segments assigned to S1
will also be assigned to S2 and S3. In
-/// stable scenario (external view matches ideal state), all segments assigned
to S1 will have the same enabled
-/// instances of [S1, S2, S3] sorted (in alphabetical order). If we always
pick the same index of enabled instances for
-/// all segments, only one of S1, S2, S3 will be picked, and all the segments
are processed by the same server. In
-/// transitioning/error scenario (external view does not match ideal state),
if a segment is down on S1, we mark all
-/// segments with the same assignment ([S1, S2, S3]) down on S1 to ensure that
we always route the segments to the same
-/// replica-group.
+/// Adaptive routing preserves that guarantee without explicit partition or
mirror-set metadata. The inherited
Review Comment:
The rework is the right call, and this javadoc does a good job of stating
why it works. Two assumptions are now load-bearing and worth naming explicitly
here, because neither lives in this file:
1. **One ranking snapshot per query.** If the parent ever re-ranked per
segment, same-partition segments could diverge.
2. **The strictly-less-than tie-break** in
`ReplicaGroupInstanceSelector.selectServers()` (see my comment there).
I verified the argument itself: for old segments in one instance set,
`candidates = idealState - unavailableInstances` — the online filter is
implied, since `unavailableInstances` is the union of non-online instances over
every old segment in that set — so all old segments in a partition get
identical, identically-ordered lists. New segments get the same instance list
with per-segment online flags, and selection keys off instance identity, not
the flag. `refreshSegmentStates()` then filters by `_enabledInstances`
uniformly. It holds; I'd just like the two dependencies written down so a
future refactor of the parent doesn't quietly break the subclass with no test
in this file failing.
##########
pom.xml:
##########
@@ -2061,7 +2061,7 @@
<plugin>
<groupId>com.diffplug.spotless</groupId>
<artifactId>spotless-maven-plugin</artifactId>
- <version>2.46.1</version>
+ <version>3.0.0</version>
Review Comment:
Thanks for dropping the cleanthat engine — a version bump is the right shape
of fix. But this is still a build-wide change riding along in a routing PR, and
a major version bump is a bigger one than the engine switch it replaces:
spotless 3.0.0 can change formatting defaults and tool requirements across all
~100 modules, and if it misbehaves it'll get bisected to this PR.
I'd land it separately, or stay on 2.46.1 and pin `googleJavaFormat`'s
version inside `removeUnusedImports` — the minimal change that unblocks the
record. Not blocking the approval, but I don't think this PR should be the
thing that moves the repo's formatter.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/StrictReplicaGroupInstanceSelector.java:
##########
@@ -18,44 +18,52 @@
*/
package org.apache.pinot.broker.routing.instanceselector;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
+import
org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+
/// Instance selector for strict replica-group routing strategy.
///
-/// ```
-/// The strict replica-group routing strategy always routes the query to the
instances within the same replica-group.
-/// (Note that the replica-group information is derived from the ideal state
of the table, where the instances are
-/// sorted alphabetically in the instance state map, so the replica-groups in
the instance selector might not match the
-/// replica-groups in the instance partitions). The goal of this algorithm is
to ensure that segments from the same
-/// partition are never served from multiple different instances. The
instances in a replica-group should have all the
-/// online segments (segments with ONLINE/CONSUMING instances in the ideal
state and selected by the pre-selector)
-/// available (ONLINE/CONSUMING in the external view) in order to serve
queries. If any segment is unavailable in the
-/// replica-group, we mark the whole replica-group down and not serve queries
with this replica-group.
-///
-/// The selection algorithm is the same as {@link
ReplicaGroupInstanceSelector}, and will always evenly distribute the
-/// traffic to all replica-groups that have all online segments available.
+/// The strict replica-group routing strategy always routes same-partition
segments to the same instance. During
+/// routing state construction, [#updateSegmentMapsForUpsertTable(IdealState,
ExternalView, Set, Map)] removes from
+/// every segment in a partition any replica that is unavailable for any old
segment in that partition. Consequently,
+/// all same-partition segments have identical, ordered candidate identities.
///
-/// The algorithm relies on the mirror segment assignment from replica-group
segment assignment strategy. With mirror
-/// segment assignment, any server in one replica-group will always have a
corresponding server in other replica-groups
-/// that have the same segments assigned. For example, if S1 is a server in
replica-group 1, and it has mirror server S2
-/// in replica-group 2 and S3 in replica-group 3. All segments assigned to S1
will also be assigned to S2 and S3. In
-/// stable scenario (external view matches ideal state), all segments assigned
to S1 will have the same enabled
-/// instances of [S1, S2, S3] sorted (in alphabetical order). If we always
pick the same index of enabled instances for
-/// all segments, only one of S1, S2, S3 will be picked, and all the segments
are processed by the same server. In
-/// transitioning/error scenario (external view does not match ideal state),
if a segment is down on S1, we mark all
-/// segments with the same assignment ([S1, S2, S3]) down on S1 to ensure that
we always route the segments to the same
-/// replica-group.
+/// Adaptive routing preserves that guarantee without explicit partition or
mirror-set metadata. The inherited
+/// selector takes one ranking snapshot for the query and deterministically
chooses the best candidate from each
+/// segment's ordered list. Identical filtered candidate lists, the same
ranking snapshot, and deterministic list-order
+/// tie-breaking therefore produce identical selections for every segment in a
partition. Different partitions may
+/// independently choose different replicas.
///
-/// Note that new segments won't be used to exclude instances from serving
when the segment is unavailable.
-/// ```
+/// New segments do not exclude a candidate when that segment is unavailable;
they remain optional so that the broker or
+/// server can skip them if necessary.
public class StrictReplicaGroupInstanceSelector extends
ReplicaGroupInstanceSelector {
@Override
void updateSegmentMaps(IdealState idealState, ExternalView externalView,
Set<String> onlineSegments,
Map<String, Long> newSegmentCreationTimeMap) {
super.updateSegmentMapsForUpsertTable(idealState, externalView,
onlineSegments, newSegmentCreationTimeMap);
}
+
+ @Override
+ public InstanceMapping select(List<String> segments, int requestId,
+ SegmentStates segmentStates, Map<String, String> queryOptions) {
+ ServerSelectionContext ctx = new ServerSelectionContext(queryOptions,
_config);
+ if (_adaptiveServerSelector != null && _priorityPoolInstanceSelector !=
null) {
+ if (ctx.isUseFixedReplica()) {
+ throw new IllegalArgumentException(
Review Comment:
This throws for the config-level setting too, not just the query option —
`ServerSelectionContext` falls back to `InstanceSelectorConfig`, which
`InstanceSelectorFactory` populates from `pinot.broker.use.fixed.replica` and
`routingConfig.getUseFixedReplica()`.
So an operator who set either of those (default is `false`, so this is
opt-in, but it's a documented feature) plus adaptive routing gets an
`IllegalArgumentException` on **every** query against a strict-RG table.
`calculateRoutes()` isn't wrapped in a try at
`BaseSingleStageBrokerRequestHandler:782`, so it surfaces as an internal error
rather than a `QueryErrorCode`, and the only remedy is a config change plus a
restart. Before this PR that combination worked — fixed replica simply won.
And fixed replica is *safe* under your new argument: with identical
per-partition candidate lists, `candidates.get((tableNameHash + replicaOffset)
% numCandidates)` yields the same index for every segment in a partition, so
the strict invariant holds.
Suggestion: handle it in `InstanceSelectorFactory` exactly like the legacy
upsert/dedup block you added below — if strict RG + adaptive +
`useFixedReplica`, log a warning and null out
`effectiveAdaptiveServerSelector`. If you specifically want to reject the
*query option*, it should be a `BadQueryRequestException`/`QueryErrorCode` and
must not fire for the config or table-level setting. The test name
(`…RejectsFixedReplicaQueryOption`) reads like the query-option case is what
was actually intended.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java:
##########
@@ -124,13 +126,19 @@ private Pair<Map<String, String>, Map<String, String>>
selectServers(List<String
} 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);
+ // instance. As of 8 July 2026, this fallback is unreachable, but new
implementations could require it.
+ int bestRank = Integer.MAX_VALUE;
+ for (SegmentInstanceCandidate candidate : candidates) {
+ Integer rank = serverRankMap.get(candidate.getInstance());
+ if (rank == null) {
+ selectedInstance = candidates.get(instanceIdx);
+ break;
+ }
+ if (rank < bestRank) {
Review Comment:
The rewrite is a good change on its own — no streams, no double traversal,
and the semantics are preserved (any unranked candidate falls back to the
round-robin index, same as the old `anyMatch` guard).
Worth a comment saying this loop is now load-bearing for
`StrictReplicaGroupInstanceSelector`: `rank < bestRank` means the first
candidate in list order wins a tie, and candidate lists are identical per
partition, which is precisely what makes every segment of a partition resolve
to the same server. Changing `<` to `<=`, reordering candidates, or
reintroducing a shuffle would silently break the strict guarantee, and nothing
in `ReplicaGroupSelectorTest` would point at this line. Same for the `rank ==
null` branch reusing `instanceIdx` rather than a per-candidate fallback.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/StrictReplicaGroupInstanceSelector.java:
##########
@@ -18,44 +18,52 @@
*/
package org.apache.pinot.broker.routing.instanceselector;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
+import
org.apache.pinot.broker.routing.adaptiveserverselector.ServerSelectionContext;
+import org.apache.pinot.common.utils.config.QueryOptionsUtils;
+
/// Instance selector for strict replica-group routing strategy.
///
-/// ```
-/// The strict replica-group routing strategy always routes the query to the
instances within the same replica-group.
-/// (Note that the replica-group information is derived from the ideal state
of the table, where the instances are
-/// sorted alphabetically in the instance state map, so the replica-groups in
the instance selector might not match the
-/// replica-groups in the instance partitions). The goal of this algorithm is
to ensure that segments from the same
-/// partition are never served from multiple different instances. The
instances in a replica-group should have all the
-/// online segments (segments with ONLINE/CONSUMING instances in the ideal
state and selected by the pre-selector)
-/// available (ONLINE/CONSUMING in the external view) in order to serve
queries. If any segment is unavailable in the
-/// replica-group, we mark the whole replica-group down and not serve queries
with this replica-group.
-///
-/// The selection algorithm is the same as {@link
ReplicaGroupInstanceSelector}, and will always evenly distribute the
-/// traffic to all replica-groups that have all online segments available.
+/// The strict replica-group routing strategy always routes same-partition
segments to the same instance. During
+/// routing state construction, [#updateSegmentMapsForUpsertTable(IdealState,
ExternalView, Set, Map)] removes from
+/// every segment in a partition any replica that is unavailable for any old
segment in that partition. Consequently,
+/// all same-partition segments have identical, ordered candidate identities.
///
-/// The algorithm relies on the mirror segment assignment from replica-group
segment assignment strategy. With mirror
-/// segment assignment, any server in one replica-group will always have a
corresponding server in other replica-groups
-/// that have the same segments assigned. For example, if S1 is a server in
replica-group 1, and it has mirror server S2
-/// in replica-group 2 and S3 in replica-group 3. All segments assigned to S1
will also be assigned to S2 and S3. In
-/// stable scenario (external view matches ideal state), all segments assigned
to S1 will have the same enabled
-/// instances of [S1, S2, S3] sorted (in alphabetical order). If we always
pick the same index of enabled instances for
-/// all segments, only one of S1, S2, S3 will be picked, and all the segments
are processed by the same server. In
-/// transitioning/error scenario (external view does not match ideal state),
if a segment is down on S1, we mark all
-/// segments with the same assignment ([S1, S2, S3]) down on S1 to ensure that
we always route the segments to the same
-/// replica-group.
+/// Adaptive routing preserves that guarantee without explicit partition or
mirror-set metadata. The inherited
+/// selector takes one ranking snapshot for the query and deterministically
chooses the best candidate from each
+/// segment's ordered list. Identical filtered candidate lists, the same
ranking snapshot, and deterministic list-order
+/// tie-breaking therefore produce identical selections for every segment in a
partition. Different partitions may
+/// independently choose different replicas.
///
-/// Note that new segments won't be used to exclude instances from serving
when the segment is unavailable.
-/// ```
+/// New segments do not exclude a candidate when that segment is unavailable;
they remain optional so that the broker or
+/// server can skip them if necessary.
public class StrictReplicaGroupInstanceSelector extends
ReplicaGroupInstanceSelector {
@Override
void updateSegmentMaps(IdealState idealState, ExternalView externalView,
Set<String> onlineSegments,
Map<String, Long> newSegmentCreationTimeMap) {
super.updateSegmentMapsForUpsertTable(idealState, externalView,
onlineSegments, newSegmentCreationTimeMap);
}
+
+ @Override
+ public InstanceMapping select(List<String> segments, int requestId,
+ SegmentStates segmentStates, Map<String, String> queryOptions) {
+ ServerSelectionContext ctx = new ServerSelectionContext(queryOptions,
_config);
+ if (_adaptiveServerSelector != null && _priorityPoolInstanceSelector !=
null) {
+ if (ctx.isUseFixedReplica()) {
+ throw new IllegalArgumentException(
+ "useFixedReplica cannot be used when adaptive routing is enabled
for StrictReplicaGroupInstanceSelector");
+ }
+ if (QueryOptionsUtils.getNumReplicaGroupsToQuery(ctx.getQueryOptions())
!= null) {
Review Comment:
These two guards look inverted to me: the option that's rejected above is
safe, and the one preserved here is the one that breaks the invariant this PR
exists to protect.
With `numReplicaGroupsToQuery` set, `selectServers()` cycles `replicaOffset`
per segment, so consecutive segments deliberately land in different replica
groups — including segments of the same partition.
`testStrictReplicaGroupAdaptivePreservesNumReplicaGroupsToQuery` asserts
exactly that: in the fixture `seg0` and `seg1` share the candidate list
(`server_a`@rg0, `server_c`@rg1), i.e. same partition, and the expected result
routes `seg0` to `server_a` and `seg1` to `server_c`. That's the
cross-replica-group read the PR description says can miss or duplicate upsert
keys.
It's pre-existing behavior so I'm not asking you to fix it here, but the
comment calls it intentional and a test now blesses it, which reads oddly next
to a hard throw on the safe option. At minimum I'd reword this to say the path
is knowingly unsafe for upsert tables, and consider rejecting or ignoring
`numReplicaGroupsToQuery` for strict RG as a follow-up.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java:
##########
@@ -72,27 +71,45 @@ void init(TableConfig tableConfig,
ZkHelixPropertyStore<ZNRecord> propertyStore,
/// Returns the enabled server instances currently serving the table.
Set<String> getServingInstances();
+ /// Holds the result of an instance selection: `segmentToInstanceMap` maps
each segment to its selected server
+ /// instance, `optionalSegmentToInstanceMap` maps segments not yet fully
online that the server may skip, and
+ /// `unavailableSegments` lists segments that have candidates but could not
be routed. These unavailable segments are
+ /// merged into the query's unavailable-segment list and surfaced as a
`BROKER_SEGMENT_UNAVAILABLE` error.
+ record InstanceMapping(Map<String, String> segmentToInstanceMap,
+ Map<String, String> optionalSegmentToInstanceMap,
+ List<String> unavailableSegments) {
Review Comment:
After the rework, nothing in production populates `unavailableSegments`.
Grepping the PR head, the 3-arg form is constructed exactly twice: line 81
(`EMPTY`) and `InstanceSelectorTest:791` (the fake selector added to test the
dedup path). Every real selector uses the 2-arg convenience constructor.
So the third component, the `LinkedHashSet` dedup in `BaseInstanceSelector`,
and the javadoc sentence promising a `BROKER_SEGMENT_UNAVAILABLE` error all
describe a path that can't be reached today. Since the dropped-segment behavior
is gone, I'd drop the third component and the dedup and keep the record at two
— it makes the refactor commit purely mechanical, which is easier to backport.
Keeping it as an extension point for custom selectors is defensible too, but
then the javadoc should say that's what it is; as written it reads as though
some selector does this.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorFactory.java:
##########
@@ -136,8 +145,15 @@ public static InstanceSelector
getInstanceSelector(TableConfig tableConfig,
if (instanceSelector == null) {
instanceSelector = new BalancedInstanceSelector();
}
+ if (instanceSelector.getClass() == ReplicaGroupInstanceSelector.class
Review Comment:
Good catch closing the legacy upsert/dedup path. Two notes on the mechanics:
- `getClass() == ReplicaGroupInstanceSelector.class` correctly excludes the
Strict subclass, but an exact-class test also excludes any plugin-provided
subclass. That's probably what you want; it's unusual enough to deserve a
comment saying the exclusion is deliberate.
- Unlike the strict path, this disable isn't gated by the new config flag,
so an operator has no escape hatch if it regresses their latency. Reasonable
default for a correctness fix, but it's a silent behavior change for those
tables — worth a line in the PR description and release notes next to the flag.
--
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]
