Add streaming selection ORDER BY combine for physically sorted segments

[rohityadav1993]

PR flow

Adds streaming ORDER BY combine: leaf returns sorted blocks, combine k-way merges them, activates segments lazily, stops at limit+offset.

flowchart TD
  N0["Check sortedSelectionMergeMode enabled #40;F4#44; F5#44; F7#41;"]:::stAdded
  N1["Leaf#58; StreamingSelectionOrderByOperator#46;getNextBlock#40;#41; returns sorted block #40;F3#41;"]:::stAdded
  N2["Combine#58; StreamingSelectionOrderByCombineOperator#46;getNextBlock#40;#41; k#45;way merge #40;F2#41;"]:::stAdded
  N3["Activate eligible cursors via min#47;max frontier #40;F2#41;"]:::stAdded
  N4["Select leader cursor#44; advance#44; emit row #40;F2#41;"]:::stAdded
  N5["Flush block when size reached#44; return SelectionResultsBlock #40;F2#41;"]:::stAdded
  N0 -->|"decides to use leaf"| N1
  N0 -->|"decides to use combine"| N2
  N1 -->|"leaf blocks feed combine"| N2
  N2 -->|"merge loop activates eligible segments"| N3
  N3 -->|"activated cursors influence leader selection"| N4
  N4 -->|"row emission triggers block flush"| N5
  N5 -->|"flush returns block and loop continues"| N2
  classDef stAdded fill:#dafbe1,stroke:#1a7f37,color:#1f2328,stroke-width:2px
  classDef stModified fill:#fff8c5,stroke:#9a6700,color:#1f2328,stroke-width:2px
  classDef stRemoved fill:#ffebe9,stroke:#cf222e,color:#1f2328,stroke-width:2px
  classDef stUnchanged fill:#f6f8fa,stroke:#656d76,color:#1f2328,stroke-width:1px

Loading

AI-generated · Green: added · Yellow: modified · Red: removed · Gray: existing

Partial evidence: 1 file patches omitted; 3 truncated.

Diff evidence

• F2: pinot-core/src/main/java/org/apache/pinot/core/operator/combine/StreamingSelectionOrderByCombineOperator.java — after
• F3: pinot-core/src/main/java/org/apache/pinot/core/operator/query/StreamingSelectionOrderByOperator.java — after
• F4: pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java — before · after
• F5: pinot-core/src/main/java/org/apache/pinot/core/plan/SelectionPlanNode.java — before · after
• F7: pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java — before · after

• Regenerate PR flow

feature performance release-notes

Summary

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

• SelectionOrderByOperator computes the full top-K of a segment at once. The new leaf operator returns sorted rows in bounded blocks, one SelectionResultsBlock for each getNextBlock() call.
• The new combine operator does a k-way merge of these blocks. It activates a segment only when the merge needs rows from that segment.
• The combine stops when it has limit + offset rows. It does not read the remaining segments.

The feature is opt-in. With the default option value (OFF), the server uses the existing path.

Design doc: https://docs.google.com/document/d/1KMwCKYjCJdy83kvoNcXtkWENf603R6WSRUj-jjMaRYg/edit?tab=t.0

Benchmarks (harness and report; can be contributed if useful): rohityadav1993#1

Problem

An unbounded leaf-stage ORDER BY goes to MinMaxValueBasedSelectionOrderByCombineOperator (the MinMax combine). This shape occurs below a sorted merge join input. A query can also send it directly. The MinMax combine merges the rows of all segments into one block before it returns a row.

flowchart LR
  S1[Segment 1] --> M["MinMax combine<br/>keeps all rows in one block"]
  S2[Segment 2] --> M
  SN[Segment N] --> M
  M --> B[One large block]
  M -.->|large data| E["ThreadAccountant stops the query<br/>(EarlyTerminationException)"]
  E -.->|broker shows| C["Cancelled by sender"]

Loading

With a large data volume, the merge can use more CPU than the leaf-stage budget. Then ThreadAccountant raises EarlyTerminationException in SelectionOperatorUtils.mergeWithOrdering(). The broker shows this error as Cancelled by sender. This message does not give the real cause. This is the first item of Challenge 1 in #18667.

Approach

flowchart LR
  subgraph Leaf["Per segment: StreamingSelectionOrderByOperator"]
    L1["Segment 1 cursor<br/>block, block, ..."]
    L2["Segment 2 cursor<br/>block, block, ..."]
    LN["Segment N cursor<br/>(not activated yet)"]
  end
  L1 --> H["StreamingSelectionOrderByCombineOperator<br/>k-way heap merge"]
  L2 --> H
  LN -.->|"activated only when the merge<br/>frontier reaches its min/max"| H
  H --> O["Bounded output blocks<br/>(sortedSelectionMergeBlockSize rows)"]

Loading

