Allow segment upload via Metadata in MergeRollup Minion task #9825

KKcorps · 2022-11-17T16:48:03Z

Users need to use the same configs as SegmentGenerationAndPushTask to switch to metadata push.
e.g.
pushType: METADATA

zhtaoxiang · 2022-11-17T17:46:45Z

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

@@ -242,6 +255,10 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
            new BasicHeader(FileUploadDownloadClient.CustomHeaders.SEGMENT_ZK_METADATA_CUSTOM_MAP_MODIFIER,
                segmentZKMetadataCustomMapModifier.toJsonString());

+        URI outputSegmentTarURI = moveSegmentToOutputPinotFS(pinotTaskConfig.getConfigs(), convertedTarredSegmentFile);


I feel that we should check push mode first before moving the segment. Otherwise, this middle step is not needed.

makes sense

+1
As @zhtaoxiang mentioned, we should NOT move the segments to deep storage when we do the segment tar file push. In that case, the controller will move the segment to the deep storage during the upload phase.

...minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionPushUtils.java

zhtaoxiang · 2022-11-17T18:09:49Z

...src/main/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java

+          singleFileGenerationTaskConfig.put(
+              BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI, outputSegmentDirURI.toString());
+        }
+        if ((outputDirURI == null) || (pushMode == null)) {


When push mode is MetadataPush, we should set the default location data_dir/tablename when OUTPUT_SEGMENT_DIR_URI is not set by users. Most of the time, users will not need to set it.

zhtaoxiang · 2022-11-17T18:22:06Z

...src/main/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java

+        String downloadUrl = downloadURLList[0];
+        URI downloadURI = URI.create(downloadUrl);
+        URI outputDirURI = null;
+        if (!downloadURI.getScheme().contentEquals("http")) {


Maybe I misunderstand something, but I don't fully follow the logic here. Why is the outputDir decided by the downloadUrl?

Because the segment needs to be copied to some place that is accessible by the controller in case of metadata push.
If user doesn't set the outputDirUri then the best place would be to copy it in deep store. The download url refers to deep store path and hence using it to derive the metadata.

If that's the goal, then we just need to use controllerConfig.getDataDir()/tableName. What do you think?

That makes more sense.

Added the commit to use dataDir.

...pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/TaskUtils.java

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

snleee · 2022-11-18T05:30:26Z

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

@@ -242,6 +255,10 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
            new BasicHeader(FileUploadDownloadClient.CustomHeaders.SEGMENT_ZK_METADATA_CUSTOM_MAP_MODIFIER,
                segmentZKMetadataCustomMapModifier.toJsonString());

+        URI outputSegmentTarURI = moveSegmentToOutputPinotFS(pinotTaskConfig.getConfigs(), convertedTarredSegmentFile);


+1
As @zhtaoxiang mentioned, we should NOT move the segments to deep storage when we do the segment tar file push. In that case, the controller will move the segment to the deep storage during the upload phase.

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

snleee · 2022-11-18T05:39:40Z

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

@@ -242,6 +256,10 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
            new BasicHeader(FileUploadDownloadClient.CustomHeaders.SEGMENT_ZK_METADATA_CUSTOM_MAP_MODIFIER,
                segmentZKMetadataCustomMapModifier.toJsonString());

+        URI outputSegmentTarURI = moveSegmentToOutputPinotFS(pinotTaskConfig.getConfigs(), convertedTarredSegmentFile);
+        LOGGER.info("Moved generated segment from [{}] to location: [{}]",


What happens if the Minion somehow got killed (e.g. restart) after finish the segment move to outputPinotFS but before finishing the metadata push to controller?

I think that this can potentially leave some files in the deep storage without any Pinot side segment metadata pointing those files. This will potentially become the problem unless we have the cleanup mechanism. Do we have any design/solution to address this?

One solution can be to always mention a seperate OUTPUT_DIR and always clean that dir in case the task fails. Even if we fail to clean the outputDir it should not be a problem if it is seperate from the deep store.

Personally, I think this may not be a problem. Because Minion task should be idempotent, a later retry should generate segments with the same name and the existing ones will be reused or be override.

We should not assume that the later retry will always come to finish this particular task. Similarly, the controller can be down and fail the metadata update. If the controller recovery happens after the minion's retry attempt fails, I think that this will cause some files to sit at the deep storage.

codecov-commenter · 2022-11-21T11:35:17Z

Codecov Report

Merging #9825 (72efe9e) into master (e307106) will increase coverage by 42.36%.
The diff coverage is 42.63%.

@@              Coverage Diff              @@
##             master    #9825       +/-   ##
=============================================
+ Coverage     28.06%   70.43%   +42.36%     
- Complexity       53     4984     +4931     
=============================================
  Files          1949     1981       +32     
  Lines        104632   106370     +1738     
  Branches      15847    16117      +270     
=============================================
+ Hits          29362    74918    +45556     
+ Misses        72395    26235    -46160     
- Partials       2875     5217     +2342

Flag	Coverage Δ
integration1	`24.99% <42.63%> (-0.23%)`	⬇️
integration2	`24.45% <28.68%> (+0.02%)`	⬆️
unittests1	`67.96% <0.00%> (?)`
unittests2	`15.78% <9.30%> (?)`

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...he/pinot/segment/local/utils/SegmentPushUtils.java	`11.97% <0.00%> (+11.97%)`	⬆️
...che/pinot/plugin/minion/tasks/MinionTaskUtils.java	`40.90% <40.90%> (ø)`
.../tasks/BaseMultipleSegmentsConversionExecutor.java	`80.00% <51.85%> (-9.86%)`	⬇️
...nandpush/SegmentGenerationAndPushTaskExecutor.java	`67.88% <75.00%> (ø)`
...on/tasks/mergerollup/MergeRollupTaskGenerator.java	`84.89% <100.00%> (+7.92%)`	⬆️
...gments/RealtimeToOfflineSegmentsTaskGenerator.java	`94.85% <100.00%> (+24.26%)`	⬆️
...andpush/SegmentGenerationAndPushTaskGenerator.java	`33.87% <100.00%> (ø)`
...ery/aggregation/DoubleAggregationResultHolder.java	`60.00% <0.00%> (-15.00%)`	⬇️
...apache/pinot/common/function/FunctionRegistry.java	`81.63% <0.00%> (-5.21%)`	⬇️
...nction/DistinctCountBitmapAggregationFunction.java	`47.66% <0.00%> (-5.19%)`	⬇️
... and 1391 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

zhtaoxiang · 2022-11-21T17:39:54Z

...minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java

+  public static Map<String, String> getPushTaskConfig(String tableName, Map<String, String> taskConfigs,
+      ClusterInfoAccessor clusterInfoAccessor) {
+    try {
+      URI outputDirURI = URI.create(clusterInfoAccessor.getDataDir() + "/" + tableName);


I think we should check push mode first before deciding the outputDir?

If push mode is tar, we simply need to build segment locally and tar push to Pinot.

Let me know if I misunderstood something

It is not really used in the case when pushMode is TAR.

if (BatchConfigProperties.SegmentPushType.valueOf(pushMode.toUpperCase()) != BatchConfigProperties.SegmentPushType.TAR) { outputSegmentTarURI = moveSegmentToOutputPinotFS(configs, convertedTarredSegmentFile); LOGGER.info("Moved generated segment from [{}] to location: [{}]", convertedTarredSegmentFile, outputSegmentTarURI); } else { outputSegmentTarURI = convertedTarredSegmentFile.toURI(); } ...... case TAR: try (PinotFS pinotFS = MinionTaskUtils.getLocalPinotFs()) { SegmentPushUtils.pushSegments( spec, pinotFS, Arrays.asList(outputSegmentTarURI.toString()), headers, parameters); } catch (RetriableOperationException | AttemptsExceededException e) { throw new RuntimeException(e); } break;

I agree that it is not really used in the case when push mode is TAR. However, I feel the current logic is a little bit hard to follow if we don't have the full context.

If we can check push mode first then decide what to do, it's easier to follow. WDYT?

Cool. Made the changes. A lot of the code used here has been borrowed from SegmentGenerationAndPushTask. I will also raise another PR to port some of these changes from here to there.

zhtaoxiang · 2022-11-30T07:50:15Z

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentPushUtils.java

@@ -97,6 +97,49 @@ public static URI generateSegmentTarURI(URI dirURI, URI fileURI, String prefix,
  public static void pushSegments(SegmentGenerationJobSpec spec, PinotFS fileSystem, List<String> tarFilePaths)
      throws RetriableOperationException, AttemptsExceededException {
    String tableName = spec.getTableSpec().getTableName();
+    AuthProvider authProvider = AuthProviderUtils.makeAuthProvider(spec.getAuthToken());


just curious: why do we need to create those 3 new methods? It seems that there is no logic change?

We need to pass this custom header in the new flow
Older methods don't allow passing new headers

// Set segment ZK metadata custom map modifier into HTTP header to modify the segment ZK metadata SegmentZKMetadataCustomMapModifier segmentZKMetadataCustomMapModifier = getSegmentZKMetadataCustomMapModifier(pinotTaskConfig, segmentConversionResult); Header segmentZKMetadataCustomMapModifierHeader = new BasicHeader(FileUploadDownloadClient.CustomHeaders.SEGMENT_ZK_METADATA_CUSTOM_MAP_MODIFIER, segmentZKMetadataCustomMapModifier.toJsonString());

...ot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGenerator.java

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

...rc/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

zhtaoxiang · 2022-11-30T19:46:08Z

...minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java

+  public static Map<String, String> getPushTaskConfig(String tableName, Map<String, String> taskConfigs,
+      ClusterInfoAccessor clusterInfoAccessor) {
+    try {
+      URI outputDirURI = URI.create(clusterInfoAccessor.getDataDir() + "/" + tableName);


I agree that it is not really used in the case when push mode is TAR. However, I feel the current logic is a little bit hard to follow if we don't have the full context.

If we can check push mode first then decide what to do, it's easier to follow. WDYT?

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

zhtaoxiang · 2022-12-02T17:54:47Z

...minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/MinionTaskUtils.java

+
+        if (!isLocalOutputDir(outputDirURIScheme)) {
+          singleFileGenerationTaskConfig.put(BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI, outputDirURI.toString());
+          singleFileGenerationTaskConfig.put(BatchConfigProperties.PUSH_MODE, pushMode);


nit: should we set push_mode to metadata when the provided one is URI? and we should log this change.

zhtaoxiang

LGTM

snleee

Please address the following issue.

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

dismissing approval

…atibility

snleee

Let's fix the test. LGTM other than minor comments. Thank you for working on this!

snleee · 2022-12-05T21:07:51Z

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

+          String uploadURL = taskConfigs.get(MinionConstants.UPLOAD_URL_KEY);
+          SegmentConversionUtils.uploadSegment(taskConfigs, headers, parameters, tableNameWithType, segmentName,
+              uploadURL, tarFile);
+        } catch (RetriableOperationException | AttemptsExceededException e) {


Why do we need to catch specific exceptions and wrap it with RuntimeException here? Our top caller function's definition: executeTask() throw Exception

snleee · 2022-12-05T21:10:10Z

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java

+                SegmentPushUtils.getSegmentUriToTarPathMap(outputSegmentDirURI, pushJobSpec,
+                    new String[]{outputSegmentTarURI.toString()});
+            SegmentPushUtils.sendSegmentUriAndMetadata(spec, outputFileFS, segmentUriToTarPathMap, headers, parameters);
+          } catch (RetriableOperationException | AttemptsExceededException e) {


Kartik Khare added 5 commits November 17, 2022 11:29

Add code to push metadata in minion tasks

a0a425a

Fix metadata push

8944ea4

Add config for overwrite dir

7280b64

Fix download url to generate outputDir

ce6c8fe

Fix linting

0cd020b

zhtaoxiang reviewed Nov 17, 2022

View reviewed changes

Remove MinionPushUtils class

2bed0e6

KKcorps changed the title ~~Allow segment upload via Metadata in MergeRollup Minion tas~~ Allow segment upload via Metadata in MergeRollup Minion task Nov 18, 2022

snleee reviewed Nov 18, 2022

View reviewed changes

Kartik Khare added 6 commits November 18, 2022 11:36

Do not copy in case of tar push

cced16b

Remove duplicate code for filesystem

c691c0c

Add metadata push to realtime to offline task

b868117

Use data dir as output dir

169332f

Add tests for realtime to offline metadata push

198c296

use controller data dir for output

4486ae8

zhtaoxiang reviewed Nov 21, 2022

View reviewed changes

zhtaoxiang reviewed Nov 30, 2022

View reviewed changes

Remove redundant code for URI push

1ffb275

snleee reviewed Nov 30, 2022

View reviewed changes

zhtaoxiang reviewed Nov 30, 2022

View reviewed changes

zhtaoxiang reviewed Dec 2, 2022

View reviewed changes

Enforce outputDir in metadata push

72efe9e

KKcorps force-pushed the minion_metadata_push branch from 370bc3d to 72efe9e Compare December 2, 2022 18:51

zhtaoxiang approved these changes Dec 3, 2022

View reviewed changes

snleee previously approved these changes Dec 5, 2022

View reviewed changes

...c/main/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java Outdated Show resolved Hide resolved

snleee self-requested a review December 5, 2022 18:09

Use SegmentConversionUtils method to push TAR files for backward comp…

8b4dd1a

…atibility

snleee approved these changes Dec 5, 2022

View reviewed changes

Do not catch exception since it is already handled

ee76795

KKcorps merged commit fb48288 into apache:master Dec 6, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Allow segment upload via Metadata in MergeRollup Minion task #9825

Allow segment upload via Metadata in MergeRollup Minion task #9825

KKcorps commented Nov 17, 2022 •

edited

Loading

zhtaoxiang Nov 17, 2022

KKcorps Nov 17, 2022

snleee Nov 18, 2022

zhtaoxiang Nov 17, 2022

zhtaoxiang Nov 17, 2022

KKcorps Nov 17, 2022

zhtaoxiang Nov 18, 2022

KKcorps Nov 21, 2022

KKcorps Nov 21, 2022

snleee Nov 18, 2022

snleee Nov 18, 2022 •

edited

Loading

KKcorps Nov 18, 2022

zhtaoxiang Nov 18, 2022

snleee Nov 18, 2022

codecov-commenter commented Nov 21, 2022 •

edited

Loading

zhtaoxiang Nov 21, 2022

KKcorps Nov 30, 2022

zhtaoxiang Nov 30, 2022

KKcorps Dec 2, 2022

zhtaoxiang Nov 30, 2022

KKcorps Nov 30, 2022 •

edited

Loading

zhtaoxiang Nov 30, 2022

zhtaoxiang Dec 2, 2022

zhtaoxiang left a comment

snleee left a comment •

edited

Loading

snleee left a comment

snleee Dec 5, 2022

snleee Dec 5, 2022

Allow segment upload via Metadata in MergeRollup Minion task #9825

Allow segment upload via Metadata in MergeRollup Minion task #9825

Conversation

KKcorps commented Nov 17, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

snleee Nov 18, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Nov 21, 2022 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KKcorps Nov 30, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhtaoxiang left a comment

Choose a reason for hiding this comment

snleee left a comment • edited Loading

Choose a reason for hiding this comment

snleee left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KKcorps commented Nov 17, 2022 •

edited

Loading

snleee Nov 18, 2022 •

edited

Loading

codecov-commenter commented Nov 21, 2022 •

edited

Loading

KKcorps Nov 30, 2022 •

edited

Loading

snleee left a comment •

edited

Loading