MSE: spool fan-out copies aggregation intermediates even when every receiver stage is remote
- Dominant language
- Java
- Stars
- 6.1k
- Forks
- 1.5k
- Avg merge
- 2d 55m
- Merged PRs (30d)
- 182
Description
## Background
[PR #19353](https://github.com/apache/pinot/pull/19353) fixed an NPE that occurred with `useSpools = true`. `BroadcastExchange` is the only exchange that routes one block instance to more than one destination. It now copies blocks that carry aggregation intermediate results in `OBJECT` columns. Local mailboxes deliver on-heap blocks by reference, and downstream operators mutate those objects in place.
That fix skips the copy for remote destinations, because a remote destination serializes the block on the calling thread and never shares the mutable object. This issue is the follow-up to [a review comment on that PR](https://github.com/apache/pinot/pull/19353#discussion_r3902651419): the skip never applies to spools, which are the case the fix was written for.
## Problem 1: the remote skip never applies to a spool
A multi-send (spool) node builds one inner exchange per receiver stage. It wraps each inner exchange as a `BlockExchange.BlockExchangeSendingMailbox`. `isLocal()` on that wrapper returns `true` unconditionally, whatever the inner exchange's own mailboxes are. Every outer destination of a spool is such a wrapper, so `BroadcastExchange#route` takes the local branch for all of them.
A receiver stage with no worker on the sending server therefore still gets a copy. The inner exchange serializes that copy immediately, on the same thread, and no receiver ever mutates it. Each copy costs one serialization plus one deserialization of every non-null `OBJECT` cell, for each extra receiver stage.
Ordinary broadcast edges do not carry `OBJECT` columns today, so the remote branch is currently unreachable in practice.
## Problem 2: `isLocal()` carries two meanings
- `BlockExchange#sendBlock` reads it as "do not split the block here".
- `BroadcastExchange#route` reads it as "this destination can give the block to a receiver by reference".
The two answers differ for `BlockExchangeSendingMailbox`. Only the conservative direction keeps the code correct: an unnecessary copy is safe, a missing copy corrupts data.
## Suggested fix
1. Add a predicate to `SendingMailbox`, for example `deliversByReference()`:
- `InMemorySendingMailbox` returns `true`.
- `GrpcSendingMailbox` returns `false`.
- `BlockExchangeSendingMailbox` returns `true` if any mailbox of its inner exchange returns `true`.
2. Use the new predicate in `BroadcastExchange#route`. Then `isLocal()` keeps only the meaning that `sendBlock` needs.
The delegation must be an OR over the inner mailboxes. An inner `HashExchange` builds a new block for each destination, but those blocks hold the same cell objects. One local worker in a receiver stage is therefore enough to require a copy.
## Also: state the serialization contract on the interface
`BroadcastExchange#route` gives the original block to remote destinations before it gives the block to a local one. This is safe only because `send(MseBlock.Data)` serializes the block before it returns. `GrpcSendingMailbox` documents this, but `route` depends on it for every implementation that does not deliver by reference. The requirement belongs on `SendingMailbox#send`, where implementers can see it.
## Related
`BlockExchangeSendingMailbox#isLocal()` also makes the outer exchange skip the splitter, and the inner exchanges get `BlockSplitter.NO_OP` (`MailboxSendOperator#getBlockExchange`). Multi-send blocks to remote receivers are therefore never split against `MAX_MAILBOX_CONTENT_SIZE_BYTES`. This is a separate defect with the same cause, and one change can correct both.
Contributor guide
Research direction
Start at BroadcastExchange#route and SendingMailbox#send, then inspect isLocal() in BlockExchangeSendingMailbox and the delivery behavior of InMemorySendingMailbox and GrpcSendingMailbox. Trace how MailboxSendOperator#getBlockExchange handles the splitter for multi-send blocks. Done means reference delivery, serialization guarantees, and remote spool splitting follow the stated contracts without unnecessary OBJECT-cell copies.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems, performance
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 64/100