Class	Change
StreamingSelectionOrderByOperator (new)	Reads the sorted forward index in order and returns sorted blocks. It does not put the full segment into a priority queue. For a multi-column ORDER BY, it sorts each run of equal leading values in a second pass. A segment with no rows that match returns one empty block with the DataSchema, then null. Thus the consumer always gets the schema.
StreamingSelectionOrderByCombineOperator (new)	Does a k-way heap merge of the segment cursors and returns bounded blocks. It acquires and releases segments one at a time. The merge loop checks termination, deadline and resource usage at intervals, the same as BaseStreamingCombineOperator. Segments are usually near-disjoint on the leading column, so one cursor supplies long runs of rows. Thus the current heap leader stays out of the priority queue. The combine offers it again only when the leader changes or the loop returns.
SelectionPlanNode, CombinePlanNode, InstancePlanMakerImplV2, QueryContext	Select the new operators only on streaming paths (MSE leaf, gRPC streaming reduce), as the mode decision below shows. The SSE blocking path always keeps the MinMax combine. It calls nextBlock() one time, so the streaming combine gives no benefit there and adds its per-cursor memory cost.

If the schema of a segment block is different from the merge schema (this can occur during a reload), the combine drops that block. It attaches a processing error to the response and continues the merge. SelectionOrderByResultsBlockMerger also drops blocks with a different schema.

Opt-in query options

flowchart TD
  Q[Selection ORDER BY query] --> P{"Streaming path?<br/>(MSE leaf or gRPC streaming)"}
  P -->|no| MM[MinMax combine]
  P -->|yes| MODE{sortedSelectionMergeMode}
  MODE -->|"OFF (default)"| MM
  MODE -->|ON| ST[Streaming combine]
  MODE -->|AUTO| X{"Leading ORDER BY<br/>is a column?"}
  X -->|no| MM
  X -->|yes| D{"DESC without<br/>allowReverseOrder?"}
  D -->|yes| MM
  D -->|no| R{"Sorted-segment ratio >=<br/>sortedSelectionMergeAutoMinSortedRatio?<br/>(segment metadata only)"}
  R -->|no| MM
  R -->|yes| ST

Loading

Option	Default	Meaning
sortedSelectionMergeMode	OFF	OFF, ON or AUTO. ON always uses the streaming combine, without a sortedness check. This includes a leading ORDER BY expression that is not a column (useful for benchmarks). AUTO uses the streaming combine only if at least sortedSelectionMergeAutoMinSortedRatio of the queried segments are physically sorted on the leading ORDER BY column. AUTO decides from segment metadata, before it acquires a segment.
sortedSelectionMergeAutoMinSortedRatio	0.8	The minimum fraction of sorted segments for AUTO to use the streaming combine. Not measured yet. A benchmark sweep can set it.
sortedSelectionMergeBlockSize	10000	The number of rows in each output block (Server.DEFAULT_SORTED_SELECTION_MERGE_BLOCK_SIZE).

Reviewers replaced the first version, a boolean option sortedSelectionMergeEnabled, with the three-value sortedSelectionMergeMode. The boolean did not check sortedness on the combine side, and it always disabled maxExecutionThreads.

Server-wide defaults. These two server keys replace the built-in defaults for queries that do not set the option:

• pinot.server.query.executor.sorted.selection.merge.block.size
• pinot.server.query.executor.sorted.selection.merge.auto.min.sorted.ratio

The server validates them at startup, the same as numGroupsLimit. The server reads them, so they apply to the MSE leaf and the gRPC streaming path without a broker change. sortedSelectionMergeMode has no server default. You set it for each query.

DESC behavior

DataSourceMetadata.isSorted() tells only about ascending physical order. A descending scan must reverse the docId set. SelectionPlanNode does this only if allowReverseOrder is set (default false). Without this option, ReverseDocIdSetOperator.initializeBitmap() reads all docIds that match into a RoaringBitmap first. Then memory is not bounded.

Mode	DESC leading ORDER BY, allowReverseOrder not set	DESC, allowReverseOrder = true
AUTO	MinMax combine. AUTO never puts the streaming combine over leaves that materialize.	Streaming combine if the ratio check passes
ON	Streaming combine (for benchmarks)	Streaming combine

Null handling

