wangyum opened a new pull request, #59027:
URL: https://github.com/apache/spark/pull/59027
### What changes were proposed in this pull request?
Add a broadcast range join for range-predicate joins that today fall back to
`BroadcastNestedLoopJoinExec`. It is off by default
(`spark.sql.join.broadcastRangeJoin.enabled`).
`ExtractRangeJoinKeys` recognizes three shapes. When more than one matches,
point-in-range wins, then interval overlap, then a partial range.
- **Point-in-range.** One side is a point and the other is `(low, high)`,
for example `p >= lo AND p <= hi` or `p BETWEEN lo AND hi`. The keys are
`(point, point)` and `(low, high)`. The build side is an interval index, probed
with `overlapping(point, point)`.
- **Interval overlap.** Both sides are `(low, high)`, for example `a.lo <
b.hi AND b.lo < a.hi`. The keys are `(low, high)` on each side. The build side
is an interval index, probed with `overlapping(low, high)`.
- **Partial range.** One cross-side inequality, one column per side, for
example `a < b` or `a > b`. The keys are those two columns. The build side is a
sorted point index. `<` and `>` share that index. The join scans `upTo` or
`from` according to which side holds the lower bound.
The index returns a superset. Inclusivity, extra conjuncts, and collation
stay on the original condition and are evaluated for each candidate. Null keys
are not matches. An equi-join key still wins, so an equality plus a range
predicate stays a hash join.
Supported join types are inner, left outer, right outer, left semi, and left
anti. The preserved side is streamed. Full outer stays a broadcast nested loop
join. A `BROADCAST` hint is honored only when that build side is supported.
`SHUFFLE_REPLICATE_NL` stays a cartesian product. Types the index can order
(including a UDT ordered as its SQL type, and collated strings) are recognized.
Arrays, intervals, and other orderable types stay a nested loop join.
`BroadcastExchangeExec` builds the index through `RangeBroadcastMode`. Under
AQE, `LogicalQueryStageStrategy` rebuilds the operator when that mode matches.
Bound-key nullability is normalized so a stage rewrite still satisfies
`BroadcastDistribution`.
This is a broadcast index for a side that already fits the broadcast
threshold. It is not a sort-merge range join for two large sides, and there is
no `RANGE_JOIN` hint.
### Why are the changes needed?
A range predicate such as an IP lookup or an interval overlap has no
equality key, so Spark plans `BroadcastNestedLoopJoinExec` and compares every
streamed row with every build row. A broadcast index probes that same small
side in logarithmic time and then rechecks the original predicate.
### Does this PR introduce _any_ user-facing change?
Yes, behind a new configuration that defaults to false, so plans are
unchanged unless it is turned on. Compared with released Spark, the
configuration does not exist. On master, with it enabled, a supported range
predicate is planned as `BroadcastRangeJoinExec` instead of
`BroadcastNestedLoopJoinExec`.
```sql
SET spark.sql.join.broadcastRangeJoin.enabled = true;
SELECT s.id, l.zip_id
FROM signin_ip s
JOIN ip_lookup l
ON s.ip_add_int BETWEEN l.begin_ip AND l.end_ip;
```
Answers match the nested loop join. Full outer, an equality key, an
unsupported type, and a non-deterministic key keep the previous plan.
Overlapping build intervals make the index larger than the build table, and the
existing `spark.sql.maxBroadcastTableSize` check still applies to the index.
### How was this patch tested?
Unit and SQL tests:
- `ExtractRangeJoinKeysSuite`: point-in-range, overlap, partial range, and
rejected types.
- `RangeIndexSuite`: interval sweep, keys that compare equal but are not
identical, and point-index bounds.
- `RangeJoinSuite`: boundary inclusivity, both build sides, outer, semi,
anti, partial `upTo`/`from`, and interval overlap.
- `RangeJoinSQLSuite`: the conf off and on, `BETWEEN`, hints, join types, a
residual predicate, collation, AQE, and whole-stage codegen.
- `AdaptiveQueryExecSuite`: broadcast reuse, nullability after AQE,
build-side validation, and partition coalescing.
- Golden plans in `range-join.sql` and `range-join-disabled.sql`.
Local benchmark against broadcast nested loop join
([`RangeJoinBenchmark`](https://issues.apache.org/jira/secure/attachment/13084756/RangeJoinBenchmark.scala),
Apple M2 Max, OpenJDK 17). Selective cases probe 400,000 rows against 4,000
build rows. Best-time ratio:
| Case | Nested loop | Range join | Ratio |
|---|---:|---:|---:|
| Point-in-range inner, build ranges | 3793 ms | 57 ms | 66.6X |
| Point-in-range left / right outer | 4492 / 4467 ms | 56 / 55 ms | 80.6X /
81.2X |
| Point-in-range left semi / left anti | 2076 / 3917 ms | 50 / 41 ms | 41.9X
/ 94.5X |
| Interval overlap inner, build right | 3809 ms | 37 ms | 103.1X |
| Interval overlap left / right outer | 4022 / 4307 ms | 51 / 54 ms | 79.1X
/ 80.0X |
| Interval overlap left semi / left anti | 3911 / 4198 ms | 34 / 37 ms |
115.3X / 113.8X |
| Partial `<` / `>`, selective build side | 2749 / 2689 ms | 71 / 67 ms |
38.8X / 40.4X |
| Point-in-range, date / timestamp | 3877 / 3754 ms | 50 / 51 ms | 76.9X /
73.3X |
| Point-in-range, decimal | 9898 ms | 75 ms | 131.9X |
| Point-in-range, `UNICODE_CI` string | 212142 ms | 1464 ms | 144.9X |
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor (Grok 4.7)
--
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]
