rohityadav1993 opened a new pull request, #19120:
URL: https://github.com/apache/pinot/pull/19120
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 exceeds 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 that
is 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) performs a k-way heap
merge across the
per-segment operators and emits bounded blocks rather than one
materialized result. Segments are
acquired and released incrementally, and the merge loop performs periodic
termination /
deadline / resource-usage sampling, matching
`BaseStreamingCombineOperator`.
- **`SelectionPlanNode`**, **`CombinePlanNode`**,
**`InstancePlanMakerImplV2`** and **`QueryContext`**
select these operators only when the option is set and the sortedness
precondition holds; any
unmet precondition falls 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 behaviour 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
64 tests, all green:
| Test class | Count |
|---|---|
| `StreamingSelectionOrderByOperatorTest` (new) | 13 |
| `StreamingSelectionOrderByCombineOperatorTest` (new) | 15 |
| `CombineSlowOperatorsTest` (extended) | 11 |
| `QueryOptionsUtilsTest` (extended) | 25 |
`CombineSlowOperatorsTest.testStreamingSelectionOrderByCombineOperatorHonorsDeadline`
pins the
deadline behaviour specifically: an already-expired deadline must yield an
`ExceptionResultsBlock`
before any child operator is driven.
### Known gaps
- **No integration test.** This PR touches no file under
`pinot-integration-tests`; the broker-side
reduce, `OFFSET` handling, and the streaming response path are covered
only at operator level.
Happy to add one here if reviewers prefer that over a follow-up.
- The multi-column path rebuilds the second-pass operator chain per emitted
equal-prefix run rather
than per segment, which is a cost cliff on high-cardinality leading sort
columns
(e.g. `ORDER BY ts, id`). Correct, but worth a follow-up.
### Commits
This PR contains **2 commits**: the feature, and a separate follow-up commit
applying review fixes
(termination/deadline checks in the merge loop, schema-mismatch reporting,
segment release on
`Error` as well as `Exception`, scan-cost accounting, and the query-option
naming above). They are
kept separate so the fixes are reviewable on their own; happy to squash
before merge.
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]
