Jackie-Jiang commented on code in PR #19180:
URL: https://github.com/apache/pinot/pull/19180#discussion_r4251542468
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -317,6 +334,134 @@ public Map<String, Object> getSegmentMetadata(
}
}
+ @PUT
+ @Path("segments/{tableNameWithType}/{segmentName}/metadata")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType",
action = Actions.Table.UPLOAD_SEGMENT)
+ @Authenticate(AccessType.UPDATE)
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Update the custom map in the ZK metadata for a
segment",
+ notes = "Updates only the segment ZK metadata custom map without
uploading or refreshing the segment")
+ @ApiResponses(value = {
+ @ApiResponse(code = 200, message = "Success"),
+ @ApiResponse(code = 400, message = "Invalid table name, CRC, or custom
map modifier"),
+ @ApiResponse(code = 404, message = "Table or segment not found"),
+ @ApiResponse(code = 409, message = "Segment metadata changed
concurrently"),
+ @ApiResponse(code = 412, message = "Segment CRC does not match"),
+ @ApiResponse(code = 413, message = "Custom map modifier or resulting ZK
metadata is too large")
+ })
+ public SuccessResponse updateSegmentZKMetadataCustomMap(
+ @ApiParam(value = "Table name with type", required = true, example =
"myTable_OFFLINE")
+ @PathParam("tableNameWithType") String tableNameWithType,
+ @ApiParam(value = "Name of the segment", required = true)
@PathParam("segmentName") @Encoded String segmentName,
+ @ApiParam(value = "Expected segment CRC", required = true)
@HeaderParam(HttpHeaders.IF_MATCH)
+ String expectedCrcString,
+ @ApiParam(value = "Custom map modifier", required = true) InputStream
customMapModifierInputStream,
+ @Context HttpHeaders headers) {
+ tableNameWithType = DatabaseUtils.translateTableName(tableNameWithType,
headers);
+ segmentName = decodePathSegment(segmentName);
+ if (tableNameWithType.contains("/") || tableNameWithType.contains("\\") ||
segmentName.isEmpty()
+ || segmentName.contains("/") || segmentName.contains("\\")) {
+ throw new ControllerApplicationException(LOGGER, "Invalid table or
segment path", Status.BAD_REQUEST);
+ }
+ if (TableNameBuilder.getTableTypeFromTableName(tableNameWithType) == null)
{
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Table type not provided with table name: %s",
tableNameWithType), Status.BAD_REQUEST);
+ }
+ String customMapModifierJson =
readCustomMapModifier(customMapModifierInputStream);
+
+ long expectedCrc;
+ try {
+ expectedCrc = Long.parseLong(expectedCrcString);
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, "Missing or invalid
If-Match segment CRC", Status.BAD_REQUEST,
+ e);
+ }
+
+ SegmentZKMetadataCustomMapModifier customMapModifier;
+ try {
+ customMapModifier = new
SegmentZKMetadataCustomMapModifier(customMapModifierJson);
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, "Invalid segment ZK
metadata custom map modifier",
+ Status.BAD_REQUEST, e);
+ }
+
+ ZNRecord segmentMetadataRecord =
+
_pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType,
segmentName);
+ if (segmentMetadataRecord == null) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Failed to find segment: %s in table: %s",
segmentName, tableNameWithType), Status.NOT_FOUND);
+ }
+ SegmentZKMetadata segmentZKMetadata = new
SegmentZKMetadata(segmentMetadataRecord);
+ if (segmentZKMetadata.getSegmentUploadStartTime() > 0) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment upload is in progress for segment: %s in
table: %s", segmentName,
+ tableNameWithType), Status.CONFLICT);
+ }
+ if (segmentZKMetadata.getCrc() != expectedCrc) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment CRC does not match for segment: %s in table:
%s", segmentName, tableNameWithType),
+ Status.PRECONDITION_FAILED);
+ }
+
segmentZKMetadata.setCustomMap(customMapModifier.modifyMap(segmentZKMetadata.getCustomMap()));
+ validateSegmentZKMetadataSize(segmentZKMetadata);
+ if
(!_pinotHelixResourceManager.updateZkMetadataWithoutDataChange(tableNameWithType,
segmentZKMetadata,
+ segmentMetadataRecord.getVersion())) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment metadata changed concurrently for segment: %s
in table: %s", segmentName,
+ tableNameWithType), Status.CONFLICT);
+ }
+ return new SuccessResponse(
+ String.format("Successfully updated ZK metadata for segment: %s in
table: %s", segmentName,
+ tableNameWithType));
+ }
+
+ private static String readCustomMapModifier(@Nullable InputStream
inputStream) {
+ if (inputStream == null) {
+ throw new ControllerApplicationException(LOGGER, "Missing segment ZK
metadata custom map modifier",
+ Status.BAD_REQUEST);
+ }
+ try {
+ byte[] bytes = ByteStreams.toByteArray(ByteStreams.limit(inputStream,
MAX_CUSTOM_MAP_MODIFIER_SIZE_BYTES + 1L));
Review Comment:
Non-blocking question: what is the basis for the new 64 KiB request limit? A
bounded read is reasonable, but this is a new API-specific limit, separate from
the size limit on the resulting ZK record, and existing upload callers do not
use this helper. Could we document the supported request limit and explain why
64 KiB is appropriate, or reuse a controller request-size policy if one
applies? JSON shape/type validation should remain regardless.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -317,6 +334,134 @@ public Map<String, Object> getSegmentMetadata(
}
}
+ @PUT
+ @Path("segments/{tableNameWithType}/{segmentName}/metadata")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType",
action = Actions.Table.UPLOAD_SEGMENT)
+ @Authenticate(AccessType.UPDATE)
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Update the custom map in the ZK metadata for a
segment",
+ notes = "Updates only the segment ZK metadata custom map without
uploading or refreshing the segment")
+ @ApiResponses(value = {
+ @ApiResponse(code = 200, message = "Success"),
+ @ApiResponse(code = 400, message = "Invalid table name, CRC, or custom
map modifier"),
+ @ApiResponse(code = 404, message = "Table or segment not found"),
+ @ApiResponse(code = 409, message = "Segment metadata changed
concurrently"),
+ @ApiResponse(code = 412, message = "Segment CRC does not match"),
+ @ApiResponse(code = 413, message = "Custom map modifier or resulting ZK
metadata is too large")
+ })
+ public SuccessResponse updateSegmentZKMetadataCustomMap(
+ @ApiParam(value = "Table name with type", required = true, example =
"myTable_OFFLINE")
+ @PathParam("tableNameWithType") String tableNameWithType,
+ @ApiParam(value = "Name of the segment", required = true)
@PathParam("segmentName") @Encoded String segmentName,
+ @ApiParam(value = "Expected segment CRC", required = true)
@HeaderParam(HttpHeaders.IF_MATCH)
+ String expectedCrcString,
+ @ApiParam(value = "Custom map modifier", required = true) InputStream
customMapModifierInputStream,
+ @Context HttpHeaders headers) {
+ tableNameWithType = DatabaseUtils.translateTableName(tableNameWithType,
headers);
+ segmentName = decodePathSegment(segmentName);
+ if (tableNameWithType.contains("/") || tableNameWithType.contains("\\") ||
segmentName.isEmpty()
Review Comment:
Non-blocking question: can the table/segment path checks reuse shared
validation, or can we document why a local check is needed? Rejecting '/'
before constructing the ZK path has a concrete safety purpose and should be
retained. The empty-name and backslash rules are additional name policy
introduced only for this API; existing metadata GET/reset paths do not have
this same guard. Please clarify the intended contract and whether this belongs
in a shared path-component validator, taking care to preserve valid
existing/realtime segment names. A broad cleanup of existing endpoints need not
block this API.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -317,6 +334,134 @@ public Map<String, Object> getSegmentMetadata(
}
}
+ @PUT
+ @Path("segments/{tableNameWithType}/{segmentName}/metadata")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType",
action = Actions.Table.UPLOAD_SEGMENT)
+ @Authenticate(AccessType.UPDATE)
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Update the custom map in the ZK metadata for a
segment",
+ notes = "Updates only the segment ZK metadata custom map without
uploading or refreshing the segment")
+ @ApiResponses(value = {
+ @ApiResponse(code = 200, message = "Success"),
+ @ApiResponse(code = 400, message = "Invalid table name, CRC, or custom
map modifier"),
+ @ApiResponse(code = 404, message = "Table or segment not found"),
+ @ApiResponse(code = 409, message = "Segment metadata changed
concurrently"),
+ @ApiResponse(code = 412, message = "Segment CRC does not match"),
+ @ApiResponse(code = 413, message = "Custom map modifier or resulting ZK
metadata is too large")
+ })
+ public SuccessResponse updateSegmentZKMetadataCustomMap(
+ @ApiParam(value = "Table name with type", required = true, example =
"myTable_OFFLINE")
+ @PathParam("tableNameWithType") String tableNameWithType,
+ @ApiParam(value = "Name of the segment", required = true)
@PathParam("segmentName") @Encoded String segmentName,
+ @ApiParam(value = "Expected segment CRC", required = true)
@HeaderParam(HttpHeaders.IF_MATCH)
+ String expectedCrcString,
+ @ApiParam(value = "Custom map modifier", required = true) InputStream
customMapModifierInputStream,
+ @Context HttpHeaders headers) {
+ tableNameWithType = DatabaseUtils.translateTableName(tableNameWithType,
headers);
+ segmentName = decodePathSegment(segmentName);
+ if (tableNameWithType.contains("/") || tableNameWithType.contains("\\") ||
segmentName.isEmpty()
+ || segmentName.contains("/") || segmentName.contains("\\")) {
+ throw new ControllerApplicationException(LOGGER, "Invalid table or
segment path", Status.BAD_REQUEST);
+ }
+ if (TableNameBuilder.getTableTypeFromTableName(tableNameWithType) == null)
{
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Table type not provided with table name: %s",
tableNameWithType), Status.BAD_REQUEST);
+ }
+ String customMapModifierJson =
readCustomMapModifier(customMapModifierInputStream);
+
+ long expectedCrc;
+ try {
+ expectedCrc = Long.parseLong(expectedCrcString);
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, "Missing or invalid
If-Match segment CRC", Status.BAD_REQUEST,
+ e);
+ }
+
+ SegmentZKMetadataCustomMapModifier customMapModifier;
+ try {
+ customMapModifier = new
SegmentZKMetadataCustomMapModifier(customMapModifierJson);
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, "Invalid segment ZK
metadata custom map modifier",
+ Status.BAD_REQUEST, e);
+ }
+
+ ZNRecord segmentMetadataRecord =
+
_pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType,
segmentName);
+ if (segmentMetadataRecord == null) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Failed to find segment: %s in table: %s",
segmentName, tableNameWithType), Status.NOT_FOUND);
+ }
+ SegmentZKMetadata segmentZKMetadata = new
SegmentZKMetadata(segmentMetadataRecord);
+ if (segmentZKMetadata.getSegmentUploadStartTime() > 0) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment upload is in progress for segment: %s in
table: %s", segmentName,
+ tableNameWithType), Status.CONFLICT);
+ }
+ if (segmentZKMetadata.getCrc() != expectedCrc) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment CRC does not match for segment: %s in table:
%s", segmentName, tableNameWithType),
+ Status.PRECONDITION_FAILED);
+ }
+
segmentZKMetadata.setCustomMap(customMapModifier.modifyMap(segmentZKMetadata.getCustomMap()));
+ validateSegmentZKMetadataSize(segmentZKMetadata);
+ if
(!_pinotHelixResourceManager.updateZkMetadataWithoutDataChange(tableNameWithType,
segmentZKMetadata,
+ segmentMetadataRecord.getVersion())) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment metadata changed concurrently for segment: %s
in table: %s", segmentName,
+ tableNameWithType), Status.CONFLICT);
+ }
+ return new SuccessResponse(
+ String.format("Successfully updated ZK metadata for segment: %s in
table: %s", segmentName,
+ tableNameWithType));
+ }
+
+ private static String readCustomMapModifier(@Nullable InputStream
inputStream) {
+ if (inputStream == null) {
+ throw new ControllerApplicationException(LOGGER, "Missing segment ZK
metadata custom map modifier",
+ Status.BAD_REQUEST);
+ }
+ try {
+ byte[] bytes = ByteStreams.toByteArray(ByteStreams.limit(inputStream,
MAX_CUSTOM_MAP_MODIFIER_SIZE_BYTES + 1L));
+ if (bytes.length > MAX_CUSTOM_MAP_MODIFIER_SIZE_BYTES) {
+ throw new ControllerApplicationException(LOGGER, "Segment ZK metadata
custom map modifier is too large",
+ Status.REQUEST_ENTITY_TOO_LARGE);
+ }
+ return new String(bytes, StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new ControllerApplicationException(LOGGER, "Failed to read segment
ZK metadata custom map modifier",
+ Status.BAD_REQUEST, e);
+ }
+ }
+
+ private static String decodePathSegment(String pathSegment) {
+ // URIUtils uses form decoding, where '+' represents a space. In a path,
'+' is a literal character. Jersey can
+ // supply either the raw or decoded path parameter depending on the
request URI, so protect literal plus signs
+ // before decoding percent escapes.
+ return URIUtils.decode(pathSegment.replace("+", "%2B"));
Review Comment:
Non-blocking question: why does this endpoint need different segment-name
decoding from the existing metadata GET/reset endpoints? Those use @Encoded
with URIUtils.decode; this helper additionally preserves a raw '+'. A correctly
encoded literal plus (%2B) already decodes correctly with URIUtils.decode, and
the new client helper sends %2B. Please explain the request/Jersey case that
requires the extra replacement, particularly the claim that an @Encoded
parameter can already be decoded. If path decoding needs a general fix, could
we use a shared path-specific utility with focused tests for raw '+', %2B, %20,
and %25, while checking compatibility of existing endpoints? The client's fix
to encode each component only once should be preserved.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java:
##########
@@ -317,6 +334,134 @@ public Map<String, Object> getSegmentMetadata(
}
}
+ @PUT
+ @Path("segments/{tableNameWithType}/{segmentName}/metadata")
+ @Authorize(targetType = TargetType.TABLE, paramName = "tableNameWithType",
action = Actions.Table.UPLOAD_SEGMENT)
+ @Authenticate(AccessType.UPDATE)
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Update the custom map in the ZK metadata for a
segment",
+ notes = "Updates only the segment ZK metadata custom map without
uploading or refreshing the segment")
+ @ApiResponses(value = {
+ @ApiResponse(code = 200, message = "Success"),
+ @ApiResponse(code = 400, message = "Invalid table name, CRC, or custom
map modifier"),
+ @ApiResponse(code = 404, message = "Table or segment not found"),
+ @ApiResponse(code = 409, message = "Segment metadata changed
concurrently"),
+ @ApiResponse(code = 412, message = "Segment CRC does not match"),
+ @ApiResponse(code = 413, message = "Custom map modifier or resulting ZK
metadata is too large")
+ })
+ public SuccessResponse updateSegmentZKMetadataCustomMap(
+ @ApiParam(value = "Table name with type", required = true, example =
"myTable_OFFLINE")
+ @PathParam("tableNameWithType") String tableNameWithType,
+ @ApiParam(value = "Name of the segment", required = true)
@PathParam("segmentName") @Encoded String segmentName,
+ @ApiParam(value = "Expected segment CRC", required = true)
@HeaderParam(HttpHeaders.IF_MATCH)
+ String expectedCrcString,
+ @ApiParam(value = "Custom map modifier", required = true) InputStream
customMapModifierInputStream,
+ @Context HttpHeaders headers) {
+ tableNameWithType = DatabaseUtils.translateTableName(tableNameWithType,
headers);
+ segmentName = decodePathSegment(segmentName);
+ if (tableNameWithType.contains("/") || tableNameWithType.contains("\\") ||
segmentName.isEmpty()
+ || segmentName.contains("/") || segmentName.contains("\\")) {
+ throw new ControllerApplicationException(LOGGER, "Invalid table or
segment path", Status.BAD_REQUEST);
+ }
+ if (TableNameBuilder.getTableTypeFromTableName(tableNameWithType) == null)
{
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Table type not provided with table name: %s",
tableNameWithType), Status.BAD_REQUEST);
+ }
+ String customMapModifierJson =
readCustomMapModifier(customMapModifierInputStream);
+
+ long expectedCrc;
+ try {
+ expectedCrc = Long.parseLong(expectedCrcString);
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, "Missing or invalid
If-Match segment CRC", Status.BAD_REQUEST,
+ e);
+ }
+
+ SegmentZKMetadataCustomMapModifier customMapModifier;
+ try {
+ customMapModifier = new
SegmentZKMetadataCustomMapModifier(customMapModifierJson);
+ } catch (Exception e) {
+ throw new ControllerApplicationException(LOGGER, "Invalid segment ZK
metadata custom map modifier",
+ Status.BAD_REQUEST, e);
+ }
+
+ ZNRecord segmentMetadataRecord =
+
_pinotHelixResourceManager.getSegmentMetadataZnRecord(tableNameWithType,
segmentName);
+ if (segmentMetadataRecord == null) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Failed to find segment: %s in table: %s",
segmentName, tableNameWithType), Status.NOT_FOUND);
+ }
+ SegmentZKMetadata segmentZKMetadata = new
SegmentZKMetadata(segmentMetadataRecord);
+ if (segmentZKMetadata.getSegmentUploadStartTime() > 0) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment upload is in progress for segment: %s in
table: %s", segmentName,
+ tableNameWithType), Status.CONFLICT);
+ }
+ if (segmentZKMetadata.getCrc() != expectedCrc) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment CRC does not match for segment: %s in table:
%s", segmentName, tableNameWithType),
+ Status.PRECONDITION_FAILED);
+ }
+
segmentZKMetadata.setCustomMap(customMapModifier.modifyMap(segmentZKMetadata.getCustomMap()));
+ validateSegmentZKMetadataSize(segmentZKMetadata);
+ if
(!_pinotHelixResourceManager.updateZkMetadataWithoutDataChange(tableNameWithType,
segmentZKMetadata,
+ segmentMetadataRecord.getVersion())) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Segment metadata changed concurrently for segment: %s
in table: %s", segmentName,
+ tableNameWithType), Status.CONFLICT);
+ }
+ return new SuccessResponse(
+ String.format("Successfully updated ZK metadata for segment: %s in
table: %s", segmentName,
+ tableNameWithType));
+ }
+
+ private static String readCustomMapModifier(@Nullable InputStream
inputStream) {
+ if (inputStream == null) {
+ throw new ControllerApplicationException(LOGGER, "Missing segment ZK
metadata custom map modifier",
+ Status.BAD_REQUEST);
+ }
+ try {
+ byte[] bytes = ByteStreams.toByteArray(ByteStreams.limit(inputStream,
MAX_CUSTOM_MAP_MODIFIER_SIZE_BYTES + 1L));
+ if (bytes.length > MAX_CUSTOM_MAP_MODIFIER_SIZE_BYTES) {
+ throw new ControllerApplicationException(LOGGER, "Segment ZK metadata
custom map modifier is too large",
+ Status.REQUEST_ENTITY_TOO_LARGE);
+ }
+ return new String(bytes, StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new ControllerApplicationException(LOGGER, "Failed to read segment
ZK metadata custom map modifier",
+ Status.BAD_REQUEST, e);
+ }
+ }
+
+ private static String decodePathSegment(String pathSegment) {
+ // URIUtils uses form decoding, where '+' represents a space. In a path,
'+' is a literal character. Jersey can
+ // supply either the raw or decoded path parameter depending on the
request URI, so protect literal plus signs
+ // before decoding percent escapes.
+ return URIUtils.decode(pathSegment.replace("+", "%2B"));
+ }
+
+ private static void validateSegmentZKMetadataSize(SegmentZKMetadata
segmentZKMetadata) {
+ int zkJuteMaxBufferSizeBytes =
+ Integer.getInteger(ZkSystemPropertyKeys.JUTE_MAXBUFFER,
DEFAULT_ZK_JUTE_MAX_BUFFER_SIZE_BYTES);
+ if (zkJuteMaxBufferSizeBytes <= 0) {
+ zkJuteMaxBufferSizeBytes = DEFAULT_ZK_JUTE_MAX_BUFFER_SIZE_BYTES;
+ }
+ int safeJuteRecordSizeBytes = (int) ((long) zkJuteMaxBufferSizeBytes *
ZK_RECORD_SIZE_LIMIT_PERCENT / 100);
Review Comment:
Non-blocking question: do we need a separate full-record size policy here?
The normal Helix write already serializes the ZNRecord and enforces its
serializer write limit, accounting for compression. This precheck serializes it
again before the actual write and additionally caps it at 90% of
jute.maxbuffer, making this endpoint stricter than existing metadata writes.
What failure case requires the 10% reserve and early serialization? If the goal
is only to return a clear 413, could we rely on the existing limit and
translate the size failure appropriately? If the extra policy is necessary,
please document the rationale and consider shared enforcement. Removing the
precheck should not turn an oversized request into a misleading 409 concurrency
error.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
