apache / apache/pinot

MSE: spool fan-out copies aggregation intermediates even when every receiver stage is remote

Open
#19,427 0 comments 0 reactions 0 assignees View on GitHub
enhancement multi-stage performance
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.