With enableNullHandling, the combine uses the min/max of a segment for pruning only if the leading column is non-null. Segment metadata gives this flag (ColumnMetadata.isNonNull(), #19072).

Leading column in the segment	Prune by min/max	Counts as sorted for AUTO
Flagged non-null	Yes	Yes (if physically sorted)
Has nulls, or no flag (built before #19072 and not reloaded)	No: always activates	No

The combine makes this decision before it acquires a segment. It never reads the null value vector. AUTO counts a segment only if its leaf can stream. Thus null handling does not make AUTO select the streaming combine over leaves that cannot stream. (Known limit: with allowReverseOrder set, a DESC docId set that cannot be reversed falls back without a signal. A TODO in InstancePlanMakerImplV2 records this.)

Consuming segments

A consuming (MutableSegment) segment never gets min/max bounds. Its cursor activates at the start of the merge.

Before the fix, this sequence gave rows in the wrong order:

sequenceDiagram
  participant C as Streaming combine
  participant M as Consuming segment
  participant I as Ingestion
  C->>M: constructor reads min/max (snapshot: 100..199)
  I->>M: index new row (value 5)
  Note over C: merge emits rows from other segments, up to the frontier
  C->>M: activate (prefetch): plan and doc count built now
  M-->>C: head row = 5
  Note over C: row 5 is emitted after larger rows

Loading

MutableSegmentImpl copies the current min/max when the combine reads them. With server prefetch (AcquireReleaseColumnsSegmentOperator), the segment plan and its doc count are built later, at activation. A row indexed between these two points can be outside the bounds. The merge then emitted that row too late, after rows with larger values (ASC). The set of top-N rows stayed correct. Only the order was wrong.

The fix gives a consuming segment no bounds. Thus the combine always acquires and reads it, with no frontier skip. Few segments consume at one time. Each one holds one block until its cursor drains. The cost is small.

Without prefetch, all plans run before the combine reads the bounds, so that path did not have the problem. The MinMax combine uses min/max only to prune and then sorts all rows, so it does not have the problem. ConsumingSegmentBoundsStalenessTest covers ASC and DESC with dictionary and raw columns. Its MinMax cases are a baseline.

Tied rows under a single-column ORDER BY

If segment minimums are equal on the only ORDER BY column, the combine defers the tied segments. It does not activate them all at once. The MinMax combine already does this. The benchmark below found the case.

Effect: if a tied group crosses the LIMIT, the result can contain a different member of that group than with OFF. Rows that are equal on all ORDER BY expressions have no defined order. Thus this agrees with the existing contract. With two or more ORDER BY expressions, the combine defers nothing. A tie on the first column can hide a row that sorts earlier on the second column.

Memory bound

Shape	Memory for each active cursor	MinMax combine
Sorted column covers the full ORDER BY	One block of at most min(limit + offset, MAX_DOC_PER_CALL) rows (MAX_DOC_PER_CALL = 10,000)	O(limit + offset) rows for all shapes
Other shapes (multi-column ORDER BY, unsorted children)	One 10,000-row scan block plus a run buffer of at most limit + offset rows

Total streaming memory = memory for each cursor × number of cursors active at the same time, plus one output block.

• Large LIMIT: the streaming combine uses less memory.
• Many segments that overlap and must all be active together: the streaming combine can use more memory (see the benchmark under Known gaps).

A cap on the number of active cursors gives incorrect results. Memory controls for high overlap are a follow-up.

Observability

CombineSelectOrderbyStreaming shows these explain attributes: blockSize, frontierPruning, deferTiedCursors, numSegments, numSortedSegments. On the MSE path, they show with explainAskingServers=true. numSortedSegments is metadata sortedness. It does not show for an expression ORDER BY.

Stat	Segment the merge never activates	Activated segment with no rows that match
numSegmentsProcessed	Counted (as with the MinMax combine)	Counted
numSegmentsMatched	Not counted (no docs scanned)	Not counted

Thus numSegmentsProcessed - numSegmentsMatched = segments not activated + activated segments with no matches. A separate count needs a new DataTable MetadataKey with broker and MSE aggregation. That change widens the wire format, so it is a TODO in the code.

No behavior change when the option is off

With OFF (the default), planning and execution use the existing path. The server does not construct the new operators. No existing option, plan node or wire format changes. The two new server keys only set defaults for the new options.

nextBlock() convention for follow-up PRs

The consumer calls this operator many times. The operator returns null at end of stream. This agrees with neither of the two contracts that Operator.nextBlock() documents now. The follow-up PRs use this guarantee: a segment with no rows that match always returns one block with the schema before its terminal null. Thus a consumer always knows the DataSchema of a child. A separate PR will add this as a third case to the Operator.nextBlock() Javadoc. That change affects the contract of all implementations.

Tests

Unit tests

165 tests in eight classes. All pass.

Test class	Count
StreamingSelectionOrderByCombineOperatorTest	59
StreamingSelectionOrderByOperatorTest	30
QueryOptionsUtilsTest	34
InstancePlanMakerImplV2Test	11
CombineSlowOperatorsTest	10
ConsumingSegmentBoundsStalenessTest	8 (4 streaming, 4 MinMax baseline)
SelectionCombineOperatorTest	7
CombinePlanNodeTest	6

The combine tests also cover:

• Segment lifecycle. The combine releases each acquired segment exactly one time. This is true:
  • when the merge drains all cursors;
  • when LIMIT stops the merge with cursors still active;
  • on stop(), also when called two times;
  • when a child throws an exception or an Error.
• A schema mismatch during reload.
• Deferral of tied minimums.
• Null pruning for each segment.
• A consuming segment that indexes rows between combine construction and activation, with prefetch.
• The explain attributes.

Each of these tests was checked to fail on an operator with the guarded code removed.

CombineSlowOperatorsTest.testStreamingSelectionOrderByCombineOperatorHonorsDeadline checks deadline behavior. With an expired deadline, the combine must return an ExceptionResultsBlock before it calls a child operator. The test has a timeOut, so a regression fails the test and does not stop the build.

Local cluster test

The receiver above the leaf stage shows the size of the blocks it gets from this combine. The full stack ran with PR2 (#19121, option streamingSortedMailboxReceive=true) on a colocated sorted join. The table had 3,144,172 docs in 2 segments, on 4 servers with 2 replicas. The test ran with the hint join_strategy='sorted' set, and again with the hint removed.

Validation: streaming sorted selection

Sorted merge join query:

SET useMultistageEngine = true;
SET sortedSelectionMergeMode = 'ON';
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: both join inputs are a PinotLogicalSortExchange(isSortOnSender=true) above a LogicalSort on a filtered scan of mytable. A top-level LogicalSort(fetch=10) is above the join.

Full stageStats: https://gist.github.com/rohityadav1993/2c0b8f5bb37dc4de9df1cde967a35dc8

Result stageStats (the query returns 10 rows):

• SORTED_MERGE_JOIN has emittedRows: 3629.
• Its two inputs are the MAILBOX_RECEIVE operators of stage 3 and stage 4. Each has kWayMergeUsed: true and emittedRows: 20000.
• 20,000 rows = two blocks of the default block size (10,000), one for each of the 2 segments. The total for both join inputs is 40,000 rows.
• Each receive gets data from an EMPTY_MAILBOX_SEND with no stats. This is a known PR3 issue: early termination does not propagate, so that child has empty stats. It will be raised separately.

Same query without streaming: sortedSelectionMergeMode not set, streamingSortedMailboxReceive=false, enableTrace=true, no join_strategy hint, and SET maxRowsInJoin = 10485770; to bypass the hash join guardrail. The MAILBOX_RECEIVE stageStats shows emittedRows: 3144172. The full dataset is materialized, not streamed in bounded blocks.

Join input MAILBOX_RECEIVE	Rows received
Streaming (ON, sorted merge join)	20,000 per input (40,000 total)
Not streaming (hash join)	3,144,172

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

Remote cluster test

Not done in this PR. Later PRs will include sorted merge join benchmarks and cluster tests.

Known gaps

• No integration test. No SSE query path uses this operator yet. Integration tests will come with the MSE consumer.

• Benchmark (harness and full report: Benchmark harness for the streaming selection ORDER BY combine rohityadav1993/pinot#1). Setup: 100 segments × 50,000 rows, one server-side combine, ON compared with OFF with 10 threads, segment overlap DISJOINT, PARTIAL and FULL. The heap values are the minimum -Xmx at which the query completes. They are a provisioning bound, not measured live memory.

  LIMIT	Wall time ON / OFF	Heap
  100,000 and more	0.07 to 0.75 (ON faster in all cases)	ON needs 2x to 14x less. Its need stays flat as LIMIT increases.
  1,000,000		OFF about 950 MB; ON less than 64 MB (DISJOINT) and 237 MB (FULL)

  ON uses less CPU in 17 of 18 two-column cases. The benchmark uses only immutable segments.

• Adverse cases. At FULL overlap and LIMIT less than 100,000, the merge must read all segments that overlap:

  • ON (one merge thread) is slower than OFF (10 threads). It is up to 7x slower at LIMIT 10,000 with a two-column ORDER BY, and 10x slower with one column at K=1.
  • At FULL overlap and LIMIT 10,000 with a two-column ORDER BY, ON needs about 1.27x more heap.

• AUTO does not look at overlap. It decides only from the sorted-segment ratio. Thus it can select ON in the adverse cases. An overlap-aware fallback is a follow-up.

• Not swept yet: sortedSelectionMergeAutoMinSortedRatio (default 0.8, not measured) and sortedSelectionMergeBlockSize (the harness uses 10,000; measure it with the MSE receiver).

Part of #18667.

Follow-up PRs (work in progress):

• Add streaming k-way merge to SortedMailboxReceiveOperator #19121 (MSE ordered leaf above the k-way merge receive of [MSE] Add opt-in sender-sorted merging for global ordered windows #19396)
• Add streaming sorted merge join with colocation support #19122 (sorted merge join)

[codecov-commenter]

Codecov Report

❌ Patch coverage is 94.19355% with 36 lines in your changes missing coverage. Please review.
✅ Project coverage is 68.34%. Comparing base (2cd9a2f) to head (ba2b193).
⚠️ Report is 37 commits behind head on master.

Files with missing lines	Patch %	Lines
...rator/query/StreamingSelectionOrderByOperator.java	92.30%	11 Missing and 8 partials ⚠️
...bine/StreamingSelectionOrderByCombineOperator.java	95.92%	2 Missing and 7 partials ⚠️
...pinot/core/plan/maker/InstancePlanMakerImplV2.java	94.02%	2 Missing and 2 partials ⚠️
...va/org/apache/pinot/core/plan/CombinePlanNode.java	33.33%	0 Missing and 2 partials ⚠️
.../org/apache/pinot/core/plan/SelectionPlanNode.java	94.11%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #19120       +/-   ##
=============================================
+ Coverage     39.77%   68.34%   +28.57%     
- Complexity     1449     1450        +1     
=============================================
  Files          3518     3522        +4     
  Lines        228478   229534     +1056     
  Branches      36201    36454      +253     
=============================================
+ Hits          90881   156883    +66002     
+ Misses       129364    60389    -68975     
- Partials       8233    12262     +4029     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	68.34% <94.19%> (+28.57%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	68.34% <94.19%> (+28.57%)	⬆️
unittests	68.34% <94.19%> (+28.57%)	⬆️
unittests1	58.34% <94.19%> (?)
unittests2	39.84% <3.06%> (+0.06%)	⬆️

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]

Still out of scope. When the segment is sorted on only a prefix of the ORDER BY and each prefix value has few rows, the per-run sort pass dominates. Results stay correct, and the path is opt-in. Tracked as a follow-up with [10]'s levers.

[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

[gortiz]

Thanks for this — I went through the merge algorithm carefully and I think the core is sound. I traced the k-way merge, the run-based tail sort, the frontier pruning argument and the two-phase docId round-trip, and I did not find a wrong-results bug on the happy path. The pruning invariant (bound.compareTo(head[0]), activate on tie) is correct, cursor heads are stable while sitting in the heap, and the per-segment limit+offset cap across runs is right. The parity-based test suite is genuinely good — real segments, compared against the existing operators, both null-handling modes, mixed sorted/unsorted, DESC fallback.

One thing I think deserves more prominence in the description than it gets: for a segment sorted on the leading order-by column with no tail to sort, the per-cursor block is capped at MAX_DOC_PER_CALL regardless of LIMIT, whereas MinMaxValueBasedSelectionOrderByCombineOperator's accumulator is O(limit). For the unbounded sort under a sorted merge join — this PR's actual target — that is a large memory win, independent of streaming or pruning. Worth stating explicitly.

My main comment is not about the algorithm but about when it gets selected. CombinePlanNode gates only on the query option, never on sortedness, while start()/stop() are no-ops — so _numTasks / maxExecutionThreads becomes dead and the whole leaf stage serialises on the consumer thread. When no segment is sorted, every activate() is a full segment scan run one after another. And ORDER BY … DESC hits this by default, because DEFAULT_ALLOW_REVERSE_ORDER is false, so no segment ever gets the streaming operator while the combine still switches. Details in [1], [2], [3].

The four I would most like to see addressed before merge:

• [1] gate on sortedness (a tri-valued on/off/auto would also let us force it for benchmarking, and let expressions through)
• [3] the DESC default
• [5] the Operator.nextBlock() contract — this one matters most, because PR2/PR3 inherit the convention. There is a concrete three-part fix that also deletes resolveDataSchema() entirely
• [6] the empty-block asymmetry, which is a silent-truncation risk

The rest ([4], [7]–[15]) are fine as follow-ups or as-you-like. [11] is a straightforward perf win we would probably want before the sorted-join PRs land.

One design question, not blocking and probably aimed at PR3: this adds an SSE leaf operator that the MSE planner has no representation for. The consequences are that the option is query-wide rather than per-stage, the choice is invisible to the MSE explain plan, and — most importantly — a sorted merge join whose inputs are intermediate stages can never benefit, since the operator only exists at the leaf. The collation information already exists in the planner as RelCollation; it just gets flattened into orderByExpressions before it reaches the server. Would it be worth adding a streaming sort node to the broker planner, so the leaf case becomes a special case of a general mechanism rather than a separate one? Happy to discuss separately.

[gortiz]

[1/15] Could we make the option tri-valued (on/off/auto)? The gate never checks sortedness, and this operator disables maxExecutionThreads.

The PR description says the new operators are chosen 'only when the option is set and the sortedness precondition holds'. That holds in SelectionPlanNode (which needs sortedColumnsPrefixSize > 0), but not here -- this gate only checks the hint, isSelectionQuery, limit != 0 and 'first order-by is an identifier'.

That matters because start() is a no-op in the new operator, so startProcess() never runs -- and that is the only thing that submits to _executorService. _numTasks = min(numSegments, maxExecutionThreads) (BaseCombineOperator:91) becomes dead, and the merge runs entirely on the consumer thread. When no segment is physically sorted on the leading column, every child is a plain SelectionOrderByOperator and SegmentCursor.activate() then performs a full segment scan plus top-K synchronously, one segment after another. On the SSE path that takes a query from maxExecutionThreads down to one thread.

Separately, the gate refuses the case where the first order-by is an expression, which we would like to be able to force even when it is slower, so we can benchmark it.

Would you consider making sortedSelectionMergeEnabled tri-valued -- on / off / auto?

• on: take this path unconditionally, expressions included (materialising early if needed). Useful for benchmarking and A/B.
• off: today's default.
• auto: apply a heuristic.

The heuristic can be cheap and needs no extra I/O: DataSourceMetadata.isSorted() is readable without acquiring the segment (the combine constructor at :158 already reads min/max off the same metadata), it is the same predicate SelectionPlanNode.getSortedColumnsPrefix() uses, and segment overlap falls out of the min/max array the constructor already sorts. Deciding from metadata would also let SegmentCursor drop isStreamingChild(), which today has to materializeChildOperator() -- and therefore acquire -- just to answer the question.

There is precedent for the shape a few lines above at :158: SequentialSortedGroupByCombineOperator is chosen over the parallel SortedGroupByCombineOperator behind a segment-count threshold with a server-config default (DEFAULT_SORT_AGGREGATE_SEQUENTIAL_COMBINE_NUM_SEGMENTS_THRESHOLD). Same trade-off, same file.

[rohityadav1993]

Done in f3f2db3. Replaced the boolean with sortedSelectionMergeMode (OFF/ON/AUTO, default OFF), following the getJoinOverflowMode idiom.

ON forces the streaming path unconditionally, including non-identifier order-bys. AUTO takes it when >= sortedSelectionMergeAutoMinSortedRatio (default 0.8, unmeasured) of segments are physically sorted, decided from metadata alone, no acquire needed.

Resolved once in InstancePlanMakerImplV2, not independently per gate - CombinePlanNode.run() builds leaves before its own gate runs, so a per-gate AUTO decision would arrive too late (same issue as comment [2]).

Known gap, not fixed here: AUTO's metadata-only sortedness check does not see nulls in the leading column, so with null handling on it can resolve ON while the leaf's fuller check declines to stream - correct results, just no speedup. Pinned by testAutoIgnoresNullsInTheLeadingColumn. Would like a go-ahead before filing an issue to track it.

Tests: 62 combine/operator tests + 30 in QueryOptionsUtilsTest covering mode/ratio parsing, AUTO thresholds, ON forcing, and the gap above.

[gortiz]

[2/15] On the SSE path the operator cannot stream at all -- consider restricting this PR to _streamer != null.

Here _streamer == null, so streaming=false, so _blockSize = Integer.MAX_VALUE and the if (_streaming && _outputRows.size() >= _blockSize) flush never fires. InstanceResponseOperator calls nextBlock() once and expects the complete result, so there is nowhere to emit incrementally.

The result is that _outputRows grows to limit+offset -- exactly like MinMaxValueBasedSelectionOrderByCombineOperator's accumulator -- and on top of that every activated cursor stays alive holding its own materialised block. So this call site takes the memory profile of the new merge, loses the parallel segment processing of the operator it replaces, and gains nothing, because a single block comes out at the end either way.

Given the PR targets the MSE leaf stage, would it make sense to restrict this change to the _streamer != null branch and leave SSE on MinMaxValueBased? That would also make StreamingSelectionOrderByCombineOperator.resolveDataSchema() unreachable -- see comment [5].

[rohityadav1993]

Accepted, done in f3f2db3. Deleted CombinePlanNode's non-streamer branch; the streaming combine only appears when _streamer != null. The blocking SSE path keeps MinMaxValueBasedSelectionOrderByCombineOperator unconditionally.

Also went further than asked: InstancePlanMakerImplV2.makeInstancePlan now clears sortedSelectionMergeMode for the whole blocking plan, since SelectionPlanNode gates the leaf on the same option independently and has no way to know which combine it feeds - without this the option would have been half-honored (streaming leaf, blocking combine, no actual streaming).

Verified this pairing was not a correctness bug (row counts/contents matched the baseline in every shape I tried), but removing it is still right for coherence, and because [10]/[15] propose changes that would break that accidental correctness.

Reworded per your note: gate is "streaming-capable paths" (_streamer != null), not "MSE only".

[gortiz]

[3/15] With default options a DESC query never reaches this branch, but the combine still switches -- worst of both.

DataSourceMetadata.isSorted() means ascending physical order, so DESC needs the docId scan reversed. getSortedByProject() only attempts projectOperator.withOrder(DESC) when QueryOptionsUtils.isReverseOrderAllowed() is true, and DEFAULT_ALLOW_REVERSE_ORDER is false (CommonConstants:860).

So for ORDER BY x DESC with default options: withOrder is never attempted, isCompatibleWith(DESC) is false for a normal ascending scan, StreamingSelectionOrderByOperator is never constructed, and every segment falls through to SelectionPartiallyOrderedByDescOperation. Meanwhile CombinePlanNode:136 checks none of this and still installs StreamingSelectionOrderByCombineOperator.

Net effect of sortedSelectionMergeEnabled=true on any DESC query with default options: every child is a single materialised block AND the merge is serial -- strictly worse than MinMaxValueBased on both axes, and it is the default rather than a corner case. It also means the duplicated project build below fires on every DESC segment.

Could the streaming path imply allowReverseOrder on this branch, or the combine-side gate become aware that no child ended up streaming? Either way, DESC behaviour is worth a line in the PR description -- it is not mentioned today.

[rohityadav1993]

Done in f3f2db3, but not via your first suggestion: ReverseDocIdSetOperator.initializeBitmap() drains the whole matching docId set into a RoaringBitmap up front when there's no bitmap index to reuse, which is likely why allowReverseOrder defaults false. Implying it from the streaming branch would silently trade away the bounded memory this feature is for, so reversal stays an explicit user lever.

What changed: InstancePlanMakerImplV2.isSortedEnoughForStreamingMerge is now direction-aware - DESC without allowReverseOrder resolves AUTO to OFF, in one place both gates read. ON still force-enables DESC as a benchmark lever.

Tests: testAutoDoesNotSelectStreamingForDescWithoutReverseOrder, testAutoSelectsStreamingForDescWithReverseOrder, testAutoHonoursTheMinSortedRatioThresholdForDesc, testAutoOnlyChecksTheLeadingOrderByDirection.

Residual, not fixed here: allowReverseOrder=true doesn't guarantee a reversed scan (getSortedByProject silently falls back if withOrder(DESC) throws), so AUTO can still resolve ON over materialized children in that case. Same shape as the [1] null-handling gap; deferred pending a go-ahead to track it.

PR description now has a DESC behavior paragraph.

[gortiz]

[4/15] The discarded project operator is built twice and never closed.

getSortedByProject() runs a full ProjectPlanNode -- filter plan construction, index reader lookups, possibly a withOrder(DESC) wrapper -- and the ProjectionOperator it produces is Closeable. On this fall-through we drop streamingProjectOperator and build a second one over all expressions, paying the plan-build cost twice and leaking whatever the first allocated.

When expressions.size() == numOrderByExpressions the two calls are identical, so it is pure duplicated work. And per comment [3], this path is the default for every DESC query.

Could the compatibility decision be hoisted so the project is built once -- or the existing instance reused for the SelectionPartiallyOrderedByDescOperation branch when the expression lists match?

[rohityadav1993]

Only the duplicate build is fixed in f3f2db3 - the leak is not. The DESC fall-through now reuses streamingProjectOperator when it already covers the full expression list; the wider order-by-prefix case still builds a second one.

On the leak: ProjectPlanNode#run already has a pre-existing "TODO: figure out a way to close this operator" covering every call site in the module, so a narrow fix here would not actually close it - left as-is.

Tests: testDescIncompatibleFallbackBuildsTheProjectOnce (counts data source lookups, 4 vs 3 baseline) and testDescIncompatibleFallbackMatchesTheMaterializedPlan (pins schema/rows across reuse, wider-rebuild, zero-match cases).

[gortiz]

[5/15] Undocumented third nextBlock() contract. Suggest: document it, and always emit at least one schema-carrying block.

Operator.nextBlock() documents two contracts by layer: 'operators above projection phase (aggregation, selection, combine etc.) should only be called once, and will return a non-null block', versus 'operators in projection phase (docIdSet, projection, transformExpression) can be called multiple times, and will return non-empty block or null if no more documents available'. This is a selection operator that is called repeatedly and returns null, so it matches neither.

The javadoc is already partly stale -- MultiStageOperator implements Operator and is also called repeatedly -- but note MSE terminates with a SuccessMseBlock (MseBlock.Eos), never null. Every results-layer stream in the codebase ends with a block: BaseStreamingCombineOperator uses LAST_RESULTS_BLOCK worker-side and a MetadataResultsBlock driver-side, and SelectionOnlyOperator:156 returns a block unconditionally.

I would NOT switch to a sentinel here though: BaseStreamingCombineOperator.processSegments() already does while ((resultsBlock = operator.nextBlock()) != null) for children where isChildOperatorSingleBlock() is false. So null-as-EOS is already the assumed SSE contract -- it is just undocumented.

What null costs us today is the schema: a zero-match segment signals EOS before ever emitting one, which is the only reason StreamingSelectionOrderByCombineOperator.resolveDataSchema() exists.

Suggested fix, three parts:

1. Update Operator.nextBlock() to document a third case -- multi-block results operators may be called repeatedly; MSE terminates with MseBlock.Eos, SSE streaming children terminate with null and must emit at least one block carrying the DataSchema. (Possibly a separate PR, since it touches every implementor's contract.)

2. Honour that guarantee here. Track whether any block was emitted; on the first exhausted call return a schema-carrying empty block instead of null:

   if (rows == null || rows.isEmpty()) {
   _exhausted = true;
   if (_emittedAnyBlock) {
   return null;
   }
   if (_twoPhase && _dataSchema == null) {
   fetchNonOrderByColumns(List.of());
   }
   return new SelectionResultsBlock(_dataSchema, List.of(), _comparator, _queryContext);
   }
   _emittedAnyBlock = true;

No extra scanning: single-phase already has _dataSchema from the constructor (:190), and for two-phase fetchNonOrderByColumns(List.of()) builds the TransformOperator, iterates zero blocks, and still reaches if (_dataSchema == null) _dataSchema = buildTwoPhaseDataSchema(transformOperator).

3. Then delete StreamingSelectionOrderByCombineOperator.resolveDataSchema(). SegmentCursor.pullBlock() captures the schema before its empty-rows check, then loops for streaming children and exhausts on null -- no change needed there.

Related: nextSortedRows() also needs the empty-block fix in comment [6], otherwise a mid-stream zero-doc block still truncates the segment.

This is worth stating explicitly in the PR description, since PR2 and PR3 inherit the convention.

[rohityadav1993]

Done in f3f2db3, steps 2 and 3. Step 1 (the Operator.nextBlock() javadoc) I'd rather do as its own PR since it touches every implementor's contract - will link once opened.

Step 2: a zero-match segment now emits one empty schema-carrying block before its terminal null, instead of null immediately.
Step 3: resolveDataSchema() deleted.

One correction: I don't think step 3 was enabled by step 2 as framed. Rows only ever reach _outputRows through pullBlock(), which already sets _dataSchema first, and a true zero-match query never calls flushDataBlock() at all. Negative control confirmed it: disabling step 2 and deleting resolveDataSchema() together still passed all 30 combine tests. So the invariant held by avoidance before, and by construction now - the deletion was always safe, step 2 just makes it stay safe under future changes.

Not fixed here: a zero-match query (not just segment) still returns no schema, since the combine then exits through MetadataResultsBlock, whose getDataSchema() is hardcoded null. That's a convention shared by every BaseStreamingCombineOperator subclass, so I left it as a pinned gap (testEmptyResultOnStreamingPath asserts assertNull(result._schema)) rather than a cross-cutting fix here.

[gortiz]

[13/15] The null merger is safe only by luck -- isQuerySatisfied() would NPE. Consider overriding it, and dropping the unused ExecutorService.

super(null, ...) works because BaseStreamingCombineOperator.createQuerySatisfiedTracker() returns null and the overridden getNextBlock() never reaches isQuerySatisfied(), which dereferences _resultsBlockMerger. processSegments() is already overridden to fail loud -- could isQuerySatisfied() get the same treatment (return false, or throw)? Then a future change to the base streaming loop fails at the seam rather than with an NPE at query time.

Also, the constructor still takes an ExecutorService it never uses, since start() is a no-op. Worth either dropping it or adding a line saying it is kept for signature parity.

[rohityadav1993]

Done in f3f2db3. isQuerySatisfied() now throws IllegalStateException, matching processSegments()'s existing fail-loud behavior.

Kept the executor param rather than dropping it - removing it would just pass null up to BaseCombineOperator._executorService, the same latent-NPE shape one level up. Comment on the constructor explains why.

New test testBaseWorkerEntryPointsFailLoud covers both seams.

[gortiz]

[14/15] Schema mismatch drops the rest of the segment, not just one block -- and there is no test for it.

On mismatch pullBlock() returns false, which exhausts the cursor. So a segment whose first block matched but whose later block diverges contributes a partial prefix, and the query still returns successfully with only a MERGE_RESPONSE error attached.

That is close to SelectionOrderByResultsBlockMerger, but not identical: the merger drops one block, this drops the remainder of the segment. Worth a comment saying the difference is intentional.

The PR description calls this out as a deliberate design decision, but StreamingSelectionOrderByCombineOperatorTest has no coverage for it. Could we add a case with two segments whose schemas diverge, asserting both the dropped rows and the attached error?

[rohityadav1993]

Done in e3a908a. Added a comment on pullBlock(): it drops the same unit as the merger, where a block is a whole segment. Since a reload swaps the whole segment, a mismatch normally starts at the first block and both drop everything; they only differ on a mid-segment block, where keeping a clean prefix is better than a prefix and a suffix with a hole.

testSchemaMismatchDropsTheRestOfTheSegmentAndIsReportedOnce makes two segments diverge mid-stream and checks the kept rows, the untouched segments, sorted output, and a single deduplicated error.

[gortiz]

[15/15] Default is 10_000 here but 1000 in the PR description; also please benchmark block sizes and add a cluster config.

Three things:

1. The PR description's validation section says 'the 1000 default DEFAULT_SORTED_SELECTION_MERGE_BLOCK_SIZE', while the code says 10_000. The reported emittedRows figures derive from it, so worth reconciling.

2. Could we get benchmark numbers across a range of block sizes? This constant trades per-block overhead against how much the downstream receiver buffers, and there is no data in the PR supporting 10_000 specifically.

3. Every other constant in this Broker block -- DEFAULT_MSE_STREAMING_GROUP_BY_FLUSH_THRESHOLD two lines above -- is paired with a CONFIG_OF_* key so an operator can change the default without rewriting every query. Either add one here, or reuse the block size selection already uses elsewhere rather than introducing a second, unconfigurable one.

[rohityadav1993]

1. Reconciled in the description; the default was always 10000.

2. Deferred. The bench harness (Benchmark harness for the streaming selection ORDER BY combine rohityadav1993/pinot#1) pins 10000 today, and the trade-off is best measured with PR2's receiver (Add streaming k-way merge to SortedMailboxReceiveOperator #19121) downstream.

3. Done in e3a908a as server config: pinot.server.query.executor.sorted.selection.merge.block.size, plus ...auto.min.sorted.ratio for AUTO, validated at init and applied when the query does not set the option (like numGroupsLimit). Server-side, so it reaches the MSE leaf and gRPC streaming without broker changes; the constant moved to Server. Kept separate from other block sizes since it sizes the blocks streamed downstream, not the segment scan.

[gortiz]

[test] Strong parity suite. Two gaps: schema mismatch, and the segment acquire/release lifecycle.

The parity approach here is good -- real segments, compared against the existing operator, both null-handling modes, mixed sorted/unsorted, pruning, the DESC fallback. That covers the merge logic well.

Two things not covered:

1. The data-schema-mismatch drop in SegmentCursor.pullBlock() -- see comment [14].
2. The acquire-on-activate / release-on-exhaust lifecycle. Nothing asserts that stop(), or an exception mid-merge, releases every cursor still holding a segment. That is the subtlest part of the design and the one that pins a segment for the lifetime of the server if it regresses. A mock IndexSegment counting acquire/release calls would pin it cheaply.

[rohityadav1993]

Both covered in e3a908a. Mismatch: see [14].

Lifecycle: instead of a mock segment, each real leaf is wrapped in a counting AcquireReleaseColumnsSegmentOperator (as the prefetch path plans it), so the combine's real calls are exercised. Tests check every acquire is released exactly once on full drain, on limit with cursors open, on stop() (including repeated), and on a child Exception or Error. Each was watched failing with its release removed.

[gortiz]

[test] The null 'ready' latch makes a regression fail with an NPE rather than the intended assertion.

The point of this test is that no child operator is ever driven, so null is only safe while that assertion holds. If the deadline check regresses, SlowOperator runs and NPEs on the latch, and the failure reads as a test bug rather than 'the deadline was not observed'. Passing a fresh CountDownLatch(1) that is simply never awaited keeps the failure mode legible.

[rohityadav1993]

Fixed in f3f2db3 - passes a fresh CountDownLatch(1) that's never awaited instead of null.

Correction: this can't actually NPE - the null guard predates this PR and still serves other tests. The real regression mode is worse for CI legibility: SlowOperator sleeps 3,600,000 ms, so a deadline regression hangs this test for an hour instead of failing. Not adding a @test timeout for that since it changes test semantics and wasn't asked for, but can if you'd like.

[rohityadav1993]

Follow-up: the offered timeOut is in (e3a908a), so a regression fails the deadline test instead of hanging the build.

[rohityadav1993]

@gortiz, @xiangfu0, requesting for a review, all blocking review comments are addressed and most of the non blocking comments have been addressed.