J-HowHuang commented on code in PR #18174:
URL: https://github.com/apache/pinot/pull/18174#discussion_r3082279792
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -926,17 +926,18 @@ protected void
setZkOperationTimeIfAvailable(ImmutableSegment segment, @Nullable
}
}
- private void reloadSegments(List<SegmentDataManager> segmentDataManagers,
IndexLoadingConfig indexLoadingConfig,
+ private void reloadSegmentDataManagersInParallel(List<SegmentDataManager>
segmentDataManagers,
Review Comment:
I think we can keep the original name, and mention the parallel to this
method's javadoc if you feel like.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/IndexLoadingConfig.java:
##########
@@ -119,6 +119,12 @@ public IndexLoadingConfig() {
this(null, null, null);
}
+ public IndexLoadingConfig copy() {
+ IndexLoadingConfig copy = new
IndexLoadingConfig(_instanceDataManagerConfig, _tableConfig, _schema);
+ copy.setTableDataDir(_tableDataDir);
+ return copy;
+ }
Review Comment:
Let's not have this method. This doesn't copy everything in this object
(e.g. `_readMode` is not copied here), which would fail to meet the semantic of
"copy". Unless you make sure everything is honored while performing the copy.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/BaseTableDataManager.java:
##########
@@ -926,17 +926,18 @@ protected void
setZkOperationTimeIfAvailable(ImmutableSegment segment, @Nullable
}
}
- private void reloadSegments(List<SegmentDataManager> segmentDataManagers,
IndexLoadingConfig indexLoadingConfig,
+ private void reloadSegmentDataManagersInParallel(List<SegmentDataManager>
segmentDataManagers,
boolean forceDownload, String reloadJobId)
throws Exception {
+ IndexLoadingConfig indexLoadingConfigTemplate = fetchIndexLoadingConfig();
List<String> failedSegments = new ArrayList<>();
AtomicReference<Throwable> sampleException = new AtomicReference<>();
CompletableFuture.allOf(segmentDataManagers.stream().map(segmentDataManager ->
CompletableFuture.runAsync(() -> {
String segmentName = segmentDataManager.getSegmentName();
try {
_segmentReloadSemaphore.acquire(segmentName, _logger);
try {
- reloadSegment(segmentDataManager, indexLoadingConfig, forceDownload);
+ reloadSegment(segmentDataManager, indexLoadingConfigTemplate.copy(),
forceDownload);
Review Comment:
If we have 100k of segments then we'll have as many copies of
`IndexLoadingConfig` objects here, with only the segment tier aren't the same.
Segment tiers are expected to be only a few per server. Can we have, for
example, a map from tier to index loading config so we only need to create as
many copies as the amount of tiers?
--
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]
