Copy blocks with mutable agg intermediates per local receiver in MSE broadcast exchange - #19353
Copy blocks with mutable agg intermediates per local receiver in MSE broadcast exchange#19353yashmayya wants to merge 2 commits into
Conversation
…spool exchange Queries with useSpools=true fail with an NPE (or return corrupted results) when the spooled stage's output contains mutable aggregation intermediate results, e.g. funnelStepDurationStats: Cannot read field "_timestamp" because "o" is null (FunnelStepEvent.compareTo, called from PriorityQueue sift during merge) Root cause: a multi-send (spool) MailboxSendNode fans each block out to the per-receiver-stage exchanges through a plain BroadcastExchange, which routes the very same block instance to every destination. Local (same-JVM) mailboxes then deliver the on-heap rows by reference, so two receiver stages running on the same server observe the same intermediate result objects (e.g. the same PriorityQueue<FunnelStepEvent>). Both consumers mutate those objects: AggregationFunction#merge is allowed to mutate its arguments (the funnel implementation does addAll on the left one), and extractFinalResult drains the queue. Two operator chains mutating/draining the same PriorityQueue corrupts its heap array (null slots -> NPE in compareTo) or silently produces wrong results even without concurrency (the second consumer sees a drained queue). This never happens without spools because hash/singleton exchanges route each row to exactly one destination; only the spool fan-out delivers the same rows to more than one consumer stage. Fix: route multi-send blocks through a new SpoolBroadcastExchange that gives every active receiver stage except the first its own copy of blocks carrying mutable cells (OBJECT columns, i.e. aggregation intermediate results). The copies round-trip the OBJECT cells through the aggregation function intermediate result serde (the same mechanism used to ship them across servers) and are made before the original is handed to a local receiver that could start mutating it. Blocks without OBJECT columns - the common spool case - are still shared by reference with zero overhead, and serialized blocks are read-only so they are shared too.
yashmayya
added
bug
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19353 +/- ##
============================================
+ Coverage 57.71% 67.48% +9.77%
- Complexity 7 1430 +1423
============================================
Files 2659 3486 +827
Lines 159219 224000 +64781
Branches 26113 35339 +9226
============================================
+ Hits 91896 151173 +59277
- Misses 59512 60822 +1310
- Partials 7811 12005 +4194
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
gortiz
left a comment
gortiz
left a comment
There was a problem hiding this comment.
Two structural points on where this fix lives and what copy() actually guarantees. The diagnosis and the mechanics look right — these are about scope and API shape rather than the correctness of what's here.
| @Override | ||
| protected void route(List<SendingMailbox> destinations, MseBlock.Data block) { | ||
| int numDestinations = destinations.size(); | ||
| if (numDestinations == 1 || !mayContainMutableCells(block)) { |
There was a problem hiding this comment.
Spooling isn't really the defect here — in-place mutation of a payload that an exchange handed to more than one consumer is. A plain BROADCAST to a stage with two workers on the same server reproduces this with no spool involved.
Looking at the other exchanges: SingletonExchange asserts a single mailbox, RandomExchange picks one destination per block, and HashExchange partitions rows disjointly into fresh blocks. BroadcastExchange.route is the only other route() that hands the same MseBlock.Data instance to N mailboxes — and MailboxSendOperator creates one SendingMailbox per receiver worker, so any two workers of that stage co-located on one server get in-memory mailboxes sharing the block by reference. Same instance, same two consumers mutating it, same corruption.
So the invariant this PR relies on is a two-way one: no edge that duplicates a block ever carries a type that a downstream operator mutates in place. Today that holds only because the exchange above a partial aggregate is always hash/singleton (a broadcast there would double-count), which is a planner property now asserted in a runtime class's javadoc. Two ways to make it structural instead:
- Move the copy into
BroadcastExchange(or intoBlockExchange#sendBlock, for local mailboxes after the first).mayContainMutableCells()is a scan of a cached array (DataSchemacachesgetStoredColumnDataTypes()), so this is near-free, it covers the multi-worker broadcast case, and it makesSpoolBroadcastExchangeunnecessary along with the unenforced invariant documented inBroadcastExchange. - Fix the mutation contract instead —
AggregationFunction#mergemutating its left argument andextractFinalResultdraining the accumulator are what make a shared block unsafe. That's the deeper fix, but it's a deliberate perf choice, so (1) is the realistic one.
Either way, could you state in the PR description why the multi-worker broadcast case is out of scope, rather than leaving it implied?
There was a problem hiding this comment.
Hm that's a good point, I'll restructure this to remove the spool broadcast exchange and fold it into the regular one. I disagree with the second suggestion of fixing the aggregation function merge mutation contract because as said it'll have a big perf impact.
| return this; | ||
| } | ||
|
|
||
| /// Returns a copy of this block that does not share any mutable cell values with this block. |
There was a problem hiding this comment.
copy() promises more than it delivers: it copies OBJECT cells and shares every other mutable cell.
"a copy of this block that does not share any mutable cell values" isn't quite what the method does. OBJECT is one of roughly ten stored types backed by a mutable Java object: MAP holds a live Map, BYTES a ByteArray over a byte[], and every *_ARRAY type an int[]/long[]/String[]/Object[]. ColumnDataType.UUID's own javadoc in DataSchema already flags that its placeholder "wraps a mutable 16-byte array".
The javadoc's justification — "cells of all other column types are effectively immutable" — is a statement about current operator behaviour, not about the types, and it's exactly the assumption that will rot. The day an operator sorts an array cell in place or merges into a MAP cell, this method keeps silently sharing it and the bug comes back in a form nobody will connect to this code.
Suggestion: have the method take an EnumSet<ColumnDataType> of the types to copy, so the caller — which knows what operators are downstream — makes that decision explicitly, and the assumption becomes an argument someone has to look at rather than a hidden invariant. It also gives the BroadcastExchange case above somewhere to express a wider policy if it ever needs one. At minimum, renaming to copyAggregationIntermediates() / copyObjectColumns() would stop the name overpromising.
There was a problem hiding this comment.
I think rename + clarification is the better option here, I'll do that.
…haring one block across mailboxes
yashmayya
changed the title
3 participants
Problem
Queries fail with an NPE when
useSpools = trueand the spooled stage outputs mutable aggregation intermediate results (for example,funnelStepDurationStats):The failure is not deterministic. The same query can also return wrong results without an error.
Root cause
BroadcastExchangeroutes the same block instance to every destination, and local (same-JVM) mailboxes deliver on-heap rows by reference. It is the only exchange that duplicates a block: hash partitions rows disjointly, and singleton/random pick one destination per block. Two receivers on the same server therefore see the same intermediate result objects — for example, the samePriorityQueue<FunnelStepEvent>.Both receivers mutate these objects.
AggregationFunction#mergecan mutate its arguments (the funnel implementation callsaddAllon the left one), andextractFinalResultdrains the queue. Two operator chains that mutate the samePriorityQueuecorrupt its heap array. This causes null slots and the NPE inFunnelStepEvent#compareTo. Without concurrent access, the second receiver sees a drained queue and silently returns wrong results.Today this only fires through spools: a multi-send node fans each block out to its receiver-stage exchanges through a broadcast, and identical partial-aggregation stages get deduplicated into one spooled stage whose output rows hold the queues. But the same exposure exists for any broadcast to two co-located workers, if a broadcast edge ever carries intermediate results. The fix therefore covers all broadcasts, not only spools.
Fix
BroadcastExchange#routenow copies blocks that carry aggregation intermediate results (OBJECT columns) instead of sharing them:RowHeapDataBlock#copyObjectColumns. It clones the row arrays and copies non-null OBJECT cells through the aggregation function's intermediate result serde — the same mechanism that ships these objects across servers.This makes the safety structural: the one exchange that duplicates blocks never hands the same mutable object to two receivers, for spools and for regular multi-worker broadcasts alike. No planner invariant is needed.
Performance
Testing
BroadcastExchangeTestcovers copy isolation, per-column aggregation function mapping, null cells, remote-before-local ordering, sharing of no-OBJECT and serialized blocks, early termination, and the missing-agg-functions precondition.WindowFunnelTestregression test runs a funnel GROUP BY CTE that feeds two different consumers underuseSpools = true, with the spooled partial aggregation in a leaf stage and in an intermediate stage (below a window function). Without the fix it fails with the NPE above (asPriorityQueue.peek() is nullandIndex -1 out of boundsvariants of the same corruption). The test also asserts that the plan contains a spool, so it cannot pass vacuously if the planner stops deduplicating the shared subtree.Out of scope
Making
AggregationFunction#mergeandextractFinalResultnon-mutating would remove the root hazard, but mutation in place is a deliberate performance choice that the single-stage engine also relies on. This PR keeps that contract and isolates the receivers instead.