J-HowHuang commented on code in PR #19861:
URL: https://github.com/apache/pinot/pull/19861#discussion_r4280464879
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -1085,6 +1104,100 @@ protected void
setZkOperationTimeIfAvailable(ImmutableSegment segment, @Nullable
}
}
+ /// Returns the subset of `segmentDataManagers` that [#isReloadNeeded]
reports as out of date, evaluating the check
+ /// in parallel on the reload executor. Skipped segments are counted in the
reload job status (as successful) when
+ /// `reloadJobId` is set. A segment is kept (i.e. reloaded as usual) when
its check fails, so an error never turns
+ /// into a silently skipped reload.
+ ///
+ /// The check runs against the currently loaded copy without holding the
segment lock. Skipping on that snapshot is
+ /// safe: anything that later changes the segment (a replace or another
reload) loads it against the then-current
+ /// config, and a segment found to need reload still goes through the
regular reload path, which re-evaluates
+ /// everything under the lock.
+ @VisibleForTesting
+ List<SegmentDataManager>
selectSegmentsNeedingReload(List<SegmentDataManager> segmentDataManagers,
+ IndexLoadingConfig indexLoadingConfig, @Nullable String reloadJobId)
+ throws Exception {
+ boolean[] needReload = new boolean[segmentDataManagers.size()];
+ CompletableFuture<?>[] futures = new
CompletableFuture<?>[segmentDataManagers.size()];
+ for (int i = 0; i < segmentDataManagers.size(); i++) {
+ int index = i;
+ SegmentDataManager segmentDataManager = segmentDataManagers.get(i);
+ futures[i] = CompletableFuture.runAsync(() -> {
+ String segmentName = segmentDataManager.getSegmentName();
+ // Hold a reference so the segment cannot be destroyed while its
directory is inspected. If it is already
+ // being destroyed it was replaced or offloaded concurrently; keep it
so the regular reload path handles it.
+ if (!segmentDataManager.increaseReferenceCount()) {
+ needReload[index] = true;
+ return;
+ }
+ try {
+ needReload[index] = isReloadNeeded(segmentDataManager,
indexLoadingConfig);
+ } catch (Exception e) {
+ _logger.warn("Caught exception while checking whether segment: {}
needs reload, reloading it", segmentName,
+ e);
+ needReload[index] = true;
+ } finally {
+ releaseSegment(segmentDataManager);
+ }
+ }, _segmentReloadExecutor);
+ }
+ CompletableFuture.allOf(futures).get();
+
+ List<SegmentDataManager> segmentsToReload = new ArrayList<>();
+ List<String> skippedSegments = new ArrayList<>();
+ for (int i = 0; i < segmentDataManagers.size(); i++) {
+ if (needReload[i]) {
+ segmentsToReload.add(segmentDataManagers.get(i));
+ } else {
+ skippedSegments.add(segmentDataManagers.get(i).getSegmentName());
+ }
+ }
+ if (!skippedSegments.isEmpty()) {
+ _logger.info("Skipping reload for {}/{} segments that are already up to
date", skippedSegments.size(),
+ segmentDataManagers.size());
+ _logger.debug("Skipped reload for up-to-date segments: {}",
skippedSegments);
+ if (reloadJobId != null) {
+ _reloadJobStatusCache.recordSkipped(reloadJobId,
skippedSegments.size());
+ }
+ }
+ return segmentsToReload;
+ }
+
+ /// Returns whether reloading the segment would change what is loaded, i.e.
whether the regular reload path would
+ /// download a new copy (completed segment whose CRC differs from ZK), move
it to a different tier, or preprocess it
+ /// under the given index loading config. Consuming segments are never
considered as needing reload, as reloading
+ /// them means force committing.
+ protected boolean isReloadNeeded(SegmentDataManager segmentDataManager,
IndexLoadingConfig indexLoadingConfig)
Review Comment:
Done. `isSegmentStale` is left unchange, it has quite its own logic
--
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]
