Share instance selector candidate lists between segments with equal assignments - #19785

PR flow

Sharing candidate lists for old segments via canonical instance state maps

flowchart TD
  N0["Get IS and EV maps for segment #40;F1#41;"]:::stModified
  N1["Canonicalize #40;IS#44;EV#41; pair #40;F1#41;"]:::stModified
  N2["Get cached candidate list or compute and cache #40;F1#41;"]:::stModified
  N3["Set segment state with candidate list #40;F1#41;"]:::stModified
  N0 -->|"using the maps"| N1
  N1 -->|"accessing candidate list"| N2
  N2 -->|"passing candidate list"| N3
  classDef stAdded fill:#dafbe1,stroke:#1a7f37,color:#1f2328,stroke-width:2px
  classDef stModified fill:#fff8c5,stroke:#9a6700,color:#1f2328,stroke-width:2px
  classDef stRemoved fill:#ffebe9,stroke:#cf222e,color:#1f2328,stroke-width:2px
  classDef stUnchanged fill:#f6f8fa,stroke:#656d76,color:#1f2328,stroke-width:1px

AI-generated · Green: added · Yellow: modified · Red: removed · Gray: existing

• Regenerate PR flow

Summary

Part of #19782.

Within one instance selector rebuild, segments with equal ideal state and external view instance state maps now share one list of routing candidates. In a realtime table, all segments of a stream partition usually have the same servers and states, so a large table has about as many distinct assignments as partitions. The rebuild also stops building a TreeMap and a TreeSet per segment, and stops copying candidate lists when all candidates are enabled.

The PR also adds the SLF4J marker ROUTING_NEW_SEGMENTS to the two "Got N new segments" INFO logs, so operators can drop them with a log4j2 MarkerFilter.

Changes

• convertToSortedMap becomes sortByKeyIfNeeded. It returns the map itself when its iteration order is already sorted (a linear check), and copies it only otherwise. The Helix maps are LinkedHashMap, so the old SortedMap shortcut never applied. ZooKeeper maps are usually written in key order, so the copy is rare.
• computeOldSegmentCandidates and computeNewSegmentCandidates build the candidates in one pass over the sorted ideal state map, without a TreeSet of online instances.
• getEnabledCandidatesAndAddToServingInstances returns the input list itself when every candidate is enabled.
• InstanceStateMapsCache maps each distinct pair of (ideal state map, external view map), compared by content, to one canonical InstanceStateMaps. This entry keeps the candidate list of the first segment. For ReplicaGroupInstanceSelector.updateSegmentMapsForUpsertTable, it also keeps the online instances and a flag that tells if the pair already added its unavailable instances.
• The cache lives for one rebuild. It holds at most 4096 pairs. Past that limit, each segment gets its own list, as before.

These do not change: the candidate order, the replica ids, the rules for new and old segments, the unavailable segments, the serving instances, and the replica health metrics.

Why shared lists are safe: the lists are read-only

The broker publishes the candidate lists to queries through SegmentStates, so a shared list must never change after the rebuild. These are all the consumers:

No consumer calls add, remove, set, sort or clear. The rebuild adds elements only to lists that it creates, before it stores them. SegmentInstanceCandidate is immutable (all fields are final).

As evidence, we ran the full pinot-broker test suite with a temporary change that wraps every candidate list in Collections.unmodifiableList. All 555 tests passed. The wrapper is not part of this PR, because it adds one object per list.

Concurrency

This PR keeps the concurrency rule of #19782. InstanceStateMapsCache, InstanceStateMaps, and their online instances and flags are mutable but private to the rebuild thread. The only shared objects that queries see are the candidate lists, which are read-only (see above). Thus one list can be shared between many segments, and between the old and the new SegmentStates.

Compatibility

• There are no config, API, metric or wire format changes. The routing results are the same as before.

• The two INFO logs keep the same level, message and arguments. To drop them, add this filter to the log4j2 config of the broker (for example, on the logger for BaseInstanceSelector). The log4j2 configs that Pinot ships do not change.

  <MarkerFilter marker="ROUTING_NEW_SEGMENTS" onMatch="DENY" onMismatch="NEUTRAL"/>

Tests

• InstanceSelectorTest.testSharedCandidateListsMatchPerSegmentLists, for the balanced, replica-group and strict replica-group selectors. It compares a sharing selector with a per-segment selector: candidates, unavailable segments, serving instances, replica health, and selection for 6 request ids. A third selector with a cache limit of 1 mixes shared and per-segment lists. The test then enables a disabled instance and compares again. The data has unsorted maps, an OFFLINE replica in the external view, a disabled instance, a segment missing from the external view, and new segments.
• InstanceSelectorTest.testSortByKeyIfNeeded: a sorted LinkedHashMap and a TreeMap come back unchanged. An unsorted map comes back as a sorted copy.

Local results: org.apache.pinot.broker.routing (18 classes) runs 256 tests with 0 failures, including the 91 tests in InstanceSelectorTest. Spotless, license and checkstyle checks pass for pinot-common and pinot-broker.

Measured effect

Synthetic realtime table with 380k segments, one instance selector rebuild (times are noisy):

We also replayed the assignment changes of a real 380k-segment realtime table. With this PR, the full rebuild of all routing components allocates 80 MB per change instead of 280 MB.