Hi all,
I would like to discuss an MPP DataExchange change for queries that produce a
very large `TsBlock`, especially when one `Object` or BLOB value is tens or
hundreds of MB. The proposed implementation limits the payload of each
DataExchange response and fetches one serialized `TsBlock` over multiple RPCs
when necessary.
## Background
The current `getDataBlock` path serializes every requested `TsBlock` and puts
the complete binary values into one `TGetDataBlockResponse`. This is normally
safe because upstream operators produce relatively small blocks. However, a
single large value must remain in one row, so `max_tsblock_size_in_bytes`
cannot always prevent an individual `TsBlock` from becoming very large.
The complete response can temporarily exist in the Thrift frame buffer, the
client response object, and the SourceHandle. Because MPP clients are borrowed
from a pool and the underlying connections remain alive, a large response can
also leave an oversized RPC buffer retained by the pooled connection and lead
to memory issues.
## Proposed change
The change keeps the existing `getDataBlock` RPC. It does not add a new RPC
method.
The DataNode configuration is:
```properties
# The maximum payload size of one MPP data exchange RPC response
# effectiveMode: hot_reload
# Datatype: long, Unit: byte
mpp_data_exchange_max_payload_size_in_bytes=4,194,304
mpp_data_exchange_max_payload_size_in_bytes=4,194,304
```
This value limits the sum of the binary payloads in one response.
The Thrift structures are extended with one optional `i32 offset` field:
```thrift
struct TGetDataBlockRequest {
1: required TFragmentInstanceId sourceFragmentInstanceId
2: required i32 startSequenceId
3: required i32 endSequenceId
4: required i32 index
5: optional i32 offset
}
struct TGetDataBlockResponse {
1: required list<binary> tsBlocks
2: optional i32 offset
}
```
The request offset is the starting byte position in the serialized `TsBlock`
for `startSequenceId`. The first request carries offset `0`; subsequent
requests use the offset returned by the preceding response.
The response contains complete serialized `TsBlock`s and, at most, one trailing
fragment. If the last list element is a fragment, `response.offset` is set to
the next byte position. If `response.offset` is not set, all returned elements
are complete blocks.
For a sequence range `[startSequenceId, endSequenceId)`, the server fills the
response up to the configured payload limit:
```text
for each sequence:
load or reuse the serialized TsBlock
calculate the remaining bytes from request.offset for the first sequence
append the complete block when it fits
otherwise append the remaining payload as the last fragment
return the next offset
```
`SinkChannel` serializes a sequence only once and stores the serialized
`ByteBuffer` until the corresponding TsBlock is acknowledged, aborted, or
closed. Later fragment requests use `duplicate()`/`slice()` and do not
serialize the TsBlock again.
`SourceHandle` uses `getDataBlockWithFragments` for remote pulls. It keeps
`nextSequenceId`, the current `offset`, completed blocks, and the current
partial block outside the RPC retry loop. Consequently, a failed retry does not
fetch completed blocks or already received fragments again.
## Compatibility and scope
- No new RPC method is introduced.
- A request without `offset` retains the existing complete-block response path
for compatibility during rollout.
- New clients always set offset, including `0`, and use the bounded-payload
path.
- The offset is `i32`, matching the maximum addressable serialized `ByteBuffer`
range in Java.
- The change affects MPP DataExchange transport only.
I would appreciate feedback from yours.
Thanks,
Weihao Li
