xiangfu0 commented on code in PR #20106:
URL: https://github.com/apache/pinot/pull/20106#discussion_r4559225556
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/tablestate/TableStateUtils.java:
##########
@@ -82,30 +91,162 @@ public static List<String>
getSegmentsInGivenStateForThisInstance(HelixManager h
public static boolean isAllSegmentsLoaded(HelixManager helixManager, String
tableNameWithType) {
List<String> onlineSegments =
getSegmentsInGivenStateForThisInstance(helixManager,
tableNameWithType, SegmentStateModel.ONLINE);
- if (onlineSegments.isEmpty()) {
- LOGGER.info("No ONLINE segment found for table: {}", tableNameWithType);
- return true;
- }
+ return areSegmentsLoaded(helixManager, tableNameWithType, onlineSegments);
+ }
- // Check if ideal state and current state matches for all segments
assigned to the current instance
+ /// Checks the current assignment before a consumer starts. Earlier ONLINE
segments of the same partition must be
+ /// loaded, as must uploaded segments from that partition. Resolve
partitions from names or segment metadata,
+ /// matching upsert and dedup loading; unresolved segments are
conservatively included. The consumer and its
+ /// successors are excluded to avoid waiting on themselves.
+ ///
+ /// @param isLocallyReady identifies segments whose metadata is already
ready locally despite a pending ONLINE
+ /// transition, such as a predecessor that finished
consumption and released its permit
+ public static boolean isReadyToConsume(HelixManager helixManager, String
tableNameWithType, LLCSegmentName consumer,
+ Predicate<String> isLocallyReady) {
HelixDataAccessor dataAccessor = helixManager.getHelixDataAccessor();
- PropertyKey.Builder keyBuilder = dataAccessor.keyBuilder();
- String instanceName = helixManager.getInstanceName();
- LiveInstance liveInstance =
dataAccessor.getProperty(keyBuilder.liveInstance(instanceName));
- if (liveInstance == null) {
- LOGGER.warn("Failed to find live instance for instance: {}",
instanceName);
+ IdealState idealState =
dataAccessor.getProperty(dataAccessor.keyBuilder().idealStates(tableNameWithType));
+ if (idealState == null) {
+ LOGGER.warn("Failed to find ideal state for table: {}",
tableNameWithType);
return false;
}
- String sessionId = liveInstance.getEphemeralOwner();
- CurrentState currentState =
- dataAccessor.getProperty(keyBuilder.currentState(instanceName,
sessionId, tableNameWithType));
- if (currentState == null) {
- LOGGER.warn("Failed to find current state for instance: {}, sessionId:
{}, table: {}", instanceName, sessionId,
- tableNameWithType);
+ String instanceName = helixManager.getInstanceName();
+ Map<String, String> consumerAssignment =
idealState.getInstanceStateMap(consumer.getSegmentName());
+ String consumerState = consumerAssignment != null ?
consumerAssignment.get(instanceName) : null;
+ if (!SegmentStateModel.CONSUMING.equals(consumerState) &&
!SegmentStateModel.ONLINE.equals(consumerState)) {
return false;
}
+
+ List<String> requiredSegments = new ArrayList<>();
+ Function<String, Integer> partitionIdFromMetadata =
+ segmentName -> getPartitionIdFromMetadata(helixManager,
tableNameWithType, segmentName);
+ for (Map.Entry<String, Map<String, String>> entry :
idealState.getRecord().getMapFields().entrySet()) {
+ if
(!SegmentStateModel.ONLINE.equals(entry.getValue().get(instanceName))) {
+ continue;
+ }
+ String segmentName = entry.getKey();
+ if (isRequiredHistory(segmentName, consumer, partitionIdFromMetadata) &&
!isLocallyReady.test(segmentName)) {
+ requiredSegments.add(segmentName);
+ }
+ }
+ return areSegmentsLoaded(helixManager, tableNameWithType,
requiredSegments);
+ }
+
+ /// Reads the table assignment, this live instance's current state, and
partition metadata for custom-named ONLINE
+ /// segments once for evaluation by multiple consumers. Subsequent Helix
updates cannot mutate the snapshot.
+ /// A cached snapshot must only reject readiness; callers must perform a
fresh check before authorizing consumption
+ /// from a cached result.
+ public static ReadinessSnapshot getReadinessSnapshot(HelixManager
helixManager, String tableNameWithType) {
+ HelixDataAccessor dataAccessor = helixManager.getHelixDataAccessor();
+ IdealState idealState =
dataAccessor.getProperty(dataAccessor.keyBuilder().idealStates(tableNameWithType));
+ if (idealState == null) {
+ LOGGER.warn("Failed to find ideal state for table: {}",
tableNameWithType);
+ return new ReadinessSnapshot(Map.of(), Map.of(), Map.of());
+ }
+ String instanceName = helixManager.getInstanceName();
+ Map<String, String> idealStates = new HashMap<>();
+ Map<String, Integer> metadataPartitionIds = new HashMap<>();
+ for (Map.Entry<String, Map<String, String>> entry :
idealState.getRecord().getMapFields().entrySet()) {
+ String segmentName = entry.getKey();
+ String state = entry.getValue().get(instanceName);
+ if (state != null) {
+ idealStates.put(segmentName, state);
+ if (SegmentStateModel.ONLINE.equals(state) &&
SegmentUtils.getPartitionIdFromSegmentName(segmentName) == null) {
+ Integer partitionId = getPartitionIdFromMetadata(helixManager,
tableNameWithType, segmentName);
Review Comment:
**MAJOR: one ZK read per custom-named segment, even when it's already
loaded.** This loop runs for every ONLINE custom segment on each snapshot
refresh, while holding `_readinessSnapshotLock`, and line 127 does the same on
each fresh confirmation. Probe with 1000 loaded custom segments and 8
partitions: 8000 metadata reads at startup and 1000 per rollover, versus 0 at
`33ecaff2`. On a localhost ZK, 1000 sequential reads took about 90–105 ms.
Could the partition lookup run only for segments that are neither loaded
(CURRENTSTATE ONLINE) nor locally initialized? Those are the only ones that can
block, and in steady state there are almost none.
--
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]
