Add streaming selection ORDER BY combine for physically sorted segments

[rohityadav1993]

feature performance release-notes

Summary

Instead of computing a segment's entire top-K at once like SelectionOrderByOperator, return results incrementally, one SelectionResultsBlock per getNextBlock() call, so a downstream merge/combine stage can lazily pull rows from multiple segments, stop early once it has enough global results, and optionally k-way merge-sort blocks before sending downstream.

Problem

An unbounded leaf-stage ORDER BY (the shape injected below a sorted merge join input, and also reachable directly) routes to MinMaxValueBasedSelectionOrderByCombineOperator, which merges every segment's rows into a single materialized block before returning anything. At large data volumes this can exceed the leaf stage's CPU budget and ThreadAccountant raises EarlyTerminationException inside SelectionOperatorUtils.mergeWithOrdering(), surfacing at the broker as a spurious Cancelled by sender. This is the first bullet of Challenge 1 in #18667.

This PR adds a streaming alternative for segments that are physically sorted on the leading ORDER BY column.

Approach

• StreamingSelectionOrderByOperator (new): emits sorted blocks incrementally for a segment physically sorted on the leading ORDER BY column, walking the sorted forward index in order instead of filling a priority queue with the whole segment. Multi-column ORDER BY is handled by a second pass over each equal-prefix run.
• StreamingSelectionOrderByCombineOperator (new): k-way heap merge across the per-segment operators, emitting bounded blocks instead of one materialized result. Segments are acquired/released incrementally; the merge loop does periodic termination/deadline/resource-usage sampling, matching BaseStreamingCombineOperator.
• SelectionPlanNode, CombinePlanNode, InstancePlanMakerImplV2, QueryContext: select these new operators only when the option is set and the sortedness precondition holds; otherwise fall back to the existing operators.
• A data-schema mismatch between segments (possible mid-reload) is reported as a processing exception rather than merged blindly, mirroring SelectionOrderByResultsBlockMerger.

Opt-in query options

Option	Default	Meaning
sortedSelectionMergeEnabled	false	Use the streaming selection ORDER BY combine when the sortedness precondition holds
sortedSelectionMergeBlockSize	10000	Rows per emitted block (Broker.DEFAULT_SORTED_SELECTION_MERGE_BLOCK_SIZE)

No behavior change when the option is off

Without sortedSelectionMergeEnabled, planning and execution take the existing path unchanged — the new operators are never constructed. No existing option, plan node, or wire format changes.

Tests

Unit tests

64 tests, all green:

Test class	Count
StreamingSelectionOrderByOperatorTest (new)	13
StreamingSelectionOrderByCombineOperatorTest (new)	15
CombineSlowOperatorsTest (extended)	11
QueryOptionsUtilsTest (extended)	25

CombineSlowOperatorsTest.testStreamingSelectionOrderByCombineOperatorHonorsDeadline pins deadline behavior: an already-expired deadline must yield an ExceptionResultsBlock before any child operator is driven.

Local cluster test

