xiangfu0 commented on code in PR #19943:
URL: https://github.com/apache/pinot/pull/19943#discussion_r4343866940
##########
pinot-common/src/main/java/org/apache/pinot/common/datatable/DataTableImplV4.java:
##########
@@ -478,14 +481,25 @@ public byte[] toBytes()
// NOTE: We ignore metadata serialization time in
"responseSerializationCpuTimeNs" as it's negligible while
// considering it will bring a lot code complexity.
byte[] metadataBytes = serializeMetadata();
- if (metadataBytes.length != metadataLength) {
- throw new IllegalStateException(
- "Metadata length changed during serialization from " +
metadataLength + " to " + metadataBytes.length);
+ // Readers parse the whole table as one buffer, so the total must fit in
an int as well.
+ long totalLength = sectionsLength + Integer.BYTES + metadataBytes.length;
+ if (totalLength > Integer.MAX_VALUE) {
+ throw new IllegalStateException("Serialized data table size " +
totalLength + " exceeds the 2 GB limit");
}
- byteBuffer.putInt(metadataBytes.length);
- byteBuffer.put(metadataBytes);
- assert !byteBuffer.hasRemaining();
- return bytes;
+ ByteBuffer metadata = ByteBuffer.allocate(Integer.BYTES +
metadataBytes.length);
+ metadata.putInt(metadataBytes.length);
+ metadata.put(metadataBytes);
+
+ CompoundDataBuffer.Builder builder = new
CompoundDataBuffer.Builder(ByteOrder.BIG_ENDIAN, false);
Review Comment:
Good call building with `owner=false` here — it means closing the returned
buffer won't release the table's own fixed/variable sections, so the table
keeps ownership. Combined with the heap-backed builder pages, there's nothing
to release and GC handles the lifetime once Netty/gRPC are done. This is the
key bit that makes sharing the buffers safe.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/QueryScheduler.java:
##########
@@ -74,6 +78,19 @@ public QueryScheduler(PinotConfiguration config, String
instanceId, QueryExecuto
_threadAccountant = threadAccountant;
_latestQueryTime = latestQueryTime;
_resourceManager = resourceManager;
+ // submit() and submitResponse() call each other by default. Fail here
instead of with a StackOverflowError on the
+ // first query.
+ Preconditions.checkState(overrides("submit") ||
overrides("submitResponse"),
+ "%s must override submitResponse(ServerQueryRequest) or the deprecated
submit(ServerQueryRequest)",
+ getClass().getName());
+ }
+
+ private boolean overrides(String methodName) {
+ try {
+ return getClass().getMethod(methodName,
ServerQueryRequest.class).getDeclaringClass() != QueryScheduler.class;
Review Comment:
Minor: this check now turns "subclass overrides neither method" from a
compile error (old abstract `submit`) into a constructor-time
`IllegalStateException`. That's a reasonable trade-off and you call it out in
the PR. `getMethod` also only sees public methods, so a subclass that declared
`submit`/`submitResponse` as protected wouldn't be detected — not possible here
since both are public, just noting the assumption.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingResponseUtils.java:
##########
@@ -47,6 +52,19 @@ public static Server.ServerResponse
getNonStreamingResponse(DataTable dataTable)
private static Server.ServerResponse getResponse(DataTable dataTable, String
responseType)
throws IOException {
return
Server.ServerResponse.newBuilder().putMetadata(Response.MetadataKeys.RESPONSE_TYPE,
responseType)
- .setPayload(ByteString.copyFrom(dataTable.toBytes())).build();
+ .setPayload(toByteString(dataTable.toDataBuffer())).build();
+ }
+
+ /// Wraps the buffers of a serialized data table in one [ByteString] without
copying them. The buffers must not change
+ /// after this call, which holds for a serialized data table.
+ private static ByteString toByteString(DataBuffer dataBuffer) {
+ List<ByteBuffer> buffers = new ArrayList<>();
+ dataBuffer.appendAsByteBuffers(buffers);
+ List<ByteString> byteStrings = new ArrayList<>(buffers.size());
+ for (ByteBuffer buffer : buffers) {
+ byteStrings.add(UnsafeByteOperations.unsafeWrap(buffer));
Review Comment:
`unsafeWrap` is safe here only because these buffers are never mutated after
the table is built, which the comment above correctly notes. Worth keeping in
mind that gRPC may re-serialize the payload on a retry, so the DataTable's
buffers must stay immutable for the whole lifetime of this `ByteString` — which
holds for a serialized DataTable. No change needed, just flagging the contract.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -310,16 +313,21 @@ private void sendErrorResponse(ChannelHandlerContext ctx,
long requestId, String
}
/// Send a response (either query results or exception) back to broker as
response to the query request.
+ ///
+ /// The response buffers are wrapped in one composite Netty buffer without
copying them.
private void sendResponse(ChannelHandlerContext ctx, long requestId, String
tableNameWithType,
- long queryArrivalTimeMs, byte[] serializedDataTable) {
+ long queryArrivalTimeMs, DataBuffer response) {
long sendResponseStartTimeMs = System.currentTimeMillis();
int queryProcessingTimeMs = (int) (sendResponseStartTimeMs -
queryArrivalTimeMs);
-
ctx.writeAndFlush(Unpooled.wrappedBuffer(serializedDataTable)).addListener(f ->
{
+ long responseSizeBytes = response.size();
+ List<ByteBuffer> responseBuffers = new ArrayList<>();
+ response.appendAsByteBuffers(responseBuffers);
+ ctx.writeAndFlush(Unpooled.wrappedBuffer(responseBuffers.toArray(new
ByteBuffer[0]))).addListener(f -> {
Review Comment:
The composite `Unpooled.wrappedBuffer(ByteBuffer[])` avoids the heap copy,
nice. Netty still copies heap→direct before the socket write (as the PR notes),
so the win is removing the extra full-size `byte[]`, not the socket-level copy.
For a zero-length response this would wrap an empty array, but a serialized
DataTable always has a header + metadata so that can't happen — fine as-is.
--
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]
