Enable Consistent Data Push for Standalone Segment Push Job Runners #9295

yuanbenson · 2022-08-29T21:20:54Z

Description:

This PR addresses #9268 for the segment push job runners under the standalone execution framework: SegmentMetadataPushJobRunner, SegmentTarPushJobRunner, and SegmentUriPushJobRunner.

This is accomplished by introducing a new class ConsistentDataPushUtils which contains APIs and helpers for *PushJobRunner(s) to call to invoke the consistent push protocol.

Since there are large overlaps in the code for all of the *PushJobRunner(s), also took this opportunity to refactor and extract the common logic out to BaseSegmentPushJobRunner abstract class.

To enable consistent data push, this PR also introduces a new boolean config in table config under
TableConfig->IngestionConfig->BatchIngestionConfig->consistentDataPush.

Users can enable consistent data push by setting the consistentDataPush config to true as below before invoking ingestion jobs,

...
    "batchIngestionConfig": {
      "segmentIngestionType": "REFRESH",
      "segmentIngestionFrequency": "DAILY",
      "consistentDataPush": true
    },
...

which will

In the segment generation phase: inject timestamps to segment names (via segment postfix) in order to prevent segment name conflicts from overwriting existing segments directly for the REFRESH usecase and
In the segment push/upload phase: wrap the segments replacement protocol around segments upload which will achieve atomic switching between old and new segments data. In the case of failure, revert and abort the swap, which will ensure broker only routes to the old segments.

Testing Done:

Added new test testUploadAndQueryWithConsistentPush in SegmentUploadIntegrationTest, which

Runs SegmentMetadataPushJobRunner with consistent push enabled.
[] -> [v1 segments]
Checks that the segment lineage entry is in expected and completed state.
Checks that count stars return expected outputs.
Runs SegmentTarPushJobRunner with consistent push enabled.
[v1 segments] -> [v2 segments]
Checks again that the segment lineage entry is in expected and completed state.
Checks again that count stars return expected outputs (that we have successfully bulk replaced the original set of segments).

snleee

Partial Review

snleee · 2022-08-30T03:57:34Z

...ion-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java

@@ -81,10 +89,15 @@ protected List<String> getBloomFilterColumns() {
    return null;
  }

+  @BeforeMethod
+  public void setUpTest()


I think that this is not actually necessary.

We can call TestUtils.ensureDirectoriesExistAndEmpty() function in setUp() and clean up those in cleanUp()

For 2 different tests, we can use different table names.

Calling TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir) in setUp is insufficient as there is another test case testUploadAndQuery that leaves data in the tmp directories.

Using two table names will also lead to more code for now, since it seems like BaseClusterIntegrationTest is written in a way that assumes one table/config per subclass, e.g. getTableName, createOfflineTableConfig.

Hence, I prefer to keep the test structure as is with the before and after method annotations, unless this is an anti-pattern. Another way is to create a different test class altogether, but combining the two here to not prevent slow down from too many tests.

snleee · 2022-08-30T03:58:06Z

...ion-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java

@@ -217,10 +358,15 @@ public Boolean apply(@Nullable Void aVoid) {
    }, 100L, 300_000, "Failed to load " + countStarResult + " documents", true);
  }

+  @AfterMethod


This is also not necessary if we use different table names.

...ion-tests/src/test/java/org/apache/pinot/integration/tests/SegmentUploadIntegrationTest.java

...n/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/BaseSegmentPushJobRunner.java

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

...n/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/BaseSegmentPushJobRunner.java

codecov-commenter · 2022-08-30T23:27:47Z

Codecov Report

Merging #9295 (deae0ca) into master (0f4bcfc) will decrease coverage by 0.03%.
The diff coverage is 26.96%.

❗ Current head deae0ca differs from pull request most recent head f3aa9fb. Consider uploading reports for the commit f3aa9fb to get more accurate results

@@             Coverage Diff              @@
##             master    #9295      +/-   ##
============================================
- Coverage     69.80%   69.76%   -0.04%     
+ Complexity     4777     4703      -74     
============================================
  Files          1875     1878       +3     
  Lines         99860    99930      +70     
  Branches      15194    15192       -2     
