gortiz commented on code in PR #19785:
URL: https://github.com/apache/pinot/pull/19785#discussion_r4257488121
##########
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:
Done, thanks.
##########
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:
Done, thanks.
##########
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:
Done, thanks.
##########
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:
Done: it now returns a HashSet-backed Set, and I split the ternary.
--
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]
