Add streaming sorted merge join with colocation support - #19122
Draft
rohityadav1993 wants to merge 3 commits into
Draft
Add streaming sorted merge join with colocation support#19122rohityadav1993 wants to merge 3 commits into
rohityadav1993 wants to merge 3 commits into
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19122 +/- ##
============================================
+ Coverage 65.49% 65.60% +0.11%
Complexity 1423 1423
============================================
Files 3430 3435 +5
Lines 218010 219177 +1167
Branches 34648 34920 +272
============================================
+ Hits 142784 143796 +1012
- Misses 63666 63758 +92
- Partials 11560 11623 +63
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:
|
rohityadav1993
force-pushed
the
oss/pr3-sorted-merge-join
branch
2 times, most recently
from
August 5, 2026 19:28
05adc5a to
f62bb17
Compare
An unbounded leaf-stage ORDER BY (as injected for sorted merge join inputs) routes to MinMaxValueBasedSelectionOrderByCombineOperator, which merges every segment's rows into a single block before returning anything. At large data volumes this exceeds the leaf stage's CPU budget and ThreadAccountant raises EarlyTerminationException inside SelectionOperatorUtils.mergeWithOrdering(), surfacing to the broker as a spurious "Cancelled by sender". This adds a streaming alternative, opt-in via the `streamingSelectionOrderBy` query option: - StreamingSelectionOrderByOperator emits sorted blocks incrementally for a segment that is physically sorted on the leading ORDER BY column, reading the sorted forward index in order instead of building a priority queue. - StreamingSelectionOrderByCombineOperator performs a k-way heap merge across segment operators and emits bounded blocks (`streamingSelectionOrderByBlockSize`, default 10000) rather than one materialized result. - SelectionPlanNode and CombinePlanNode select these operators when the option is set and the sortedness precondition holds; otherwise behaviour is unchanged. Part of apache#18667.
SortedMailboxReceiveOperator accumulates every row from all senders and then sorts, even when each sender's stream is already sorted on the collation keys (noted as a TODO in the class). That defeats streaming and makes the receive stage's peak memory proportional to the full input. When the sending stage is known to produce sorted output, the receiver now performs a k-way heap merge across mailboxes and emits bounded sorted blocks as data arrives: - BlockingMultiStreamConsumer gains a StreamHandle<T> abstraction with awaitDataOrTerminal()/poll(), so a consumer can peek the head of each stream without draining it — the primitive the heap merge needs. - PlanFragmenter/PinotLogicalQueryPlanner propagate sender-side collation so the receiver can tell whether its inputs are individually sorted. - Opt-in via the `streamingSortedMailboxReceive` query option, with `streamingSortedMailboxReceiveBlockSize` controlling emitted block size. Without the option, the existing accumulate-then-sort path is used unchanged. Part of apache#18667.
Hash join materializes the entire right side into a hash table before probing, so peak memory grows linearly with right-side row count, and multi-column keys fall back to ObjectLookupTable with a composite key allocation per row. When both inputs are already sorted on the join keys, a merge join avoids the hash table entirely and streams. - SortedMergeJoinOperator performs a two-pointer merge with lazy block reads (one block per side held in memory), type-dispatched key comparison, non-equi residual filter support, split equi-only/filter paths, maxRowsInJoin overflow handling (THROW/BREAK) on both the buffered right-key run and the emitted rows, periodic termination checks, and early-termination propagation so a downstream LIMIT stops the join instead of completing the full cross-product of buffered input. - Selected via /*+ joinOptions(join_strategy='sorted') */, carried through the plan as JoinStrategy.SORTED. - PinotJoinExchangeNodeInsertRule injects a LogicalSort below the sort exchange so both inputs arrive globally sorted, not merely sorted per sender. Join keys are collated NULLS LAST, matching the operator's key comparator. - Colocation: joinOptions(is_colocated_by_join_keys='true') sets the existing prePartitioned field on the sort exchange, reusing the hash-join prePartitioned machinery, so co-partitioned inputs get a direct 1:1 exchange (receive fanIn drops to 1, no cross-server shuffle) with no changes needed in MailboxAssignmentVisitor or WorkerManager. Part of apache#18667.
rohityadav1993
force-pushed
the
oss/pr3-sorted-merge-join
branch
from
August 5, 2026 19:56
f62bb17 to
3d2219d
Compare
Jackie-Jiang
added
release-notes
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Hash join materializes the entire right side into a hash table before probing, so peak memory grows
linearly with right-side row count, and multi-column keys fall back to
ObjectLookupTablewith acomposite key allocation per row. When both inputs are already sorted on the join keys, a merge join
avoids the hash table entirely and streams. This is the Sorted Merge join proposal in #18667.
Approach
SortedMergeJoinOperator(new) performs a two-pointer merge holding one block per side, withtype-dispatched key comparison, non-equi residual filter support, split equi-only and filtered
paths,
maxRowsInJoinoverflow handling (THROW/BREAK) applied to both emitted rows andthe buffered right run, periodic deadline/termination sampling driven by a monotonic row counter,
and early-termination propagation so a downstream
LIMITstops the join rather than completing thecross-product of buffered input.
/*+ joinOptions(join_strategy='sorted') */, carried through the plan asJoinNode.JoinStrategy.SORTEDand the new proto valueJoinStrategy.SORTED = 3.RelToPlanNodeConverterrestricts it toINNERandLEFTjoins with at least one equi key.PinotJoinExchangeNodeInsertRuleinjects aLogicalSortbelow the sort exchange on each side,so inputs arrive globally sorted rather than merely sorted per sender. Collation is
ASCENDING NULLS LAST, and the merge's comparator is null-aware with matching nulls-lastsemantics, so a null-key row terminates a run and falls through to the outer loop.
Distribution-type hints are rejected rather than ignored —
broadcastin particular would breakthe merge's partitioning assumption — mirroring how the
lookupbranch rejects them.joinOptions(is_colocated_by_join_keys='true')hint viaPinotHintOptions.JoinHintOptions.isColocatedByJoinKeys(join)and passes it as theprePartitionedargument of the existingPinotLogicalSortExchange.create(...)overload. No newfield or plumbing is added by this PR — co-partitioned inputs then get a direct 1:1 exchange with
no changes in
MailboxAssignmentVisitororWorkerManager.MultiStageOperator.Type.SORTED_MERGE_JOIN(id 17) and anInStageStatsTreeBuildercase, so the operator appears under its own name instageStats.LOOKUPis now an explicit case there anddefault:throws, replacing anassert-guardedfall-through that would have silently mislabelled a future strategy in production.
No behaviour change without the hint
Joins default to
JoinStrategy.HASH.HASH = 0is the proto default, so a plan from anolder-version broker carrying no
joinStrategyfield deserializes to the existing behaviour.PlanNodeDeserializerthrows on an unknown strategy rather than degrading toHASH, keeping itsymmetric with
PlanNodeSerializer— a silent degrade would run e.g. anAS_OFplan as a hash joinwith its
matchConditiondropped.Tests
SortedMergeJoinOperatorTest(new)QueryCompilationTest(extended)testColocatedSortedMergeJoinIsPrePartitionedandtestNonColocatedSortedMergeJoinIsNotPrePartitionedPlanNodeSerDeTest(extended)testJoinStrategySerDe(iterates everyJoinStrategy, so it fails when a future strategy is added without serde wiring) andtestUnknownJoinStrategyFailsFastSortedMergeJoin.json(new, viaResourceBasedQueriesTest)pinot-query-plannersuiteKnown gaps
only, and a hash join produces the identical multiset for every one of them. They are correctness
coverage, not routing coverage.
assertions above; it has been validated on a cluster (below), not in CI.
SortedMergeJoinOperatordoes not extendBaseJoinOperatorand re-implements roughly 120 lines ofhint and option parsing, which has already begun to drift from the base. Worth consolidating in a
follow-up.
join_strategy='sorted'on its own still routes the receive side through accumulate-then-sortunless
streamingSortedMailboxReceiveis also set. The join is correct either way; only thestreaming property of the receive stage is lost.
Cluster verification
Two separate runs, on different builds. Both used an 87.8M-doc, 56-segment table.
On the current commit (4 servers, non-colocated, funnel-shaped join): results are byte-identical
to the hash-join baseline across all 196 output buckets. The join operator's
selfExecutionTimeMsis44 ms versus 2,079 ms for hash join, and
timeBuildingHashTableMs(1,412 ms) disappears entirely —the operator holds no hash table. End-to-end this query shape is ~1.28x slower than hash join
(median 907 ms vs 708 ms), because the sorted plan post-filters where hash join pushes the
time-bucket equality into
ON. The win here is capability and bounded memory, not wall clock.On an earlier, pre-rebase build (table Murmur-partitioned 128 ways on the join key), enabling
colocation:
fanInfanOutdeserializedBytesThis colocation measurement has not been re-taken on the current commit; the colocation code path
is unchanged in substance since (it now reuses the upstream
prePartitionedfield instead of alocally added one). The logical
EXPLAINis unchanged either way — the difference appears only inthe dispatched plan and
stageStats.Caveats for colocation: the query must carry the table partition hints so
WorkerManagercanvalidate hint-vs-actual partition info;
partition_sizemust divide the actual partition count; andon datasets with empty partitions a smaller
partition_sizeis needed to route throughassignMultiplePartitionsPerWorker, sinceassignOnePartitionPerWorkerrequires a segment for everypartition (a pre-existing limitation).
Stacking
Part of #18667.