This PR's leaf-stage combine has no external effect on its own that an explain plan can surface — it's observable in the block sizes the sorted-mailbox-merge receiver (PR2: #19121) reports consuming. Verified by exercising the full stack with streamingSortedMailboxReceive=true (later PR) on a colocated sorted join (/*+ joinOptions(join_strategy='sorted', is_colocated_by_join_keys='true') */, with join_strategy='sorted' both set and removed), against 3,144,172 docs across 2 segments in a local table on 4 servers / 2 replicas.

Validation: streaming sorted selection

Sorted merge join query:

SET useMultistageEngine = true;
SET sortedSelectionMergeEnabled = true;
SET streamingSortedMailboxReceive = true;
 
SELECT
  /*+ joinOptions(join_strategy='sorted', is_colocated_by_join_keys='true') */
  a.correlation_id,
  a.occurred_at_min,
  b.occurred_at_min
FROM mytable AS a
  JOIN mytable AS b ON a.correlation_id = b.correlation_id
WHERE a.occurred_at_min BETWEEN 1777327200 and 1777328100
LIMIT 10

Explain plan: a top-level LogicalSort(fetch=10) over a PinotLogicalSortExchange, feeding a LogicalJoin whose two inputs are each PinotLogicalSortExchange(isSortOnSender=true) wrapping a LogicalSort over a filtered scan of mytable — i.e. both join inputs are sorted before the sort-exchange.

Full stageStats: https://gist.github.com/rohityadav1993/2c0b8f5bb37dc4de9df1cde967a35dc8
Result stageStats (query returns 10 rows): top MAILBOX_RECEIVE (stage-1 output) → MAILBOX_SEND/SORT_OR_LIMIT → a second MAILBOX_RECEIVE (fanIn 2) → MAILBOX_SEND/SORT_OR_LIMIT/TRANSFORM → SORTED_MERGE_JOIN (emittedRows 3629), whose two inputs are MAILBOX_RECEIVE at stage 3 and stage 4, each with kWayMergeUsed: true and emittedRows: 20000. Each of those receives from an EMPTY_MAILBOX_SEND with no stats (flagged in the PR as a bug to raise separately in PR3: early termination isn't propagated, resulting in empty stats for that child).

Note: stage 3 receiving 20000 events from each side implies the 1000 default DEFAULT_SORTED_SELECTION_MERGE_BLOCK_SIZE and 2 workers (2 segments); total across both sides is 40000 emittedRows.

Corresponding non-streaming sorted-selection query (same query shape but sortedSelectionMergeEnabled=false, streamingSortedMailboxReceive=false, enableTrace=true, join_strategy hint removed, and requiring SET maxRowsInJoin = 10485770; to bypass hashjoin guardrails): the accompanying MAILBOX_RECEIVE stageStats shows emittedRows: 3144172 — significantly higher than the streaming case, since the whole dataset is materialized rather than streamed in bounded blocks.

Conclusion: achieving an efficient sorted merge join requires a corresponding streaming sorted-selection operator at the leaf stage.

stageStats showed the two join-input MAILBOX_RECEIVE operators (one per server holding a matching partition) each reporting kWayMergeUsed: true and closing exactly one emittedRows: 10000 block (streamingSortedMailboxReceiveBlockSize's default), for 20,000 rows fetched in bounded blocks across the 2 servers.

Remote cluster test

Skipped; thorough benchmarks with sorted merge join and cluster tests will be captured in later PRs.

Known gaps

• No integration test added: this operator isn't wired into any SSE query path yet. Integration tests will follow once the merger operator is added.

Part of #18667.

Follow up PRs wip:
#19121
#19122

[codecov-commenter]

Codecov Report

❌ Patch coverage is 80.89172% with 90 lines in your changes missing coverage. Please review.
✅ Project coverage is 66.61%. Comparing base (a8b207e) to head (3bc0311).
⚠️ Report is 76 commits behind head on master.

Files with missing lines	Patch %	Lines
...bine/StreamingSelectionOrderByCombineOperator.java	72.41%	39 Missing and 17 partials ⚠️
...rator/query/StreamingSelectionOrderByOperator.java	88.26%	21 Missing and 6 partials ⚠️
...va/org/apache/pinot/core/plan/CombinePlanNode.java	55.55%	1 Missing and 3 partials ⚠️
...pinot/core/plan/maker/InstancePlanMakerImplV2.java	71.42%	1 Missing and 1 partial ⚠️
.../org/apache/pinot/core/plan/SelectionPlanNode.java	93.33%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19120      +/-   ##
============================================
+ Coverage     65.49%   66.61%   +1.12%     
  Complexity     1423     1423              
============================================
  Files          3430     3443      +13     
  Lines        218010   218789     +779     
  Branches      34648    34846     +198     
============================================
+ Hits         142784   145747    +2963     
+ Misses        63666    61290    -2376     
- Partials      11560    11752     +192     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	?
java-25	66.61% <80.89%> (+1.12%)	⬆️
temurin	66.61% <80.89%> (+1.12%)	⬆️
unittests	66.61% <80.89%> (+1.12%)	⬆️
unittests1	57.21% <80.89%> (+0.35%)	⬆️
unittests2	38.79% <0.42%> (+0.94%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[rohityadav1993]

Self review: This is going to be inefficient when segment sort expression is not same as orderBy experssion and there are too few rows per key.

Consciously keeping it out of scope for now to keep the logic simple for happy path.

[rohityadav1993]

Hi @gortiz, could you help with reviewing this 1st of the 3 PRs.

Here is flow chart of the new selection operator being added to help with the review

flowchart TD
    A["getNextBlock()"] --> B{"_exhausted?"}
    B -- yes --> Z["return null"]
    B -- no --> C{"_tailToSort?"}

    C -- "no (no unsorted tail)" --> D["nextSortedRows()"]
    C -- "yes (tail must be sorted per-run)" --> E["nextRun()"]

    subgraph NSR["nextSortedRows() — pass-through mode"]
        D --> D1{"remaining = numRowsToKeep - numRowsEmitted <= 0?"}
        D1 -- yes --> D2["return null"]
        D1 -- no --> D3["_projectOperator.nextBlock()"]
        D3 --> D4{"block == null?"}
        D4 -- yes --> D2
        D4 -- no --> D5["build BlockValSets + RowBasedBlockValueFetcher\nfor phase1 expressions"]
        D5 --> D6["materializeRow() for first min(numDocs, remaining) rows"]
        D6 --> D7["numRowsEmitted += rows.size()\nreturn rows"]
    end

    subgraph NR["nextRun() — buffer-and-sort-per-run mode"]
        E --> E1{"remaining <= 0?"}
        E1 -- yes --> E2["return null"]
        E1 -- no --> E3{"_pendingRow == null?"}
        E3 -- yes --> E4["_pendingRow = nextRow()"]
        E4 --> E5{"still null?"}
        E5 -- yes --> E2
        E3 -- no --> E6
        E5 -- no --> E6["clear _runHeap;\nseed with _pendingRow as runFirstRow"]
        E6 --> E7["loop: row = nextRow()"]
        E7 --> E8{"primaryComparator(row, runFirstRow) == 0?"}
        E8 -- yes --> E9["add row to _runHeap (bounded to numRowsToKeep)"]
        E9 --> E7
        E8 -- no --> E10["stash row as _pendingRow (next run's first row)\nbreak loop"]
        E10 --> E11["drainAscending(_runHeap)"]
        E11 --> E12{"rows.size() > remaining?"}
        E12 -- yes --> E13["truncate to first 'remaining' rows"]
        E12 -- no --> E14["numRowsEmitted += rows.size()\nreturn rows"]
        E13 --> E14
    end

    subgraph NEXTROW["nextRow() — forward scan cursor"]
        F["nextRow()"] --> F1{"current block exhausted?"}
        F1 -- yes --> F2{"_projectExhausted?"}
        F2 -- yes --> F3["return null"]
        F2 -- no --> F4["_projectOperator.nextBlock()"]
        F4 --> F5{"block == null?"}
        F5 -- yes --> F6["_projectExhausted = true\nreturn null"]
        F5 -- no --> F7["rebuild fetcher/docIds/nullBitmaps\nfor new block; reset _currentPos=0"]
        F7 --> F1
        F1 -- no --> F8["materializeRow() at _currentPos++\nreturn row"]
    end

    E4 -.calls.-> F
    E7 -.calls.-> F

    D7 --> G{"_twoPhase?"}
    E14 --> G
    G -- yes --> H["fetchNonOrderByColumns(rows)"]
    G -- no --> I["_dataSchema already built\n(buildSinglePhaseDataSchema in ctor)"]

    subgraph PHASE2["fetchNonOrderByColumns() — two-phase second pass"]
        H --> H1["collect docIds bitmap from rows"]
        H1 --> H2["sort a docId-ordered view sharing same row instances"]
        H2 --> H3["BitmapDocIdSetOperator.ascending(docIds)\n→ ProjectionOperator → TransformOperator"]
        H3 --> H4["pull transformOperator blocks,\nfill non-order-by values into rows in place"]
        H4 --> H5{"_dataSchema == null?"}
        H5 -- yes --> H6["buildTwoPhaseDataSchema()"]
        H5 -- no --> H7["done"]
        H6 --> H7
    end

    H7 --> J
    I --> J["new SelectionResultsBlock(_dataSchema, rows, _comparator, _queryContext)"]
    J --> K["return block"]

Loading