Re: [PR] [SPARK-8682][SQL] Support range join for range-predicate joins [spark]

[via GitHub]

cloud-fan commented on code in PR #59027:
URL: https://github.com/apache/spark/pull/59027#discussion_r4367132555


##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1220,6 +1220,22 @@ object SQLConf {
     .bytesConf(ByteUnit.BYTE)
     .createWithDefaultString("10MB")
 
+  val BROADCAST_RANGE_JOIN_ENABLED = 
buildConf("spark.sql.join.broadcastRangeJoin.enabled")
+    .internal()
+    .doc("When true, joins with range predicates (such as point-in-range, 
interval-overlap, " +
+      "or single inequalities) are planned as broadcast range joins instead of 
broadcast " +
+      "nested loop joins if the build side is eligible for broadcast. 
Supported join types " +
+      "include inner, left outer, right outer, left semi, and left anti. 
Broadcast range joins " +
+      "build an index on the broadcast side to significantly accelerate 
selective range probes, " +
+      "but perform worse than nested loop joins when most, but not all, of the 
build intervals " +

Review Comment:
   **Nit (P3):** I built this head and ran the committed staggered shape (2,000 
build intervals, each either ending before the probe window or spanning it, 
10,000 probes) at lower fractions of spanning intervals, spread across start 
order, against the same query with the flag off (local[1], JDK 17, AQE off, 
best of five, two runs). Range-join speedup over the nested loop join:
   
   | intervals reaching the probe | double keys | long keys |
   |---|---:|---:|
   | 0% | 3.0X-3.1X | 2.2X |
   | 2% | 1.35X-1.47X | 1.8X |
   | 5% | 0.77X-0.98X | 1.25X-1.40X |
   | 10% | 0.63X-0.75X | 1.04X-1.10X |
   | 25% | 0.63X-0.82X | 0.84X-0.98X |
   | 50% | 0.53X-0.59X | 0.83X-0.87X |
   
   A subtree that contains even one interval reaching the probe can't be 
pruned, so a small minority of long or open-ended intervals keeps most of the 
tree alive, and boxed keys pay an `Ordering[Any]` comparison at every visited 
node. This sentence is close for long keys but not for boxed keys. Could the 
doc state the boundary per key family, backed by benchmark rows at about 5%, 
10%, and 25%?
   
   **Verification:**
   - **Inspection:** Every slow-regime statement in the config doc agrees with 
committed benchmark rows that cover interleaved overlap at about 5%, 10%, 25%, 
50%, and 90% for long and double keys in all three JDK results files.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeRelation.scala:
##########
@@ -0,0 +1,1240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import java.io.{Externalizable, ObjectInput, ObjectOutput}
+
+import scala.collection.mutable.ArrayBuffer
+
+import com.esotericsoftware.kryo.{Kryo, KryoSerializable}
+import com.esotericsoftware.kryo.io.{Input, Output}
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{BoundReference, Expression, 
InterpretedProjection}
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildSide}
+import org.apache.spark.sql.catalyst.planning.{IntervalOverlapJoin, 
PartialRangeJoin, PointInRangeJoin, RangeJoinKind}
+import org.apache.spark.sql.catalyst.plans.physical.BroadcastMode
+import org.apache.spark.sql.catalyst.types.PhysicalDataType
+import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DateType, 
Decimal, DecimalType, IntegerType, LongType, ShortType, TimestampNTZType, 
TimestampType}
+import org.apache.spark.util.{KnownSizeEstimation, SizeEstimator, Utils}
+
+/**
+ * Broadcast payload of a range join. An index only names candidate rows.
+ * [[BroadcastRangeJoinExec]] evaluates the original join condition to filter
+ * candidates. Keys that map to one order-preserving long are stored raw and
+ * compared unboxed ([[LongIntervalIndex]], [[LongPointIndex]]); every other
+ * supported type is compared as boxed values ([[IntervalIndex]],
+ * [[PointIndex]]). [[RangeRelationBroadcastMode]] picks the family from the
+ * build keys' data types.
+ *
+ *  - The interval indexes answer "which ranges overlap this window?"
+ *  - The point indexes answer "which points lie on this side of a bound?"
+ */
+private[execution] sealed trait RangeRelation extends KnownSizeEstimation with 
Serializable {
+  def keyType: DataType
+}
+
+/**
+ * Which [[RangeRelation]] a broadcast materializes.
+ *
+ * A point index has no direction, so one broadcast of a column serves both
+ * `<` and `>`. The join records which side holds the lower bound and scans
+ * accordingly.
+ */
+private[execution] sealed abstract class RangeIndexKind {
+  def matches(rangeJoinKind: RangeJoinKind): Boolean = 
RangeIndexKind(rangeJoinKind) == this
+}
+
+private[execution] case object IntervalIndexKind extends RangeIndexKind
+
+private[execution] case object PointIndexKind extends RangeIndexKind
+
+private[execution] object RangeIndexKind {
+  def apply(rangeJoinKind: RangeJoinKind): RangeIndexKind = rangeJoinKind 
match {
+    case PointInRangeJoin | IntervalOverlapJoin => IntervalIndexKind
+    case _: PartialRangeJoin => PointIndexKind
+  }
+}
+
+/**
+ * The boxed-key index family's shared ordering cache, derived from `keyType`
+ * on first comparison and kept out of the serialized form, so each executor
+ * re-derives its own instance instead of receiving one per broadcast. The
+ * long index family compares raw keys and has no ordering to cache.
+ */
+private[joins] sealed trait BoxedKeyOrdering extends RangeRelation {
+  @transient
+  @volatile private[this] var _ordering: Ordering[Any] = _
+
+  private[joins] def ordering: Ordering[Any] = {
+    var ord = _ordering
+    if (ord == null) {
+      ord = PhysicalDataType.ordering(keyType)
+      _ordering = ord
+    }
+    ord
+  }
+}
+
+/**
+ * The long index family's shared key conversion, derived from `keyType` on
+ * first probe and kept out of the serialized form, so each executor derives
+ * its own converter after deserialization. The interpreted probe path calls
+ * [[toLong]] once per streamed key; generated probe code keeps its own
+ * conversion, emitted at plan time.
+ */
+private[joins] sealed trait LongKeyConversion extends RangeRelation {
+  @transient
+  @volatile private[this] var _toLong: Any => Long = _
+
+  /**
+   * One evaluated streamed key to its order-preserving long. `RangePredicate`
+   * gives both sides of a range predicate one type after analysis, so the
+   * streamed key type is this index's `keyType`.
+   */
+  private[joins] def toLong(key: Any): Long = {
+    var f = _toLong
+    if (f == null) {
+      f = RangeIndex.requiredLongKey(keyType).value
+      _toLong = f
+    }
+    f(key)
+  }
+}
+
+private[joins] object RangeIndex {
+
+  /**
+   * Capacity for a build buffer from the driver's row-count hint, capped at 
2^20 for any
+   * larger hint (well under a hint that large) and equal to the hint itself 
below that.
+   * `BroadcastExchangeExec` only checks the row count against its row-count
+   * limit (up to half a billion for this mode) before calling `transform`, 
and only
+   * checks the built relation's byte size afterward, so sizing a buffer 
directly off an
+   * inflated hint can allocate hundreds of megabytes before a single row is 
read - ahead
+   * of the size check that is supposed to catch an over-large broadcast. 
Starting small
+   * and letting the buffer grow from there bounds only the initial 
allocation; the final
+   * size still tracks the actual row count, which is what the post-build size 
check sees.
+   */
+  def initialCapacity(sizeHint: Option[Long]): Int =
+    sizeHint.map(n => math.min(n, 1L << 20).toInt).getOrElse(64)
+
+  /** Accessor reading a single field from `InternalRow` as a boxed value (or 
null). */
+  def getValue(dt: DataType, ordinal: Int): InternalRow => Any = {
+    val accessor = InternalRow.getAccessor(dt)
+    (input: InternalRow) => accessor(input, ordinal)
+  }
+
+  /**
+   * One key type's bridge to an order-preserving long: reading a projected 
build
+   * field, converting an evaluated streamed value, and turning generated Java 
for
+   * a typed key into generated Java for a long. The three agree, so a build 
key
+   * and a probe key of the same type become the same long.
+   */
+  final case class LongKey(
+      field: (InternalRow, Int) => Long,
+      value: Any => Long,
+      code: String => String)
+
+  /**
+   * An order-preserving long for a key type that has one, or None when the 
type
+   * must stay on the generic boxed index: variable-length keys, floating 
point,
+   * whose SQL equality for -0.0 and NaN has no bit-preserving order, and
+   * decimals wider than `Decimal.MAX_LONG_DIGITS`, whose unscaled value does
+   * not fit a long. This is the only place a key type is specialized; the long
+   * indexes and the generated probe code both consult it.
+   */
+  def longKey(dt: DataType): Option[LongKey] = {
+    def key(field: (InternalRow, Int) => Long, value: Any => Long): LongKey =
+      LongKey(field, value, v => s"((long) $v)")
+    dt match {
+      case _: BooleanType =>
+        Some(LongKey(
+          (r, o) => if (r.getBoolean(o)) 1L else 0L,
+          v => if (v.asInstanceOf[Boolean]) 1L else 0L,
+          v => s"($v ? 1L : 0L)"))
+      case _: ByteType =>
+        Some(key((r, o) => r.getByte(o).toLong, v => 
v.asInstanceOf[Byte].toLong))
+      case _: ShortType =>
+        Some(key((r, o) => r.getShort(o).toLong, v => 
v.asInstanceOf[Short].toLong))
+      case _: IntegerType | _: DateType =>
+        Some(key((r, o) => r.getInt(o).toLong, v => 
v.asInstanceOf[Int].toLong))
+      case _: LongType | _: TimestampType | _: TimestampNTZType =>
+        Some(key((r, o) => r.getLong(o), v => v.asInstanceOf[Long]))
+      case d: DecimalType if d.precision <= Decimal.MAX_LONG_DIGITS =>
+        Some(LongKey(
+          (r, o) => r.getDecimal(o, d.precision, d.scale).toUnscaledLong,
+          v => v.asInstanceOf[Decimal].toUnscaledLong,
+          v => s"$v.toUnscaledLong()"))
+      case _ => None
+    }
+  }
+
+  /**
+   * Every key maps to one order-preserving long. 
[[RangeRelationBroadcastMode]] picks
+   * the index family by this test and [[BroadcastRangeJoinExec]] picks its 
probe
+   * code by the same one, so the two cannot drift apart. `RangePredicate` 
gives
+   * both sides of a range predicate one type after analysis, so testing the
+   * build keys also settles the streamed keys.
+   */
+  def canMapKeysToLong(keys: Seq[Expression]): Boolean =
+    keys.forall(k => longKey(k.dataType).isDefined)
+
+  /** longKey for a type the caller has routed to the long family. */
+  def requiredLongKey(dt: DataType): LongKey = {
+    longKey(dt).getOrElse(
+      throw new IllegalStateException(s"$dt has no order-preserving long 
key."))
+  }
+}
+
+/**
+ * Array-backed candidate scan. The index hands rows to the join in array 
order,
+ * without a closure or a tuple hop: the per-pair overhead then matches a bare
+ * loop over a broadcast row array. One instance per probe; never serialized.
+ */
+private[joins] final class RowSlice(
+    rows: Array[InternalRow],
+    from: Int,
+    until: Int) extends Iterator[InternalRow] {
+  private var i = from
+  def hasNext: Boolean = i < until
+  def next(): InternalRow = {
+    val r = rows(i)
+    i += 1
+    r
+  }
+}
+
+private[joins] object RowSlice {
+  /** A shared empty scan: a null probe bound matches nothing. */
+  val empty: RowSlice = new RowSlice(new Array[InternalRow](0), 0, 0)
+}
+
+// ===== Boxed-key indexes ====================================================
+//
+// Keys are the boxed values `RangeIndex.getValue` reads, compared through
+// `BoxedKeyOrdering`. Floating point, strings, binary, and decimals wider
+// than 18 digits stay here: these types have no order-preserving
+// single-long form, which is the property that routes a key type to the
+// long indexes below (`RangeIndex.longKey`).
+
+/**
+ * Augmented interval tree of build-side intervals. Lows and highs sit in
+ * parallel arrays sorted by low bound, with a row array in the same order
+ * feeding the scan path and `maxHighIdx(mid)` indexing the maximum high
+ * bound in the span rooted at mid. Each key and row object is owned once:
+ * the tree holds one reference per value plus the single row array. Java
+ * serialization keeps these aliases; Kryo's reference resolver does not
+ * track primitive wrappers, so a key shared by `lows` and `highs`, such as
+ * the one key of a degenerate build, can deserialize as two objects.
+ * Subtrees whose max high is below the probe window, or whose low starts
+ * after it, are pruned during search, and a window covering every interval
+ * skips the walk entirely.
+ */
+private[joins] class IntervalIndex private (
+    private[this] var _keyType: DataType,
+    private[this] var lows: Array[Any],
+    private[this] var highs: Array[Any],
+    private[this] var maxHighIdx: Array[Int],
+    private[this] var rows: Array[InternalRow])
+  extends BoxedKeyOrdering with Externalizable with KryoSerializable {
+
+  // Needed for Java/Kryo deserialization; see readExternal/read below.
+  def this() = this(null, null, null, null, null)
+
+  def keyType: DataType = _keyType
+
+  // All-covering window bounds. `lows` is sorted, so the last entry is the
+  // largest low: it bounds every low from above, and coreHigh is the minimum
+  // high, which bounds every high from below. A window overlaps every stored
+  // interval exactly when it holds probeLow <= coreHigh and probeHigh >= 
coreLow,
+  // whether or not the intervals share a common core.
+  // Recomputed in `read` too, since `lows`/`highs` are reassigned there.
+  private[this] var coreLow: Any = IntervalIndex.coreLow(lows)
+  private[this] var coreHigh: Any = IntervalIndex.coreHigh(highs, ordering)
+
+  /**
+   * Measured over the retained arrays, not the index object: SizeEstimator
+   * would consult `estimatedSize` for a KnownSizeEstimation and recurse.
+   * Its identity tracking charges a key shared by `lows` and `highs`, or
+   * aliased across indexes, exactly once.
+   */
+  override def estimatedSize: Long =
+    SizeEstimator.estimate(Array[AnyRef](lows, highs, maxHighIdx, rows))
+
+  def overlapping(low: Any, high: Any): Iterator[InternalRow] = {
+    if (lows.isEmpty || low == null || high == null) return Iterator.empty
+    val ord = ordering
+    val (probeLow, probeHigh) = if (ord.lteq(low, high)) (low, high) else 
(high, low)
+    // A window that covers every interval gets every row the walk would
+    // return, without visiting the tree. Returns all rows iff every interval
+    // overlaps the probe; the condition still re-checks each candidate.
+    if (ord.lteq(probeLow, coreHigh) && ord.gteq(probeHigh, coreLow)) {
+      return new RowSlice(rows, 0, rows.length)
+    }
+    new OverlapIterator(lows, highs, maxHighIdx, rows, ord, probeLow, 
probeHigh)
+  }
+
+  override def writeExternal(out: ObjectOutput): Unit = Utils.tryOrIOException 
{
+    write(out.writeUTF, out.writeInt, (o: Any) => 
out.writeObject(o.asInstanceOf[AnyRef]))
+  }
+
+  override def write(kryo: Kryo, out: Output): Unit = Utils.tryOrIOException {
+    write(out.writeString, out.writeInt,
+      (o: Any) => kryo.writeClassAndObject(out, o.asInstanceOf[AnyRef]))
+  }
+
+  private def write(
+      writeString: String => Unit,
+      writeInt: Int => Unit,
+      writeObj: Any => Unit): Unit = {
+    writeString(keyType.json)
+    val n = lows.length
+    writeInt(n)
+    var i = 0
+    while (i < n) {
+      writeObj(lows(i))
+      writeObj(highs(i))
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      writeInt(maxHighIdx(i))
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      writeObj(rows(i))
+      i += 1
+    }
+  }
+
+  override def readExternal(in: ObjectInput): Unit = Utils.tryOrIOException {
+    read(() => in.readUTF(), () => in.readInt(), () => in.readObject())
+  }
+
+  override def read(kryo: Kryo, in: Input): Unit = Utils.tryOrIOException {
+    read(() => in.readString(), () => in.readInt(), () => 
kryo.readClassAndObject(in))
+  }
+
+  private def read(
+      readString: () => String,
+      readInt: () => Int,
+      readObj: () => AnyRef): Unit = {
+    _keyType = DataType.fromJson(readString())
+    val n = readInt()
+    lows = new Array[Any](n)
+    highs = new Array[Any](n)
+    maxHighIdx = new Array[Int](n)
+    rows = new Array[InternalRow](n)
+    var i = 0
+    while (i < n) {
+      lows(i) = readObj()
+      highs(i) = readObj()
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      maxHighIdx(i) = readInt()
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      rows(i) = readObj().asInstanceOf[InternalRow]
+      i += 1
+    }
+    coreLow = IntervalIndex.coreLow(lows)
+    coreHigh = IntervalIndex.coreHigh(highs, ordering)
+  }
+}
+
+private[joins] object IntervalIndex {
+
+  // Shared by the primary constructor and `read`, which recomputes these
+  // caches after reassigning `lows`/`highs` on deserialization. `lows`/
+  // `highs` are null right after the zero-arg constructor runs, before
+  // `readExternal`/`read` populate them.
+  private[joins] def coreLow(lows: Array[Any]): Any =
+    if (lows == null || lows.isEmpty) null else lows(lows.length - 1)
+
+  // `ordering` is by-name: the zero-arg constructor calls this with `highs ==
+  // null`, before `keyType` is set, and forcing `ordering` eagerly would
+  // throw trying to derive an ordering for a not-yet-known key type.
+  private[joins] def coreHigh(highs: Array[Any], ordering: => Ordering[Any]): 
Any = {
+    if (highs == null || highs.isEmpty) {
+      null
+    } else {
+      val ord = ordering
+      var min = highs(0)
+      var i = 1
+      while (i < highs.length) {
+        if (ord.lt(highs(i), min)) min = highs(i)
+        i += 1
+      }
+      min
+    }
+  }
+
+  def build(
+      keyType: DataType,
+      buildKeys: Seq[Expression],
+      rows: Iterator[InternalRow],
+      sizeHint: Option[Long]): IntervalIndex = {
+    val buffer = new ArrayBuffer[(Any, Any, 
InternalRow)](RangeIndex.initialCapacity(sizeHint))
+    val projection = new InterpretedProjection(buildKeys)
+    val lowOf = RangeIndex.getValue(buildKeys.head.dataType, 0)
+    val ordering = PhysicalDataType.ordering(keyType)
+
+    if (buildKeys.length == 1) {
+      // A point build key is one key used as both ends, so it is copied once
+      // and serves as both bounds, matching `LongIntervalIndex.build`.
+      while (rows.hasNext) {
+        val r = rows.next()
+        val p = projection(r)
+        val rawLo = lowOf(p)
+        if (rawLo != null) {
+          val key = InternalRow.copyValue(rawLo)
+          buffer += ((key, key, r.copy()))
+        }
+      }
+    } else {
+      val highOf = RangeIndex.getValue(buildKeys(1).dataType, 1)
+      while (rows.hasNext) {
+        val r = rows.next()
+        val p = projection(r)
+        val rawLo = lowOf(p)
+        val rawHi = highOf(p)
+        if (rawLo != null && rawHi != null) {
+          val lo = InternalRow.copyValue(rawLo)
+          val hi = InternalRow.copyValue(rawHi)
+          val (normLo, normHi) = if (ordering.lteq(lo, hi)) (lo, hi) else (hi, 
lo)
+          buffer += ((normLo, normHi, r.copy()))
+        }
+      }
+    }
+
+    sortAndBuild(keyType, ordering, buffer.toArray)
+  }
+
+  private def sortAndBuild(
+      keyType: DataType,
+      ordering: Ordering[Any],
+      valid: Array[(Any, Any, InternalRow)]): IntervalIndex = {
+    java.util.Arrays.sort(valid, Ordering.by[(Any, Any, InternalRow), 
Any](_._1)(ordering))
+    val length = valid.length
+    val lows = new Array[Any](length)
+    val highs = new Array[Any](length)
+    val rows = new Array[InternalRow](length)
+    var i = 0
+    while (i < length) {
+      lows(i) = valid(i)._1
+      highs(i) = valid(i)._2
+      rows(i) = valid(i)._3
+      i += 1
+    }
+
+    // `maxHighIdx(mid)` indexes the maximum high bound in the span rooted at
+    // mid. Storing an index instead of aliasing the key keeps each key owned,
+    // and serialized, exactly once.
+    val maxHighIdx = new Array[Int](length)
+
+    def fillMaxHigh(start: Int, end: Int): Int = {
+      if (start >= end) return -1
+      val mid = (start + end) >>> 1
+      var best = mid
+      val left = fillMaxHigh(start, mid)
+      if (left >= 0 && ordering.gt(highs(left), highs(best))) best = left
+      val right = fillMaxHigh(mid + 1, end)
+      if (right >= 0 && ordering.gt(highs(right), highs(best))) best = right
+      maxHighIdx(mid) = best
+      best
+    }
+
+    fillMaxHigh(0, length)
+    new IntervalIndex(keyType, lows, highs, maxHighIdx, rows)
+  }
+}
+
+private[joins] class OverlapIterator(
+    lows: Array[Any],
+    highs: Array[Any],
+    maxHighIdx: Array[Int],
+    rows: Array[InternalRow],
+    ord: Ordering[Any],
+    probeLow: Any,
+    probeHigh: Any) extends Iterator[InternalRow] {
+
+  private[this] val depth = 32 - 
java.lang.Integer.numberOfLeadingZeros(lows.length)
+  private[this] var stackStart = new Array[Int](math.max(8, depth * 3))
+  private[this] var stackEnd = new Array[Int](math.max(8, depth * 3))
+  private[this] var stackTop = 0
+  private[this] var nextRow: InternalRow = null
+
+  push(0, lows.length)
+
+  private[joins] def stackCapacity: Int = stackStart.length
+  private[joins] def pendingSpans: Int = stackTop
+
+  private def push(s: Int, e: Int): Unit = {
+    if (stackTop >= stackStart.length) {
+      stackStart = java.util.Arrays.copyOf(stackStart, stackStart.length * 2)
+      stackEnd = java.util.Arrays.copyOf(stackEnd, stackEnd.length * 2)
+    }
+    stackStart(stackTop) = s
+    stackEnd(stackTop) = e
+    stackTop += 1
+  }
+
+  private def advance(): InternalRow = {
+    while (stackTop > 0) {
+      stackTop -= 1
+      val start = stackStart(stackTop)
+      val end = stackEnd(stackTop)
+      if (start < end) {
+        val mid = (start + end) >>> 1
+        if (ord.gteq(highs(maxHighIdx(mid)), probeLow)) {
+          if (ord.lteq(lows(mid), probeHigh)) {
+            push(mid + 1, end)
+          }
+          push(mid, -1)
+          push(start, mid)
+        }
+      } else if (end == -1) {
+        if (ord.gteq(highs(start), probeLow) && ord.lteq(lows(start), 
probeHigh)) {
+          return rows(start)
+        }
+      }
+    }
+    null
+  }
+
+  override def hasNext: Boolean = {
+    if (nextRow != null) {
+      true
+    } else {
+      nextRow = advance()
+      nextRow != null
+    }
+  }
+
+  override def next(): InternalRow = {
+    if (!hasNext) throw new NoSuchElementException("empty iterator")
+    val r = nextRow
+    nextRow = null
+    r
+  }
+}
+
+/**
+ * Sorted build-side points for single cross-side inequalities (`<`, `<=`, 
`>`, `>=`).
+ * Keys sit beside the rows they name, one owned copy each, in key order.
+ */
+private[joins] class PointIndex private (
+    private[this] var _keyType: DataType,
+    private[this] var keys: Array[Any],
+    private[this] var rows: Array[InternalRow])
+  extends BoxedKeyOrdering with Externalizable with KryoSerializable {
+
+  // Needed for Java/Kryo deserialization; see readExternal/read below.
+  def this() = this(null, null, null)
+
+  def keyType: DataType = _keyType
+
+  private[this] def length: Int = keys.length
+
+  override def estimatedSize: Long = 
SizeEstimator.estimate(Array[AnyRef](keys, rows))
+
+  /** Points whose key is less than or equal to `value`. */
+  def upTo(value: Any): Iterator[InternalRow] =
+    if (value == null) Iterator.empty else slice(0, searchBound(value, 
skipEquals = true))
+
+  /** Points whose key is greater than or equal to `value`. */
+  def from(value: Any): Iterator[InternalRow] =
+    if (value == null) Iterator.empty else slice(searchBound(value, skipEquals 
= false), length)
+
+  /**
+   * Points whose key is strictly less than `value`. For a strict partial-range
+   * predicate the original condition rejects keys equal to the probe value,
+   * so the scan can skip them: a build heavy in duplicated keys would only
+   * hand the join rows that all fail.
+   */
+  def upToExclusive(value: Any): Iterator[InternalRow] =
+    if (value == null) Iterator.empty else slice(0, searchBound(value, 
skipEquals = false))
+
+  /** Points whose key is strictly greater than `value`. */
+  def fromExclusive(value: Any): Iterator[InternalRow] =
+    if (value == null) Iterator.empty else slice(searchBound(value, skipEquals 
= true), length)
+
+  private def searchBound(value: Any, skipEquals: Boolean): Int = {
+    val ord = ordering
+    var lo = 0
+    var hi = length
+    while (lo < hi) {
+      val mid = (lo + hi) >>> 1
+      val c = ord.compare(keys(mid), value)
+      if (c < 0 || (skipEquals && c == 0)) lo = mid + 1 else hi = mid
+    }
+    lo
+  }
+
+  private def slice(from: Int, until: Int): Iterator[InternalRow] = new 
RowSlice(rows, from, until)
+
+  override def writeExternal(out: ObjectOutput): Unit = Utils.tryOrIOException 
{
+    write(out.writeUTF, out.writeInt, (o: Any) => 
out.writeObject(o.asInstanceOf[AnyRef]))
+  }
+
+  override def write(kryo: Kryo, out: Output): Unit = Utils.tryOrIOException {
+    write(out.writeString, out.writeInt,
+      (o: Any) => kryo.writeClassAndObject(out, o.asInstanceOf[AnyRef]))
+  }
+
+  private def write(
+      writeString: String => Unit,
+      writeInt: Int => Unit,
+      writeObj: Any => Unit): Unit = {
+    writeString(keyType.json)
+    val n = keys.length
+    writeInt(n)
+    var i = 0
+    while (i < n) {
+      writeObj(keys(i))
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      writeObj(rows(i))
+      i += 1
+    }
+  }
+
+  override def readExternal(in: ObjectInput): Unit = Utils.tryOrIOException {
+    read(() => in.readUTF(), () => in.readInt(), () => in.readObject())
+  }
+
+  override def read(kryo: Kryo, in: Input): Unit = Utils.tryOrIOException {
+    read(() => in.readString(), () => in.readInt(), () => 
kryo.readClassAndObject(in))
+  }
+
+  private def read(
+      readString: () => String,
+      readInt: () => Int,
+      readObj: () => AnyRef): Unit = {
+    _keyType = DataType.fromJson(readString())
+    val n = readInt()
+    keys = new Array[Any](n)
+    rows = new Array[InternalRow](n)
+    var i = 0
+    while (i < n) {
+      keys(i) = readObj()
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      rows(i) = readObj().asInstanceOf[InternalRow]
+      i += 1
+    }
+  }
+}
+
+private[joins] object PointIndex {
+
+  def build(
+      keyType: DataType,
+      buildKeys: Seq[Expression],
+      rows: Iterator[InternalRow],
+      sizeHint: Option[Long]): PointIndex = {
+    val buffer = new ArrayBuffer[(Any, 
InternalRow)](RangeIndex.initialCapacity(sizeHint))
+    val projection = new InterpretedProjection(buildKeys)
+    val lowOf = RangeIndex.getValue(buildKeys.head.dataType, 0)
+    val ordering = PhysicalDataType.ordering(keyType)
+
+    while (rows.hasNext) {
+      val r = rows.next()
+      val p = projection(r)
+      val rawKey = lowOf(p)
+      if (rawKey != null) {
+        val key = InternalRow.copyValue(rawKey)
+        buffer += ((key, r.copy()))
+      }
+    }
+
+    sortAndBuild(keyType, ordering, buffer.toArray)
+  }
+
+  private def sortAndBuild(
+      keyType: DataType,
+      ordering: Ordering[Any],
+      valid: Array[(Any, InternalRow)]): PointIndex = {
+    java.util.Arrays.sort(valid, Ordering.by[(Any, InternalRow), 
Any](_._1)(ordering))
+    val length = valid.length
+    val keys = new Array[Any](length)
+    val rows = new Array[InternalRow](length)
+    var i = 0
+    while (i < length) {
+      keys(i) = valid(i)._1
+      rows(i) = valid(i)._2
+      i += 1
+    }
+    new PointIndex(keyType, keys, rows)
+  }
+}
+
+// ===== Long-specialized indexes for keys that map to one long ==============
+//
+// Indexes for the key types `RangeIndex.longKey` maps to one order-preserving
+// long, stored as raw longs converted at the build boundary. The key 
comparison
+// itself touches no Spark types once built: the tree walk and the binary 
searches
+// compare raw longs, and generated probe code passes long keys without boxing.
+// The row payload still does - `rows: Array[InternalRow]` and the serialized
+// `_keyType: DataType` are retained alongside the long arrays. Fixed width 
alone
+// is not the routed property: floating point is fixed-width but stays boxed,
+// while compact decimals route here as unscaled longs.
+// `RangeRelationBroadcastMode` picks between the two families once, from the
+// build keys' data types.
+
+/**
+ * Augmented interval tree of build-side intervals, on raw keys. Lows and 
highs sit in
+ * parallel arrays sorted by low bound, with a row array in the same order 
feeding the
+ * scan path and `maxHighs(mid)` holding the maximum high bound in the span 
rooted at
+ * mid - the value itself, unlike [[IntervalIndex.maxHighIdx]], which stores 
an index
+ * into `highs` because boxed values there are shared references worth not 
duplicating.
+ * `highs` here is already an unboxed `Array[Long]`, so storing the max value 
directly
+ * costs 4 bytes/row more than an Int index would (a Long vs. an Int), in 
exchange for
+ * comparing it without the extra `highs(idx)` indirection on every node a 
probe visits.
+ */
+private[joins] class LongIntervalIndex private (
+    private[this] var _keyType: DataType,
+    private[this] var lows: Array[Long],
+    private[this] var highs: Array[Long],
+    private[this] var maxHighs: Array[Long],
+    private[this] var rows: Array[InternalRow])
+  extends LongKeyConversion with Externalizable with KryoSerializable {
+
+  // Needed for Java/Kryo deserialization; see readExternal/read below.
+  def this() = this(null, null, null, null, null)
+
+  def keyType: DataType = _keyType
+
+  // All-covering window bounds on raw keys: `lows` is sorted, so the last 
entry
+  // is the largest low, bounding every low from above, and coreHigh is the
+  // minimum high, bounding every high from below. A window overlaps every
+  // stored interval exactly when it holds probeLow <= coreHigh and
+  // probeHigh >= coreLow, whether or not a common core exists.
+  // Recomputed in `read` too, since `lows`/`highs` are reassigned there.
+  private[this] var coreLow: Long = LongIntervalIndex.coreLow(lows)
+  private[this] var coreHigh: Long = LongIntervalIndex.coreHigh(highs)
+
+  override def estimatedSize: Long =
+    SizeEstimator.estimate(Array[AnyRef](lows, highs, maxHighs, rows))
+
+  def overlapping(low: Long, high: Long): Iterator[InternalRow] = {
+    if (lows.isEmpty) return Iterator.empty
+    var probeLow = low
+    var probeHigh = high
+    if (probeLow > probeHigh) {
+      val t = probeLow
+      probeLow = probeHigh
+      probeHigh = t
+    }
+    // A window that covers every interval gets every row the walk would
+    // return, without visiting the tree. Returns all rows iff every interval
+    // overlaps the probe; the condition still re-checks each candidate.
+    if (probeLow <= coreHigh && probeHigh >= coreLow) {
+      return new RowSlice(rows, 0, rows.length)
+    }
+    new LongOverlapIterator(lows, highs, rows, maxHighs, probeLow, probeHigh)
+  }
+
+  override def writeExternal(out: ObjectOutput): Unit = Utils.tryOrIOException 
{
+    write(out.writeUTF, out.writeInt, out.writeLong,
+      (o: Any) => out.writeObject(o.asInstanceOf[AnyRef]))
+  }
+
+  override def write(kryo: Kryo, out: Output): Unit = Utils.tryOrIOException {
+    write(out.writeString, out.writeInt, out.writeLong,
+      (o: Any) => kryo.writeClassAndObject(out, o.asInstanceOf[AnyRef]))
+  }
+
+  private def write(
+      writeString: String => Unit,
+      writeInt: Int => Unit,
+      writeLong: Long => Unit,
+      writeObj: Any => Unit): Unit = {
+    writeString(keyType.json)
+    val n = lows.length
+    writeInt(n)
+    var i = 0
+    while (i < n) {
+      writeLong(lows(i))
+      writeLong(highs(i))
+      writeLong(maxHighs(i))
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      writeObj(rows(i))
+      i += 1
+    }
+  }
+
+  override def readExternal(in: ObjectInput): Unit = Utils.tryOrIOException {
+    read(() => in.readUTF(), () => in.readInt(), () => in.readLong(), () => 
in.readObject())
+  }
+
+  override def read(kryo: Kryo, in: Input): Unit = Utils.tryOrIOException {
+    read(() => in.readString(), () => in.readInt(), () => in.readLong(),
+      () => kryo.readClassAndObject(in))
+  }
+
+  private def read(
+      readString: () => String,
+      readInt: () => Int,
+      readLong: () => Long,
+      readObj: () => AnyRef): Unit = {
+    _keyType = DataType.fromJson(readString())
+    val n = readInt()
+    lows = new Array[Long](n)
+    highs = new Array[Long](n)
+    maxHighs = new Array[Long](n)
+    rows = new Array[InternalRow](n)
+    var i = 0
+    while (i < n) {
+      lows(i) = readLong()
+      highs(i) = readLong()
+      maxHighs(i) = readLong()
+      i += 1
+    }
+    i = 0
+    while (i < n) {
+      rows(i) = readObj().asInstanceOf[InternalRow]
+      i += 1
+    }
+    coreLow = LongIntervalIndex.coreLow(lows)
+    coreHigh = LongIntervalIndex.coreHigh(highs)
+  }
+}
+
+private[joins] object LongIntervalIndex {
+
+  // Shared by the primary constructor and `read`, which recomputes these
+  // caches after reassigning `lows`/`highs` on deserialization. `lows`/
+  // `highs` are null right after the zero-arg constructor runs, before
+  // `readExternal`/`read` populate them.
+  private[joins] def coreLow(lows: Array[Long]): Long =
+    if (lows == null || lows.isEmpty) {
+      Long.MinValue
+    } else {
+      lows(lows.length - 1)
+    }
+
+  private[joins] def coreHigh(highs: Array[Long]): Long = {
+    if (highs == null || highs.isEmpty) {
+      Long.MaxValue
+    } else {
+      var min = Long.MaxValue
+      var i = 0
+      while (i < highs.length) {
+        if (highs(i) < min) min = highs(i)
+        i += 1
+      }
+      min
+    }
+  }
+
+  def build(
+      keyType: DataType,
+      buildKeys: Seq[Expression],
+      rows: Iterator[InternalRow],
+      sizeHint: Option[Long]): LongIntervalIndex = {
+    val toLongs: Seq[(InternalRow, Int) => Long] =
+      buildKeys.map(k => RangeIndex.requiredLongKey(k.dataType).field)
+    val valid = new ArrayBuffer[(Long, Long, 
InternalRow)](RangeIndex.initialCapacity(sizeHint))
+    val projection = new InterpretedProjection(buildKeys)

Review Comment:
   **Nit (P3):** `InterpretedProjection.apply` allocates a fresh `Array[Any]` 
and `GenericInternalRow` on every call and boxes each primitive key, and 
`LongKey.field` then reads it straight back with `getLong`/`getInt`. For a 
1M-row build that is 1M arrays, 1M rows, and 2M boxed longs on the driver's 
single-threaded broadcast path, which delays the broadcast stage under AQE. The 
builders at lines 399, 662, and 1106 do the same. `HashedRelation` builds its 
keys with one reused `UnsafeProjection.create(key)` (HashedRelation.scala:475). 
Could these builders do the same, keeping `InternalRow.copyValue` for string 
keys, since the reused projection overwrites its buffer, and the row `copy()`?
   
   **Verification:**
   - **Behavior:** For every supported key family, indexes built through the 
reused projection answer the same windows and bounds as before, including null 
keys, inverted intervals, and string keys backed by reused buffers.
   - **Inspection:** Each build creates its key projection once and copies only 
variable-length key values and rows per build row.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastRangeJoinExec.scala:
##########
@@ -0,0 +1,592 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.SparkException
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, 
BuildSide}
+import org.apache.spark.sql.catalyst.planning.RangeJoinKind
+import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.catalyst.plans.physical.{BroadcastDistribution, 
Distribution, Partitioning, UnspecifiedDistribution}
+import org.apache.spark.sql.execution.{CodegenSupport, ExplainUtils, SparkPlan}
+import org.apache.spark.sql.execution.metric.SQLMetrics
+
+/**
+ * Performs a range join on two tables. Supported types are inner, left outer, 
right outer,
+ * left semi, and left anti. The preserved side is streamed: unmatched build 
rows are not
+ * emitted, because the broadcast is copied to every partition. See 
`ExtractRangeJoinKeys`
+ * for the exact set of join-condition shapes that are recognized as a range 
join.
+ *
+ * The build side is an index. Point-in-range and interval overlap both build 
an
+ * [[IntervalIndex]] and probe it with `overlapping`. A single inequality 
builds a
+ * [[PointIndex]] and scans the side of the bound the build side holds. Keys 
that
+ * map to one order-preserving long instead build [[LongIntervalIndex]] or
+ * [[LongPointIndex]], so their probes compare raw longs without boxing. Either
+ * probe returns a superset. A candidate is kept only when the original join
+ * condition is true, so inclusivity stays on the predicate. Heavy overlap can
+ * make one probe return almost every build row.
+ */
+case class BroadcastRangeJoinExec(
+    left: SparkPlan,
+    right: SparkPlan,
+    buildSide: BuildSide,
+    joinType: JoinType,
+    leftKeys: Seq[Expression],
+    rightKeys: Seq[Expression],
+    condition: Option[Expression],
+    rangeJoinKind: RangeJoinKind)
+  extends JoinCodegenSupport {
+
+  if (condition.isEmpty) {
+    throw SparkException.internalError("BroadcastRangeJoinExec requires a join 
condition.")
+  }
+
+  override lazy val metrics = Map(
+    "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output 
rows"))
+
+  /** BuildRight means the right relation is the broadcast relation. */
+  private val (streamed, broadcast, streamedKeys, buildKeys) = buildSide match 
{
+    case BuildRight => (left, right, leftKeys, rightKeys)
+    case BuildLeft => (right, left, rightKeys, leftKeys)
+  }
+
+  override def simpleStringWithNodeId(): String = {
+    val opId = ExplainUtils.getOpId(this)
+    s"$nodeName $joinType $buildSide ($opId)".trim
+  }
+
+  override def requiredChildDistribution: Seq[Distribution] = {
+    val mode = RangeRelationBroadcastMode(
+      BindReferences.bindReferences(buildKeys, broadcast.output),
+      RangeIndexKind(rangeJoinKind))
+    buildSide match {
+      case BuildLeft =>
+        BroadcastDistribution(mode) :: UnspecifiedDistribution :: Nil
+      case BuildRight =>
+        UnspecifiedDistribution :: BroadcastDistribution(mode) :: Nil
+    }
+  }
+
+  override def outputPartitioning: Partitioning = streamed.outputPartitioning
+
+  override def outputOrdering: Seq[SortOrder] = streamed.outputOrdering
+
+  override def output: Seq[Attribute] = {
+    joinType match {
+      case _: InnerLike =>
+        left.output ++ right.output
+      case LeftOuter =>
+        left.output ++ right.output.map(_.withNullability(true))
+      case RightOuter =>
+        left.output.map(_.withNullability(true)) ++ right.output
+      case LeftSemi | LeftAnti =>
+        left.output
+      case t =>
+        throw SparkException.internalError(s"BroadcastRangeJoin should not 
take $t as the JoinType")
+    }
+  }
+
+  // The point-index scan used to probe this join's build side. See 
`PointScan.forJoin`.
+  private lazy val pointScan: PointScan = PointScan.forJoin(rangeJoinKind, 
buildSide)
+
+  private def perPartition(
+      relation: Broadcast[RangeRelation])(
+      f: BroadcastRangeJoinExec.ProbeContext => Iterator[InternalRow]): 
RDD[InternalRow] = {
+    val boundStreamKeys = BindReferences.bindReferences(streamedKeys, 
streamed.output)
+    val cond = condition.get
+    val schema = left.output ++ right.output
+    val numBuildFields = broadcast.output.length
+    val kind = rangeJoinKind
+    val side = buildSide
+    streamed.execute().mapPartitionsInternal { streamedIter =>
+      val candidateFn = BroadcastRangeJoinExec.makeCandidateFn(
+        relation.value, boundStreamKeys, PointScan.forJoin(kind, side))
+      f(new BroadcastRangeJoinExec.ProbeContext(
+        streamedIter,
+        Predicate.create(cond, schema).eval _,
+        new JoinedRow,
+        new GenericInternalRow(numBuildFields),
+        candidateFn,
+        side))
+    }
+  }
+
+  private def innerJoin(relation: Broadcast[RangeRelation]): RDD[InternalRow] 
= {
+    perPartition(relation) { p =>
+      p.stream.flatMap { row =>
+        p.candidates(row).flatMap { buildRow =>
+          val joined = p.join(row, buildRow)
+          if (p.condition(joined)) Iterator.single(joined) else Iterator.empty

Review Comment:
   **Nit (P3):** `flatMap` with `Iterator.single(joined)` allocates one 
iterator per accepted pair, and because flatMap holds it in its 
current-iterator field, escape analysis can't remove it. 
`BroadcastNestedLoopJoinExec.innerJoin` does the same work with 
`buildRows.iterator.map(r => joinedRow(streamedRow, 
r)).filter(boundCondition)`. `p.candidates(row).map(p.join(row, 
_)).filter(p.condition)` returns the same rows here, since the result 
projection copies each row before the iterator advances. This only matters on 
the interpreted path: whole-stage codegen disabled, or the join falling out of 
it.
   
   **Verification:**
   - **Behavior:** With whole-stage codegen disabled, inner range joins for 
every key family and build side return the nested-loop result.
   - **Inspection:** The interpreted inner-join path creates no iterator per 
accepted candidate.



##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/RangeJoinSQLSuite.scala:
##########
@@ -0,0 +1,793 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.planning.{IntervalOverlapJoin, 
LessPartialRangeJoin}
+import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftOuter, 
LeftSemi}
+import org.apache.spark.sql.execution._
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.DoubleType
+
+/**
+ * Planner and end-to-end coverage for broadcast range join.
+ *
+ * Which condition is recognized: `ExtractRangeJoinKeysSuite`.
+ * What the index returns: `RangeIndexSuite`.
+ * What the operator returns: `RangeJoinSuite`.
+ * This suite checks that the planner chooses the operator, and that the SQL
+ * result matches broadcast nested loop join where that comparison is the 
point.
+ */
+class RangeJoinSQLSuite extends QueryTest with SharedSparkSession with 
AdaptiveSparkPlanHelper {
+
+  private val pointCond =
+    "s.ip_add_int >= l.BEGIN_IP2LONG AND s.ip_add_int <= l.END_IP2LONG"
+
+  private def findRangeJoin(plan: SparkPlan): Seq[BroadcastRangeJoinExec] = {
+    collect(plan) { case j: BroadcastRangeJoinExec => j }
+  }
+
+  /** `n` range joins sit in one whole-stage, not in nested stages or input 
adapters. */
+  private def oneCodegenStageOwns(plan: SparkPlan, n: Int): Boolean = {
+    exists(plan) {
+      case w: WholeStageCodegenExec =>
+        def count(p: SparkPlan): Int = p match {
+          case _: InputAdapter | _: WholeStageCodegenExec => 0
+          case j: BroadcastRangeJoinExec => 1 + j.children.map(count).sum
+          case other => other.children.map(count).sum
+        }
+        count(w.child) == n
+      case _ => false
+    }
+  }
+
+  /** The range join is compiled into a whole-stage, not hidden under an input 
adapter. */
+  private def rangeJoinIsCodegen(plan: SparkPlan): Boolean = {
+    exists(plan) {
+      case w: WholeStageCodegenExec =>
+        def contains(p: SparkPlan): Boolean = p match {
+          case _: InputAdapter => false
+          case _: BroadcastRangeJoinExec => true
+          case other => other.children.exists(contains)
+        }
+        contains(w.child)
+      case _ => false
+    }
+  }
+
+  private def findBroadcastNestedLoopJoin(plan: SparkPlan): 
Seq[BroadcastNestedLoopJoinExec] = {
+    collect(plan) { case j: BroadcastNestedLoopJoinExec => j }
+  }
+
+  private def findCartesianProduct(plan: SparkPlan): Seq[CartesianProductExec] 
= {
+    collect(plan) { case c: CartesianProductExec => c }
+  }
+
+  private def withRangeJoin(extra: (String, String)*)(f: => Unit): Unit = {
+    withSQLConf((Seq(SQLConf.BROADCAST_RANGE_JOIN_ENABLED.key -> "true") ++ 
extra): _*)(f)
+  }
+
+  /**
+   * One query per key family and join type that a range join can plan, shared 
by
+   * "whole-stage codegen covers each range-join probe" and "interpreted 
execution
+   * covers each range-join probe" so the two coverage lists cannot drift 
apart.
+   */
+  private def probeCoverageQueries: Seq[String] = Seq(
+    s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON $pointCond",
+    s"SELECT s.id, l.zip_id FROM signin_ip s LEFT JOIN ip_lookup l ON 
$pointCond",
+    s"SELECT s.id, l.zip_id FROM signin_ip s RIGHT JOIN ip_lookup l ON 
$pointCond",
+    s"SELECT s.id FROM signin_ip s LEFT SEMI JOIN ip_lookup l ON $pointCond",
+    s"SELECT s.id FROM signin_ip s LEFT ANTI JOIN ip_lookup l ON $pointCond",
+    "SELECT p.p, r.lo, r.hi FROM dec_points p JOIN dec_ranges r ON p.p >= r.lo 
AND p.p <= r.hi",
+    "SELECT /*+ BROADCAST(r) */ p.s, r.lo, r.hi FROM str_points p JOIN 
str_ranges r " +
+      "ON p.s >= r.lo AND p.s <= r.hi",
+    "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a JOIN intervals_b b " +
+      "ON a.lo < b.hi AND b.lo < a.hi",
+    "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a LEFT JOIN intervals_b b 
ON a.hi < b.lo",
+    "SELECT /*+ BROADCAST(a) */ a.lo, b.hi FROM intervals_a a JOIN intervals_b 
b ON a.lo < b.hi",
+    // DOUBLE keys stay on the boxed interval index (see 
setupDoubleIntervalViews), unlike
+    // the INT two-key overlap above, which is long-specialized.
+    "SELECT a.lo, a.hi, b.lo, b.hi FROM dbl_ranges_a a JOIN dbl_ranges_b b " +
+      "ON a.lo < b.hi AND b.lo < a.hi",
+    // Double and string keys stay on the boxed point index, so these queries 
reach
+    // its generated probe code (or, interpreted, the per-index dispatch in
+    // BroadcastRangeJoinExec.makeCandidateFn), which int partial ranges never 
do.
+    "SELECT p.d, b.lo FROM dbl_partial_points p JOIN dbl_partial_bounds b ON 
p.d < b.lo",

Review Comment:
   **Nit (P3):** These four boxed partial-range queries are all `<`/`>`, so 
`case points: PointIndex` in `makeCandidateFn` only ever runs 
`upToExclusive`/`fromExclusive`. The inclusive operator checks in 
`RangeJoinSuite` use int keys, which take the separate `LongPointIndex` arm, 
and `RangeIndexSuite` calls `PointIndex.upTo`/`from` directly. If the boxed 
`UpTo` arm mapped to `upToExclusive`, a codegen-off `p.d <= b.lo` would drop 
every build row equal to the probe, which the condition recheck can't recover. 
Could the list add `<=`/`>=` on DOUBLE and STRING in both directions, with 
probe values equal to build keys?
   
   **Verification:**
   - **Regression:** With whole-stage codegen enabled and disabled, 
less-or-equal and greater-or-equal range joins on double and string keys whose 
probe values equal build keys return the nested-loop result. The check fails if 
the boxed inclusive scan is replaced by its exclusive form.



##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastRangeJoinExec.scala:
##########
@@ -0,0 +1,592 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.SparkException
+import org.apache.spark.broadcast.Broadcast
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, 
BuildSide}
+import org.apache.spark.sql.catalyst.planning.RangeJoinKind
+import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.catalyst.plans.physical.{BroadcastDistribution, 
Distribution, Partitioning, UnspecifiedDistribution}
+import org.apache.spark.sql.execution.{CodegenSupport, ExplainUtils, SparkPlan}
+import org.apache.spark.sql.execution.metric.SQLMetrics
+
+/**
+ * Performs a range join on two tables. Supported types are inner, left outer, 
right outer,
+ * left semi, and left anti. The preserved side is streamed: unmatched build 
rows are not
+ * emitted, because the broadcast is copied to every partition. See 
`ExtractRangeJoinKeys`
+ * for the exact set of join-condition shapes that are recognized as a range 
join.
+ *
+ * The build side is an index. Point-in-range and interval overlap both build 
an
+ * [[IntervalIndex]] and probe it with `overlapping`. A single inequality 
builds a
+ * [[PointIndex]] and scans the side of the bound the build side holds. Keys 
that
+ * map to one order-preserving long instead build [[LongIntervalIndex]] or
+ * [[LongPointIndex]], so their probes compare raw longs without boxing. Either
+ * probe returns a superset. A candidate is kept only when the original join
+ * condition is true, so inclusivity stays on the predicate. Heavy overlap can
+ * make one probe return almost every build row.
+ */
+case class BroadcastRangeJoinExec(
+    left: SparkPlan,
+    right: SparkPlan,
+    buildSide: BuildSide,
+    joinType: JoinType,
+    leftKeys: Seq[Expression],
+    rightKeys: Seq[Expression],
+    condition: Option[Expression],
+    rangeJoinKind: RangeJoinKind)
+  extends JoinCodegenSupport {
+
+  if (condition.isEmpty) {
+    throw SparkException.internalError("BroadcastRangeJoinExec requires a join 
condition.")
+  }
+
+  override lazy val metrics = Map(
+    "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output 
rows"))
+
+  /** BuildRight means the right relation is the broadcast relation. */
+  private val (streamed, broadcast, streamedKeys, buildKeys) = buildSide match 
{
+    case BuildRight => (left, right, leftKeys, rightKeys)
+    case BuildLeft => (right, left, rightKeys, leftKeys)
+  }
+
+  override def simpleStringWithNodeId(): String = {
+    val opId = ExplainUtils.getOpId(this)
+    s"$nodeName $joinType $buildSide ($opId)".trim
+  }
+
+  override def requiredChildDistribution: Seq[Distribution] = {
+    val mode = RangeRelationBroadcastMode(
+      BindReferences.bindReferences(buildKeys, broadcast.output),
+      RangeIndexKind(rangeJoinKind))
+    buildSide match {
+      case BuildLeft =>
+        BroadcastDistribution(mode) :: UnspecifiedDistribution :: Nil
+      case BuildRight =>
+        UnspecifiedDistribution :: BroadcastDistribution(mode) :: Nil
+    }
+  }
+
+  override def outputPartitioning: Partitioning = streamed.outputPartitioning
+
+  override def outputOrdering: Seq[SortOrder] = streamed.outputOrdering
+
+  override def output: Seq[Attribute] = {
+    joinType match {
+      case _: InnerLike =>
+        left.output ++ right.output
+      case LeftOuter =>
+        left.output ++ right.output.map(_.withNullability(true))
+      case RightOuter =>
+        left.output.map(_.withNullability(true)) ++ right.output
+      case LeftSemi | LeftAnti =>
+        left.output
+      case t =>
+        throw SparkException.internalError(s"BroadcastRangeJoin should not 
take $t as the JoinType")
+    }
+  }
+
+  // The point-index scan used to probe this join's build side. See 
`PointScan.forJoin`.
+  private lazy val pointScan: PointScan = PointScan.forJoin(rangeJoinKind, 
buildSide)
+
+  private def perPartition(
+      relation: Broadcast[RangeRelation])(
+      f: BroadcastRangeJoinExec.ProbeContext => Iterator[InternalRow]): 
RDD[InternalRow] = {
+    val boundStreamKeys = BindReferences.bindReferences(streamedKeys, 
streamed.output)
+    val cond = condition.get
+    val schema = left.output ++ right.output
+    val numBuildFields = broadcast.output.length
+    val kind = rangeJoinKind
+    val side = buildSide
+    streamed.execute().mapPartitionsInternal { streamedIter =>
+      val candidateFn = BroadcastRangeJoinExec.makeCandidateFn(
+        relation.value, boundStreamKeys, PointScan.forJoin(kind, side))
+      f(new BroadcastRangeJoinExec.ProbeContext(
+        streamedIter,
+        Predicate.create(cond, schema).eval _,
+        new JoinedRow,
+        new GenericInternalRow(numBuildFields),
+        candidateFn,
+        side))
+    }
+  }
+
+  private def innerJoin(relation: Broadcast[RangeRelation]): RDD[InternalRow] 
= {
+    perPartition(relation) { p =>
+      p.stream.flatMap { row =>
+        p.candidates(row).flatMap { buildRow =>
+          val joined = p.join(row, buildRow)
+          if (p.condition(joined)) Iterator.single(joined) else Iterator.empty
+        }
+      }
+    }
+  }
+
+  /**
+   * Emit each candidate the condition accepts. When none were, emit the stream
+   * row once with the build side padded by nulls.
+   *
+   * The loop test checks `streamRow` before `stream.hasNext`. A whole-stage
+   * stream reuses one `UnsafeRow`, and `hasNext` fills that buffer. It must
+   * not run while the current row is still being joined.
+   */
+  private def outerJoin(relation: Broadcast[RangeRelation]): RDD[InternalRow] 
= {
+    perPartition(relation) { p =>
+      new Iterator[InternalRow] {
+        private var streamRow: InternalRow = null
+        private var matches: Iterator[InternalRow] = Iterator.empty
+        private var foundMatch: Boolean = false
+        private var resultRow: InternalRow = null
+
+        private def findNext(): Boolean = {
+          while (streamRow != null || p.stream.hasNext) {

Review Comment:
   **Nit (P3):** The ordering here is load-bearing. When the join falls out of 
whole-stage codegen (output wider than `spark.sql.codegen.maxFields`, or a 
`CodegenFallback` conjunct) while its streamed child stays code-generated, 
`hasNext` refills the child's shared `UnsafeRow`. Every interpreted-path test 
turns whole-stage codegen off globally, though, so the streamed child never 
reuses a buffer, and reordering this to `p.stream.hasNext || streamRow != null` 
would keep every test green. A VALUES view doesn't help either: 
`CollapseCodegenStages` never makes a `LocalTableScanExec` a whole-stage root, 
and it hands out its own rows. Could you add left and right outer cases with 
codegen on, the join excluded through the field limit, a streamed side computed 
over `range()` so it reuses its output row, and several matches per streamed 
row?
   
   **Verification:**
   - **Regression:** With whole-stage codegen enabled, the range join excluded 
from it, and its streamed side produced by a code-generated operator that 
reuses its output row, left and right outer range joins whose streamed rows 
each match several build rows return the nested-loop result. The check fails if 
the stream is advanced while candidates for the current row remain.



##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/RangeIndexSuite.scala:
##########
@@ -0,0 +1,1086 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, 
ObjectInputStream, ObjectOutputStream}
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.BoundReference
+import org.apache.spark.sql.catalyst.types.PhysicalDataType
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Unit tests for the range-join indexes: the boxed pair ([[IntervalIndex]],
+ * [[PointIndex]]), the raw-long pair ([[LongIntervalIndex]], 
[[LongPointIndex]]),
+ * and the broadcast-mode routing that picks between the two families.
+ * Operator-level behavior is covered by `RangeJoinSuite`.
+ */
+class RangeIndexSuite extends SparkFunSuite {
+
+  private def row(id: Int): InternalRow = InternalRow(id)
+
+  private def span(lo: Any, hi: Any, id: Int): (Any, Any, InternalRow) = (lo, 
hi, row(id))
+
+  private def buildIntervals(spans: Seq[(Any, Any, InternalRow)]): 
IntervalIndex =
+    buildIntervals(IntegerType, spans)
+
+  /**
+   * Builds the boxed interval index through the production build. The row
+   * layout puts the asserted id at ordinal 0, exactly where a real build row
+   * carries its columns; the keys bind to ordinals 1 and 2.
+   */
+  private def buildIntervals(
+      keyType: DataType,
+      spans: Seq[(Any, Any, InternalRow)]): IntervalIndex = {
+    val keys = Seq(
+      BoundReference(1, keyType, nullable = true),
+      BoundReference(2, keyType, nullable = true))
+    IntervalIndex.build(keyType, keys,
+      spans.iterator.map(t => InternalRow(t._3.getInt(0), t._1, t._2)),
+      Some(spans.size.toLong))
+  }
+
+  /** The point-index build with the same row layout: id at 0, key at 1. */
+  private def buildPoints(keyType: DataType, keyed: Seq[(Any, InternalRow)]): 
PointIndex = {
+    val key = Seq(BoundReference(1, keyType, nullable = true))
+    PointIndex.build(keyType, key,
+      keyed.iterator.map(t => InternalRow(t._2.getInt(0), t._1)),
+      Some(keyed.size.toLong))
+  }
+
+  private def ids(rows: Iterator[InternalRow]): List[Int] = 
rows.map(_.getInt(0)).toList
+
+  private def probeIds(index: IntervalIndex, low: Any, high: Any): List[Int] =
+    ids(index.overlapping(low, high))
+
+  /**
+   * An iterator that hands back the same mutable row on every `next()` call, 
each field
+   * set to the current index. Mirrors how Spark's row iterators commonly 
reuse a single
+   * buffer (e.g. `UnsafeRow` deserialization) instead of allocating a fresh 
row per record.
+   */
+  private def reusingRowIterator(
+      fields: Int,
+      n: Int): (org.apache.spark.sql.catalyst.expressions.GenericInternalRow,
+      Iterator[InternalRow]) = {
+    val shared = new 
org.apache.spark.sql.catalyst.expressions.GenericInternalRow(fields)
+    val iter = new Iterator[InternalRow] {
+      private var i = 0
+      override def hasNext: Boolean = i < n
+      override def next(): InternalRow = {
+        var c = 0
+        while (c < fields) { shared.update(c, i); c += 1 }
+        i += 1
+        shared
+      }
+    }
+    (shared, iter)
+  }
+
+  private def reusingByteBufferStringIterator(
+      n: Int): (Array[Byte], Iterator[InternalRow]) = {
+    val sharedBytes = new Array[Byte](4)
+    val sharedRow = new 
org.apache.spark.sql.catalyst.expressions.GenericInternalRow(2)
+    val iter = new Iterator[InternalRow] {
+      private var i = 0
+      override def hasNext: Boolean = i < n
+      override def next(): InternalRow = {
+        val s = f"$i%04d"
+        java.lang.System.arraycopy(s.getBytes("UTF-8"), 0, sharedBytes, 0, 4)
+        val utf8 = UTF8String.fromAddress(sharedBytes, 
Platform.BYTE_ARRAY_OFFSET, 4)
+        sharedRow.update(0, utf8)
+        sharedRow.update(1, i)
+        i += 1
+        sharedRow
+      }
+    }
+    (sharedBytes, iter)
+  }
+
+  private def javaRoundTrip(index: IntervalIndex): (Array[Byte], 
IntervalIndex) = {
+    val bytes = new ByteArrayOutputStream()
+    val out = new ObjectOutputStream(bytes)
+    out.writeObject(index)
+    out.close()
+    val raw = bytes.toByteArray
+    val in = new ObjectInputStream(new ByteArrayInputStream(raw))
+    try {
+      (raw, in.readObject().asInstanceOf[IntervalIndex])
+    } finally {
+      in.close()
+    }
+  }
+
+  test("build keeps a point, orders an inverted interval, and drops a null 
bound") {
+    val index = buildIntervals(Seq(
+      span(5, 5, 5),
+      span(1, 3, 1),
+      span(3, 1, 3),
+      (null, 1, row(1))))
+    // (3, 1) is stored as [1, 3]. The null bound is absent.
+    assert(ids(index.overlapping(2, 2)).sorted == List(1, 3))
+    assert(ids(index.overlapping(5, 5)) == List(5))
+    assert(ids(index.overlapping(4, 4)).isEmpty)
+  }
+
+  test("IntervalIndex.build defends against a build-side iterator reusing its 
row buffer") {
+    // If build() captured keys or rows by reference instead of copying them, 
every stored
+    // entry would collapse to the iterator's last value once the shared 
buffer is mutated
+    // again after the build-side iterator is drained -- the hazard guarded 
against is any
+    // producer whose rows or projected keys alias a reused buffer (e.g. a 
typed UDF result),
+    // not specifically BroadcastExchangeExec: its own row iterator decodes a 
fresh UnsafeRow
+    // per record.
+    val n = 50
+    val buildKeys = Seq(
+      BoundReference(0, IntegerType, nullable = true),
+      BoundReference(1, IntegerType, nullable = true))
+    val (shared, rows) = reusingRowIterator(fields = 3, n)
+    val index = IntervalIndex.build(IntegerType, buildKeys, rows, Some(n))
+
+    // Simulate the caller reusing the buffer after the build-side iterator is 
exhausted.
+    shared.update(0, -1)
+    shared.update(1, -1)
+    shared.update(2, -1)
+
+    (0 until n).foreach { i =>
+      val payloads = index.overlapping(i, i).map(_.getInt(2)).toList
+      assert(payloads == List(i), s"interval $i was corrupted by later reuse 
of the row buffer")
+    }
+  }
+
+  test("IntervalIndex.build defends against string keys sharing a reused byte 
buffer") {
+    val n = 50
+    val buildKeys = Seq(
+      BoundReference(0, StringType, nullable = true),
+      BoundReference(0, StringType, nullable = true))
+    val (sharedBytes, rows) = reusingByteBufferStringIterator(n)
+    val index = IntervalIndex.build(StringType, buildKeys, rows, Some(n))
+
+    java.util.Arrays.fill(sharedBytes, 'X'.toByte)
+
+    (0 until n).foreach { i =>
+      val key = UTF8String.fromString(f"$i%04d")
+      val matching = index.overlapping(key, key).toList
+      assert(matching.size == 1, s"interval $i was corrupted by shared buffer 
reuse")
+      assert(matching.head.getInt(1) == i)
+    }
+  }
+
+  test("a one-key interval build owns a single copy of the point key") {
+    // One build key is one key used as both ends. The build must copy it once
+    // and serve both bounds from that copy, instead of copying twice.
+    val n = 50
+    val oneKey = Seq(BoundReference(0, StringType, nullable = true))
+    val (sharedBytes, rows) = reusingByteBufferStringIterator(n)
+    val index = IntervalIndex.build(StringType, oneKey, rows, Some(n))
+
+    java.util.Arrays.fill(sharedBytes, 'X'.toByte)
+
+    (0 until n).foreach { i =>
+      val key = UTF8String.fromString(f"$i%04d")
+      val matching = index.overlapping(key, key).toList
+      assert(matching.size == 1, s"interval $i was corrupted by shared buffer 
reuse")
+      assert(matching.head.getInt(1) == i)
+    }
+
+    // SizeEstimator's identity tracking charges the shared bounds once, so the
+    // estimate stays below the same build made from two independently copied 
keys.
+    val twoKeys = Seq(
+      BoundReference(0, StringType, nullable = true),
+      BoundReference(0, StringType, nullable = true))
+    val (_, moreRows) = reusingByteBufferStringIterator(n)
+    val twoCopy = IntervalIndex.build(StringType, twoKeys, moreRows, Some(n))
+    assert(index.estimatedSize < twoCopy.estimatedSize)
+  }
+
+  test("an inverted build interval reaches the windows the condition accepts") 
{
+    // Int keys route to the long index, so this also pins swap normalization 
there.
+    val index = RangeRelationBroadcastMode(
+      Seq(BoundReference(0, IntegerType, nullable = true),
+        BoundReference(1, IntegerType, nullable = true)),
+      IntervalIndexKind)
+      .transform(Array(InternalRow(3, 1), InternalRow(0, 5)))
+      .asInstanceOf[LongIntervalIndex]
+    assert(index.overlapping(0L, 4L).map(_.getInt(0)).toSet == Set(3, 0))
+    assert(index.overlapping(4L, 9L).map(_.getInt(0)).toList == List(0))
+  }
+
+  test("interval tree returns each overlapping row once") {
+    // Spans arrive out of low order: [10, 20] then [0, 5].
+    val unsorted = buildIntervals(Seq(span(10, 20, 0), span(0, 5, 1)))
+    assert(probeIds(unsorted, 3, 3) == List(1))
+    assert(probeIds(unsorted, 15, 15) == List(0))
+
+    // Probe 50 is a build key. Row 0 spans it; row 1 starts there.
+    val spanning = buildIntervals(Seq(span(0, 100, 0), span(50, 200, 1)))
+    assert(probeIds(spanning, 50, 50).toSet == Set(0, 1))
+    assert(probeIds(spanning, 25, 25) == List(0))
+    assert(probeIds(spanning, 150, 150) == List(1))
+
+    val duplicated = buildIntervals(Seq(span(0, 100, 0), span(0, 100, 1)))
+    assert(probeIds(duplicated, 50, 50).sorted == List(0, 1))
+
+    val nested = buildIntervals(Seq(
+      span(0, 100, 0), span(10, 20, 1), span(10, 50, 2), span(90, 110, 3)))
+    def assertOnce(low: Int, expected: Set[Int]): Unit = {
+      val actual = probeIds(nested, low, low)
+      assert(actual.toSet == expected, s"probe $low -> $actual")
+      assert(actual.size == expected.size, s"probe $low duplicated $actual")
+    }
+    assertOnce(0, Set(0))
+    assertOnce(10, Set(0, 1, 2))
+    assertOnce(15, Set(0, 1, 2))
+    assertOnce(30, Set(0, 2))
+    assertOnce(95, Set(0, 3))
+    assertOnce(105, Set(3))
+    assertOnce(200, Set.empty)
+
+    assert(probeIds(buildIntervals(Seq.empty), 1, 1).isEmpty)
+
+    val withNull = buildIntervals(Seq(span(0, 10, 0), span(5, 5, 1), (null, 4, 
row(9))))
+    assert(probeIds(withNull, null, 5).isEmpty)
+    assert(probeIds(withNull, 5, null).isEmpty)
+    // The null bound is dropped, so two intervals remain and the estimate 
covers both.
+    val one = buildIntervals(Seq(span(0, 10, 0)))
+    assert(withNull.estimatedSize > one.estimatedSize)
+
+    // At an interior endpoint, the range that ends and the one that starts
+    // are each a candidate once.
+    val chained = buildIntervals(Seq(span(0, 1, 0), span(1, 2, 1), span(2, 3, 
2)))
+    assert(probeIds(chained, 0, 0) == List(0))
+    assert(probeIds(chained, 1, 1).sorted == List(0, 1))
+    assert(probeIds(chained, 2, 2).sorted == List(1, 2))
+    assert(probeIds(chained, 3, 3) == List(2))
+
+    // A probe whose endpoints are reversed is the closed window between them.
+    // [10, 25] meets [0, 30], [15, 40], and the point at 15.
+    val inverted = buildIntervals(Seq(span(0, 30, 0), span(15, 40, 1), 
span(15, 15, 2)))
+    assert(probeIds(inverted, 25, 10).sorted == List(0, 1, 2))
+    assert(probeIds(inverted, 10, 20).sorted == List(0, 1, 2))
+  }
+
+  test("interval tree is correct under heavy build-side overlap") {
+    // Every interval nests inside the one before it, so a point in the middle
+    // hits about half of the rows. The tree still stores one node per row.

Review Comment:
   **Nit (P3):** With `span(i, 2 * n - i, i)` for i < 300, every interval 
contains 300, so a probe in the middle hits all 300 rows. It's the probe at `n 
/ 2 = 150` that hits about half (151). Could the comment say that?
   
   **Verification:**
   - **Inspection:** The comment's statement of how many rows the middle and 
quarter-domain probes hit matches the nested-interval fixture it documents.



##########
sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala:
##########
@@ -632,6 +632,26 @@ class SQLMetricsSuite extends SharedSparkSession with 
SQLMetricsTestUtils
     }
   }
 
+  test("BroadcastRangeJoin metrics") {
+    withSQLConf(SQLConf.BROADCAST_RANGE_JOIN_ENABLED.key -> "true") {
+      withTempView("rangeProbe", "rangeBuild") {
+        sql("CREATE OR REPLACE TEMP VIEW rangeProbe(p) AS VALUES (5), (15), 
(25)")
+        sql("CREATE OR REPLACE TEMP VIEW rangeBuild(lo, hi) AS VALUES (0, 10), 
(10, 20), (20, 30)")
+        // ... -> BroadcastRangeJoin(nodeId = 1) -> TungstenProject(nodeId = 0)
+        val query = "SELECT /*+ BROADCAST(b) */ p.p, b.lo, b.hi FROM 
rangeProbe p " +
+          "JOIN rangeBuild b ON p.p >= b.lo AND p.p <= b.hi"

Review Comment:
   **Nit (P3):** Thanks for adding this. With probes 5/15/25 against 
[0,10]/[10,20]/[20,30], each probe surfaces exactly one candidate through the 
tree walk and every candidate passes, so moving `$numOutput.add(1)` in 
`codegenInner` outside `$checkCondition` would still report 3. The outer 
(including the null-padded row), left semi, and left anti increments in 
`codegenOuter`, `codegenLeftExistence`, and the interpreted map in `doExecute` 
are still never read. Could the test add a fixture where some candidates fail 
the condition, plus left and right outer joins with unmatched streamed rows and 
left semi/anti joins, with whole-stage codegen on and off? Only the reported 
counts are at stake, not query results.
   
   This extends an [existing 
discussion](https://github.com/apache/spark/pull/59027#pullrequestreview-5628337351).
   
   **Verification:**
   - **Regression:** With whole-stage codegen enabled and disabled, the range 
join reports a 'number of output rows' equal to the rows it returns for: inner 
joins whose candidates include rows the condition rejects, left and right outer 
joins with unmatched streamed rows, and left semi and left anti joins. The 
assertion fails if any path counts a rejected candidate, double-counts a 
null-padded row, or skips an output row.



##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/planning/ExtractRangeJoinKeysSuite.scala:
##########
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.planning
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint, 
LocalRelation}
+import org.apache.spark.sql.types.{DayTimeIntervalType, DoubleType, 
IntegerType}
+
+/**
+ * Plan-level coverage for [[ExtractRangeJoinKeys]]: which condition shapes are
+ * recognized, which `RangeJoinKind` tag is produced, and what residual 
remains.
+ */
+class ExtractRangeJoinKeysSuite extends PlanTest {
+
+  private val points = LocalRelation($"p".int)
+  private val ranges = LocalRelation($"lo".int, $"hi".int)
+  private val intervalsA = LocalRelation($"lo".int, $"hi".int)
+  private val intervalsB = LocalRelation($"lo".int, $"hi".int)
+
+  private val p = points.output.head
+  private val rLo = ranges.output(0)
+  private val rHi = ranges.output(1)
+  private val aLo = intervalsA.output(0)
+  private val aHi = intervalsA.output(1)
+  private val bLo = intervalsB.output(0)
+  private val bHi = intervalsB.output(1)
+
+  private def join(left: LocalRelation, right: LocalRelation, cond: 
Expression): Join =
+    Join(left, right, Inner, Some(cond), JoinHint.NONE)
+
+  private def assertKeys(
+      left: LocalRelation,
+      right: LocalRelation,
+      cond: Expression,
+      expected: RangeJoinKind,
+      leftKeys: Seq[Expression],
+      rightKeys: Seq[Expression]): Unit = {
+    join(left, right, cond) match {
+      case ExtractRangeJoinKeys(_, actualLeft, actualRight, rangeJoinKind) =>
+        assert(rangeJoinKind === expected)
+        assert(actualLeft === leftKeys)
+        assert(actualRight === rightKeys)
+      case other => fail(s"expected $expected, got $other")
+    }
+  }
+
+  private def assertUnrecognized(plan: Join): Unit = {
+    assert(ExtractRangeJoinKeys.unapply(plan).isEmpty)
+  }
+
+  test("point-in-range keys are the point and the two bounds") {
+    // Inclusivity and conjunct order stay on the condition. BETWEEN lowers to 
a With,
+    // but RewriteWithExpression resolves it to a plain And of comparisons 
before physical
+    // planning runs, so by the time this extractor sees the condition it is 
already one
+    // of these shapes.
+    val inclusive = And(GreaterThanOrEqual(p, rLo), LessThanOrEqual(p, rHi))
+    val exclusive = And(GreaterThan(p, rLo), LessThan(p, rHi))
+    val highFirst = And(LessThanOrEqual(p, rHi), GreaterThan(p, rLo))
+    val lowFirst = And(GreaterThanOrEqual(p, rLo), LessThan(p, rHi))
+    Seq(inclusive, exclusive, highFirst, lowFirst).foreach { cond =>
+      assertKeys(points, ranges, cond, PointInRangeJoin, Seq(p), Seq(rLo, rHi))
+    }
+    assertKeys(
+      points, ranges, And(inclusive, Not(EqualTo(p, rLo))),
+      PointInRangeJoin, Seq(p), Seq(rLo, rHi))
+  }
+
+  test("point-in-range tie-break: the earliest candidate pair wins") {
+    // Three predicates give three candidate pairs under 
predicates.combinations(2):
+    // (P0, P1), (P0, P2), (P1, P2). (P0, P1) and (P0, P2) are each 
independently a
+    // valid point-in-range join -- p bounded by (rLo3, rHi3), and separately 
by
+    // (rLo3, rHi2) -- while (P1, P2) shares no endpoint. combinations(2) 
enumerates

Review Comment:
   **Nit (P3):** After `RangePredicate` normalization, `p <= rHi3` and `p <= 
rHi2` both have `p` as their low bound, so they do share an endpoint. They 
aren't a point-in-range pair because neither one's high equals the other's low, 
and the interval-overlap regrouping fails `place` because each side would mix 
both relations. Could the comment say that?
   
   **Verification:**
   - **Inspection:** The comment's explanation of why the later predicate pair 
is rejected matches the extractor's chaining and placement rules for those 
predicates.



##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/RangeJoinSQLSuite.scala:
##########
@@ -0,0 +1,793 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.planning.{IntervalOverlapJoin, 
LessPartialRangeJoin}
+import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftOuter, 
LeftSemi}
+import org.apache.spark.sql.execution._
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.DoubleType
+
+/**
+ * Planner and end-to-end coverage for broadcast range join.
+ *
+ * Which condition is recognized: `ExtractRangeJoinKeysSuite`.
+ * What the index returns: `RangeIndexSuite`.
+ * What the operator returns: `RangeJoinSuite`.
+ * This suite checks that the planner chooses the operator, and that the SQL
+ * result matches broadcast nested loop join where that comparison is the 
point.
+ */
+class RangeJoinSQLSuite extends QueryTest with SharedSparkSession with 
AdaptiveSparkPlanHelper {
+
+  private val pointCond =
+    "s.ip_add_int >= l.BEGIN_IP2LONG AND s.ip_add_int <= l.END_IP2LONG"
+
+  private def findRangeJoin(plan: SparkPlan): Seq[BroadcastRangeJoinExec] = {
+    collect(plan) { case j: BroadcastRangeJoinExec => j }
+  }
+
+  /** `n` range joins sit in one whole-stage, not in nested stages or input 
adapters. */
+  private def oneCodegenStageOwns(plan: SparkPlan, n: Int): Boolean = {
+    exists(plan) {
+      case w: WholeStageCodegenExec =>
+        def count(p: SparkPlan): Int = p match {
+          case _: InputAdapter | _: WholeStageCodegenExec => 0
+          case j: BroadcastRangeJoinExec => 1 + j.children.map(count).sum
+          case other => other.children.map(count).sum
+        }
+        count(w.child) == n
+      case _ => false
+    }
+  }
+
+  /** The range join is compiled into a whole-stage, not hidden under an input 
adapter. */
+  private def rangeJoinIsCodegen(plan: SparkPlan): Boolean = {
+    exists(plan) {
+      case w: WholeStageCodegenExec =>
+        def contains(p: SparkPlan): Boolean = p match {
+          case _: InputAdapter => false
+          case _: BroadcastRangeJoinExec => true
+          case other => other.children.exists(contains)
+        }
+        contains(w.child)
+      case _ => false
+    }
+  }
+
+  private def findBroadcastNestedLoopJoin(plan: SparkPlan): 
Seq[BroadcastNestedLoopJoinExec] = {
+    collect(plan) { case j: BroadcastNestedLoopJoinExec => j }
+  }
+
+  private def findCartesianProduct(plan: SparkPlan): Seq[CartesianProductExec] 
= {
+    collect(plan) { case c: CartesianProductExec => c }
+  }
+
+  private def withRangeJoin(extra: (String, String)*)(f: => Unit): Unit = {
+    withSQLConf((Seq(SQLConf.BROADCAST_RANGE_JOIN_ENABLED.key -> "true") ++ 
extra): _*)(f)
+  }
+
+  /**
+   * One query per key family and join type that a range join can plan, shared 
by
+   * "whole-stage codegen covers each range-join probe" and "interpreted 
execution
+   * covers each range-join probe" so the two coverage lists cannot drift 
apart.
+   */
+  private def probeCoverageQueries: Seq[String] = Seq(
+    s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON $pointCond",
+    s"SELECT s.id, l.zip_id FROM signin_ip s LEFT JOIN ip_lookup l ON 
$pointCond",
+    s"SELECT s.id, l.zip_id FROM signin_ip s RIGHT JOIN ip_lookup l ON 
$pointCond",
+    s"SELECT s.id FROM signin_ip s LEFT SEMI JOIN ip_lookup l ON $pointCond",
+    s"SELECT s.id FROM signin_ip s LEFT ANTI JOIN ip_lookup l ON $pointCond",
+    "SELECT p.p, r.lo, r.hi FROM dec_points p JOIN dec_ranges r ON p.p >= r.lo 
AND p.p <= r.hi",
+    "SELECT /*+ BROADCAST(r) */ p.s, r.lo, r.hi FROM str_points p JOIN 
str_ranges r " +
+      "ON p.s >= r.lo AND p.s <= r.hi",
+    "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a JOIN intervals_b b " +
+      "ON a.lo < b.hi AND b.lo < a.hi",
+    "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a LEFT JOIN intervals_b b 
ON a.hi < b.lo",
+    "SELECT /*+ BROADCAST(a) */ a.lo, b.hi FROM intervals_a a JOIN intervals_b 
b ON a.lo < b.hi",
+    // DOUBLE keys stay on the boxed interval index (see 
setupDoubleIntervalViews), unlike
+    // the INT two-key overlap above, which is long-specialized.
+    "SELECT a.lo, a.hi, b.lo, b.hi FROM dbl_ranges_a a JOIN dbl_ranges_b b " +
+      "ON a.lo < b.hi AND b.lo < a.hi",
+    // Double and string keys stay on the boxed point index, so these queries 
reach
+    // its generated probe code (or, interpreted, the per-index dispatch in
+    // BroadcastRangeJoinExec.makeCandidateFn), which int partial ranges never 
do.
+    "SELECT p.d, b.lo FROM dbl_partial_points p JOIN dbl_partial_bounds b ON 
p.d < b.lo",
+    "SELECT p.d, b.lo FROM dbl_partial_bounds b JOIN dbl_partial_points p ON 
p.d > b.lo",
+    "SELECT p.s, b.lo FROM str_partial_points p JOIN str_partial_bounds b ON 
p.s < b.lo",
+    "SELECT p.s, b.lo FROM str_partial_bounds b JOIN str_partial_points p ON 
p.s > b.lo")
+
+  private def setupIpLookupViews(): Unit = {
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW signin_ip(id, ip_add_int) AS VALUES
+        |  ('1', 1467630666L),
+        |  ('2', 1186929717L),
+        |  ('3', 3501544352L),
+        |  ('31', 3501544320L),
+        |  ('32', 3501544383L),
+        |  ('33', 3501544330L),
+        |  ('34', 3501544390L),
+        |  ('4', 1626243295L),
+        |  ('41', 1626243072L),
+        |  ('42', 1626243327L),
+        |  ('5', 1206355995L),
+        |  ('5', 1206355995L),
+        |  ('6', CAST(NULL AS BIGINT)),
+        |  ('7', 0L)
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW ip_lookup(
+        |    BEGIN_IP2LONG, END_IP2LONG, TWO_CHAR_CNTRY_CD, zip_id) AS VALUES
+        |  (1467630592L, 1467630847L, 'de', 2380520L),
+        |  (1186929664L, 1186930175L, 'us', 6723936L),
+        |  (1186929664L, 1186930175L, 'us', 6723936L),
+        |  (3501544320L, 3501544383L, 'ca', 12804631L),
+        |  (3501544330L, 3501544390L, 'ca', 12804631001L),
+        |  (1626243072L, 1626243327L, 'us', 8849753L),
+        |  (1206355968L, 1206356223L, 'us', 10138801L),
+        |  (CAST(NULL AS BIGINT), 1206356223L, 'us', 10138801L),
+        |  (1206355968L, CAST(NULL AS BIGINT), 'us', 10138801L),
+        |  (CAST(NULL AS BIGINT), CAST(NULL AS BIGINT), 'us', 10138801L)
+        |""".stripMargin)
+  }
+
+  private def setupIntervalViews(): Unit = {
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW intervals_a(lo, hi) AS VALUES
+        |  (-1, 0), (0, 1), (0, 2), (1, 5)
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW intervals_b(lo, hi) AS VALUES
+        |  (-2, -1), (-4, -2), (1, 3), (5, 7)
+        |""".stripMargin)
+  }
+
+  /**
+   * Two-column DOUBLE ranges, so a two-key interval-overlap query here builds 
a boxed
+   * `IntervalIndex` (DoubleType has no bit-preserving long encoding) with 
both bounds
+   * supplied by the streamed side - the `case intervals: IntervalIndex` 
branch of
+   * `BroadcastRangeJoinExec.makeCandidateFn` that `intervals_a`/`intervals_b` 
(INT,
+   * long-specialized) never reaches.
+   */
+  private def setupDoubleIntervalViews(): Unit = {
+    // An unsuffixed literal like `1.0` resolves to DECIMAL, not DOUBLE (only 
`1.0D` or a
+    // cast does), and DECIMAL(2, 1) maps to LongIntervalIndex, defeating the 
point of this
+    // view. The `D` suffix keeps these columns on the boxed IntervalIndex 
path.
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dbl_ranges_a(lo, hi) AS VALUES
+        |  (-1.0D, 0.0D), (0.0D, 1.0D), (0.0D, 2.0D), (1.0D, 5.0D)
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dbl_ranges_b(lo, hi) AS VALUES
+        |  (-2.0D, -1.0D), (-4.0D, -2.0D), (1.0D, 3.0D), (5.0D, 7.0D)
+        |""".stripMargin)
+    assert(sql("SELECT lo, hi FROM 
dbl_ranges_a").schema.fields.forall(_.dataType == DoubleType))
+    assert(sql("SELECT lo, hi FROM 
dbl_ranges_b").schema.fields.forall(_.dataType == DoubleType))
+  }
+
+  private def setupDecimalViews(): Unit = {
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dec_ranges(lo, hi) AS VALUES
+        |  (CAST(0.00 AS DECIMAL(10, 2)), CAST(2.50 AS DECIMAL(10, 2))),
+        |  (CAST(3.00 AS DECIMAL(10, 2)), CAST(5.00 AS DECIMAL(10, 2)))
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dec_points(p) AS VALUES
+        |  (CAST(1.25 AS DECIMAL(10, 2))),
+        |  (CAST(3.00 AS DECIMAL(10, 2))),
+        |  (CAST(9.00 AS DECIMAL(10, 2)))
+        |""".stripMargin)
+  }
+
+  private def setupCollatedStrings(): Unit = {
+    // 'M' and 'm' compare equal under UNICODE_CI and differ as bytes.
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW str_ranges(lo, hi) AS VALUES
+        |  (CAST('a' AS STRING COLLATE UNICODE_CI), CAST('a' AS STRING COLLATE 
UNICODE_CI)),
+        |  (CAST('b' AS STRING COLLATE UNICODE_CI), CAST('b' AS STRING COLLATE 
UNICODE_CI)),
+        |  (CAST('c' AS STRING COLLATE UNICODE_CI), CAST('M' AS STRING COLLATE 
UNICODE_CI)),
+        |  (CAST('m' AS STRING COLLATE UNICODE_CI), CAST('z' AS STRING COLLATE 
UNICODE_CI))
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW str_points(s) AS VALUES
+        |  (CAST('M' AS STRING COLLATE UNICODE_CI)),
+        |  (CAST('m' AS STRING COLLATE UNICODE_CI))
+        |""".stripMargin)
+  }
+
+  /** Points and bounds for single-inequality joins on boxed key families. */
+  private def setupPartialRangeViews(): Unit = {
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dbl_partial_points(d) AS VALUES
+        |  (0.5), (3.0), (6.0),
+        |  (CAST(NULL AS DOUBLE))
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dbl_partial_bounds(lo) AS VALUES
+        |  (1.0), (4.0), (7.5)
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW str_partial_points(s) AS VALUES
+        |  ('a'), ('m'), ('z')
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW str_partial_bounds(lo) AS VALUES
+        |  ('b'), ('n')
+        |""".stripMargin)
+  }
+
+  private def setupDateAndTimestampViews(): Unit = {
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dt_ranges(lo, hi) AS VALUES
+        |  (DATE'2024-01-01', DATE'2024-01-31'),
+        |  (DATE'2024-06-01', DATE'2024-06-30')
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW dt_points(d) AS VALUES
+        |  (DATE'2024-01-15'),
+        |  (DATE'2024-05-01'),
+        |  (DATE'2024-06-15')
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW ts_ranges(lo, hi) AS VALUES
+        |  (TIMESTAMP'2024-01-01 00:00:00', TIMESTAMP'2024-01-01 12:00:00')
+        |""".stripMargin)
+    sql(
+      """
+        |CREATE OR REPLACE TEMP VIEW ts_points(t) AS VALUES
+        |  (TIMESTAMP'2024-01-01 06:00:00'),
+        |  (TIMESTAMP'2024-01-01 18:00:00')
+        |""".stripMargin)
+  }
+
+  test("range join is disabled by default and falls back to nested loop join") 
{
+    setupIpLookupViews()
+    val df = sql(s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON 
$pointCond")
+    assert(findRangeJoin(df.queryExecution.executedPlan).isEmpty)
+    assert(findBroadcastNestedLoopJoin(df.queryExecution.executedPlan).size == 
1)
+  }
+
+  test("point-in-range matches the nested loop result") {
+    setupIpLookupViews()
+    val query = s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON 
$pointCond"
+    val expected = sql(query).collect()
+    withRangeJoin() {
+      val df = sql(query)
+      assert(findRangeJoin(df.queryExecution.executedPlan).size == 1)
+      checkAnswer(df, expected)
+    }
+  }
+
+  test("BETWEEN is planned as a point-in-range join") {
+    setupIpLookupViews()
+    val query =
+      "SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l " +
+        "ON s.ip_add_int BETWEEN l.BEGIN_IP2LONG AND l.END_IP2LONG"
+    val expected = sql(query).collect()
+    withRangeJoin(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> "false") {
+      val df = sql(query)
+      val planned = findRangeJoin(df.queryExecution.executedPlan)
+      assert(planned.size == 1)
+      // BETWEEN lowers to one point expression. The other side keeps (low, 
high).
+      val keyCounts = Set(planned.head.leftKeys.length, 
planned.head.rightKeys.length)
+      assert(keyCounts == Set(1, 2))
+      checkAnswer(df, expected)
+    }
+  }
+
+  test("hints select range join, nested loop, or cartesian") {
+    setupIpLookupViews()
+    setupIntervalViews()
+    val expected = sql(
+      s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON 
$pointCond").collect()
+    withRangeJoin() {
+      val broadcast = sql(
+        "SELECT /*+ BROADCAST(l) */ s.id, l.zip_id FROM signin_ip s JOIN 
ip_lookup l " +
+          s"ON $pointCond")
+      val joins = findRangeJoin(broadcast.queryExecution.executedPlan)
+      assert(joins.size == 1 && joins.head.buildSide == BuildRight)
+      checkAnswer(broadcast, expected)
+
+      val dualHintInner = sql(
+        "SELECT /*+ BROADCAST(s), BROADCAST(l) */ s.id, l.zip_id " +
+          s"FROM signin_ip s JOIN ip_lookup l ON $pointCond")
+      val dualJoins = findRangeJoin(dualHintInner.queryExecution.executedPlan)
+      assert(dualJoins.size == 1 && dualJoins.head.buildSide == BuildRight)
+      checkAnswer(dualHintInner, expected)
+
+      val dualHintLeftOuter = sql(
+        "SELECT /*+ BROADCAST(s), BROADCAST(l) */ s.id, l.zip_id " +
+          s"FROM signin_ip s LEFT JOIN ip_lookup l ON $pointCond")
+      val dualOuterJoins = 
findRangeJoin(dualHintLeftOuter.queryExecution.executedPlan)
+      assert(dualOuterJoins.size == 1 && dualOuterJoins.head.buildSide == 
BuildRight)
+
+      val unsupported = sql(
+        "SELECT /*+ BROADCAST(a) */ a.lo, b.hi FROM intervals_a a LEFT JOIN 
intervals_b b " +
+          "ON a.lo < b.hi")
+      val loops = 
findBroadcastNestedLoopJoin(unsupported.queryExecution.executedPlan)
+      assert(findRangeJoin(unsupported.queryExecution.executedPlan).isEmpty)
+      assert(loops.size == 1 && loops.head.buildSide == BuildLeft)
+
+      val cartesian = sql(
+        "SELECT /*+ SHUFFLE_REPLICATE_NL(s) */ s.id, l.zip_id " +
+          s"FROM signin_ip s JOIN ip_lookup l ON $pointCond")
+      val plan = cartesian.queryExecution.executedPlan
+      assert(findRangeJoin(plan).isEmpty)
+      assert(findCartesianProduct(plan).size == 1)
+    }
+  }
+
+  test("supported join types match nested loop; full outer does not") {
+    setupIpLookupViews()
+    Seq(
+      s"SELECT s.id, l.zip_id FROM signin_ip s LEFT JOIN ip_lookup l ON 
$pointCond",
+      s"SELECT s.id, l.zip_id FROM signin_ip s RIGHT JOIN ip_lookup l ON 
$pointCond",
+      s"SELECT s.id FROM signin_ip s LEFT SEMI JOIN ip_lookup l ON $pointCond",
+      s"SELECT s.id FROM signin_ip s LEFT ANTI JOIN ip_lookup l ON $pointCond"
+    ).foreach { query =>
+      val expected = sql(query).collect()
+      withRangeJoin() {
+        val df = sql(query)
+        assert(findRangeJoin(df.queryExecution.executedPlan).size == 1, query)
+        checkAnswer(df, expected)
+      }
+    }
+    val fullOuter =
+      s"SELECT s.id, l.zip_id FROM signin_ip s FULL OUTER JOIN ip_lookup l ON 
$pointCond"
+    withRangeJoin() {
+      assert(findRangeJoin(sql(fullOuter).queryExecution.executedPlan).isEmpty)
+    }
+  }
+
+  test("partial range left outer broadcasts the right side") {
+    setupIntervalViews()
+    // No b.lo is strictly greater than a.hi (5), so the left row (1, 5) is 
kept with nulls.
+    val query =
+      "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a LEFT JOIN intervals_b 
b " +
+        "ON a.hi < b.lo"
+    val expected = sql(query).collect()
+    assert(expected.exists(_.anyNull))
+    withRangeJoin() {
+      val df = sql(query)
+      val planned = findRangeJoin(df.queryExecution.executedPlan)
+      assert(planned.size == 1)
+      assert(planned.head.joinType == LeftOuter)
+      assert(planned.head.buildSide == BuildRight)
+      assert(planned.head.rangeJoinKind == LessPartialRangeJoin(strict = true))
+      checkAnswer(df, expected)
+    }
+  }
+
+  test("strict inequalities over duplicated build keys match the nested loop 
join") {
+    // 10,000 build rows over 100 dates, 100 per date. Under `<`, a point

Review Comment:
   **Nit (P3):** `dup_dates` and `probe_dates` are `id % 100` and `id` from 
`range()`, so BIGINT, and the candidate-count test in `RangeJoinSuite` (lines 
306-308) builds IntegerType rows while also talking about dates. Behavior is 
unaffected, but it reads like DateType coverage of the strict scans. Could the 
names and comments say keys or values instead?
   
   **Verification:**
   - **Inspection:** The view names, test comments, and fixture descriptions of 
both duplicated-key strict-scan tests describe the integer key type the 
fixtures use and make no date-type coverage claim.



-- 
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: reviews-unsubscr...@spark.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org