xiangfu0 commented on code in PR #19861:
URL: https://github.com/apache/pinot/pull/19861#discussion_r4278286585
##########
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)
+ throws Exception {
+ if (segmentDataManager instanceof RealtimeSegmentDataManager) {
Review Comment:
Consuming segments always report `false` here, so `onlyIfNeeded` never
force-commits them even though a plain reload would. That seems intentional
(and the javadoc calls it out), just flagging it so it's a conscious choice —
an operator using `onlyIfNeeded` to roll out a config won't commit consuming
segments.
##########
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)
+ throws Exception {
+ if (segmentDataManager instanceof RealtimeSegmentDataManager) {
+ return false;
+ }
+ IndexSegment segment = segmentDataManager.getSegment();
+ if (!(segment instanceof ImmutableSegmentImpl)) {
+ return true;
+ }
+ String segmentName = segmentDataManager.getSegmentName();
+ SegmentZKMetadata zkMetadata = fetchZKMetadata(segmentName);
+ if (shouldDownloadForCrcMismatch(zkMetadata,
segment.getSegmentMetadata())) {
+ return true;
+ }
+ String segmentTier = getSegmentCurrentTier(segmentName);
Review Comment:
Minor: we already hold `segmentDataManager` here, so
`segmentDataManager.getTier()` would give the current tier directly instead of
re-reading `_segmentDataManagerMap` via `getSegmentCurrentTier(segmentName)`.
Using the manager we hold avoids a tiny TOCTOU window (map entry could be
swapped between calls). Falls back to reload safely either way, so purely a
cleanup.
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/ControllerJobStatusResource.java:
##########
@@ -92,15 +92,21 @@ public String
reloadJobStatus(@PathParam("tableNameWithType") String tableNameWi
}
}
+ ReloadJobStatus jobStatus = reloadJobId != null ?
_serverReloadJobStatusCache.getJobStatus(reloadJobId) : null;
+ if (jobStatus != null) {
+ // Segments skipped by an only-if-needed reload keep their old load
time but are up to date, so they count as
+ // successful. Cap at the total in case a skipped segment was reloaded
again by a later operation.
+ successCount = (int) Math.min(successCount +
jobStatus.getSkippedCount(), totalSegmentCount);
Review Comment:
The `Math.min(..., totalSegmentCount)` cap is the right guard. One residual
edge: a skipped segment whose `loadTimeMs >= reloadJobSubmissionTimestamp`
(reloaded by some other op just after the job started) would be counted in both
`successCount` and `skippedCount`. The cap keeps the total sane, but success
could momentarily look higher than reality. Probably fine to leave, just noting
it.
##########
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);
Review Comment:
Nice that `isReloadNeeded` runs on the same `_segmentReloadExecutor` and the
caller blocks on `allOf(...).get()`, matching the existing batch-reload
pattern. One thing to confirm for yourself: for a full-table `onlyIfNeeded`
reload this now submits N check tasks and blocks the pool, then
`reloadSegments` submits another N — sequential, not nested, so no
self-starvation. Worth a quick mental check that no caller already holds a
reload-pool thread when entering here.
--
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]
