You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "abhishekbafna (via GitHub)" <gi...@apache.org> on 2025/07/01 13:30:31 UTC
[PR] Implement parallel segment download and untar for improved performance [pinot]
abhishekbafna opened a new pull request, #16249:
URL: https://github.com/apache/pinot/pull/16249
- This pull request improves the efficiency and scalability of segment downloading and extraction by introducing parallel processing.
- The key changes include replacing sequential segment processing with a multithreaded approach, adding a new helper method for parallel execution, and updating the data structures to support thread safety.
label: performance
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2204332057
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -190,35 +195,26 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
String downloadURLString = taskConfigs.get(MinionConstants.DOWNLOAD_URL_KEY);
String[] downloadURLs = downloadURLString.split(MinionConstants.URL_SEPARATOR);
AuthProvider authProvider = AuthProviderUtils.makeAuthProvider(taskConfigs.get(MinionConstants.AUTH_TOKEN));
- LOGGER.info("Start executing {} on table: {}, input segments: {} with downloadURLs: {}, uploadURL: {}", taskType,
- tableNameWithType, inputSegmentNames, downloadURLString, uploadURL);
File tempDataDir = new File(new File(MINION_CONTEXT.getDataDir(), taskType), "tmp-" + UUID.randomUUID());
Preconditions.checkState(tempDataDir.mkdirs());
+ int numRecords;
+ List<File> inputSegmentDirs = new ArrayList<>(downloadURLs.length);
+ Map<String, SegmentMetadata> segmentMetadataMap = Collections.synchronizedMap(new HashMap<>(downloadURLs.length));
+ int nThreads = Math.min(getParallelism(taskConfigs), downloadURLs.length);
+ LOGGER.info(
+ "Start executing {} on table: {}, input segments: {} with downloadURLs: {}, uploadURL: {}, thread pool size:{}",
+ taskType, tableNameWithType, inputSegmentNames, downloadURLString, uploadURL, nThreads);
try {
- List<File> inputSegmentDirs = new ArrayList<>();
- int numRecords = 0;
-
- for (int i = 0; i < downloadURLs.length; i++) {
- String segmentName = segmentNames[i];
- // Download and decompress the segment file
- _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURLs[i]
- + " (" + (i + 1) + " out of " + downloadURLs.length + ")");
- File indexDir;
- try {
- indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURLs[i], taskType,
- tempDataDir, "_" + i);
- } catch (Exception e) {
- LOGGER.error("Failed to download segment from download url: {}", downloadURLs[i], e);
- _minionMetrics.addMeteredTableValue(tableNameWithType, MinionMeter.SEGMENT_DOWNLOAD_FAIL_COUNT, 1L);
- _eventObserver.notifyTaskError(_pinotTaskConfig, e);
- throw e;
+ if (nThreads <= 1) {
+ for (int index = 0; index < downloadURLs.length; index++) {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, segmentMetadataMap, segmentNames.length);
}
- inputSegmentDirs.add(indexDir);
-
- reportSegmentDownloadMetrics(indexDir, tableNameWithType, taskType);
- SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
- numRecords += segmentMetadata.getTotalDocs();
+ } else {
+ parallelDownloadAndUntarSegments(nThreads, tableNameWithType, taskType, segmentNames, downloadURLs,
+ tempDataDir, segmentMetadataMap);
}
+ numRecords = processSegmentMetadata(segmentNames, segmentMetadataMap, inputSegmentDirs);
Review Comment:
nit : processSegmentMetadata is doing 2 things - calculating numRecords and modifying an input variable inputSegmentDirs.
I think it'l be cleaner to just create 2 methods getInputSegmentDirs(segmentMetadataMap, segmentNames) and getNumRecords(segmentMetadataMap)
##########
pinot-minion/src/main/java/org/apache/pinot/minion/MinionConf.java:
##########
@@ -31,6 +31,13 @@ public class MinionConf extends PinotConfiguration {
public static final String MINION_TASK_PROGRESS_MANAGER_CLASS = "pinot.minion.taskProgressManager.class";
public static final int DEFAULT_END_REPLACE_SEGMENTS_SOCKET_TIMEOUT_MS = 10 * 60 * 1000; // 10 mins
+ /**
+ * The number of threads to use for downloading segments from the deepstore.
+ * This is a global setting that applies to all tasks of BaseMultipleSegmentsConversionExecutor class.
Review Comment:
nit - lets document that this config is applied at a task level and total parallelism will be this factor * number of parallel tasks running.
that will useful to clarify in documentation because this config is at a minion level and can be easily misinterpreted to be a minion level factor
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2192128700
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +349,75 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int getThreadPoolSize(Map<String, String> taskConfigs) {
+ int nThreads = _minionConf.getProperty(Minion.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
+ MinionConstants.DEFAULT_SEGMENT_DOWNLOAD_THREAD_POOL_SIZE);
+ nThreads = Integer.parseInt(taskConfigs.getOrDefault(MinionConstants.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
+ String.valueOf(nThreads)));
+ return nThreads;
+ }
+
+ private void parallelDownloadAndUntarSegments(int nThreads, String tableNameWithType, String taskType,
+ String[] segmentNames, String[] downloadURLs, File tempDataDir, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+
+ ExecutorService executorService = null;
+ int length = downloadURLs.length;
+ try {
+ executorService = Executors.newFixedThreadPool(nThreads);
+ List<Future<Void>> futures = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ int index = i;
+ futures.add(executorService.submit(() -> {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, recordCounter, inputSegmentDirs);
+ return null;
+ }));
+
+ // Wait for all downloads to complete and cancel other tasks if any download fails
+ for (Future<Void> future : futures) {
Review Comment:
+1, tests failures are related to this
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2204618568
##########
pinot-minion/src/main/java/org/apache/pinot/minion/MinionConf.java:
##########
@@ -31,6 +31,13 @@ public class MinionConf extends PinotConfiguration {
public static final String MINION_TASK_PROGRESS_MANAGER_CLASS = "pinot.minion.taskProgressManager.class";
public static final int DEFAULT_END_REPLACE_SEGMENTS_SOCKET_TIMEOUT_MS = 10 * 60 * 1000; // 10 mins
+ /**
+ * The number of threads to use for downloading segments from the deepstore.
+ * This is a global setting that applies to all tasks of BaseMultipleSegmentsConversionExecutor class.
Review Comment:
It should not be the case. As the default value for `DEFAULT_SEGMENT_DOWNLOAD_PARALLELISM = 1` config is set to 1 and it can not be interpreted that only 1 segment download would happen across the tasks.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2185135111
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -195,30 +201,20 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
File tempDataDir = new File(new File(MINION_CONTEXT.getDataDir(), taskType), "tmp-" + UUID.randomUUID());
Preconditions.checkState(tempDataDir.mkdirs());
try {
- List<File> inputSegmentDirs = new ArrayList<>();
- int numRecords = 0;
-
- for (int i = 0; i < downloadURLs.length; i++) {
- String segmentName = segmentNames[i];
- // Download and decompress the segment file
- _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURLs[i]
- + " (" + (i + 1) + " out of " + downloadURLs.length + ")");
- File indexDir;
- try {
- indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURLs[i], taskType,
- tempDataDir, "_" + i);
- } catch (Exception e) {
- LOGGER.error("Failed to download segment from download url: {}", downloadURLs[i], e);
- _minionMetrics.addMeteredTableValue(tableNameWithType, MinionMeter.SEGMENT_DOWNLOAD_FAIL_COUNT, 1L);
- _eventObserver.notifyTaskError(_pinotTaskConfig, e);
- throw e;
+ AtomicInteger recordCount = new AtomicInteger(0);
+ List<File> inputSegmentDirs = Collections.synchronizedList(new ArrayList<>(downloadURLs.length));
+ int nThreads = Integer.parseInt(taskConfigs.getOrDefault(MinionConstants.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
Review Comment:
> would it be better to use minionConf here rather than task config ? if its part of task config, we will have to modify all task configs for all tables to make use of this feature.
This is correct but on that other hand task level would allow for different config for different tasks. Provides much more control than a single global config.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2185134422
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +348,67 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private void parallelDownloadAndUntarSegments(int nThreads, String tableNameWithType, String taskType,
+ String[] segmentNames, String[] downloadURLs, File tempDataDir, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+
+ ExecutorService executorService = null;
+ int length = downloadURLs.length;
+ try {
+ executorService = Executors.newFixedThreadPool(nThreads);
+ List<Future<Void>> futures = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ int index = i;
+ futures.add(executorService.submit(() -> {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, recordCounter, inputSegmentDirs);
+ return null;
+ }));
+
+ // Wait for all downloads to complete and cancel other tasks if any download fails
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (Exception e) {
+ // Cancel all other download tasks
+ for (Future<Void> f : futures) {
+ f.cancel(true);
+ }
+ throw e;
+ }
+ }
+ }
+ } finally {
+ if (executorService != null) {
+ executorService.shutdown();
+ }
+ }
+ }
+
+ private void downloadAndUntarSegment(String tableNameWithType, String taskType,
+ String segmentName, String downloadURL, File tempDataDir, int index, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+ try {
+ _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURL
+ + " (" + (index + 1) + " out of " + inputSegmentDirs.size() + ")");
+ // Download and decompress the segment file
+ File indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURL, taskType,
+ tempDataDir, "_" + index);
+ reportSegmentDownloadMetrics(indexDir, tableNameWithType, taskType);
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(indexDir);
+ // Ensure segment directory placement is at same index as the segment name in the inputSegmentNames
+ inputSegmentDirs.set(index, indexDir);
+ recordCounter.addAndGet(segmentMetadata.getTotalDocs());
+ } catch (Exception e) {
Review Comment:
I think, this is better. This keep the logic together and ensure that `inputSegmentDirs` and `recordCounter` only happens when everything above succeed.
The `new SegmentMetadataImpl(indexDir);` also throes IOException and other.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #16249:
URL: https://github.com/apache/pinot/pull/16249#issuecomment-3024265170
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/16249?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `0%` with `38 lines` in your changes missing coverage. Please review.
> Project coverage is 33.37%. Comparing base [(`1a476de`)](https://app.codecov.io/gh/apache/pinot/commit/1a476dea72ebf1c15e14ded28ae60c385f71b2c3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`fdbba50`)](https://app.codecov.io/gh/apache/pinot/commit/fdbba5029f3cb676fb4102ed35c67b8df145c9de?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 377 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/16249?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../tasks/BaseMultipleSegmentsConversionExecutor.java](https://app.codecov.io/gh/apache/pinot/pull/16249?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-minion-tasks%2Fpinot-minion-builtin-tasks%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Fminion%2Ftasks%2FBaseMultipleSegmentsConversionExecutor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1taW5pb24tdGFza3MvcGlub3QtbWluaW9uLWJ1aWx0aW4tdGFza3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3Bpbm90L3BsdWdpbi9taW5pb24vdGFza3MvQmFzZU11bHRpcGxlU2VnbWVudHNDb252ZXJzaW9uRXhlY3V0b3IuamF2YQ==) | 0.00% | [38 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/16249?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (1a476de) and HEAD (fdbba50). Click for more details.
>
> <details><summary>HEAD has 68 uploads less than BASE</summary>
>
>| Flag | BASE (1a476de) | HEAD (fdbba50) |
>|------|------|------|
>|java-21|10|1|
>|unittests|6|1|
>|unittests1|3|0|
>|skip-bytebuffers-true|4|0|
>|temurin|15|1|
>|unittests2|3|1|
>|skip-bytebuffers-false|8|0|
>|java-11|5|0|
>|integration|9|0|
>|integration2|3|0|
>|integration1|3|0|
>|custom-integration1|3|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #16249 +/- ##
=============================================
- Coverage 62.90% 33.37% -29.53%
+ Complexity 1386 656 -730
=============================================
Files 2867 2962 +95
Lines 163354 171198 +7844
Branches 24952 26212 +1260
=============================================
- Hits 102755 57140 -45615
- Misses 52847 109457 +56610
+ Partials 7752 4601 -3151
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.37% <0.00%> (-29.45%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [skip-bytebuffers-true](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.37% <0.00%> (-29.53%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.37% <0.00%> (-29.53%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/16249/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `33.37% <0.00%> (-0.20%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/16249?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2184695459
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -195,30 +201,20 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
File tempDataDir = new File(new File(MINION_CONTEXT.getDataDir(), taskType), "tmp-" + UUID.randomUUID());
Preconditions.checkState(tempDataDir.mkdirs());
try {
- List<File> inputSegmentDirs = new ArrayList<>();
- int numRecords = 0;
-
- for (int i = 0; i < downloadURLs.length; i++) {
- String segmentName = segmentNames[i];
- // Download and decompress the segment file
- _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURLs[i]
- + " (" + (i + 1) + " out of " + downloadURLs.length + ")");
- File indexDir;
- try {
- indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURLs[i], taskType,
- tempDataDir, "_" + i);
- } catch (Exception e) {
- LOGGER.error("Failed to download segment from download url: {}", downloadURLs[i], e);
- _minionMetrics.addMeteredTableValue(tableNameWithType, MinionMeter.SEGMENT_DOWNLOAD_FAIL_COUNT, 1L);
- _eventObserver.notifyTaskError(_pinotTaskConfig, e);
- throw e;
+ AtomicInteger recordCount = new AtomicInteger(0);
+ List<File> inputSegmentDirs = Collections.synchronizedList(new ArrayList<>(downloadURLs.length));
+ int nThreads = Integer.parseInt(taskConfigs.getOrDefault(MinionConstants.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
Review Comment:
we can also set nThreads = max(downloadURLs.length, nThreads) so that if tasks download just 1 segment, then they don't create a new thread for it.
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +348,67 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private void parallelDownloadAndUntarSegments(int nThreads, String tableNameWithType, String taskType,
+ String[] segmentNames, String[] downloadURLs, File tempDataDir, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+
+ ExecutorService executorService = null;
+ int length = downloadURLs.length;
+ try {
+ executorService = Executors.newFixedThreadPool(nThreads);
+ List<Future<Void>> futures = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ int index = i;
+ futures.add(executorService.submit(() -> {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, recordCounter, inputSegmentDirs);
+ return null;
+ }));
+
+ // Wait for all downloads to complete and cancel other tasks if any download fails
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (Exception e) {
+ // Cancel all other download tasks
+ for (Future<Void> f : futures) {
+ f.cancel(true);
+ }
+ throw e;
+ }
+ }
+ }
+ } finally {
+ if (executorService != null) {
+ executorService.shutdown();
+ }
+ }
+ }
+
+ private void downloadAndUntarSegment(String tableNameWithType, String taskType,
+ String segmentName, String downloadURL, File tempDataDir, int index, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+ try {
+ _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURL
+ + " (" + (index + 1) + " out of " + inputSegmentDirs.size() + ")");
+ // Download and decompress the segment file
+ File indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURL, taskType,
+ tempDataDir, "_" + index);
+ reportSegmentDownloadMetrics(indexDir, tableNameWithType, taskType);
+ SegmentMetadata segmentMetadata = new SegmentMetadataImpl(indexDir);
+ // Ensure segment directory placement is at same index as the segment name in the inputSegmentNames
+ inputSegmentDirs.set(index, indexDir);
+ recordCounter.addAndGet(segmentMetadata.getTotalDocs());
+ } catch (Exception e) {
Review Comment:
nit - we need to catch the exception only for downloadSegmentToLocalAndUntar right ? we should avoid catching the exception for the entire block of code ?
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -195,30 +201,20 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
File tempDataDir = new File(new File(MINION_CONTEXT.getDataDir(), taskType), "tmp-" + UUID.randomUUID());
Preconditions.checkState(tempDataDir.mkdirs());
try {
- List<File> inputSegmentDirs = new ArrayList<>();
- int numRecords = 0;
-
- for (int i = 0; i < downloadURLs.length; i++) {
- String segmentName = segmentNames[i];
- // Download and decompress the segment file
- _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURLs[i]
- + " (" + (i + 1) + " out of " + downloadURLs.length + ")");
- File indexDir;
- try {
- indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURLs[i], taskType,
- tempDataDir, "_" + i);
- } catch (Exception e) {
- LOGGER.error("Failed to download segment from download url: {}", downloadURLs[i], e);
- _minionMetrics.addMeteredTableValue(tableNameWithType, MinionMeter.SEGMENT_DOWNLOAD_FAIL_COUNT, 1L);
- _eventObserver.notifyTaskError(_pinotTaskConfig, e);
- throw e;
+ AtomicInteger recordCount = new AtomicInteger(0);
+ List<File> inputSegmentDirs = Collections.synchronizedList(new ArrayList<>(downloadURLs.length));
+ int nThreads = Integer.parseInt(taskConfigs.getOrDefault(MinionConstants.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
Review Comment:
would it be better to use minionConf here rather than task config ? if its part of task config, we will have to modify all task configs for all tables to make use of this feature.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2193431697
##########
pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java:
##########
@@ -72,6 +72,11 @@ private MinionConstants() {
*/
public static final int DEFAULT_MAX_ATTEMPTS_PER_TASK = 1;
+ /**
+ * Segment download thread pool size to be set at task level.
+ */
+ public static final String SEGMENT_DOWNLOAD_THREAD_POOL_SIZE = "segmentDownloadThreadPoolSize";
Review Comment:
Consider `segmentDownloadParallelism` which is easier for users to understand IMO
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2179700785
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -195,30 +200,9 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
File tempDataDir = new File(new File(MINION_CONTEXT.getDataDir(), taskType), "tmp-" + UUID.randomUUID());
Preconditions.checkState(tempDataDir.mkdirs());
try {
- List<File> inputSegmentDirs = new ArrayList<>();
- int numRecords = 0;
-
- for (int i = 0; i < downloadURLs.length; i++) {
- String segmentName = segmentNames[i];
- // Download and decompress the segment file
- _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURLs[i]
- + " (" + (i + 1) + " out of " + downloadURLs.length + ")");
- File indexDir;
- try {
- indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURLs[i], taskType,
- tempDataDir, "_" + i);
- } catch (Exception e) {
- LOGGER.error("Failed to download segment from download url: {}", downloadURLs[i], e);
- _minionMetrics.addMeteredTableValue(tableNameWithType, MinionMeter.SEGMENT_DOWNLOAD_FAIL_COUNT, 1L);
- _eventObserver.notifyTaskError(_pinotTaskConfig, e);
- throw e;
- }
- inputSegmentDirs.add(indexDir);
-
- reportSegmentDownloadMetrics(indexDir, tableNameWithType, taskType);
- SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
- numRecords += segmentMetadata.getTotalDocs();
- }
+ List<File> inputSegmentDirs = Collections.synchronizedList(new ArrayList<>());
Review Comment:
I think there are ordering requirements on inputSegmentDirs. These should be ordered based on order of correposponding downloadURLs / segmentNames.
the controller orders the input segmentNames based on sequence id.
And then upsert executor creates recordReaderFileConfigs in the same order and uses that in SegmentProcessorFramework.
The order can be important for various edge cases. I am not fully sure if its ok to break the order but it would be safe to not break the order.
So if we can ensure the inputSegmentDirs are in the same order as downloadURLs / segmentNames that will help avoid regression.
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +336,59 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int parallelDownloadAndUntarSegments(String[] downloadURLs, String[] segmentNames, String tableNameWithType,
+ String taskType, File tempDataDir, List<File> inputSegmentDirs)
+ throws Exception {
+ AtomicInteger numRecords = new AtomicInteger(0);
+ List<Future<Void>> futures = new ArrayList<>();
+
+ int nThreads = Math.min(downloadURLs.length, Runtime.getRuntime().availableProcessors());
Review Comment:
I think instead of Runtime.getRuntime().availableProcessors() we should use _minionConf here.
because there will be many parallel tasks running together. dependent on numConcurrentTasksPerInstance. so if each task uses N threads, it can lead to starvation. Also each task type will have its own numConcurrentTasksPerInstance defined .
So I think safer to use a minion conf.
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +336,59 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int parallelDownloadAndUntarSegments(String[] downloadURLs, String[] segmentNames, String tableNameWithType,
+ String taskType, File tempDataDir, List<File> inputSegmentDirs)
+ throws Exception {
+ AtomicInteger numRecords = new AtomicInteger(0);
+ List<Future<Void>> futures = new ArrayList<>();
+
+ int nThreads = Math.min(downloadURLs.length, Runtime.getRuntime().availableProcessors());
Review Comment:
actually for OSS, we should avoid changing current behaviour by default. So by default we shouldn't create a threadpool but serially process it (same as today).
If config is present, we use the threadpool with configured number of threads.
Also I think better to use a common thread pool which is part of the BaseMultipleSegmentsConversionExecutor class. So that every call to executeTask() doesn't re-create the thread pool. We create this thread pool only if config is set.
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +336,59 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int parallelDownloadAndUntarSegments(String[] downloadURLs, String[] segmentNames, String tableNameWithType,
+ String taskType, File tempDataDir, List<File> inputSegmentDirs)
+ throws Exception {
+ AtomicInteger numRecords = new AtomicInteger(0);
+ List<Future<Void>> futures = new ArrayList<>();
+
+ int nThreads = Math.min(downloadURLs.length, Runtime.getRuntime().availableProcessors());
+ ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
+ try {
+ for (int i = 0; i < downloadURLs.length; i++) {
+ final int index = i;
+ futures.add(executorService.submit(() -> {
+ String segmentName = segmentNames[index];
+ String downloadUrl = downloadURLs[index];
+
+ _eventObserver.notifyProgress(_pinotTaskConfig,
+ String.format("Downloading and decompressing segment from: %s (%d out of %d)", downloadUrl, index + 1,
+ downloadURLs.length));
+
+ try {
+ File indexDir = downloadSegmentToLocalAndUntar(
+ tableNameWithType, segmentName, downloadUrl, taskType, tempDataDir, "_" + index);
+ inputSegmentDirs.add(indexDir);
+ reportSegmentDownloadMetrics(indexDir, tableNameWithType, taskType);
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
+ numRecords.addAndGet(segmentMetadata.getTotalDocs());
+ } catch (Exception e) {
Review Comment:
the catch is required only for downloadSegmentToLocalAndUntar right ?
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +336,59 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int parallelDownloadAndUntarSegments(String[] downloadURLs, String[] segmentNames, String tableNameWithType,
Review Comment:
I think better for this method to return List<SegmentMetadataImpl> as thats more intuitive for a method called "parallelDownloadAndUntarSegments". and the caller can iterate and populate numRecords.
this will also help in creating the same order of inputSegmentDirs as "segmentNames" as we can use SegmentMetadataImpl.getName() and sort based on index of segmentNames for that segment name
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +336,59 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int parallelDownloadAndUntarSegments(String[] downloadURLs, String[] segmentNames, String tableNameWithType,
+ String taskType, File tempDataDir, List<File> inputSegmentDirs)
+ throws Exception {
+ AtomicInteger numRecords = new AtomicInteger(0);
+ List<Future<Void>> futures = new ArrayList<>();
+
+ int nThreads = Math.min(downloadURLs.length, Runtime.getRuntime().availableProcessors());
+ ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
Review Comment:
after adding the configs, can we add / modify a test case that parallelly downloads multiple segments ?
because its possible many sub methods like downloadSegmentToLocalAndUntar / reportSegmentDownloadMetrics aren't thread safe and a test case can help catch such issues
Even if they're thread safe now, a test case that downloads multiple segemnts parallely can help catch other issues that might exist now or in future
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +336,59 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int parallelDownloadAndUntarSegments(String[] downloadURLs, String[] segmentNames, String tableNameWithType,
+ String taskType, File tempDataDir, List<File> inputSegmentDirs)
+ throws Exception {
+ AtomicInteger numRecords = new AtomicInteger(0);
+ List<Future<Void>> futures = new ArrayList<>();
+
+ int nThreads = Math.min(downloadURLs.length, Runtime.getRuntime().availableProcessors());
+ ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
+ try {
Review Comment:
nit - we can remove this try block ?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2184536706
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +348,67 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private void parallelDownloadAndUntarSegments(int nThreads, String tableNameWithType, String taskType,
+ String[] segmentNames, String[] downloadURLs, File tempDataDir, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+
+ ExecutorService executorService = null;
+ int length = downloadURLs.length;
+ try {
+ executorService = Executors.newFixedThreadPool(nThreads);
+ List<Future<Void>> futures = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ int index = i;
+ futures.add(executorService.submit(() -> {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, recordCounter, inputSegmentDirs);
+ return null;
+ }));
+
+ // Wait for all downloads to complete and cancel other tasks if any download fails
Review Comment:
Could you add a log.info- "Waiting n number of parallel downloads to complete"
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +348,67 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private void parallelDownloadAndUntarSegments(int nThreads, String tableNameWithType, String taskType,
+ String[] segmentNames, String[] downloadURLs, File tempDataDir, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+
+ ExecutorService executorService = null;
+ int length = downloadURLs.length;
+ try {
+ executorService = Executors.newFixedThreadPool(nThreads);
+ List<Future<Void>> futures = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ int index = i;
+ futures.add(executorService.submit(() -> {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, recordCounter, inputSegmentDirs);
+ return null;
+ }));
+
+ // Wait for all downloads to complete and cancel other tasks if any download fails
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (Exception e) {
+ // Cancel all other download tasks
+ for (Future<Void> f : futures) {
Review Comment:
log.error that we are cancelling downloads due to failure
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on PR #16249:
URL: https://github.com/apache/pinot/pull/16249#issuecomment-3034930135
can we also add details to the PR of results of testing and how much total time to run a particular task has decreased through this change ?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2192549595
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +349,75 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int getThreadPoolSize(Map<String, String> taskConfigs) {
+ int nThreads = _minionConf.getProperty(Minion.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
+ MinionConstants.DEFAULT_SEGMENT_DOWNLOAD_THREAD_POOL_SIZE);
+ nThreads = Integer.parseInt(taskConfigs.getOrDefault(MinionConstants.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
+ String.valueOf(nThreads)));
+ return nThreads;
+ }
+
+ private void parallelDownloadAndUntarSegments(int nThreads, String tableNameWithType, String taskType,
+ String[] segmentNames, String[] downloadURLs, File tempDataDir, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+
+ ExecutorService executorService = null;
+ int length = downloadURLs.length;
+ try {
+ executorService = Executors.newFixedThreadPool(nThreads);
+ List<Future<Void>> futures = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ int index = i;
+ futures.add(executorService.submit(() -> {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, recordCounter, inputSegmentDirs);
+ return null;
+ }));
+
+ // Wait for all downloads to complete and cancel other tasks if any download fails
+ for (Future<Void> future : futures) {
Review Comment:
Thank you. This was a miss.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2191785562
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -1424,6 +1424,12 @@ public static class Minion {
public static final char TASK_LIST_SEPARATOR = ',';
public static final String CONFIG_OF_ALLOW_DOWNLOAD_FROM_SERVER = "pinot.minion.task.allow.download.from.server";
public static final String DEFAULT_ALLOW_DOWNLOAD_FROM_SERVER = "false";
+
+ /**
+ * The number of threads to use for downloading segments from the deepstore.
+ * This is a global setting that applies to all tasks of BaseMultipleSegmentsConversionExecutor class.
+ */
+ public static final String SEGMENT_DOWNLOAD_THREAD_POOL_SIZE = "pinot.minion.task.segmentDownloadThreadPoolSize";
Review Comment:
this can be moved to MinionConf itself where other such minion confs are defined ?
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -352,6 +349,75 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}
}
+ private int getThreadPoolSize(Map<String, String> taskConfigs) {
+ int nThreads = _minionConf.getProperty(Minion.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
+ MinionConstants.DEFAULT_SEGMENT_DOWNLOAD_THREAD_POOL_SIZE);
+ nThreads = Integer.parseInt(taskConfigs.getOrDefault(MinionConstants.SEGMENT_DOWNLOAD_THREAD_POOL_SIZE,
+ String.valueOf(nThreads)));
+ return nThreads;
+ }
+
+ private void parallelDownloadAndUntarSegments(int nThreads, String tableNameWithType, String taskType,
+ String[] segmentNames, String[] downloadURLs, File tempDataDir, AtomicInteger recordCounter,
+ List<File> inputSegmentDirs)
+ throws Exception {
+
+ ExecutorService executorService = null;
+ int length = downloadURLs.length;
+ try {
+ executorService = Executors.newFixedThreadPool(nThreads);
+ List<Future<Void>> futures = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ int index = i;
+ futures.add(executorService.submit(() -> {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, recordCounter, inputSegmentDirs);
+ return null;
+ }));
+
+ // Wait for all downloads to complete and cancel other tasks if any download fails
+ for (Future<Void> future : futures) {
Review Comment:
this should be outside the above for loop right ?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on PR #16249:
URL: https://github.com/apache/pinot/pull/16249#issuecomment-3049701855
> LGTM other than few small changes. I think we should also ensure a test case goes through parallel download path (with more than 1 segment) and share performance results of parallel download
Modified existing test cases to go down the parallel download path.
```
22:34:24.943 INFO [BaseMultipleSegmentsConversionExecutor] [TaskStateModelFactory-task_thread-0] Start executing RealtimeToOfflineSegmentsTask on table: myTable2_REALTIME, input segments: myTable2__0__5__20250708T1704Z,myTable2__1__5__20250708T1704Z with downloadURLs: http://localhost:20000/segments/myTable2/myTable2__0__5__20250708T1704Z,http://localhost:20000/segments/myTable2/myTable2__1__5__20250708T1704Z, uploadURL: http://localhost:20000/segments, thread pool size:2
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish merged PR #16249:
URL: https://github.com/apache/pinot/pull/16249
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Implement parallel segment download and untar for improved performance [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #16249:
URL: https://github.com/apache/pinot/pull/16249#discussion_r2204607770
##########
pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java:
##########
@@ -190,35 +195,26 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
String downloadURLString = taskConfigs.get(MinionConstants.DOWNLOAD_URL_KEY);
String[] downloadURLs = downloadURLString.split(MinionConstants.URL_SEPARATOR);
AuthProvider authProvider = AuthProviderUtils.makeAuthProvider(taskConfigs.get(MinionConstants.AUTH_TOKEN));
- LOGGER.info("Start executing {} on table: {}, input segments: {} with downloadURLs: {}, uploadURL: {}", taskType,
- tableNameWithType, inputSegmentNames, downloadURLString, uploadURL);
File tempDataDir = new File(new File(MINION_CONTEXT.getDataDir(), taskType), "tmp-" + UUID.randomUUID());
Preconditions.checkState(tempDataDir.mkdirs());
+ int numRecords;
+ List<File> inputSegmentDirs = new ArrayList<>(downloadURLs.length);
+ Map<String, SegmentMetadata> segmentMetadataMap = Collections.synchronizedMap(new HashMap<>(downloadURLs.length));
+ int nThreads = Math.min(getParallelism(taskConfigs), downloadURLs.length);
+ LOGGER.info(
+ "Start executing {} on table: {}, input segments: {} with downloadURLs: {}, uploadURL: {}, thread pool size:{}",
+ taskType, tableNameWithType, inputSegmentNames, downloadURLString, uploadURL, nThreads);
try {
- List<File> inputSegmentDirs = new ArrayList<>();
- int numRecords = 0;
-
- for (int i = 0; i < downloadURLs.length; i++) {
- String segmentName = segmentNames[i];
- // Download and decompress the segment file
- _eventObserver.notifyProgress(_pinotTaskConfig, "Downloading and decompressing segment from: " + downloadURLs[i]
- + " (" + (i + 1) + " out of " + downloadURLs.length + ")");
- File indexDir;
- try {
- indexDir = downloadSegmentToLocalAndUntar(tableNameWithType, segmentName, downloadURLs[i], taskType,
- tempDataDir, "_" + i);
- } catch (Exception e) {
- LOGGER.error("Failed to download segment from download url: {}", downloadURLs[i], e);
- _minionMetrics.addMeteredTableValue(tableNameWithType, MinionMeter.SEGMENT_DOWNLOAD_FAIL_COUNT, 1L);
- _eventObserver.notifyTaskError(_pinotTaskConfig, e);
- throw e;
+ if (nThreads <= 1) {
+ for (int index = 0; index < downloadURLs.length; index++) {
+ downloadAndUntarSegment(tableNameWithType, taskType, segmentNames[index], downloadURLs[index],
+ tempDataDir, index, segmentMetadataMap, segmentNames.length);
}
- inputSegmentDirs.add(indexDir);
-
- reportSegmentDownloadMetrics(indexDir, tableNameWithType, taskType);
- SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
- numRecords += segmentMetadata.getTotalDocs();
+ } else {
+ parallelDownloadAndUntarSegments(nThreads, tableNameWithType, taskType, segmentNames, downloadURLs,
+ tempDataDir, segmentMetadataMap);
}
+ numRecords = processSegmentMetadata(segmentNames, segmentMetadataMap, inputSegmentDirs);
Review Comment:
I think, it is alright.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org