You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "rajagopr (via GitHub)" <gi...@apache.org> on 2024/07/13 00:39:25 UTC
[PR] Added support to upload segments in batch mode with METADATA upload type [pinot]
rajagopr opened a new pull request, #13597:
URL: https://github.com/apache/pinot/pull/13597
### What
Added support to upload segments in batch mode when METADATA upload type is used. The new API endpoint `/segmentList` is introduced which supports this capability.
### Why
Segment assignment was observed to be a bottleneck when multiple segments were getting updated at the same time. This was due to the fact that each segment could be updated only after acquiring the table update lock. The issue significantly affected ingestion and segment refresh when multiple minions (~100) were working at the same time.
### How
With this change a bunch of segments could be updated in one-go after acquiring the table update lock.
### Testing
Added integration test which adds multiple segments in batch mode.
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "kishoreg (via GitHub)" <gi...@apache.org>.
kishoreg commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1679749949
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -555,6 +750,67 @@ public void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
}
}
+ @POST
+ @ManagedAsync
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Path("/segmentList")
Review Comment:
we should document the convention for naming api's. +1 on /segments. upload is a verb and it should not be part of the URI.
https://medium.com/@nadinCodeHat/rest-api-naming-conventions-and-best-practices-1c4e781eb6a5
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1681464297
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -555,6 +750,67 @@ public void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
}
}
+ @POST
+ @ManagedAsync
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Path("/segmentList")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Cluster.UPLOAD_SEGMENT)
+ @Authenticate(AccessType.CREATE)
+ @ApiOperation(value = "Upload a segment", notes = "Upload a segment as binary")
Review Comment:
@klsince and @swaminathanmanish: As @Jackie-Jiang explained, the batch upload would work along with the segment start/end replace. The guarantee `all or nothing` is provided by the segment start/end replace protocol? I will call this out in the docs.
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2228612936
> What was the previous api to upload a single segment?
@kishoreg : Previous apis are `/segments` and `/v2/segments` – both of them would upload a single segment.
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2232159425
> > I suppose the scenario you are trying to optimize is when each minion uploads more than one segment (presumably two, to split data over a time boundary?) and hundreds of minions working on multiple such segments? If not, can you please explain the use case a bit more?
> > Also, can you use the start/end replace segments primitive that is already built?
>
> @mcvsubbu : Sharing additional context below.
>
> The SegmentRefreshTask (custom minion task) helps align Pinot segments to the table configs. The task can do things like: 1) Break down a large segment into multiple small segments 2) Combine multiple small segments into one large segment 3) Sort the data based on the sort column – if sort column changed. 4) Re-partition the data etc. The task extends on the abstract class BaseMultipleSegmentsConversionExecutor which can operate on multiple segments. The task aligns the Pinot segments to the table configs over multiple invocations. After each round of invocation, the segments (or subset of segments) would be a step closer to the table configuration.
>
> When we encountered issues, each minion task was operating on upto twenty segments, there were about ~50K segments in the table and about 200 minions were operating at the same time. Once the SegmentConversionResults are generated the minion attempts to upload the new segments and there is a configured timeout of about ten minutes to upload the segment. Upon closer inspection, we found that the contention is happening while updating the IdealState and the threads are waiting to acquire the table update lock. Since, many of the uploads timeout, this would result in a flood of retries which worsens the problem.
>
> Due to the contention issue described above we are unable to increase the number of minions and the overall SegmentRefreshTask would take multiple days to complete.
>
> **cc:** @swaminathanmanish (in-case you have more details to share)
>
> To the point of making use start/end replace segments, the BaseMultipleSegmentsConversionExecutor would already take care of it right?
Thanks a lot Ragesh for the detailed context. Yes this effort is to minimize the Ideal State updates instead of doing one segment at a time. When there are 100's of minions, problem worsens and we see long wait times (orders of many minutes) for these updates to happen and sometimes results in minion timeouts as well. And then we end up reducing the minions to mitigate the problem which limits horizontal scalability of minions/task takes longer to complete.
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2226667081
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/13597?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 `1.88679%` with `52 lines` in your changes missing coverage. Please review.
> Project coverage is 46.24%. Comparing base [(`59551e4`)](https://app.codecov.io/gh/apache/pinot/commit/59551e45224f1535c4863fd577622b37366ccc97?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`30455c2`)](https://app.codecov.io/gh/apache/pinot/commit/30455c2b86315702c86ee53e7033b359d32cdfd4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 750 commits behind head on master.
| [Files](https://app.codecov.io/gh/apache/pinot/pull/13597?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...he/pinot/segment/local/utils/SegmentPushUtils.java](https://app.codecov.io/gh/apache/pinot/pull/13597?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Futils%2FSegmentPushUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC91dGlscy9TZWdtZW50UHVzaFV0aWxzLmphdmE=) | 0.00% | [49 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/13597?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...he/pinot/spi/ingestion/batch/spec/PushJobSpec.java](https://app.codecov.io/gh/apache/pinot/pull/13597?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fingestion%2Fbatch%2Fspec%2FPushJobSpec.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvaW5nZXN0aW9uL2JhdGNoL3NwZWMvUHVzaEpvYlNwZWMuamF2YQ==) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/13597?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 (59551e4) and HEAD (30455c2). Click for more details.
>
> <details><summary>HEAD has 53 uploads less than BASE</summary>
>
>| Flag | BASE (59551e4) | HEAD (30455c2) |
>|------|------|------|
>|integration|7|0|
>|integration2|3|0|
>|temurin|12|1|
>|java-21|7|1|
>|skip-bytebuffers-true|3|1|
>|skip-bytebuffers-false|7|0|
>|unittests|5|1|
>|unittests1|2|1|
>|java-11|5|0|
>|unittests2|3|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #13597 +/- ##
=============================================
- Coverage 61.75% 46.24% -15.52%
- Complexity 207 1102 +895
=============================================
Files 2436 1949 -487
Lines 133233 103649 -29584
Branches 20636 16746 -3890
=============================================
- Hits 82274 47928 -34346
- Misses 44911 52088 +7177
+ Partials 6048 3633 -2415
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/13597/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/13597/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/13597/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/13597/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/13597/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/13597/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/13597/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `46.24% <1.88%> (-15.39%)` | :arrow_down: |
| [skip-bytebuffers-false](https://app.codecov.io/gh/apache/pinot/pull/13597/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/13597/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `46.24% <1.88%> (+18.51%)` | :arrow_up: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/13597/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `46.24% <1.88%> (-15.52%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/13597/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `46.24% <1.88%> (-15.51%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/13597/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `46.24% <1.88%> (-0.66%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/13597/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
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/13597?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).
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "mcvsubbu (via GitHub)" <gi...@apache.org>.
mcvsubbu commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2231327374
I suppose the scenario you are trying to optimize is when each minion uploads more than one segment (presumably two, to split data over a time boundary?) and hundreds of minions working on multiple such segments? If not, can you please explain the use case a bit more?
Also, can you use the start/end replace segments primitive that is already built?
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2231572794
@mcvsubbu We want to optimize the scenario where multiple clients trying to push segments, each with multiple segments to push, not necessary from minion. The bottleneck right now is that each segment push needs to lock the IS. The idea is to batch multiple IS changes into one to reduce the contention.
The batch upload should work along with start/end replace. Currently the flow is start replace -> push, push, push, push, ... -> end replace. With the new API, we can reduce the amount of push requests
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "klsince (via GitHub)" <gi...@apache.org>.
klsince commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1678288823
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -555,6 +750,67 @@ public void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
}
}
+ @POST
+ @ManagedAsync
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Path("/segmentList")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Cluster.UPLOAD_SEGMENT)
+ @Authenticate(AccessType.CREATE)
+ @ApiOperation(value = "Upload a segment", notes = "Upload a segment as binary")
Review Comment:
update the notes "Upload a batch of segments"
high level question: is this batch uploading API atomic? I see there is logic to clean up uploaded segments upon any failure in the middle, so it's atomic in terms of the uploading action, but I assume it's not atomic as to the ongoing queries, as I didn't see SegmentLineage or similar mechanism was used here. But anyway, might be helpful to comment the API about how things work upon failures.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -403,6 +413,173 @@ private SuccessResponse uploadSegment(@Nullable String tableName, TableType tabl
}
}
+ // Method used to update a list of segments in batch mode with the METADATA upload type.
+ private SuccessResponse uploadSegments(String tableName, TableType tableType,
+ FormDataMultiPart multiParts, boolean enableParallelPushProtection,
+ boolean allowRefresh, HttpHeaders headers, Request request) {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ String tableNameWithType = tableType == TableType.OFFLINE
+ ? TableNameBuilder.OFFLINE.tableNameWithType(rawTableName)
+ : TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ if (tableConfig == null) {
+ throw new ControllerApplicationException(LOGGER, "Failed to fetch table config for table: " + tableNameWithType,
+ Response.Status.BAD_REQUEST);
+ }
+
+ String clientAddress;
+ try {
+ clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
+ } catch (UnknownHostException ex) {
+ throw new ControllerApplicationException(LOGGER, "Failed to resolve hostname from input request",
+ Response.Status.BAD_REQUEST, ex);
+ }
+
+ String uploadTypeStr = extractHttpHeader(headers, FileUploadDownloadClient.CustomHeaders.UPLOAD_TYPE);
+ FileUploadType uploadType = getUploadType(uploadTypeStr);
+ if (!FileUploadType.METADATA.equals(uploadType)) {
+ throw new ControllerApplicationException(LOGGER, "Unsupported upload type: " + uploadTypeStr,
+ Response.Status.BAD_REQUEST);
+ }
+
+ String crypterClassNameInHeader = extractHttpHeader(headers, FileUploadDownloadClient.CustomHeaders.CRYPTER);
+ String ingestionDescriptor = extractHttpHeader(headers, CommonConstants.Controller.INGESTION_DESCRIPTOR);
+ ControllerFilePathProvider provider = ControllerFilePathProvider.getInstance();
+ List<SegmentUploadMetadata> segmentUploadMetadataList = new ArrayList<>();
+ List<File> tempEncryptedFiles = new ArrayList<>();
+ List<File> tempDecryptedFiles = new ArrayList<>();
+ List<File> tempSegmentDirs = new ArrayList<>();
+ List<String> segmentNames = new ArrayList<>();
+
+ for (BodyPart bodyPartFromReq: multiParts.getBodyParts()) {
+ FormDataBodyPart bodyPart = (FormDataBodyPart) bodyPartFromReq;
+ String segmentName = bodyPart.getContentDisposition().getFileName();
+ segmentNames.add(segmentName);
+ if (StringUtils.isEmpty(segmentName)) {
+ throw new ControllerApplicationException(LOGGER,
+ "filename is a required field within the multipart object for METADATA batch upload mode.",
+ Response.Status.BAD_REQUEST);
+ }
+ File tempEncryptedFile;
+ File tempDecryptedFile;
+ File tempSegmentDir;
+ try {
+ String sourceDownloadURIStr = extractHttpHeader(headers,
+ CommonConstants.Controller.SEGMENT_URI_HTTP_HEADER_PREFIX + segmentName);
+ if (StringUtils.isEmpty(sourceDownloadURIStr)) {
+ throw new ControllerApplicationException(LOGGER,
+ "'DOWNLOAD_URI' is required as a field within the multipart object for METADATA batch upload mode.",
+ Response.Status.BAD_REQUEST);
+ }
+ // The downloadUri for putting into segment zk metadata
+ String segmentDownloadURIStr = sourceDownloadURIStr;
+
+ String tempFileName = TMP_DIR_PREFIX + UUID.randomUUID();
+ tempEncryptedFile = new File(provider.getFileUploadTempDir(), tempFileName + ENCRYPTED_SUFFIX);
+ tempEncryptedFiles.add(tempEncryptedFile);
+ tempDecryptedFile = new File(provider.getFileUploadTempDir(), tempFileName);
+ tempDecryptedFiles.add(tempDecryptedFile);
+ tempSegmentDir = new File(provider.getUntarredFileTempDir(), tempFileName);
+ tempSegmentDirs.add(tempSegmentDir);
+ boolean encryptSegment = StringUtils.isNotEmpty(crypterClassNameInHeader);
+ File destFile = encryptSegment ? tempEncryptedFile : tempDecryptedFile;
+ // override copySegmentToFinalLocation if override provided in headers:COPY_SEGMENT_TO_DEEP_STORE
+ // else set to false for backward compatibility
+ String copySegmentToDeepStore =
+ extractHttpHeader(headers, FileUploadDownloadClient.CustomHeaders.COPY_SEGMENT_TO_DEEP_STORE);
+ boolean copySegmentToFinalLocation = Boolean.parseBoolean(copySegmentToDeepStore);
+ createSegmentFileFromBodyPart(bodyPart, destFile);
+ // Include the un-tarred segment size when available
+ long segmentSizeInBytes = getSegmentSizeFromHeaders(segmentName, headers);
+ if (segmentSizeInBytes < 0) {
+ // Use the tarred segment size as an approximation.
+ segmentSizeInBytes = getSegmentSizeFromFile(sourceDownloadURIStr);
+ }
+ if (encryptSegment) {
+ decryptFile(crypterClassNameInHeader, tempEncryptedFile, tempDecryptedFile);
+ }
+
+ String metadataProviderClass = DefaultMetadataExtractor.class.getName();
+ SegmentMetadata segmentMetadata = getSegmentMetadata(tempDecryptedFile, tempSegmentDir, metadataProviderClass);
+ LOGGER.info("Processing upload request for segment: {} of table: {} with upload type: {} from client: {}, "
+ + "ingestion descriptor: {}", segmentName, tableNameWithType, uploadType, clientAddress, ingestionDescriptor
+ );
+
+ // Validate segment
+ if (tableConfig.getIngestionConfig() == null || tableConfig.getIngestionConfig().isSegmentTimeValueCheck()) {
+ SegmentValidationUtils.validateTimeInterval(segmentMetadata, tableConfig);
+ }
+ long untarredSegmentSizeInBytes = 0L;
+ if (segmentSizeInBytes > 0) {
+ untarredSegmentSizeInBytes = segmentSizeInBytes;
+ }
+ SegmentValidationUtils.checkStorageQuota(segmentName, untarredSegmentSizeInBytes, tableConfig,
+ _controllerConf, _storageQuotaChecker);
+
+ // Encrypt segment
+ String crypterNameInTableConfig = tableConfig.getValidationConfig().getCrypterClassName();
+ Pair<String, File> encryptionInfo =
+ encryptSegmentIfNeeded(tempDecryptedFile, tempEncryptedFile, encryptSegment,
+ crypterClassNameInHeader, crypterNameInTableConfig, segmentName, tableNameWithType);
+ File segmentFile = encryptionInfo.getRight();
+
+ // Update download URI if controller is responsible for moving the segment to the deep store
+ URI finalSegmentLocationURI = null;
+ if (copySegmentToFinalLocation) {
+ URI dataDirURI = provider.getDataDirURI();
+ String dataDirPath = dataDirURI.toString();
+ String encodedSegmentName = URIUtils.encode(segmentName);
+ String finalSegmentLocationPath = URIUtils.getPath(dataDirPath, rawTableName, encodedSegmentName);
+ if (dataDirURI.getScheme().equalsIgnoreCase(CommonConstants.Segment.LOCAL_SEGMENT_SCHEME)) {
+ segmentDownloadURIStr = URIUtils.getPath(provider.getVip(), "segments", rawTableName, encodedSegmentName);
+ } else {
+ segmentDownloadURIStr = finalSegmentLocationPath;
+ }
+ finalSegmentLocationURI = URIUtils.getUri(finalSegmentLocationPath);
+ }
+ SegmentUploadMetadata segmentUploadMetadata = new SegmentUploadMetadata(segmentDownloadURIStr,
+ sourceDownloadURIStr, finalSegmentLocationURI, segmentSizeInBytes, segmentMetadata, encryptionInfo);
+ segmentUploadMetadataList.add(segmentUploadMetadata);
+ LOGGER.info("Using segment download URI: {} for segment: {} of table: {} (move segment: {})",
+ segmentDownloadURIStr, segmentFile, tableNameWithType, copySegmentToFinalLocation);
+ } catch (Exception ex) {
+ cleanupTempFiles(tempEncryptedFiles, tempDecryptedFiles, tempSegmentDirs);
+ _controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_SEGMENT_UPLOAD_ERROR,
+ segmentUploadMetadataList.size());
+ throw new ControllerApplicationException(LOGGER, "Exception while processing segments to upload: "
+ + ex.getMessage(), Response.Status.INTERNAL_SERVER_ERROR, ex);
+ }
+ }
+
+ try {
+ ZKOperator zkOperator = new ZKOperator(_pinotHelixResourceManager, _controllerConf, _controllerMetrics);
+ zkOperator.completeSegmentsOperations(tableNameWithType, uploadType, enableParallelPushProtection, allowRefresh,
+ headers, segmentUploadMetadataList);
+ return new SuccessResponse("Successfully uploaded segments: " + segmentNames + " of table: "
+ + tableNameWithType);
+ } catch (Exception ex) {
+ _controllerMetrics.addMeteredGlobalValue(ControllerMeter.CONTROLLER_SEGMENT_UPLOAD_ERROR,
+ segmentUploadMetadataList.size());
+ throw new ControllerApplicationException(LOGGER, "Exception while uploading segments: " + ex.getMessage(),
+ Response.Status.INTERNAL_SERVER_ERROR, ex);
+ } finally {
+ cleanupTempFiles(tempEncryptedFiles, tempDecryptedFiles, tempSegmentDirs);
Review Comment:
why not combine the two try-catch blocks into one, and share this `finally { cleanupTempFiles(); } `
btw, would it possible to process all segments inside one temp dir, then we just need to clean this root temp dir up in the end, and no need to track all temp dirs used by segments separately
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -295,13 +302,18 @@ private SuccessResponse uploadSegment(@Nullable String tableName, TableType tabl
extractHttpHeader(headers, FileUploadDownloadClient.CustomHeaders.COPY_SEGMENT_TO_DEEP_STORE);
copySegmentToFinalLocation = Boolean.parseBoolean(copySegmentToDeepStore);
createSegmentFileFromMultipart(multiPart, destFile);
+ PinotFS pinotFS = null;
Review Comment:
good catch, may consider to cut a small PR to land this fix first.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -555,6 +750,67 @@ public void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
}
}
+ @POST
+ @ManagedAsync
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Path("/segmentList")
Review Comment:
maybe consider `v3/segments` as there is `v2/segments`
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1678445677
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -555,6 +750,67 @@ public void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
}
}
+ @POST
+ @ManagedAsync
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Path("/segmentList")
Review Comment:
I'm not a big fan of `v2`, `v3` etc. I prefer having a separate API just for batch upload
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "jtao15 (via GitHub)" <gi...@apache.org>.
jtao15 commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2231850381
@rajagopr Thank you for the interesting PR. I'd like to take a look soon.
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1680289873
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -555,6 +750,67 @@ public void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
}
}
+ @POST
+ @ManagedAsync
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Path("/segmentList")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableName", action = Actions.Cluster.UPLOAD_SEGMENT)
+ @Authenticate(AccessType.CREATE)
+ @ApiOperation(value = "Upload a segment", notes = "Upload a segment as binary")
Review Comment:
> update the notes "Upload a batch of segments"
>
> high level question: is this batch uploading API atomic? I see there is logic to clean up uploaded segments upon any failure in the middle, so it's atomic in terms of the uploading action, but I assume it's not atomic as to the ongoing queries, as I didn't see SegmentLineage or similar mechanism was used here. But anyway, might be helpful to comment the API about how things work upon failures.
Good point. This is all or nothing API if I understand @rajagopr ?
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1681484482
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2327,6 +2327,82 @@ public void assignTableSegment(String tableNameWithType, String segmentName) {
}
}
+ // Assign a list of segments in batch mode
+ public void assignTableSegments(String tableNameWithType, List<String> segmentNames) {
+ Map<String, String> segmentZKMetadataPathMap = new HashMap<>();
+ for (String segmentName: segmentNames) {
+ String segmentZKMetadataPath = ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType,
+ segmentName);
+ segmentZKMetadataPathMap.put(segmentName, segmentZKMetadataPath);
+ }
+ // Assign instances for the segment and add it into IdealState
+ try {
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
+
+ Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap =
+ fetchOrComputeInstancePartitions(tableNameWithType, tableConfig);
+
+ // Initialize tier information only in case direct tier assignment is configured
+ if (_enableTieredSegmentAssignment && CollectionUtils.isNotEmpty(tableConfig.getTierConfigsList())) {
+ List<Tier> sortedTiers = TierConfigUtils.getSortedTiersForStorageType(tableConfig.getTierConfigsList(),
+ TierFactory.PINOT_SERVER_STORAGE_TYPE, _helixZkManager);
+ for (String segmentName: segmentNames) {
+ // Update segment tier to support direct assignment for multiple data directories
+ updateSegmentTargetTier(tableNameWithType, segmentName, sortedTiers);
+ InstancePartitions tierInstancePartitions = TierConfigUtils.getTieredInstancePartitionsForSegment(
+ tableNameWithType, segmentName, sortedTiers, _helixZkManager);
+ if (tierInstancePartitions != null && TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
+ // Override instance partitions for offline table
+ LOGGER.info("Overriding with tiered instance partitions: {} for segment: {} of table: {}",
+ tierInstancePartitions, segmentName, tableNameWithType);
+ instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.OFFLINE, tierInstancePartitions);
+ }
+ }
+ }
+
+ SegmentAssignment segmentAssignment =
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig, _controllerMetrics);
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
Review Comment:
Added the logs. The metrics would be too sparse since this is not a continuous activity and only happens when segments are uploaded?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2327,6 +2327,82 @@ public void assignTableSegment(String tableNameWithType, String segmentName) {
}
}
+ // Assign a list of segments in batch mode
+ public void assignTableSegments(String tableNameWithType, List<String> segmentNames) {
+ Map<String, String> segmentZKMetadataPathMap = new HashMap<>();
+ for (String segmentName: segmentNames) {
+ String segmentZKMetadataPath = ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType,
+ segmentName);
+ segmentZKMetadataPathMap.put(segmentName, segmentZKMetadataPath);
+ }
+ // Assign instances for the segment and add it into IdealState
+ try {
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
+
+ Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap =
+ fetchOrComputeInstancePartitions(tableNameWithType, tableConfig);
+
+ // Initialize tier information only in case direct tier assignment is configured
+ if (_enableTieredSegmentAssignment && CollectionUtils.isNotEmpty(tableConfig.getTierConfigsList())) {
+ List<Tier> sortedTiers = TierConfigUtils.getSortedTiersForStorageType(tableConfig.getTierConfigsList(),
+ TierFactory.PINOT_SERVER_STORAGE_TYPE, _helixZkManager);
+ for (String segmentName: segmentNames) {
+ // Update segment tier to support direct assignment for multiple data directories
+ updateSegmentTargetTier(tableNameWithType, segmentName, sortedTiers);
+ InstancePartitions tierInstancePartitions = TierConfigUtils.getTieredInstancePartitionsForSegment(
+ tableNameWithType, segmentName, sortedTiers, _helixZkManager);
+ if (tierInstancePartitions != null && TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
+ // Override instance partitions for offline table
+ LOGGER.info("Overriding with tiered instance partitions: {} for segment: {} of table: {}",
+ tierInstancePartitions, segmentName, tableNameWithType);
+ instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.OFFLINE, tierInstancePartitions);
+ }
+ }
+ }
+
+ SegmentAssignment segmentAssignment =
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig, _controllerMetrics);
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
+ Map<InstancePartitionsType, InstancePartitions> finalInstancePartitionsMap = instancePartitionsMap;
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, idealState -> {
+ assert idealState != null;
+ for (String segmentName: segmentNames) {
+ Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
+ if (currentAssignment.containsKey(segmentName)) {
+ LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
+ tableNameWithType);
+ } else {
+ List<String> assignedInstances =
+ segmentAssignment.assignSegment(segmentName, currentAssignment, finalInstancePartitionsMap);
+ LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
+ tableNameWithType);
+ currentAssignment.put(segmentName, SegmentAssignmentUtils.getInstanceStateMap(assignedInstances,
+ SegmentStateModel.ONLINE));
+ }
+ }
+ return idealState;
+ });
+ LOGGER.info("Added segments: {} to IdealState for table: {}", segmentNames, tableNameWithType);
+ }
+ } catch (Exception e) {
+ LOGGER.error(
+ "Caught exception while adding segments: {} to IdealState for table: {}, deleting segments ZK metadata",
+ segmentNames, tableNameWithType, e);
+ for (Map.Entry<String, String> segmentZKMetadataPathEntry: segmentZKMetadataPathMap.entrySet()) {
+ String segmentName = segmentZKMetadataPathEntry.getKey();
+ String segmentZKMetadataPath = segmentZKMetadataPathEntry.getValue();
+ if (_propertyStore.remove(segmentZKMetadataPath, AccessOption.PERSISTENT)) {
+ LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
+ } else {
+ LOGGER.error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName,
Review Comment:
Yes, the original logic doesn't retry as well. I'm not sure if we have reason to believe that retry would make the remove operation succeed. It looks like the helix library itself does some error handling once the remove method is called.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentPushUtils.java:
##########
@@ -277,6 +278,8 @@ public static void sendSegmentUriAndMetadata(SegmentGenerationJobSpec spec, Pino
Map<String, String> segmentUriToTarPathMap, List<Header> headers, List<NameValuePair> parameters)
throws Exception {
String tableName = spec.getTableSpec().getTableName();
+ Map<String, File> segmentMetadataFileMap = new HashMap<>();
+ Map<String, String> segmentUriPathMap = new HashMap<>();
Review Comment:
If we introduce a limit, then we need to invoke this API in batches. For example, if we have 100 segments to upload and if we have set the limit on batch size as 20, then we have to call this API 5 times. As of now, we are dependent on the SegmentConversionResult size and expect the size to be a reasonable number. Let's discuss and set reasonable limits.
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1680255251
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2327,6 +2327,82 @@ public void assignTableSegment(String tableNameWithType, String segmentName) {
}
}
+ // Assign a list of segments in batch mode
+ public void assignTableSegments(String tableNameWithType, List<String> segmentNames) {
+ Map<String, String> segmentZKMetadataPathMap = new HashMap<>();
+ for (String segmentName: segmentNames) {
+ String segmentZKMetadataPath = ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType,
+ segmentName);
+ segmentZKMetadataPathMap.put(segmentName, segmentZKMetadataPath);
+ }
+ // Assign instances for the segment and add it into IdealState
+ try {
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
+
+ Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap =
+ fetchOrComputeInstancePartitions(tableNameWithType, tableConfig);
+
+ // Initialize tier information only in case direct tier assignment is configured
+ if (_enableTieredSegmentAssignment && CollectionUtils.isNotEmpty(tableConfig.getTierConfigsList())) {
+ List<Tier> sortedTiers = TierConfigUtils.getSortedTiersForStorageType(tableConfig.getTierConfigsList(),
+ TierFactory.PINOT_SERVER_STORAGE_TYPE, _helixZkManager);
+ for (String segmentName: segmentNames) {
+ // Update segment tier to support direct assignment for multiple data directories
+ updateSegmentTargetTier(tableNameWithType, segmentName, sortedTiers);
+ InstancePartitions tierInstancePartitions = TierConfigUtils.getTieredInstancePartitionsForSegment(
+ tableNameWithType, segmentName, sortedTiers, _helixZkManager);
+ if (tierInstancePartitions != null && TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
+ // Override instance partitions for offline table
+ LOGGER.info("Overriding with tiered instance partitions: {} for segment: {} of table: {}",
+ tierInstancePartitions, segmentName, tableNameWithType);
+ instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.OFFLINE, tierInstancePartitions);
+ }
+ }
+ }
+
+ SegmentAssignment segmentAssignment =
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig, _controllerMetrics);
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
Review Comment:
Can we record the time for this update (start/end)? We can add that log.info as to how long it takes to add batch of segments (size of the list) and also a latency metric if it'll help
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentPushUtils.java:
##########
@@ -303,62 +306,86 @@ public static void sendSegmentUriAndMetadata(SegmentGenerationJobSpec spec, Pino
} else {
segmentMetadataFile = generateSegmentMetadataFile(fileSystem, URI.create(tarFilePath));
}
- try {
- for (PinotClusterSpec pinotClusterSpec : spec.getPinotClusterSpecs()) {
- URI controllerURI;
- try {
- controllerURI = new URI(pinotClusterSpec.getControllerURI());
- } catch (URISyntaxException e) {
- throw new RuntimeException("Got invalid controller uri - '" + pinotClusterSpec.getControllerURI() + "'");
- }
- LOGGER.info("Pushing segment: {} to location: {} for table {}", segmentName, controllerURI, tableName);
- int attempts = 1;
- if (spec.getPushJobSpec() != null && spec.getPushJobSpec().getPushAttempts() > 0) {
- attempts = spec.getPushJobSpec().getPushAttempts();
- }
- long retryWaitMs = 1000L;
- if (spec.getPushJobSpec() != null && spec.getPushJobSpec().getPushRetryIntervalMillis() > 0) {
- retryWaitMs = spec.getPushJobSpec().getPushRetryIntervalMillis();
- }
- RetryPolicies.exponentialBackoffRetryPolicy(attempts, retryWaitMs, 5).attempt(() -> {
- List<Header> reqHttpHeaders = new ArrayList<>(headers);
- try {
- reqHttpHeaders.add(new BasicHeader(FileUploadDownloadClient.CustomHeaders.DOWNLOAD_URI, segmentUriPath));
- reqHttpHeaders.add(new BasicHeader(FileUploadDownloadClient.CustomHeaders.UPLOAD_TYPE,
- FileUploadDownloadClient.FileUploadType.METADATA.toString()));
- if (spec.getPushJobSpec() != null) {
- reqHttpHeaders.add(new BasicHeader(FileUploadDownloadClient.CustomHeaders.COPY_SEGMENT_TO_DEEP_STORE,
- String.valueOf(spec.getPushJobSpec().getCopyToDeepStoreForMetadataPush())));
- }
+ segmentMetadataFileMap.put(segmentName, segmentMetadataFile);
+ segmentUriPathMap.put(segmentName, segmentUriPath);
+ }
- SimpleHttpResponse response = FILE_UPLOAD_DOWNLOAD_CLIENT.uploadSegmentMetadata(
- FileUploadDownloadClient.getUploadSegmentURI(controllerURI), segmentName,
- segmentMetadataFile, reqHttpHeaders, parameters, HttpClient.DEFAULT_SOCKET_TIMEOUT_MS);
- LOGGER.info("Response for pushing table {} segment {} to location {} - {}: {}", tableName, segmentName,
- controllerURI, response.getStatusCode(), response.getResponse());
- return true;
- } catch (HttpErrorStatusException e) {
- int statusCode = e.getStatusCode();
- if (statusCode >= 500) {
- // Temporary exception
- LOGGER.warn("Caught temporary exception while pushing table: {} segment: {} to {}, will retry",
- tableName, segmentName, controllerURI, e);
- return false;
- } else {
- // Permanent exception
- LOGGER.error("Caught permanent exception while pushing table: {} segment: {} to {}, won't retry",
- tableName, segmentName, controllerURI, e);
- throw e;
- }
- }
- });
+ // perform metadata push in batch mode for every cluster
+ try {
+ for (PinotClusterSpec pinotClusterSpec : spec.getPinotClusterSpecs()) {
+ URI controllerURI;
+ try {
+ controllerURI = new URI(pinotClusterSpec.getControllerURI());
+ } catch (URISyntaxException e) {
+ throw new RuntimeException("Got invalid controller uri - '" + pinotClusterSpec.getControllerURI() + "'");
+ }
+ LOGGER.info("Pushing segments: {} to Pinot cluster: {} for table {}",
+ segmentMetadataFileMap.keySet(), controllerURI, tableName);
+ int attempts = 1;
+ if (spec.getPushJobSpec() != null && spec.getPushJobSpec().getPushAttempts() > 0) {
+ attempts = spec.getPushJobSpec().getPushAttempts();
}
- } finally {
- FileUtils.deleteQuietly(segmentMetadataFile);
+ long retryWaitMs = 1000L;
+ if (spec.getPushJobSpec() != null && spec.getPushJobSpec().getPushRetryIntervalMillis() > 0) {
+ retryWaitMs = spec.getPushJobSpec().getPushRetryIntervalMillis();
+ }
+ RetryPolicies.exponentialBackoffRetryPolicy(attempts, retryWaitMs, 5).attempt(() -> {
+ List<Header> reqHttpHeaders = new ArrayList<>(headers);
+ try {
+ addHeaders(segmentUriPathMap, spec, reqHttpHeaders);
+ URI segmentUploadURI = getSegmentUploadURI(spec.getPushJobSpec(), controllerURI);
+ SimpleHttpResponse response = FILE_UPLOAD_DOWNLOAD_CLIENT.uploadSegmentMetadataFiles(segmentUploadURI,
+ segmentMetadataFileMap, reqHttpHeaders, parameters, HttpClient.DEFAULT_SOCKET_TIMEOUT_MS);
+ LOGGER.info("Response for pushing table {} segments {} to location {} - {}: {}", tableName,
+ segmentMetadataFileMap.keySet(), controllerURI, response.getStatusCode(), response.getResponse());
+ return true;
+ } catch (HttpErrorStatusException e) {
+ int statusCode = e.getStatusCode();
+ if (statusCode >= 500) {
+ // Temporary exception
+ LOGGER.warn("Caught temporary exception while pushing table: {} segments: {} to {}, will retry",
+ tableName, segmentMetadataFileMap.keySet(), controllerURI, e);
+ return false;
+ } else {
+ // Permanent exception
+ LOGGER.error("Caught permanent exception while pushing table: {} segments: {} to {}, won't retry",
Review Comment:
This will be all or nothing for batch mode right?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -403,6 +415,159 @@ private SuccessResponse uploadSegment(@Nullable String tableName, TableType tabl
}
}
+ // Method used to update a list of segments in batch mode with the METADATA upload type.
+ private SuccessResponse uploadSegments(String tableName, TableType tableType, FormDataMultiPart multiParts,
+ boolean enableParallelPushProtection, boolean allowRefresh, HttpHeaders headers, Request request) {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ String tableNameWithType = tableType == TableType.OFFLINE ? TableNameBuilder.OFFLINE.tableNameWithType(rawTableName)
+ : TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
Review Comment:
Can we add a log somewhere (either here or in the caller) on the batch size that this API is taking up.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -2327,6 +2327,82 @@ public void assignTableSegment(String tableNameWithType, String segmentName) {
}
}
+ // Assign a list of segments in batch mode
+ public void assignTableSegments(String tableNameWithType, List<String> segmentNames) {
+ Map<String, String> segmentZKMetadataPathMap = new HashMap<>();
+ for (String segmentName: segmentNames) {
+ String segmentZKMetadataPath = ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType,
+ segmentName);
+ segmentZKMetadataPathMap.put(segmentName, segmentZKMetadataPath);
+ }
+ // Assign instances for the segment and add it into IdealState
+ try {
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Preconditions.checkState(tableConfig != null, "Failed to find table config for table: " + tableNameWithType);
+
+ Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap =
+ fetchOrComputeInstancePartitions(tableNameWithType, tableConfig);
+
+ // Initialize tier information only in case direct tier assignment is configured
+ if (_enableTieredSegmentAssignment && CollectionUtils.isNotEmpty(tableConfig.getTierConfigsList())) {
+ List<Tier> sortedTiers = TierConfigUtils.getSortedTiersForStorageType(tableConfig.getTierConfigsList(),
+ TierFactory.PINOT_SERVER_STORAGE_TYPE, _helixZkManager);
+ for (String segmentName: segmentNames) {
+ // Update segment tier to support direct assignment for multiple data directories
+ updateSegmentTargetTier(tableNameWithType, segmentName, sortedTiers);
+ InstancePartitions tierInstancePartitions = TierConfigUtils.getTieredInstancePartitionsForSegment(
+ tableNameWithType, segmentName, sortedTiers, _helixZkManager);
+ if (tierInstancePartitions != null && TableNameBuilder.isOfflineTableResource(tableNameWithType)) {
+ // Override instance partitions for offline table
+ LOGGER.info("Overriding with tiered instance partitions: {} for segment: {} of table: {}",
+ tierInstancePartitions, segmentName, tableNameWithType);
+ instancePartitionsMap = Collections.singletonMap(InstancePartitionsType.OFFLINE, tierInstancePartitions);
+ }
+ }
+ }
+
+ SegmentAssignment segmentAssignment =
+ SegmentAssignmentFactory.getSegmentAssignment(_helixZkManager, tableConfig, _controllerMetrics);
+ synchronized (getTableUpdaterLock(tableNameWithType)) {
+ Map<InstancePartitionsType, InstancePartitions> finalInstancePartitionsMap = instancePartitionsMap;
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, idealState -> {
+ assert idealState != null;
+ for (String segmentName: segmentNames) {
+ Map<String, Map<String, String>> currentAssignment = idealState.getRecord().getMapFields();
+ if (currentAssignment.containsKey(segmentName)) {
+ LOGGER.warn("Segment: {} already exists in the IdealState for table: {}, do not update", segmentName,
+ tableNameWithType);
+ } else {
+ List<String> assignedInstances =
+ segmentAssignment.assignSegment(segmentName, currentAssignment, finalInstancePartitionsMap);
+ LOGGER.info("Assigning segment: {} to instances: {} for table: {}", segmentName, assignedInstances,
+ tableNameWithType);
+ currentAssignment.put(segmentName, SegmentAssignmentUtils.getInstanceStateMap(assignedInstances,
+ SegmentStateModel.ONLINE));
+ }
+ }
+ return idealState;
+ });
+ LOGGER.info("Added segments: {} to IdealState for table: {}", segmentNames, tableNameWithType);
+ }
+ } catch (Exception e) {
+ LOGGER.error(
+ "Caught exception while adding segments: {} to IdealState for table: {}, deleting segments ZK metadata",
+ segmentNames, tableNameWithType, e);
+ for (Map.Entry<String, String> segmentZKMetadataPathEntry: segmentZKMetadataPathMap.entrySet()) {
+ String segmentName = segmentZKMetadataPathEntry.getKey();
+ String segmentZKMetadataPath = segmentZKMetadataPathEntry.getValue();
+ if (_propertyStore.remove(segmentZKMetadataPath, AccessOption.PERSISTENT)) {
+ LOGGER.info("Deleted segment ZK metadata for segment: {} of table: {}", segmentName, tableNameWithType);
+ } else {
+ LOGGER.error("Failed to deleted segment ZK metadata for segment: {} of table: {}", segmentName,
Review Comment:
There's no retry for deletion update to Zk? I guess we did not have this in original logic as well
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -295,13 +302,18 @@ private SuccessResponse uploadSegment(@Nullable String tableName, TableType tabl
extractHttpHeader(headers, FileUploadDownloadClient.CustomHeaders.COPY_SEGMENT_TO_DEEP_STORE);
copySegmentToFinalLocation = Boolean.parseBoolean(copySegmentToDeepStore);
createSegmentFileFromMultipart(multiPart, destFile);
+ PinotFS pinotFS = null;
try {
URI segmentURI = new URI(sourceDownloadURIStr);
- PinotFS pinotFS = PinotFSFactory.create(segmentURI.getScheme());
+ pinotFS = PinotFSFactory.create(segmentURI.getScheme());
segmentSizeInBytes = pinotFS.length(segmentURI);
} catch (Exception e) {
segmentSizeInBytes = -1;
LOGGER.warn("Could not fetch segment size for metadata push", e);
+ } finally {
+ if (pinotFS != null) {
Review Comment:
Good catch !
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentPushUtils.java:
##########
@@ -277,6 +278,8 @@ public static void sendSegmentUriAndMetadata(SegmentGenerationJobSpec spec, Pino
Map<String, String> segmentUriToTarPathMap, List<Header> headers, List<NameValuePair> parameters)
throws Exception {
String tableName = spec.getTableSpec().getTableName();
+ Map<String, File> segmentMetadataFileMap = new HashMap<>();
+ Map<String, String> segmentUriPathMap = new HashMap<>();
Review Comment:
Should we have a limit on the batch size or have users provide this via config ? The next thing we will face is that we are pushing a large batch :)
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #13597:
URL: https://github.com/apache/pinot/pull/13597#discussion_r1678167065
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -403,6 +413,173 @@ private SuccessResponse uploadSegment(@Nullable String tableName, TableType tabl
}
}
+ // Method used to update a list of segments in batch mode with the METADATA upload type.
+ private SuccessResponse uploadSegments(String tableName, TableType tableType,
+ FormDataMultiPart multiParts, boolean enableParallelPushProtection,
+ boolean allowRefresh, HttpHeaders headers, Request request) {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ String tableNameWithType = tableType == TableType.OFFLINE
+ ? TableNameBuilder.OFFLINE.tableNameWithType(rawTableName)
+ : TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ if (tableConfig == null) {
+ throw new ControllerApplicationException(LOGGER, "Failed to fetch table config for table: " + tableNameWithType,
+ Response.Status.BAD_REQUEST);
+ }
+
+ String clientAddress;
+ try {
+ clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
+ } catch (UnknownHostException ex) {
Review Comment:
(minor) I most of the places we just use `e` to represent exception. Shall we keep the same convention for consistency?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/SegmentUploadMetadata.java:
##########
@@ -0,0 +1,103 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.controller.api.upload;
+
+import java.io.File;
+import java.net.URI;
+import java.util.Objects;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.segment.spi.SegmentMetadata;
+
+
+public class SegmentUploadMetadata {
Review Comment:
Good job extracting this. Please add some javadoc explaining this class and each field
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -403,6 +413,173 @@ private SuccessResponse uploadSegment(@Nullable String tableName, TableType tabl
}
}
+ // Method used to update a list of segments in batch mode with the METADATA upload type.
+ private SuccessResponse uploadSegments(String tableName, TableType tableType,
+ FormDataMultiPart multiParts, boolean enableParallelPushProtection,
+ boolean allowRefresh, HttpHeaders headers, Request request) {
+ String rawTableName = TableNameBuilder.extractRawTableName(tableName);
+ String tableNameWithType = tableType == TableType.OFFLINE
+ ? TableNameBuilder.OFFLINE.tableNameWithType(rawTableName)
+ : TableNameBuilder.REALTIME.tableNameWithType(rawTableName);
+
+ TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
+ if (tableConfig == null) {
+ throw new ControllerApplicationException(LOGGER, "Failed to fetch table config for table: " + tableNameWithType,
+ Response.Status.BAD_REQUEST);
+ }
+
+ String clientAddress;
+ try {
+ clientAddress = InetAddress.getByName(request.getRemoteAddr()).getHostName();
+ } catch (UnknownHostException ex) {
+ throw new ControllerApplicationException(LOGGER, "Failed to resolve hostname from input request",
+ Response.Status.BAD_REQUEST, ex);
+ }
+
+ String uploadTypeStr = extractHttpHeader(headers, FileUploadDownloadClient.CustomHeaders.UPLOAD_TYPE);
+ FileUploadType uploadType = getUploadType(uploadTypeStr);
+ if (!FileUploadType.METADATA.equals(uploadType)) {
+ throw new ControllerApplicationException(LOGGER, "Unsupported upload type: " + uploadTypeStr,
+ Response.Status.BAD_REQUEST);
+ }
+
+ String crypterClassNameInHeader = extractHttpHeader(headers, FileUploadDownloadClient.CustomHeaders.CRYPTER);
+ String ingestionDescriptor = extractHttpHeader(headers, CommonConstants.Controller.INGESTION_DESCRIPTOR);
+ ControllerFilePathProvider provider = ControllerFilePathProvider.getInstance();
+ List<SegmentUploadMetadata> segmentUploadMetadataList = new ArrayList<>();
+ List<File> tempEncryptedFiles = new ArrayList<>();
+ List<File> tempDecryptedFiles = new ArrayList<>();
+ List<File> tempSegmentDirs = new ArrayList<>();
+ List<String> segmentNames = new ArrayList<>();
+
+ for (BodyPart bodyPartFromReq: multiParts.getBodyParts()) {
Review Comment:
(format) reformat the changes
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentUploadDownloadRestletResource.java:
##########
@@ -555,6 +750,67 @@ public void uploadSegmentAsMultiPart(FormDataMultiPart multiPart,
}
}
+ @POST
+ @ManagedAsync
+ @Produces(MediaType.APPLICATION_JSON)
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Path("/segmentList")
Review Comment:
Suggest making the API path `segments/{tableName}/batchUpload`. `POST /segmentList` is quite misleading to me
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java:
##########
@@ -62,52 +66,58 @@ public ZKOperator(PinotHelixResourceManager pinotHelixResourceManager, Controlle
_controllerMetrics = controllerMetrics;
}
- public void completeSegmentOperations(String tableNameWithType, SegmentMetadata segmentMetadata,
- FileUploadType uploadType, @Nullable URI finalSegmentLocationURI, File segmentFile,
- @Nullable String sourceDownloadURIStr, String segmentDownloadURIStr, @Nullable String crypterName,
- long segmentSizeInBytes, boolean enableParallelPushProtection, boolean allowRefresh, HttpHeaders headers)
+ public void completeSegmentsOperations(String tableNameWithType, FileUploadType uploadType,
Review Comment:
Consider using different method to handle single segment upload vs batch segment upload. We may extract common part as needed, but modifying the existing flow can be risky. It can cause backward incompatible if any error handling behavior is changed
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2231578183
> I suppose the scenario you are trying to optimize is when each minion uploads more than one segment (presumably two, to split data over a time boundary?) and hundreds of minions working on multiple such segments? If not, can you please explain the use case a bit more?
>
> Also, can you use the start/end replace segments primitive that is already built?
@mcvsubbu : Sharing additional context below.
The SegmentRefreshTask (custom minion task) helps align Pinot segments to the table configs. The task can do things like: 1) Break down a large segment into multiple small segments 2) Combine multiple small segments into one large segment
3) Sort the data based on the sort column – if sort column changed. 4) Re-partition the data etc. The task extends on the abstract class BaseMultipleSegmentsConversionExecutor which can operate on multiple segments. The task aligns the
Pinot segments to the table configs over multiple invocations. After each round of invocation, the segments (or subset of segments) would be a step closer to the table configuration.
When we encountered issues, each minion task was operating on upto twenty segments, there were about ~50K segments in the table and about 200 minions were operating at the same time. Once the SegmentConversionResults are generated the minion attempts to upload the new segments and there is a configured timeout of about ten minutes to upload the segment. Upon closer inspection, we found that the contention is happening while updating the IdealState and the threads are waiting to acquire the table update lock. Since, many of the uploads timeout, this would result in a flood of retries which worsens the problem.
Due to the contention issue described above we are unable to increase the number of minions and the overall SegmentRefreshTask would take multiple days to complete.
**cc:** @swaminathanmanish (in-case you have more details to share)
To the point of making use start/end replace segments, the BaseMultipleSegmentsConversionExecutor would already take care of it 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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "kishoreg (via GitHub)" <gi...@apache.org>.
kishoreg commented on PR #13597:
URL: https://github.com/apache/pinot/pull/13597#issuecomment-2227686821
What was the previous api to upload a single segment?
--
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] Added support to upload segments in batch mode with METADATA upload type [pinot]
Posted by "rajagopr (via GitHub)" <gi...@apache.org>.
rajagopr closed pull request #13597: Added support to upload segments in batch mode with METADATA upload type
URL: https://github.com/apache/pinot/pull/13597
--
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