gortiz commented on code in PR #19783:
URL: https://github.com/apache/pinot/pull/19783#discussion_r4280602787
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtils.java:
##########
@@ -76,9 +100,42 @@ public static SegmentPartitionInfo
extractPartitionInfo(String tableNameWithType
return INVALID_PARTITION_INFO;
}
- return new SegmentPartitionInfo(partitionColumn,
- PartitionFunctionFactory.getPartitionFunction(columnPartitionMetadata),
- columnPartitionMetadata.getPartitions());
+ return new SegmentPartitionInfo(partitionColumn,
getPartitionFunction(columnPartitionMetadata),
+ getPartitions(columnPartitionMetadata));
+ }
+
+ /// Returns a partition function for the given metadata, shared with the
other segments with equal metadata.
+ private static PartitionFunction
getPartitionFunction(ColumnPartitionMetadata columnPartitionMetadata) {
+ PartitionFunctionKey key = new
PartitionFunctionKey(columnPartitionMetadata.getFunctionName(),
+ columnPartitionMetadata.getNumPartitions(),
columnPartitionMetadata.getFunctionConfig());
+ PartitionFunction partitionFunction = PARTITION_FUNCTION_CACHE.get(key);
+ if (partitionFunction != null) {
+ return partitionFunction;
+ }
+ partitionFunction =
PartitionFunctionFactory.getPartitionFunction(columnPartitionMetadata);
+ if (PARTITION_FUNCTION_CACHE.size() < MAX_CACHED_PARTITION_FUNCTIONS) {
Review Comment:
Agreed. More than 1024 distinct combinations is unlikely on one broker, and
past the limit the behavior is the same as before this PR. I will add an
eviction policy if we ever see it in practice.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java:
##########
@@ -206,20 +208,58 @@ public synchronized void onAssignmentChange(IdealState
idealState, ExternalView
Set<String> onlineSegments) {
updateExplicitlySetTimeBoundary(idealState);
+ // Read the ZK metadata of the new segments in one batch
+ List<String> newSegments = new ArrayList<>();
+ List<String> newSegmentZKMetadataPaths = new ArrayList<>();
for (String segment : onlineSegments) {
// NOTE: Only update the segment end time when there are ONLINE
instances in the external view to prevent moving
// the time boundary before the new segment is picked up by the
servers
- Map<String, String> instanceStateMap = externalView.getStateMap(segment);
- if (instanceStateMap != null &&
instanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
- _endTimeMsMap.computeIfAbsent(segment, k ->
extractEndTimeMsFromSegmentZKMetadataZNRecord(segment,
- _propertyStore.get(_segmentZKMetadataPathPrefix + segment, null,
AccessOption.PERSISTENT)));
+ if (!_endTimeMsMap.containsKey(segment)) {
+ Map<String, String> instanceStateMap =
externalView.getStateMap(segment);
+ if (instanceStateMap != null &&
instanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
+ newSegments.add(segment);
+ newSegmentZKMetadataPaths.add(_segmentZKMetadataPathPrefix +
segment);
+ }
+ }
+ }
+ // NOTE: Read before changing any state, so that a failed read leaves the
state as it was. The read throws on a ZK
+ // error other than a missing node, so that the routing entry keeps
its IS/EV versions and retries.
+ List<ZNRecord> znRecords = newSegments.isEmpty() ? List.of()
+ : _propertyStore.get(newSegmentZKMetadataPaths, null,
AccessOption.PERSISTENT, true);
+ boolean recomputeMaxEndTime = false;
+ // NOTE: Removing a segment can only lower the max end time, so a full
scan is needed only when a removed segment
+ // holds the current max end time.
+ Iterator<Map.Entry<String, Long>> iterator =
_endTimeMsMap.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<String, Long> entry = iterator.next();
+ if (!onlineSegments.contains(entry.getKey())) {
+ if (entry.getValue() >= _maxEndTimeMs) {
+ recomputeMaxEndTime = true;
+ }
+ iterator.remove();
+ }
+ }
+ if (recomputeMaxEndTime) {
+ _maxEndTimeMs = computeMaxEndTimeMs();
+ }
+ int numNewSegments = newSegments.size();
+ for (int i = 0; i < numNewSegments; i++) {
+ String segment = newSegments.get(i);
+ ZNRecord znRecord = znRecords.get(i);
+ if (znRecord == null) {
+ // NOTE: The node does not exist. Do not record the segment, so that
the next assignment change reads it again.
+ LOGGER.warn("Failed to find segment ZK metadata for segment: {},
table: {}, will retry on next change", segment,
Review Comment:
Agreed. If the WARN gets noisy, I will log it once per segment.
--
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]
