rohityadav1993 opened a new pull request, #19122:
URL: https://github.com/apache/pinot/pull/19122
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. This is the **Sorted Merge
join** proposal in #18667.
### Approach
- **`SortedMergeJoinOperator`** (new) performs a two-pointer merge holding
one block per side, with
type-dispatched key comparison, non-equi residual filter support, split
equi-only and filtered
paths, `maxRowsInJoin` overflow handling (`THROW` / `BREAK`) applied to
**both** emitted rows and
the buffered right run, periodic deadline/termination sampling driven by a
monotonic row counter,
and early-termination propagation so a downstream `LIMIT` stops the join
rather than completing the
cross-product of buffered input.
- Selected via `/*+ joinOptions(join_strategy='sorted') */`, carried through
the plan as
`JoinNode.JoinStrategy.SORTED` and the new proto value
`JoinStrategy.SORTED = 3`.
`RelToPlanNodeConverter` restricts it to `INNER` and `LEFT` joins with at
least one equi key.
- **`PinotJoinExchangeNodeInsertRule`** injects a `LogicalSort` below 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-last
semantics, so a null-key row terminates a run and falls through to the
outer loop.
Distribution-type hints are rejected rather than ignored — `broadcast` in
particular would break
the merge's partitioning assumption — mirroring how the `lookup` branch
rejects them.
- **Colocation** reuses machinery that is already on master. The rule reads
the existing
`joinOptions(is_colocated_by_join_keys='true')` hint via
`PinotHintOptions.JoinHintOptions.isColocatedByJoinKeys(join)` and passes
it as the
`prePartitioned` argument of the existing
`PinotLogicalSortExchange.create(...)` overload. **No new
field or plumbing is added by this PR** — co-partitioned inputs then get a
direct 1:1 exchange with
no changes in `MailboxAssignmentVisitor` or `WorkerManager`.
- **Observability:** new `MultiStageOperator.Type.SORTED_MERGE_JOIN` (id 17)
and an
`InStageStatsTreeBuilder` case, so the operator appears under its own name
in `stageStats`.
`LOOKUP` is now an explicit case there and `default:` throws, replacing an
`assert`-guarded
fall-through that would have silently mislabelled a future strategy in
production.
### No behaviour change without the hint
Joins default to `JoinStrategy.HASH`. `HASH = 0` is the proto default, so a
plan from an
older-version broker carrying no `joinStrategy` field deserializes to the
existing behaviour.
`PlanNodeDeserializer` throws on an unknown strategy rather than degrading
to `HASH`, keeping it
symmetric with `PlanNodeSerializer` — a silent degrade would run e.g. an
`AS_OF` plan as a hash join
with its `matchCondition` dropped.
### Tests
| Test class | Count |
|---|---|
| `SortedMergeJoinOperatorTest` (new) | 26 — hash-join parity, multi-block
streaming, null keys sorted last, error propagation, non-equi conditions,
overflow modes, early termination, non-globally-sorted input |
| `QueryCompilationTest` (extended) | 228, incl.
`testColocatedSortedMergeJoinIsPrePartitioned` and
`testNonColocatedSortedMergeJoinIsNotPrePartitioned` |
| `PlanNodeSerDeTest` (extended) | 126, incl. `testJoinStrategySerDe`
(iterates every `JoinStrategy`, so it fails when a future strategy is added
without serde wiring) and `testUnknownJoinStrategyFailsFast` |
| `SortedMergeJoin.json` (new, via `ResourceBasedQueriesTest`) | 5 E2E
queries validated against H2 |
| full `pinot-query-planner` suite | 1453 |
### Known gaps
- **The E2E queries do not assert that the sorted strategy actually ran.**
All 5 validate against H2
only, and a hash join produces the identical multiset for every one of
them. They are correctness
coverage, not routing coverage.
- **The colocated pre-partitioned path is not exercised by any test** beyond
the two planner-level
assertions above; it has been validated on a cluster (below), not in CI.
- `SortedMergeJoinOperator` does not extend `BaseJoinOperator` and
re-implements roughly 120 lines of
hint 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-sort
unless `streamingSortedMailboxReceive` is also set. The join is correct
either way; only the
streaming 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
`selfExecutionTimeMs` is
44 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:**
| Metric (per join-input receive) | Colocated | Non-colocated |
|---|---|---|
| receive `fanIn` | 1 | 2 |
| send `fanOut` | 1 (direct 1:1) | 2 (all-to-all) |
| cross-server bytes | ~0 (same server) | ~50.5 MB `deserializedBytes` |
| join output rows | 6,693,571 | 6,693,571 (identical) |
This 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 `prePartitioned`
field instead of a
locally added one). The logical `EXPLAIN` is unchanged either way — the
difference appears only in
the dispatched plan and `stageStats`.
**Caveats for colocation:** the query must carry the table partition hints
so `WorkerManager` can
validate hint-vs-actual partition info; `partition_size` must divide the
actual partition count; and
on datasets with empty partitions a smaller `partition_size` is needed to
route through
`assignMultiplePartitionsPerWorker`, since `assignOnePartitionPerWorker`
requires a segment for every
partition (a pre-existing limitation).
### Stacking
> Stacked on #19121. The GitHub diff for this PR includes its parents'
commits until they merge —
> **review only the top commit.** This PR compiles independently of its
immediate parent; it is
> stacked for review ordering, since the three PRs together implement one
feature.
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]
