gortiz opened a new pull request, #19943:
URL: https://github.com/apache/pinot/pull/19943
## Summary
This PR removes the last full-size array from the single-stage (v1) server
response path. The server now sends the pages of a `DataTable` to Netty and
gRPC without copying them into one `byte[]`. The wire format does not change.
This is the second of three steps.
1. #19906 (merged): paged builder sections and an exact-size `toBytes()`.
2. This PR: send the pages to Netty and gRPC without a contiguous `byte[]`.
3. Next: check `maxServerResponseSizeBytes` while the response is built, and
sample the query resource accountant by bytes written.
## Background
Since #19906, the builder writes the fixed and variable size sections in 16
KB pages. But `DataTable.toBytes()` still copies them into one array of the
full response size, because the scheduler API returns
`ListenableFuture<byte[]>`, and the Netty and gRPC send paths take a `byte[]`.
For a large response this array is a G1 humongous object.
```
Before: pages ──copy──> byte[N] ──> Unpooled.wrappedBuffer(byte[])
(Netty)
byte[N] ──copy──> ByteString.copyFrom(byte[])
(gRPC)
After: pages ──> DataBuffer ──> Unpooled.wrappedBuffer(ByteBuffer...)
(Netty)
└─> ByteString rope of unsafeWrap(ByteBuffer)
(gRPC)
```
## Changes
- **`DataTable.toDataBuffer()`** (new). It returns the same bytes as
`toBytes()` as a `DataBuffer` that can be split in several buffers. The default
wraps `toBytes()`. `DataTableImplV4` returns one small buffer for the header
and the small sections, the fixed and variable size sections as they are, and
one small buffer for the metadata. `toBytes()` is now a copy of
`toDataBuffer()`.
- **`QueryScheduler`.** It gets a parallel API that returns `DataBuffer`:
| New | Deprecated, kept for compatibility |
|---|---|
| `submitResponse(ServerQueryRequest)` | `submit(ServerQueryRequest)` |
| `createResponseFutureTask(...)` | `createQueryFutureTask(...)` |
| `processQueryAndSerializeResponse(...)` |
`processQueryAndSerialize(...)` |
| `errorResponse(...)`, `shuttingDownResponse(...)`,
`outOfCapacityResponse(...)` | `immediateErrorResponse(...)`,
`shuttingDown(...)`, `outOfCapacity(...)` |
`submit()` is no longer abstract. By default, `submit()` copies the result
of `submitResponse()` into an array, and `submitResponse()` wraps the result of
`submit()`. A subclass must override at least one of the two. Thus a scheduler
outside this repository that overrides only `submit()` keeps working without
changes, but it still sends one contiguous array until it moves to
`submitResponse()`.
- **Schedulers.** `FCFSQueryScheduler`, `PriorityScheduler`,
`BinaryWorkloadScheduler` and `WorkloadScheduler` override `submitResponse()`.
`SchedulerQueryContext` holds a `SettableFuture<DataBuffer>`.
- **`InstanceRequestHandler`.** It calls `submitResponse()` and sends the
response with `Unpooled.wrappedBuffer(ByteBuffer...)`, which makes one
composite buffer without copies.
- **`StreamingResponseUtils` (gRPC).** It wraps each buffer with
`UnsafeByteOperations.unsafeWrap` and joins them with
`ByteString.copyFrom(Iterable<ByteString>)`. Despite its name, this method
builds a balanced rope and does not copy. The old code used
`ByteString.copyFrom(byte[])`, which was a second full-size copy.
- **`PageCacheWarmupServerQueryExecutor`.** It calls `submitResponse()`.
## Compatibility
- **Wire format.** The bytes on the wire do not change, so brokers and
servers of different versions work together.
- **Netty send.** Netty still copies a heap buffer into a direct buffer
before it writes to the socket. This copy happened before too, with the
`byte[]`, and it is off-heap.
- **Response serialization metric.** `responseSerializationCpuTimeNs` and
`responseSerMemAllocatedBytes` now measure `toDataBuffer()`, which no longer
includes a full-size copy. Thus these values are lower for large responses.
- **Java API.** All existing `QueryScheduler` methods stay, so subclasses
compile without changes. They get deprecation warnings.
- A subclass of `QueryScheduler` that overrides only `submit()` keeps
working through the default `submitResponse()`.
- A subclass of one of the built-in schedulers must override
`submitResponse()`. The built-in schedulers override it, and the server calls
it, so an override of only `submit()` there is not called. I found no such
subclass.
- The constructor now fails if a subclass overrides neither `submit()` nor
`submitResponse()`. Before, `submit()` was abstract and the compiler caught
this case.
- `SchedulerQueryContext.getResultFuture()` and `setResultFuture()` now
use `DataBuffer` instead of `byte[]`. Only the built-in priority schedulers use
this class.
## Tests
- `QueryRoutingTest#testPagedResponse` (new): a server sends a 2 MB table
split in more than 100 pages through Netty. The broker receives the same rows
and the same response size.
- `StreamingResponseUtilsTest` (new): the gRPC payload of a paged table
deserializes to the same rows, and it survives protobuf encoding and parsing.
- `DataTableSerDeTest#testValuesSpanningPages`: `toDataBuffer()` returns no
buffer larger than one page, and its bytes deserialize to the same table.
- `QuerySchedulerResponseApiTest` (new): a scheduler that overrides only
`submit()` works through `submitResponse()`, a scheduler that overrides only
`submitResponse()` works through `submit()`, and a scheduler that overrides
neither fails in the constructor.
- `QueryRoutingTest` and `PageCacheWarmupServerQueryExecutorTest` mock
`submitResponse()`. `InstanceRequestHandlerTest` and
`QuerySchedulerFactoryTest` still override only the deprecated `submit()`, so
they cover the compatibility path.
--
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]
