yashmayya opened a new pull request, #19353: URL: https://github.com/apache/pinot/pull/19353
## Problem Queries fail with an NPE when `useSpools = true` and the spooled stage outputs mutable aggregation intermediate results (for example, `funnelStepDurationStats`): ``` Cannot read field "_timestamp" because "o" is null ``` The failure is not deterministic. The same query can also return wrong results without an error. ## Root cause A spool is one `MailboxSendNode` that sends to more than one receiver stage. The send operator fans each block out through a plain `BroadcastExchange`. This exchange routes the same block instance to every receiver stage. Local (same-JVM) mailboxes deliver on-heap rows by reference. As a result, two receiver stages on the same server see the same intermediate result objects — for example, the same `PriorityQueue<FunnelStepEvent>`. Both consumers mutate these objects. `AggregationFunction#merge` can mutate its arguments (the funnel implementation calls `addAll` on the left one), and `extractFinalResult` drains the queue. Two operator chains that mutate the same `PriorityQueue` corrupt its heap array. This causes null slots and the NPE in `FunnelStepEvent#compareTo`. Without concurrent access, the second consumer sees a drained queue and silently returns wrong results. This cannot happen without spools. Hash and 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 Multi-send (spool) nodes now route blocks through a new `SpoolBroadcastExchange`. For on-heap blocks with OBJECT columns (the columns that hold aggregation intermediate results), it sends the original block to the first active receiver stage, and a copy to each other receiver stage. The new `RowHeapDataBlock#copy` clones the row arrays and copies non-null OBJECT cells through the aggregation function's intermediate result serde. This is the same mechanism that ships these objects across servers. The exchange makes the copies before it hands the original block to a local receiver, because that receiver can start to mutate it immediately. ## Performance - Only spool multi-send nodes are affected. All other exchanges are unchanged. - Blocks without OBJECT columns (plain projections — the common spool case) are still shared by reference. The only added cost is one scan of the cached column-type array per block. - Serialized blocks are read-only, so they are also shared. - When the copy applies: N receiver stages need N-1 copies (fan-out is usually 2, so one copy). Only non-null OBJECT cells pay the serde round trip. Keys and other immutable cells are shared. The blocks on this edge hold post-aggregation output (one row per group per server), and remote consumers already pay the same serde cost today. ## Testing - New `SpoolBroadcastExchangeTest` covers copy isolation, per-column aggregation function mapping, null cells, sharing of no-OBJECT and serialized blocks, early termination, and the missing-agg-functions precondition. - A new `WindowFunnelTest` regression test runs a funnel GROUP BY CTE that feeds two different consumers under `useSpools = 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 (as `PriorityQueue.peek() is null` and `Index -1 out of bounds` variants 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. -- 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]
