rohityadav1993 opened a new pull request, #19121:
URL: https://github.com/apache/pinot/pull/19121
`SortedMailboxReceiveOperator` accumulates every row from all senders and
only then sorts, even when
each sender's stream is already sorted on the collation keys — flagged as a
TODO in the class today.
That defeats streaming and makes the receive stage's peak memory
proportional to the whole input.
This is the second bullet of **Challenge 1** in #18667.
### Approach
When the planner can prove each sender emits sorted output, the receiver
performs a k-way heap merge
across mailboxes and emits bounded sorted blocks as data arrives.
- **`BlockingMultiStreamConsumer`** gains a `StreamHandle<T>` abstraction
(`awaitDataOrTerminal()` / `poll()`) so a consumer can hold the head of
each stream without
draining it — the primitive the heap merge needs. A consumer must use
either the handles or
`readMseBlockBlocking()`, never both; the mode is latched.
`relieveSiblingsOnce()` does one non-blocking poll per sibling stream per
refill, so a fast sender
with a disjoint key range cannot sit blocked on a full mailbox holding an
MSE worker thread.
- **`SortedMailboxReceiveOperator`** implements the merge with a per-mailbox
cursor that caches its
head row, and verifies the precondition at runtime: each sender's rows
must be non-decreasing, and
a violation surfaces as an error block naming the offending mailbox
instead of silently wrong
output.
- **`PlanFragmenter` / `PinotLogicalQueryPlanner` / `QueryEnvironment`**
mark a leaf selection
`ORDER BY` sender fragment as sorted-on-sender when the plan shape and
collation match.
This gate **fails closed**: `resolvesToSinglePhysicalTable()` rejects
hybrid tables, logical
tables, unknown tables, and a null `TableCache`. This matters — a scan
over a hybrid table compiles
into two `ServerQueryRequest`s that `LeafOperator` runs concurrently into
one mailbox, producing
two independently sorted runs concatenated, not a sorted stream. Empty
collations are also
rejected, so a `LIMIT` without `ORDER BY` (which compiles to a
collation-less `LogicalSort`) is
never marked sorted.
- **`BaseMailboxReceiveOperator.StatKey.K_WAY_MERGE_USED`** (new) reports
per receive operator, in
the response `stageStats`, whether the merge actually ran. Reporting is
presence-based —
`StatMap` drops `false` booleans, so `kWayMergeUsed: true` appears on the
merge path and nothing
appears on the accumulate-then-sort path. This is the only runtime
discriminator between the two
paths; both render as `type: MAILBOX_RECEIVE`.
### Opt-in query options
| Option | Default | Meaning |
|---|---|---|
| `streamingSortedMailboxReceive` | `false` | Use the k-way merge when the
planner has also proven senders are sorted |
| `streamingSortedMailboxReceiveBlockSize` | `10000` | Rows per emitted
block |
| `sortedSelectionMergeEnabled` (from the parent PR) | `false` |
Additionally required **only** for the plain leaf-selection `ORDER BY` shape,
where the rel plan contains no sort exchange |
`isSortedOnSender` is set either by a rel-level sort exchange that already
declares sender-side
sorting, or — for a plain leaf selection `ORDER BY` — by `PlanFragmenter`
under
`sortedSelectionMergeEnabled`. Only the latter case needs both options.
### Rolling-upgrade note
`K_WAY_MERGE_USED` is appended as the **last** `StatKey` constant. `StatMap`
serializes keys by
ordinal and deserializes without a bounds check, so a peer on a build
predating the key cannot
decode stage stats containing it. Because BOOLEAN keys are written only when
`true`, and the key is
only ever `true` when `streamingSortedMailboxReceive` is on, leaving the
option off (the default)
keeps a mixed-version cluster safe. This is documented on the option
constant. Bounds-checking the
ordinal in `StatMap.merge(DataInput)` is a worthwhile separate fix.
### No behaviour change when the option is off
With `streamingSortedMailboxReceive` unset, the existing
accumulate-then-sort path runs unchanged and
no new stat is emitted. The planner-side marking is additionally a no-op
under
`usePhysicalOptimizer`, since the v2 path does not go through
`PlanFragmenter`.
### Tests
| Test class | Count |
|---|---|
| `SortedMailboxReceiveOperatorTest` (extended) | 36 |
| `PlanFragmenterTest` (new) | 42 cases — accept/reject sender shapes,
hybrid / logical / unknown-table rejects, collation matching, option-off for
every shape, null-cache fail-closed |
| `SortOperatorTest` (extended) | 23 |
| `BlockingMultiStreamConsumerTest` (extended) | 9 |
| full `pinot-query-planner` suite | 1449 |
| `StreamingSortedMailboxReceiveTest` (new integration test) | 4 |
The integration test asserts `kWayMergeUsed` in `stageStats` on the merge
arm and its absence on the
baseline arm, so it cannot pass with the merge path deleted.
### What this buys, honestly
On a local cluster over an 87.8M-doc, 56-segment table, the merge is
confirmed engaged
(`kWayMergeUsed: true` on both join-input receives) and results are
identical to the
accumulate-then-sort path. **It is not a throughput win** for that workload
— enabling the mailbox
merge alone was the slowest sorted variant measured. The benefit is bounded
peak memory in the
receive stage and latency to first row, not wall clock. Row order among
equal collation keys is
unspecified and may differ between the two paths, as with any SQL `ORDER
BY`; the output multiset is
identical.
### Stacking
> Stacked on #19120. The GitHub diff for this PR includes the parent's
commits until the parent merges
> — **review only the top commit.**
>
> **This PR does not compile without its parent.** `QueryEnvironment` calls
> `QueryOptionsUtils.isSortedSelectionMergeEnabled(...)`, which the parent
PR adds. Please do not
> merge this one first.
Part of #18667.
--
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]
