Jackie-Jiang commented on code in PR #14506:
URL: https://github.com/apache/pinot/pull/14506#discussion_r1861278690
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java:
##########
@@ -846,6 +849,106 @@ public String uploadLLCSegment(
}
}
+ /**
+ * Upload a low level consumer segment to segment store and return the
segment download url, crc and
+ * other segment metadata. This endpoint is used when segment store copy is
unavailable for committed
+ * low level consumer segments.
+ * Please note that invocation of this endpoint may cause query performance
to suffer, since we tar up the segment
+ * to upload it.
+ *
+ * @see <a href="https://tinyurl.com/f63ru4sb></a>
+ * @param realtimeTableName table name with type.
+ * @param segmentName name of the segment to be uploaded
+ * @param timeoutMs timeout for the segment upload to the deep-store. If
this is negative, the default timeout
+ * would be used.
+ * @return full url where the segment is uploaded, crc, segmentName. Can add
more segment metadata in the future.
+ * @throws Exception if an error occurred during the segment upload.
+ */
+ @POST
+ @Path("/segments/{realtimeTableName}/{segmentName}/uploadV2")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Upload a low level consumer segment to segment store
and return the segment download url,"
+ + "crc and other segment metadata",
+ notes = "Upload a low level consumer segment to segment store and return
the segment download url, crc "
+ + "and other segment metadata")
+ @ApiResponses(value = {
+ @ApiResponse(code = 200, message = "Success"),
+ @ApiResponse(code = 500, message = "Internal server error", response =
ErrorInfo.class),
+ @ApiResponse(code = 404, message = "Table or segment not found",
response = ErrorInfo.class),
+ @ApiResponse(code = 400, message = "Bad request", response =
ErrorInfo.class)
+ })
+ public TableSegmentUploadV2Response uploadLLCSegmentV2(
+ @ApiParam(value = "Name of the REALTIME table", required = true)
@PathParam("realtimeTableName")
+ String realtimeTableName,
+ @ApiParam(value = "Name of the segment", required = true)
@PathParam("segmentName") String segmentName,
+ @QueryParam("uploadTimeoutMs") @DefaultValue("-1") int timeoutMs,
+ @Context HttpHeaders headers)
+ throws Exception {
+ realtimeTableName = DatabaseUtils.translateTableName(realtimeTableName,
headers);
+ LOGGER.info("Received a request to upload low level consumer segment {}
for table {}", segmentName,
+ realtimeTableName);
+
+ // Check it's realtime table
+ TableType tableType =
TableNameBuilder.getTableTypeFromTableName(realtimeTableName);
+ if (TableType.OFFLINE == tableType) {
+ throw new WebApplicationException(
+ String.format("Cannot upload low level consumer segment for OFFLINE
table: %s", realtimeTableName),
+ Response.Status.BAD_REQUEST);
+ }
+
+ // Check the segment is low level consumer segment
+ if (!LLCSegmentName.isLLCSegment(segmentName)) {
+ throw new WebApplicationException(String.format("Segment %s is not a low
level consumer segment", segmentName),
+ Response.Status.BAD_REQUEST);
+ }
+
+ String tableNameWithType =
TableNameBuilder.forType(TableType.REALTIME).tableNameWithType(realtimeTableName);
Review Comment:
(minor) Not introduced in this PR, but usually `realtimeTableName` means
table name with `_REALTIME` suffix, so we can check whether table type is
REALTIME on line 893
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -1557,22 +1558,41 @@ public void uploadToDeepStoreIfMissing(TableConfig
tableConfig, List<SegmentZKMe
// Randomly ask one server to upload
URI uri =
peerSegmentURIs.get(RANDOM.nextInt(peerSegmentURIs.size()));
- String serverUploadRequestUrl = StringUtil.join("/", uri.toString(),
"upload");
- serverUploadRequestUrl =
- String.format("%s?uploadTimeoutMs=%d", serverUploadRequestUrl,
_deepstoreUploadRetryTimeoutMs);
- LOGGER.info("Ask server to upload LLC segment {} to deep store by
this path: {}", segmentName,
- serverUploadRequestUrl);
- String tempSegmentDownloadUrl =
_fileUploadDownloadClient.uploadToSegmentStore(serverUploadRequestUrl);
- String segmentDownloadUrl =
- moveSegmentFile(rawTableName, segmentName,
tempSegmentDownloadUrl, pinotFS);
- LOGGER.info("Updating segment {} download url in ZK to be {}",
segmentName, segmentDownloadUrl);
-
- // Update segment ZK metadata by adding the download URL
- segmentZKMetadata.setDownloadUrl(segmentDownloadUrl);
+ try {
+ String serverUploadRequestUrl = StringUtil.join("/",
uri.toString(), "uploadV2");
+ serverUploadRequestUrl =
+ String.format("%s?uploadTimeoutMs=%d", serverUploadRequestUrl,
_deepstoreUploadRetryTimeoutMs);
+ LOGGER.info("Ask server to upload LLC segment {} to deep store by
this path: {}", segmentName,
+ serverUploadRequestUrl);
+ TableSegmentUploadV2Response tableSegmentUploadV2Response
+ =
_fileUploadDownloadClient.uploadToSegmentStoreV2(serverUploadRequestUrl);
+ String segmentDownloadUrl =
+ moveSegmentFile(rawTableName, segmentName,
tableSegmentUploadV2Response.getDownloadUrl(), pinotFS);
+ LOGGER.info("Updating segment {} download url in ZK to be {}",
segmentName, segmentDownloadUrl);
+ // Update segment ZK metadata by adding the download URL
+ segmentZKMetadata.setDownloadUrl(segmentDownloadUrl);
+ // Update ZK crc to that of the server segment crc if unmatched
+ if (Long.parseLong(tableSegmentUploadV2Response.getSegmentCrc())
!= segmentZKMetadata.getCrc()) {
+
segmentZKMetadata.setCrc(Long.parseLong(tableSegmentUploadV2Response.getSegmentCrc()));
Review Comment:
Let's log something here if crc changed
##########
pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/TableSegmentUploadV2Response.java:
##########
@@ -0,0 +1,47 @@
+/**
+ * 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.common.restlet.resources;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+
+public class TableSegmentUploadV2Response {
+ private final String _segmentName;
+ private final String _segmentCrc;
Review Comment:
This can be returned as `long`
Suggest simply naming it `crc`
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java:
##########
@@ -846,6 +849,106 @@ public String uploadLLCSegment(
}
}
+ /**
+ * Upload a low level consumer segment to segment store and return the
segment download url, crc and
+ * other segment metadata. This endpoint is used when segment store copy is
unavailable for committed
+ * low level consumer segments.
+ * Please note that invocation of this endpoint may cause query performance
to suffer, since we tar up the segment
+ * to upload it.
+ *
+ * @see <a href="https://tinyurl.com/f63ru4sb></a>
+ * @param realtimeTableName table name with type.
+ * @param segmentName name of the segment to be uploaded
+ * @param timeoutMs timeout for the segment upload to the deep-store. If
this is negative, the default timeout
+ * would be used.
+ * @return full url where the segment is uploaded, crc, segmentName. Can add
more segment metadata in the future.
+ * @throws Exception if an error occurred during the segment upload.
+ */
+ @POST
+ @Path("/segments/{realtimeTableName}/{segmentName}/uploadV2")
Review Comment:
I'm not a big fan of `v2`, shall we call it `uploadLLCSegment`?
--
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-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
