Jackie-Jiang commented on code in PR #19785:
URL: https://github.com/apache/pinot/pull/19785#discussion_r4253150599
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java:
##########
@@ -559,4 +613,81 @@ int getPool(String instanceID) {
/// servers and let servers decide how to handle them.
protected abstract InstanceMapping select(List<String> segments, int
requestId, SegmentStates segmentStates,
Map<String, String> queryOptions);
+
+ /// The ideal state and external view instance state maps of a segment,
compared by content. Within one rebuild,
+ /// segments with equal maps get the same routing candidates, so the
canonical instance (see
+ /// [InstanceStateMapsCache]) caches what is derived from the maps, and the
segments share it.
+ ///
+ /// Not thread safe. Only used by the rebuild thread, for the duration of
one rebuild.
+ static final class InstanceStateMaps {
+ final Map<String, String> _idealStateInstanceStateMap;
+ @Nullable
+ final Map<String, String> _externalViewInstanceStateMap;
+ private final int _hashCode;
+ // Derived from the maps, computed on first use
+ @Nullable
+ Set<String> _onlineInstances;
+ @Nullable
+ List<SegmentInstanceCandidate> _oldSegmentCandidates;
+ // Used by the strict replica-group selector to collect the unavailable
instances once per instance state maps
+ boolean _unavailableInstancesCollected;
+
+ InstanceStateMaps(Map<String, String> idealStateInstanceStateMap,
+ @Nullable Map<String, String> externalViewInstanceStateMap) {
+ _idealStateInstanceStateMap = idealStateInstanceStateMap;
+ _externalViewInstanceStateMap = externalViewInstanceStateMap;
+ _hashCode = 31 * idealStateInstanceStateMap.hashCode() +
Objects.hashCode(externalViewInstanceStateMap);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof InstanceStateMaps)) {
+ return false;
+ }
+ InstanceStateMaps that = (InstanceStateMaps) o;
+ return _hashCode == that._hashCode &&
_idealStateInstanceStateMap.equals(that._idealStateInstanceStateMap)
+ && Objects.equals(_externalViewInstanceStateMap,
that._externalViewInstanceStateMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return _hashCode;
+ }
+ }
+
+ /// Returns one canonical [InstanceStateMaps] per distinct pair of instance
state maps, so that the segments of one
+ /// rebuild with equal maps share what is derived from them.
+ ///
+ /// The cache keeps at most `maxSize` pairs. Past that bound, it returns a
new [InstanceStateMaps] for each call,
+ /// which makes each segment compute its own state, as if there was no
sharing.
Review Comment:
Small documentation correction: reaching the limit only stops caching
previously unseen pairs. Pairs already in the cache continue sharing because
the lookup precedes the size check. Please clarify that here and in the
`MAX_SHARED_CANDIDATE_LISTS` comment.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java:
##########
@@ -559,4 +613,81 @@ int getPool(String instanceID) {
/// servers and let servers decide how to handle them.
protected abstract InstanceMapping select(List<String> segments, int
requestId, SegmentStates segmentStates,
Map<String, String> queryOptions);
+
+ /// The ideal state and external view instance state maps of a segment,
compared by content. Within one rebuild,
+ /// segments with equal maps get the same routing candidates, so the
canonical instance (see
+ /// [InstanceStateMapsCache]) caches what is derived from the maps, and the
segments share it.
+ ///
+ /// Not thread safe. Only used by the rebuild thread, for the duration of
one rebuild.
+ static final class InstanceStateMaps {
+ final Map<String, String> _idealStateInstanceStateMap;
+ @Nullable
+ final Map<String, String> _externalViewInstanceStateMap;
+ private final int _hashCode;
+ // Derived from the maps, computed on first use
+ @Nullable
+ Set<String> _onlineInstances;
+ @Nullable
+ List<SegmentInstanceCandidate> _oldSegmentCandidates;
+ // Used by the strict replica-group selector to collect the unavailable
instances once per instance state maps
+ boolean _unavailableInstancesCollected;
+
+ InstanceStateMaps(Map<String, String> idealStateInstanceStateMap,
+ @Nullable Map<String, String> externalViewInstanceStateMap) {
+ _idealStateInstanceStateMap = idealStateInstanceStateMap;
+ _externalViewInstanceStateMap = externalViewInstanceStateMap;
+ _hashCode = 31 * idealStateInstanceStateMap.hashCode() +
Objects.hashCode(externalViewInstanceStateMap);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof InstanceStateMaps)) {
+ return false;
+ }
+ InstanceStateMaps that = (InstanceStateMaps) o;
+ return _hashCode == that._hashCode &&
_idealStateInstanceStateMap.equals(that._idealStateInstanceStateMap)
+ && Objects.equals(_externalViewInstanceStateMap,
that._externalViewInstanceStateMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return _hashCode;
+ }
+ }
+
+ /// Returns one canonical [InstanceStateMaps] per distinct pair of instance
state maps, so that the segments of one
+ /// rebuild with equal maps share what is derived from them.
+ ///
+ /// The cache keeps at most `maxSize` pairs. Past that bound, it returns a
new [InstanceStateMaps] for each call,
+ /// which makes each segment compute its own state, as if there was no
sharing.
+ ///
+ /// Not thread safe. Only used by the rebuild thread, for the duration of
one rebuild.
+ static final class InstanceStateMapsCache {
+ private final int _maxSize;
+ private final Map<InstanceStateMaps, InstanceStateMaps> _cache = new
HashMap<>();
+
+ InstanceStateMapsCache(int maxSize) {
+ _maxSize = maxSize;
+ }
+
+ InstanceStateMaps canonicalize(Map<String, String>
idealStateInstanceStateMap,
+ @Nullable Map<String, String> externalViewInstanceStateMap) {
+ InstanceStateMaps instanceStateMaps =
+ new InstanceStateMaps(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ if (_maxSize <= 0) {
+ return instanceStateMaps;
+ }
+ InstanceStateMaps canonical = _cache.get(instanceStateMaps);
Review Comment:
Non-blocking benchmark follow-up: please include a workload with mostly
distinct assignments, including more than 4096 pairs, against the actual pre-PR
revision. Each lookup constructs a wrapper and hashes both maps, including
misses after the cache fills; strict routing also retains uncached wrappers
until the rebuild finishes. The repeated-assignment realtime results
demonstrate the benefit, but do not establish this tradeoff. Setting
`_maxSharedCandidateLists = 0` is not the pre-PR baseline because it still
incurs wrapper construction and hashing. This is a coverage suggestion, not a
demonstrated regression; preserve lookups after saturation because existing
entries can still hit.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java:
##########
@@ -443,10 +475,31 @@ void refreshSegmentStates() {
unavailableSegments.size());
}
+ /// Returns the enabled candidates, and adds their instances to the serving
instances.
+ ///
+ /// When every candidate is enabled, returns the given list itself rather
than a copy. That publishes the list to
+ /// the query path, which is safe because candidate lists are never modified
once built. It also keeps the lists
+ /// that [#updateSegmentMaps] shares between segments shared in the
published [SegmentStates].
private List<SegmentInstanceCandidate>
getEnabledCandidatesAndAddToServingInstances(
List<SegmentInstanceCandidate> candidates, Set<String> servingInstances)
{
- List<SegmentInstanceCandidate> enabledCandidates = new
ArrayList<>(candidates.size());
- for (SegmentInstanceCandidate candidate : candidates) {
+ int numCandidates = candidates.size();
+ int numLeadingEnabledCandidates = 0;
+ while (numLeadingEnabledCandidates < numCandidates) {
+ String instance =
candidates.get(numLeadingEnabledCandidates).getInstance();
+ if (!_enabledInstances.contains(instance)) {
+ break;
+ }
+ servingInstances.add(instance);
+ numLeadingEnabledCandidates++;
+ }
+ if (numLeadingEnabledCandidates == numCandidates) {
+ return candidates;
+ }
+ List<SegmentInstanceCandidate> enabledCandidates = new
ArrayList<>(numCandidates - 1);
+ enabledCandidates.addAll(candidates.subList(0,
numLeadingEnabledCandidates));
Review Comment:
Non-blocking allocation cleanup: use an indexed loop to copy the enabled
prefix instead of `subList` + `addAll`. On JDK 25, this constructs a sublist
view and `ArrayList.addAll` calls its `toArray()`, copying the prefix into a
temporary array, even for an empty prefix. A short loop avoids that
intermediate work on the disabled-instance path. Actual savings after JIT
optimization would need measurement.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/BaseInstanceSelector.java:
##########
@@ -90,8 +93,17 @@
/// which one owns the table's gauges.
public abstract class BaseInstanceSelector implements InstanceSelector {
private static final Logger LOGGER =
LoggerFactory.getLogger(BaseInstanceSelector.class);
+ /// Tags the INFO logs that print the whole new-segment map on every rebuild
("Got N new segments: ..."). To drop
+ /// them without code changes, add to the log4j2 config:
+ /// `<MarkerFilter marker="ROUTING_NEW_SEGMENTS" onMatch="DENY"
onMismatch="NEUTRAL"/>`
+ public static final Marker NEW_SEGMENTS_MARKER =
MarkerFactory.getMarker("ROUTING_NEW_SEGMENTS");
Review Comment:
Nit: this marker can be `private static final`. All references are inside
this class, and the Log4j filter matches the marker name, so it does not need a
public Java field.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/ReplicaGroupInstanceSelector.java:
##########
@@ -208,33 +208,41 @@ void updateSegmentMapsForUpsertTable(IdealState
idealState, ExternalView externa
Map<String, Map<String, String>> idealStateAssignment =
idealState.getRecord().getMapFields();
Map<String, Map<String, String>> externalViewAssignment =
externalView.getRecord().getMapFields();
- // Get the online instances for the segments
- Map<String, Set<String>> oldSegmentToOnlineInstancesMap =
+ // Get the online instances for the segments. Segments with the same
instance state maps share one
+ // InstanceStateMaps, which caches the online instances and the old
segment candidates derived from the maps.
+ InstanceStateMapsCache instanceStateMapsCache = new
InstanceStateMapsCache(_maxSharedCandidateLists);
+ Map<String, InstanceStateMaps> oldSegmentToInstanceStateMaps =
new HashMap<>(HashUtil.getHashMapCapacity(onlineSegments.size()));
- Map<String, Set<String>> newSegmentToOnlineInstancesMap = new
HashMap<>(newSegmentMapCapacity);
+ Map<String, InstanceStateMaps> newSegmentToInstanceStateMaps = new
HashMap<>(newSegmentMapCapacity);
for (String segment : onlineSegments) {
Map<String, String> idealStateInstanceStateMap =
idealStateAssignment.get(segment);
assert idealStateInstanceStateMap != null;
Map<String, String> externalViewInstanceStateMap =
externalViewAssignment.get(segment);
- Set<String> onlineInstances;
- if (externalViewInstanceStateMap == null) {
- onlineInstances = Set.of();
- } else {
- onlineInstances = getOnlineInstances(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ InstanceStateMaps instanceStateMaps =
+ instanceStateMapsCache.canonicalize(idealStateInstanceStateMap,
externalViewInstanceStateMap);
+ if (instanceStateMaps._onlineInstances == null) {
+ instanceStateMaps._onlineInstances = externalViewInstanceStateMap ==
null ? Set.of()
+ : getOnlineInstances(idealStateInstanceStateMap,
externalViewInstanceStateMap);
Review Comment:
Optional cleanup: `getOnlineInstances` now has only this caller, and its
consumers use only `contains()` and `size()`. It could return `Set<String>`
backed by `HashSet` instead of requiring `TreeSet`; candidate order is supplied
separately by `sortByKeyIfNeeded`. I would treat this as a simpler contract
rather than claim a meaningful speedup for small replica sets. Also, the
multiline ternary here should put `?` and `:` on separate lines per the
repository convention.
--
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]