============================================
+ Hits          69706    69718      +12     
- Misses        25231    25285      +54     
- Partials       4923     4927       +4

Flag	Coverage Δ
integration1	`26.14% <13.31%> (-0.07%)`	⬇️
integration2	`24.84% <1.70%> (-0.15%)`	⬇️
unittests1	`66.99% <18.10%> (-0.10%)`	⬇️
unittests2	`15.28% <1.70%> (+0.02%)`	⬆️

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

Impacted Files	Coverage Δ
...gestion/batch/common/BaseSegmentPushJobRunner.java	`0.00% <0.00%> (ø)`
...batch/standalone/SegmentMetadataPushJobRunner.java	`0.00% <0.00%> (ø)`
...tion/batch/standalone/SegmentTarPushJobRunner.java	`0.00% <0.00%> (ø)`
...tion/batch/standalone/SegmentUriPushJobRunner.java	`0.00% <0.00%> (ø)`
...t/segment/local/utils/ConsistentDataPushUtils.java	`0.00% <0.00%> (ø)`
...ingestion/batch/spec/SegmentNameGeneratorSpec.java	`0.00% <0.00%> (ø)`
...g/apache/pinot/spi/utils/IngestionConfigUtils.java	`69.13% <0.00%> (-5.54%)`	⬇️
...spi/utils/builder/ControllerRequestURLBuilder.java	`0.00% <0.00%> (ø)`
...i/config/table/ingestion/BatchIngestionConfig.java	`60.00% <71.42%> (+2.85%)`	⬆️
...e/pinot/common/utils/FileUploadDownloadClient.java	`60.27% <74.41%> (+3.45%)`	⬆️
... and 37 more

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

pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java

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

pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java

jackjlli · 2022-08-31T00:01:02Z

pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java

