raghavyadav01 commented on code in PR #19861:
URL: https://github.com/apache/pinot/pull/19861#discussion_r4278103923
##########
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:
If a skipped segment gets reloaded or refreshed by something else while this
job is running, it's counted twice: once by load time and once in
`skippedCount`. The cap only limits the total, so the job can show 100% while a
segment that actually needed a reload is still in progress. Could we track the
skipped segment names and exclude them from the load-time count? A unit test
for this accounting would help too.
##########
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:
There are now three staleness checks: `needReloadSegments()` (preprocess
only), `isSegmentStale()` (schema/config diff), and this one (CRC + tier +
preprocess). Could `needReloadSegments()` delegate to `isReloadNeeded()` so the
reload API and the check API don't drift?
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -758,7 +772,12 @@ public void reloadSegments(List<String> segmentNames,
boolean forceDownload, Str
}
}
if (!segmentDataManagers.isEmpty()) {
- reloadSegments(segmentDataManagers, fetchIndexLoadingConfig(),
forceDownload, reloadJobId);
+ IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig();
+ List<SegmentDataManager> segmentsToReload = onlyIfNeeded &&
!forceDownload
+ ? selectSegmentsNeedingReload(segmentDataManagers,
indexLoadingConfig, reloadJobId) : segmentDataManagers;
+ if (!segmentsToReload.isEmpty()) {
+ reloadSegments(segmentsToReload, indexLoadingConfig, forceDownload,
reloadJobId);
+ }
}
if (missingSegments.isEmpty()) {
_logger.info("Reloaded segments: {} with forceDownload: {}",
segmentNames, forceDownload);
Review Comment:
nit: this still logs every requested segment as reloaded, including the
skipped ones. Maybe log `segmentsToReload` instead, like `reloadAllSegments`
does with the filtered count.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -712,41 +712,55 @@ public static void deleteSegmentFilesFromDisk(String
tableDataDir, String segmen
}
@Override
- public void reloadSegment(String segmentName, boolean forceDownload, String
reloadJobId)
+ public void reloadSegment(String segmentName, boolean forceDownload, boolean
onlyIfNeeded, String reloadJobId)
throws Exception {
Preconditions.checkState(!_shutDown,
"Table data manager is already shut down, cannot reload segment: %s in
table: %s", segmentName,
_tableNameWithType);
- _logger.info("Reloading segment: {} with forceDownload: {}", segmentName,
forceDownload);
+ _logger.info("Reloading segment: {} with forceDownload: {}, onlyIfNeeded:
{}", segmentName, forceDownload,
+ onlyIfNeeded);
SegmentDataManager segmentDataManager =
_segmentDataManagerMap.get(segmentName);
if (segmentDataManager != null) {
IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig();
+ if (onlyIfNeeded && !forceDownload &&
selectSegmentsNeedingReload(List.of(segmentDataManager),
Review Comment:
nit: for a single segment this sends one task to the reload executor and
blocks on it. Calling `isReloadNeeded` inline, with the ref-count guard, would
be simpler.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableReloadResource.java:
##########
@@ -77,6 +77,10 @@
@Path("/")
public class PinotTableReloadResource {
private static final Logger LOG =
LoggerFactory.getLogger(PinotTableReloadResource.class);
+ private static final String ONLY_IF_NEEDED_DESCRIPTION = "Reload only the
segments whose loaded copy is out of "
+ + "date (CRC changed, tier changed, or index/schema preprocessing
needed); up-to-date segments are skipped and "
+ + "counted as successful. Changes that only apply at load time still
require a full reload. Ignored when "
Review Comment:
Users can't tell what "changes that only apply at load time" covers. Could
we name a few, like read mode, upsert/dedup config, and tier storage
properties, so it's clear when not to use this flag?
--
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]
