xiangfu0 commented on code in PR #19783:
URL: https://github.com/apache/pinot/pull/19783#discussion_r4278275031
##########
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:
Minor: once the cache reaches `MAX_CACHED_PARTITION_FUNCTIONS` distinct
keys, any further distinct metadata is never cached, so those functions are
rebuilt on every call and never shared. In practice a broker is very unlikely
to see >1024 distinct (name, numPartitions, config) combinations, so this is
fine — just flagging the degenerate case in case it's worth an eviction policy
or a note later. Non-blocking.
##########
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:
Minor: a segment whose ZK node is permanently absent will log this WARN on
every assignment change (and re-read it each time). You already call this out
in the PR description as an accepted trade-off, so no action needed — if it
ever gets noisy, a throttled/once-per-segment log could help. Non-blocking.
--
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]