+      String responseString = response.getResponse();
+      JsonNode responseJsonNode = JsonUtils.stringToJsonNode(responseString);
+      Iterator<JsonNode> responseElements = responseJsonNode.elements();
+      while (responseElements.hasNext()) {


Can we encapsulate the logic of fetching the segment names into a method, or is there any existing logic doing that right now?

Could you clarify a bit more on the input parameters/ask for this method we are talking about? I'll be glad to extract that out if it does not already exist.

Ohh I see, I'll look into possibilities of reusing getSelectedSegments API that you previously brought in. That may be used when we support APPEND tables anyways! Thx.

snleee · 2022-08-31T07:28:48Z

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

+    AuthProvider authProvider = AuthProviderUtils.makeAuthProvider(spec.getAuthToken());
+    LOGGER.info("Start replace segment URIs: " + segmentsUris);
+
+    int attempts = 1;


Can we have a larger number than 1 for a retrial attempt? What's the reasoning behind of doing a retry only for once?

snleee · 2022-08-31T07:30:59Z

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

+    for (URI uri : uriToLineageEntryIdMap.keySet()) {
+      String segmentLineageEntryId = uriToLineageEntryIdMap.get(uri);
+      try {
+        FILE_UPLOAD_DOWNLOAD_CLIENT.endReplaceSegments(


can we add the retry here as well to keep the consistency with the startReplaceSegments call?

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

yuanbenson · 2022-08-31T22:51:53Z

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

+
+  private static final Logger LOGGER = LoggerFactory.getLogger(SegmentPushUtils.class);
+  private static final FileUploadDownloadClient FILE_UPLOAD_DOWNLOAD_CLIENT = new FileUploadDownloadClient();
+  private static final RetryPolicy DEFAULT_RETRY_POLICY = RetryPolicies.exponentialBackoffRetryPolicy(5, 10_000L, 2.0);


@snleee @jtao15
Please lmk if these params are appropriate.

yuanbenson · 2022-08-31T22:52:04Z

pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java

+          tableTypeToFilter, excludeReplacedSegments);
+      RequestBuilder requestBuilder = RequestBuilder.get(uri).setVersion(HttpVersion.HTTP_1_1);
+      HttpClient.setTimeout(requestBuilder, HttpClient.DEFAULT_SOCKET_TIMEOUT_MS);
+      RetryPolicies.exponentialBackoffRetryPolicy(5, 10_000L, 2.0).attempt(() -> {


Same for this..

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

jtao15 · 2022-09-01T05:36:26Z

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

+
+  private static final Logger LOGGER = LoggerFactory.getLogger(SegmentPushUtils.class);
+  private static final FileUploadDownloadClient FILE_UPLOAD_DOWNLOAD_CLIENT = new FileUploadDownloadClient();
+  private static final RetryPolicy DEFAULT_RETRY_POLICY = RetryPolicies.exponentialBackoffRetryPolicy(5, 10_000L, 2.0);


pinot-common/src/main/java/org/apache/pinot/common/utils/FileUploadDownloadClient.java

…ed BaseSegmentPushJobRunner

…ement protocl and segments GET

pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/ControllerRequestURLBuilder.java

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

...n/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/BaseSegmentPushJobRunner.java

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

…tListAPI

...n/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/BaseSegmentPushJobRunner.java

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

… abs class

yuanbenson · 2022-09-07T01:29:58Z

...rc/main/java/org/apache/pinot/plugin/ingestion/batch/standalone/SegmentTarPushJobRunner.java

-    }
+  public void uploadSegments(Map<String, String> segmentsUriToTarPathMap)
+      throws AttemptsExceededException, RetriableOperationException {
+    SegmentPushUtils.pushSegments(_spec, _outputDirFS, new ArrayList<>(segmentsUriToTarPathMap.values()));


@snleee Do you think this might cause some performance loss (in the case of having many segments to push)? If so, might need to change pushSegments to take in Collection<String> as opposed to List<String>.

For copying the elements from map.values() to List?

In our case, the worst-case scenario should be 10s thousands of segments for a table and I think it's OK to keep it this way. I don't think that it's a performance-critical path.

yuanbenson · 2022-09-07T01:30:24Z

...rc/main/java/org/apache/pinot/plugin/ingestion/batch/standalone/SegmentUriPushJobRunner.java

-    }
+  public void uploadSegments(Map<String, String> segmentsUriToTarPathMap)
+      throws AttemptsExceededException, RetriableOperationException {
+    SegmentPushUtils.sendSegmentUris(_spec, new ArrayList<>(segmentsUriToTarPathMap.keySet()));


snleee

Looks good to me otherwise! Thank you for addressing all the comments.

snleee · 2022-09-07T03:10:44Z

...n/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/BaseSegmentPushJobRunner.java

+   *                                tarPaths (values), or both may be used depending on upload mode.
+   */
+  public List<String> getSegmentsToReplace(Map<String, String> segmentsUriToTarPathMap) {
+    return SegmentPushUtils.getSegmentNames(segmentsUriToTarPathMap.values());


Let's put the logic here instead of putting in Util class. We usually extract the logic to Util class when the same code needs to be called in multiple places to avoid the duplicate code. In this case, we only need to use this function here.

Jackie-Jiang · 2022-09-07T21:28:05Z

This is a great feature! Can you please help also update the pinot documentation about this new feature? https://github.com/pinot-contrib/pinot-docs

yuanbenson changed the title ~~Consistent Data Push: Standalone Job Runners~~ Consistent Data Push: Standalone Segment Push Job Runners Aug 29, 2022

yuanbenson changed the title ~~Consistent Data Push: Standalone Segment Push Job Runners~~ Enable Consistent Data Push for Standalone Segment Push Job Runners Aug 29, 2022

yuanbenson force-pushed the consistentPush branch from 9307bac to 356183a Compare August 29, 2022 21:56

yuanbenson closed this Aug 30, 2022

yuanbenson reopened this Aug 30, 2022

snleee reviewed Aug 30, 2022

View reviewed changes

yuanbenson requested a review from snleee August 30, 2022 23:49

jackjlli reviewed Aug 31, 2022

View reviewed changes

snleee reviewed Aug 31, 2022

View reviewed changes

yuanbenson commented Aug 31, 2022

View reviewed changes

yuanbenson requested a review from snleee August 31, 2022 22:52

jtao15 reviewed Sep 1, 2022

View reviewed changes

yuanbenson added 7 commits September 6, 2022 10:10

Consistent Data Push: Standalone Job Runners

c0fd3d7

Addressed comments: Refactored SegmentUploadIntegrationTest and unifi…

eb2f0a2

…ed BaseSegmentPushJobRunner

Addressed comments

c5bc653

Addressed comment: added more robust retriest to start and end replac…

4c59b8a

…ement protocl and segments GET

Changed fixedDelayRetryPolicy to exponentialBackoffRetryPolicy

ec72fc9

trigger test

41ee12d

Addressed jtao's comment

5267dec

yuanbenson force-pushed the consistentPush branch from 89474ac to 5267dec Compare September 6, 2022 17:11

snleee reviewed Sep 6, 2022

View reviewed changes

...egment-local/src/main/java/org/apache/pinot/segment/local/utils/ConsistentDataPushUtils.java Outdated Show resolved Hide resolved

Reduced calls to controller for tableConfig reads, refactor forSegmen…

4746763

…tListAPI

yuanbenson requested a review from snleee September 6, 2022 20:54

snleee reviewed Sep 7, 2022

View reviewed changes

...n/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/BaseSegmentPushJobRunner.java Outdated Show resolved Hide resolved

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentPushUtils.java Outdated Show resolved Hide resolved

Make getSegmentsToReplace method concrete in BaseSegmentPushJobRunner…

b09d747

… abs class

yuanbenson commented Sep 7, 2022

View reviewed changes

snleee approved these changes Sep 7, 2022

View reviewed changes

yuanbenson added 2 commits September 6, 2022 20:53

Moved SegmentPushUtils#getSegmentNames into base class

d6ca192

trigger test

f3aa9fb

snleee added release-notes Referenced by PRs that need attention when compiling the next release notes refactor labels Sep 7, 2022

snleee merged commit 334c978 into apache:master Sep 7, 2022

yuanbenson mentioned this pull request Sep 7, 2022

Enable Consistent Data Push on Ingestion Jobs for REFRESH use case (standalone only) #9268

Closed

snleee mentioned this pull request Apr 19, 2023

IngestionSpec is not backward compatible (0.11.0 -> 0.12.1) #10644

Open

lrao-stripe mentioned this pull request Apr 16, 2024

Enable consistentDataPush for Spark execution framework #12941

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Enable Consistent Data Push for Standalone Segment Push Job Runners #9295

Enable Consistent Data Push for Standalone Segment Push Job Runners #9295

yuanbenson commented Aug 29, 2022 •

edited

Loading

snleee left a comment

snleee Aug 30, 2022

yuanbenson Aug 30, 2022 •

edited

Loading

snleee Aug 30, 2022

codecov-commenter commented Aug 30, 2022 •

edited

Loading

jackjlli Aug 31, 2022

yuanbenson Aug 31, 2022

snleee Aug 31, 2022

snleee Aug 31, 2022 •

edited

Loading

yuanbenson Aug 31, 2022

jtao15 Sep 1, 2022

yuanbenson Aug 31, 2022

jtao15 Sep 1, 2022

yuanbenson Sep 7, 2022

snleee Sep 7, 2022 •

edited

Loading

yuanbenson Sep 7, 2022

snleee left a comment

snleee Sep 7, 2022

Jackie-Jiang commented Sep 7, 2022

Enable Consistent Data Push for Standalone Segment Push Job Runners #9295

Enable Consistent Data Push for Standalone Segment Push Job Runners #9295

Conversation

yuanbenson commented Aug 29, 2022 • edited Loading

Description:

Testing Done:

snleee left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yuanbenson Aug 30, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov-commenter commented Aug 30, 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

snleee Aug 31, 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

snleee Sep 7, 2022 • edited Loading

Choose a reason for hiding this comment

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

Jackie-Jiang commented Sep 7, 2022

yuanbenson commented Aug 29, 2022 •

edited

Loading

yuanbenson Aug 30, 2022 •

edited

Loading

codecov-commenter commented Aug 30, 2022 •

edited

Loading

snleee Aug 31, 2022 •

edited

Loading

snleee Sep 7, 2022 •

edited

Loading