cloud-fan commented on code in PR #59027:
URL: https://github.com/apache/spark/pull/59027#discussion_r4494427358
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1220,6 +1220,25 @@ object SQLConf {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("10MB")
+ val BROADCAST_RANGE_JOIN_ENABLED =
buildConf("spark.sql.join.broadcastRangeJoin.enabled")
+ .internal()
+ .doc("When true, a non-equi join is planned as a broadcast range join
instead of a broadcast " +
+ "nested loop join when it tests whether a value falls inside a range,
whether two ranges " +
+ "overlap, or compares the two sides once with `<`, `<=`, `>`, or `>=`.
Each range key must " +
+ "be deterministic, of boolean, integer, floating-point, decimal, string,
binary, date, or " +
+ "timestamp type, and drawn entirely from one side. The broadcast side is
sorted and " +
+ "indexed once, which evaluates the range key of every broadcast row.
When probing, " +
+ "each row of the other side evaluates its range key and reads only the
part of the index " +
+ "that can match it. Keys are evaluated before any pair is tested, so a
key expression that " +
+ "throws an error can fail a query that a nested loop join would have
completed. The range " +
+ "join is much faster when that part is small, somewhat slower when it
covers most of the " +
Review Comment:
**Non-blocking (P2):** This rewrite drops most of what the previous version
of this doc said, and the doc is the only guidance on when to turn the flag on:
- It says a matching non-equi join "is planned as a broadcast range join",
but that happens only when a side the range join can build is broadcastable by
size or by a `BROADCAST` hint: either side for inner joins, the right side for
left outer, semi, and anti joins, the left side for right outer joins, and
never for full outer joins. `BROADCAST(a)` on `a LEFT JOIN b ON a.lo < b.hi`
and every full outer join keep the nested loop join, and an unhinted join with
no small side keeps its previous plan (a cartesian product for inner joins,
otherwise the nested loop join). The commit message says the rewrite covers
these prerequisites, but the text doesn't.
- "Somewhat slower when it covers most of the broadcast side" doesn't match
the committed results. Fully overlapping builds run at 1.0X, while staggered
builds on double keys are already at 0.6X at 5% overlap and 0.5X at 10% (JDK
17), and long keys slow down from about 10%. The slow case is interleaved
overlap, and the boxed key families reach it at a much smaller fraction of the
build than the long-specialized ones.
- The size-check trade-off is gone. `BroadcastExchangeExec` charges the
index's retained memory against `maxBroadcastTableSize`, so a hinted or
raised-threshold build can now fail where the nested loop join ran.
Could the doc state the planning conditions, describe the slow case as
interleaved overlap with the key-family difference (fixed ratios aren't
needed), note that fully overlapping builds run at about parity, and bring back
the size-check trade-off, while keeping the new sentences on eager key
evaluation and index-build cost?
This extends existing discussions
[1](https://github.com/apache/spark/pull/59027#issuecomment-6514980325),
[2](https://github.com/apache/spark/pull/59027#issuecomment-6544954116),
[3](https://github.com/apache/spark/pull/59027#issuecomment-6554108959),
[4](https://github.com/apache/spark/pull/59027#issuecomment-6566747464),
[5](https://github.com/apache/spark/pull/59027#discussion_r4237663576),
[6](https://github.com/apache/spark/pull/59027#discussion_r4346073069),
[7](https://github.com/apache/spark/pull/59027#discussion_r4359606313),
[8](https://github.com/apache/spark/pull/59027#discussion_r4367132555),
[9](https://github.com/apache/spark/pull/59027#discussion_r4420441244),
[10](https://github.com/apache/spark/pull/59027#pullrequestreview-5628337351).
**Verification:**
- **Inspection:** Every plan-switch condition, key-family statement, slow
case, and failure case the config doc names agrees with the planner's broadcast
and join-type rules, the long-key routing, the broadcast size check, and the
committed staggered-overlap and full-overlap benchmark rows.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -350,22 +357,33 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
Seq(joins.BroadcastHashJoinExec(leftKeys, rightKeys, LeftAnti,
BuildRight,
None, planLater(j.left), planLater(j.right), isNullAwareAntiJoin =
true))
- // If it is not an equi-join, we first look at the join hints w.r.t. the
following order:
- // 1. broadcast hint: pick broadcast nested loop join. If both sides
have the broadcast
- // hints, choose the smaller side (based on stats) to broadcast for
inner and full joins,
- // choose the left side for right join, and choose right side for
left join.
- // 2. shuffle replicate NL hint: pick cartesian product if join type
is inner like.
+ // If it is not an equi-join, a broadcast hint is considered first:
+ // 1. When range join is enabled and the condition is a range
predicate,
+ // pick a broadcast range join if the hinted build side is one
+ // BroadcastRangeJoinExec supports. Both sides hinted: the smaller
Review Comment:
**Nit (P3):** With both sides hinted, `broadcastSide` drops a side
`BroadcastRangeJoinExec` can't build before it compares sizes, so only inner
joins pick the smaller side: a left outer, semi, or anti join always builds the
right side, and a right outer join the left side, whichever is smaller.
`RangeJoinSQLSuite`'s dual-hinted left outer case asserts `BuildRight`. Read as
written, "the smaller side, when that side is supported. An unsupported hinted
side falls through" suggests that a dual-hinted left outer join with a smaller
left side falls through to the nested loop join. Could this say that an
unsupported hinted side is dropped first, so that with both sides hinted an
inner join builds the smaller side, left outer, semi, and anti joins build the
right side, and a right outer join the left side, and that only a join whose
hinted sides are all unsupported falls through?
**Verification:**
- **Inspection:** For inner, left outer, left semi, left anti, and right
outer joins with one or both sides hinted, the comment's stated build side or
fall-through matches the planner code and the existing dual-hint planner
assertions.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeRelation.scala:
##########
@@ -0,0 +1,1226 @@
+/*
+ * 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}
+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. Within a
predicate, the
+ * analyzer's comparison type coercion gives both sides one type; across the
two
+ * predicates an interval-overlap join pairs, `ExtractRangeJoinKeys`' own
same-type
+ * guards keep the streamed key on the build side's type, 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)
+
+ /**
+ * One key type's bridge to an order-preserving long: converting an evaluated
+ * key value (from either the build or the streamed side), and turning
+ * generated Java for a typed key into generated Java for a long. The two
+ * agree, so a build key and a probe key of the same type become the same
+ * long.
+ */
+ final case class LongKey(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(value: Any => Long): LongKey = LongKey(value, v => s"((long) $v)")
+ dt match {
+ case _: BooleanType =>
+ Some(LongKey(v => if (v.asInstanceOf[Boolean]) 1L else 0L, v => s"($v
? 1L : 0L)"))
+ case _: ByteType =>
+ Some(key(v => v.asInstanceOf[Byte].toLong))
+ case _: ShortType =>
+ Some(key(v => v.asInstanceOf[Short].toLong))
+ case _: IntegerType | _: DateType =>
+ Some(key(v => v.asInstanceOf[Int].toLong))
+ case _: LongType | _: TimestampType | _: TimestampNTZType =>
+ Some(key(v => v.asInstanceOf[Long]))
+ case d: DecimalType if d.precision <= Decimal.MAX_LONG_DIGITS =>
+ Some(LongKey(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. Within a predicate,
the
+ * analyzer's comparison type coercion gives both sides one type; across
predicates,
+ * `ExtractRangeJoinKeys`' own same-type guards keep build and streamed keys
aligned,
+ * 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 each build key expression evaluates to, 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 apply(
+ keyType: DataType,
+ buildKeys: Seq[Expression],
+ rows: Iterator[InternalRow],
+ sizeHint: Option[Long]): IntervalIndex = {
+ val buffer = new ArrayBuffer[(Any, Any,
InternalRow)](RangeIndex.initialCapacity(sizeHint))
+ 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.apply`.
+ val lowExpr = buildKeys.head
+ while (rows.hasNext) {
+ val r = rows.next()
+ val rawLo = lowExpr.eval(r)
+ if (rawLo != null) {
+ val key = InternalRow.copyValue(rawLo)
+ buffer += ((key, key, r.copy()))
+ }
+ }
+ } else {
+ val lowExpr = buildKeys.head
+ val highExpr = buildKeys(1)
+ while (rows.hasNext) {
+ val r = rows.next()
+ val rawLo = lowExpr.eval(r)
+ val rawHi = highExpr.eval(r)
+ 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 apply(
+ keyType: DataType,
+ buildKeys: Seq[Expression],
+ rows: Iterator[InternalRow],
+ sizeHint: Option[Long]): PointIndex = {
+ val buffer = new ArrayBuffer[(Any,
InternalRow)](RangeIndex.initialCapacity(sizeHint))
+ val keyExpr = buildKeys.head
+ val ordering = PhysicalDataType.ordering(keyType)
+
+ while (rows.hasNext) {
+ val r = rows.next()
+ val rawKey = keyExpr.eval(r)
+ 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)
Review Comment:
**Non-blocking (P2):** Every interval probe that the all-covering check
doesn't answer allocates a `LongOverlapIterator` (or an `OverlapIterator` for
boxed keys) with two stack arrays and walks the tree, and `createRangeJoin`
picks the range join however few rows the build side has. For a small dimension
table that costs more than the nested loop join's pass over a handful of rows.
In a local repro at this head (10M streamed points joined to a
`BROADCAST`-hinted build of disjoint ranges, one JVM, flag on against flag
off), an 8-row build ran at 0.77X of the nested loop join on long keys (0.81X
for a left outer join) and 0.70X on double keys. A 32-row double build was at
parity, and the range join pulled ahead from about 32 rows on long keys and 64
on double keys; a 256-row control ran at 4.5X-5.8X. Returning every row for
builds of at most 16 rows instead of walking brought the 8-row cases to
1.04X-1.08X and left the control unchanged.
Answers are unaffected, but this is the selective case the config doc calls
"much faster", for lookup tables like the tax brackets and date windows the
PR's motivation names. How much of the gap remains in a cluster job, where
scans and shuffles add their own time, wasn't measured.
**Recommended change:** For builds of at most a small number of rows, have
both interval indexes answer an overlap probe without building the walk
iterator: either return every row, as the all-covering shortcut already does,
or scan the few rows in place and return the ones that overlap. Set the cutoff
from benchmark rows for builds of 8, 16, 32, and 64 ranges on long and double
keys; a cutoff of 16 rows brought an 8-row build to parity in a local run. Add
those rows to the range join benchmark, and move the index tests that assert a
walk iterator or an exact overlap set on builds below the cutoff to larger
builds, or have them assert the small-build behavior instead.
**Why this works:** The cost on tiny builds is the per-probe walk setup and
traversal, not the condition checks: with the same 8-row build, returning every
row and letting the join's condition reject the non-matching ones ran at
1.04X-1.08X of the nested loop join, where the walk ran at 0.70X-0.81X. Every
consumer already treats the probe result as candidates and re-checks the
original condition, so skipping the walk keeps answers unchanged.
**Scope:** Give the interval indexes a small-build probe path that skips the
tree walk, and measure its cutoff.
**Compatibility:** The all-covering shortcut, the lazy walk for larger
builds, and the build, serialization, and size estimate of both interval
indexes stay as they are.
**Risks:** Builds just above the cutoff keep the walk, so a cutoff set too
low leaves part of the slowdown and one set too high gives up pruning where the
walk already wins; the cutoff is an internal constant that can be retuned, and
turning the flag off restores the nested loop join. If small builds return
every row, the index no longer returns exactly the overlapping rows for them,
so tests or a future caller that treat the probe result as exact must accept a
superset. Single-inequality joins over tiny builds were not measured, so they
may keep a smaller gap until benchmark rows show otherwise.
**Constraints:** Answers must not change: every row that overlaps the window
is still returned, null bounds still match nothing, and inverted windows are
still normalized. The interpreted and generated probe paths must take the same
small-build path.
**Exclusions:**
- Single-inequality joins over tiny builds, which use the point indexes and
were not measured, unless their benchmark rows show the same gap.
**Success:** With the flag on, a point-in-range or interval-overlap range
join over a build of a handful of rows runs at or above the nested loop join on
long and double keys, for inner and outer joins, with codegen on and off. Range
joins over larger builds keep their current speedups and answers.
**Verification:**
- **Regression:** Benchmark rows for builds of 8, 16, 32, and 64 ranges
against at least 10M streamed rows, for point-in-range inner and left outer
joins on long and double keys, show the range join at or above the nested loop
join, while the existing rows for larger builds keep their speedups.
- **Behavior:** Overlap probes on builds below, at, and just above the
cutoff give the same join answers as the nested loop join, including null
bounds, inverted intervals, and NaN and signed-zero double keys, with codegen
on and off.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/LogicalQueryStageStrategy.scala:
##########
@@ -58,6 +63,79 @@ object LogicalQueryStageStrategy extends Strategy {
case _ => false
}
+ /**
+ * Extracts the [[RangeRelationBroadcastMode]] and
[[BroadcastQueryStageExec]] carried by
+ * a broadcast query stage, if any.
+ */
+ private def broadcastRangeStage(
+ plan: LogicalPlan): Option[(RangeRelationBroadcastMode,
BroadcastQueryStageExec)] =
+ plan match {
+ case LogicalQueryStage(_, bqs: BroadcastQueryStageExec) =>
+ bqs.broadcast.outputPartitioning match {
+ case BroadcastPartitioning(m: RangeRelationBroadcastMode) =>
Some((m, bqs))
+ case _ => None
+ }
+ case _ => None
+ }
+
+ /**
+ * A range-predicate join whose left or right child is already a
range-broadcast
+ * query stage. `unapply` looks up the mode once so `apply` does not search
+ * for both a guard and the constructor.
+ */
+ private object ExtractBroadcastRangeJoinStage {
+ type ReturnType =
+ (LogicalPlan, LogicalPlan, BuildSide, JoinType, Seq[Expression],
Seq[Expression],
+ Expression, RangeJoinKind)
+
+ def unapply(plan: LogicalPlan): Option[ReturnType] = {
+ if (!SQLConf.get.broadcastRangeJoinEnabled) {
+ None
+ } else {
+ plan match {
+ case j @ Join(_, _, _, Some(cond), _) =>
+ j match {
+ case ExtractRangeJoinKeys(joinType, leftKeys, rightKeys,
rangeJoinKind) =>
+ def fromStage(
+ mode: RangeRelationBroadcastMode,
+ bqs: BroadcastQueryStageExec,
+ buildSide: BuildSide): Option[ReturnType] = {
+ if (!mode.indexKind.matches(rangeJoinKind) ||
+ !BroadcastRangeJoinExec.supports(joinType, buildSide)) {
+ None
+ } else {
+ val buildKeys = buildSide match {
+ case BuildLeft => leftKeys
+ case BuildRight => rightKeys
+ }
+ // The mode's keys are ordinals, bound at plan time
against the
+ // exchange input, which is `bqs.output`. The
`LogicalQueryStage`
+ // carries the logical fragment instead, and its output
can differ
+ // from the stage's column order after AQE re-planning, so
binding
+ // against `bqs.output` compares ordinals against the same
order
+ // the broadcast was actually keyed on.
+ val bound = BindReferences.bindReferences(buildKeys,
bqs.output)
+ if (mode == RangeRelationBroadcastMode(bound,
mode.indexKind)) {
+ Some((j.left, j.right, buildSide, joinType, leftKeys,
rightKeys, cond,
+ rangeJoinKind))
+ } else {
+ None
+ }
+ }
+ }
+ broadcastRangeStage(j.left).flatMap {
Review Comment:
**Nit (P3):** This left-stage branch has no direct test: all four new
range-stage tests in `AdaptiveQueryExecSuite` build `Join(left,
LogicalQueryStage(right, ...))`. If this branch stopped matching, for example
by returning `None` or skipping the `supports` check, build-left range joins
(inner joins with a smaller left side, right outer joins) would quietly lose
stage reuse under AQE, because JoinSelection re-plans them over the stage. No
answer would change and no test would fail. Could you add left-stage cases next
to the right-stage ones: a matching inner join planned as a build-left range
join, a left stage with different keys or a different index kind that isn't
reused, and a left outer join that isn't planned over a left stage?
**Verification:**
- **Regression:** With a range broadcast stage on the left child, a matching
inner join is planned as a build-left range join keyed on the left keys, while
a mismatched mode and an unsupported join type are not planned over that stage,
each assertion failing if the left branch returns nothing, binds the right
keys, or skips the join-type check.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/RangeIndexSuite.scala:
##########
@@ -0,0 +1,1180 @@
+/*
+ * 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 `IntervalIndex.apply`. 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(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(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.apply defends against a build-side iterator reusing its
row buffer") {
+ // If apply() 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(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.apply 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(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(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(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 and contains n, so the
probe
+ // at n hits all 300 rows; the probe at n/2 hits about half (151), since
only
+ // intervals with i <= n/2 reach that far. The tree still stores one node
per row.
+ val n = 300
+ val nested = buildIntervals((0 until n).map(i => span(i, 2 * n - i, i)))
+
+ def expected(probe: Int): Set[Int] =
+ (0 until n).filter(i => i <= probe && probe <= 2 * n - i).toSet
+
+ Seq(0, 1, n / 2, n, 2 * n - 1, 2 * n).foreach { probe =>
+ assert(probeIds(nested, probe, probe).toSet == expected(probe), s"probe
$probe")
+ }
+ }
+
+ test("broadcast of heavily overlapping intervals stays linear") {
+ // [i, i + n] for i in 0 until n. Every interval contains n. The broadcast
+ // is the flat arrays, one slot per interval, so serialization stays
linear.
+ val n = 200
+ val index = buildIntervals((0 until n).map(i => span(i, i + n, i)))
+ val (raw, restored) = javaRoundTrip(index)
+ assert(raw.length < n * 1024, s"serialized ${raw.length} bytes")
+ val probe = n
+ assert(probeIds(restored, probe, probe).toSet == probeIds(index, probe,
probe).toSet)
+ assert(probeIds(restored, probe, probe).size == n)
+ }
+
+ test("estimatedSize depends on the interval count, not the overlap depth") {
+ val n = 200
+ val one = buildIntervals(Seq(span(0, 1, 0)))
+ val disjoint = buildIntervals((0 until n).map(i => span(2 * i, 2 * i + 1,
i)))
+ val overlap = buildIntervals((0 until n).map(i => span(i, i + n, i)))
+ val again = buildIntervals((0 until n).map(i => span(2 * i, 2 * i + 1, i)))
+ assert(disjoint.estimatedSize > one.estimatedSize)
+ assert(overlap.estimatedSize > one.estimatedSize)
+ assert(disjoint.estimatedSize == again.estimatedSize)
+ // Nesting stores no extra copy of the rows. Key objects differ, so the two
+ // indexes need not be equal, but overlap stays the same order of
magnitude.
+ assert(overlap.estimatedSize < disjoint.estimatedSize * 3)
+ }
+
+ test("estimatedSize includes distinct key objects") {
+ val ints = buildIntervals(Seq(span(0, 1, 0)))
+ val wide = UTF8String.fromString("x" * 64)
+ val strings = buildIntervals(StringType, Seq((wide, wide, row(0))))
+ assert(strings.estimatedSize > ints.estimatedSize)
+ }
+
+ test("interval tree treats NaN as its own point and Infinity as a bound") {
+ // PhysicalDoubleType's ordering (SQLOrderingUtil.compareDoubles) totally
orders
+ // -Inf < ... < +Inf < NaN, with NaN comparing equal only to itself. So
[1.0, +Inf]
+ // includes +Inf under inclusive probe, and a NaN key overlaps only a
probe that
+ // also lands exactly on NaN.
+ val index = buildIntervals(DoubleType, Seq(
+ (1.0, Double.PositiveInfinity, row(1)),
+ (Double.NegativeInfinity, -1.0, row(2)),
+ (Double.NaN, Double.NaN, row(3))))
+
+ assert(ids(index.overlapping(5.0, 5.0)) == List(1))
+ assert(ids(index.overlapping(Double.PositiveInfinity,
Double.PositiveInfinity)) == List(1))
+ assert(ids(index.overlapping(-5.0, -5.0)) == List(2))
+ assert(ids(index.overlapping(Double.NegativeInfinity,
Double.NegativeInfinity)) == List(2))
+ assert(ids(index.overlapping(Double.NaN, Double.NaN)) == List(3))
+ assert(ids(index.overlapping(0.0, 0.0)).isEmpty)
+ }
+
+ test("interval index merges keys that compare equal but are not equal") {
+ // Array equals is identity, and UTF8String equals is binary, so these keys
+ // compare equal under the index ordering without being equal. The tree
+ // orders them with that comparator, so a probe on either key returns both.
+ // (Signed zero is the same kind of merge, but Double's == already treats
-0.0
+ // and 0.0 as equal, which fails this helper's != precondition - see the
+ // dedicated "-0.0 and 0.0 are treated as one key" test below instead.)
+ def assertMerged(
+ dataType: DataType,
+ low: Any,
+ leftKey: Any,
+ rightKey: Any,
+ high: Any,
+ earlier: Seq[Any]): Unit = {
+ val ordering = PhysicalDataType.ordering(dataType)
+ assert(leftKey != rightKey)
+ assert(ordering.compare(leftKey, rightKey) == 0)
+ val points = Array(
+ (leftKey, leftKey, row(0)),
+ (rightKey, rightKey, row(1)))
+ val pointIdx = buildIntervals(dataType, points.toSeq)
+ assert(ids(pointIdx.overlapping(leftKey, leftKey)).toSet == Set(0, 1))
+ assert(ids(pointIdx.overlapping(rightKey, rightKey)).toSet == Set(0, 1))
+
+ val intervals = earlier.zipWithIndex.map { case (key, i) =>
+ (key, key, row(10 + i))
+ }.toArray ++ Array(
+ (low, leftKey, row(0)),
+ (rightKey, high, row(1)))
+ val index = buildIntervals(dataType, intervals.toSeq)
+ assert(ids(index.overlapping(leftKey, leftKey)).sorted == List(0, 1))
+ assert(ids(index.overlapping(rightKey, rightKey)).sorted == List(0, 1))
+ }
+
+ val bytes = Array[Byte](1, 2)
+ val sameBytes = Array[Byte](1, 2)
+ assertMerged(
+ BinaryType,
+ Array[Byte](1, 0),
+ bytes,
+ sameBytes,
+ Array[Byte](2),
+ Seq(Array[Byte](0), Array[Byte](0, 1)))
+
+ val upper = UTF8String.fromString("M")
+ val lower = UTF8String.fromString("m")
+ assertMerged(
+ StringType("UNICODE_CI"),
+ UTF8String.fromString("c"),
+ upper,
+ lower,
+ UTF8String.fromString("z"),
+ Seq(UTF8String.fromString("a"), UTF8String.fromString("b")))
+ }
+
+ test("-0.0 and 0.0 are treated as one key") {
+ // Double's own `==` already treats -0.0 and 0.0 as equal, so
`PhysicalDataType.ordering`
+ // (which SQL requires to distinguish them from every other value, but not
from each other)
+ // merges them exactly like the UTF8String/Array[Byte] keys above: a probe
on either
+ // sign finds rows stored under both.
+ val negZero: Any = -0.0
+ val posZero: Any = 0.0
+ val ordering = PhysicalDataType.ordering(DoubleType)
+ assert(ordering.compare(negZero, posZero) == 0)
+
+ val points = Array((negZero, negZero, row(0)), (posZero, posZero, row(1)))
+ val pointIdx = buildIntervals(DoubleType, points.toSeq)
+ assert(ids(pointIdx.overlapping(negZero, negZero)).toSet == Set(0, 1))
+ assert(ids(pointIdx.overlapping(posZero, posZero)).toSet == Set(0, 1))
+
+ val intervals = Array((-5.0: Any, negZero, row(0)), (posZero, 5.0: Any,
row(1)))
+ val index = buildIntervals(DoubleType, intervals.toSeq)
+ assert(ids(index.overlapping(negZero, negZero)).sorted == List(0, 1))
+ assert(ids(index.overlapping(posZero, posZero)).sorted == List(0, 1))
+ }
+
+ test("a nested probe does not drop rows from the outer iterator") {
+ val spans = (0 until 32).map(i => span(i, i + 10, i))
+ val index = buildIntervals(spans)
+ val longIdx = RangeRelationBroadcastMode(
+ Seq(BoundReference(0, IntegerType, nullable = true),
+ BoundReference(1, IntegerType, nullable = true)),
+ IntervalIndexKind)
+ .transform(spans.map(t => InternalRow(t._1, t._2,
t._3.getInt(0))).toArray)
+ .asInstanceOf[LongIntervalIndex]
+
+ // [0, 15] does not cover every span (the core is [31, 10]), so this walks
the tree
+ // instead of taking the all-covering shortcut: proof that two concurrent
walks can
+ // run on one index, the shape two interval joins fused in one whole-stage
rely on.
+ val expectedOuter = (0 to 15).toList
+ val outer = index.overlapping(0, 15)
+ assert(outer.isInstanceOf[OverlapIterator])
+ val seen = scala.collection.mutable.ArrayBuffer.empty[Int]
+ while (outer.hasNext) {
+ seen += outer.next().getInt(0)
+ assert(ids(index.overlapping(0, 5)).sorted == (0 to 5).toList)
+ }
+ assert(seen.sorted == expectedOuter)
+
+ val longOuter = longIdx.overlapping(0L, 15L)
+ assert(longOuter.isInstanceOf[LongOverlapIterator])
+ val longSeen = scala.collection.mutable.ArrayBuffer.empty[Int]
+ while (longOuter.hasNext) {
+ longSeen += longOuter.next().getInt(2)
+ assert(longIdx.overlapping(0L, 5L).map(_.getInt(2)).toList.sorted == (0
to 5).toList)
+ }
+ assert(longSeen.sorted == expectedOuter)
+ }
+
+ test("a point probe of disjoint intervals returns only the containing row") {
+ val n = 1000
+ val index = buildIntervals((0 until n).map(i => span(i * 2, i * 2 + 1, i)))
+ assert(probeIds(index, 2, 2) == List(1))
+ assert(probeIds(index, 3, 3) == List(1))
+ assert(probeIds(index, 4, 4) == List(2))
+ assert(probeIds(index, -1, -1).isEmpty)
+ assert(probeIds(index, 100000, 100000).isEmpty)
+ }
+
+ test("point index answers both sides of a bound, including equals") {
+ // Unsorted, with two rows on key 5. Equals stay, in input order.
+ val keyed = Array[(Any, InternalRow)](
+ (8, row(8)),
+ (1, row(1)),
+ (5, row(5)),
+ (5, row(50)),
+ (3, row(3)),
+ (null, row(-1)))
+ val index = buildPoints(IntegerType, keyed.toSeq)
+ assert(ids(index.upTo(5)) == List(1, 3, 5, 50))
+ assert(ids(index.upTo(4)) == List(1, 3))
+ assert(ids(index.upTo(0)).isEmpty)
+ assert(ids(index.upTo(8)) == List(1, 3, 5, 50, 8))
+ assert(ids(index.from(5)) == List(5, 50, 8))
+ assert(ids(index.from(6)) == List(8))
+ assert(ids(index.from(9)).isEmpty)
+ assert(ids(index.from(1)) == List(1, 3, 5, 50, 8))
+ assert(index.upTo(null).isEmpty)
+ assert(index.from(null).isEmpty)
+
+ val empty = buildPoints(IntegerType, Seq.empty)
+ assert(empty.upTo(1).isEmpty)
+ assert(empty.from(1).isEmpty)
+ // SizeEstimator charges the two empty arrays' shells, and the five kept
+ // points on top of them.
+ assert(empty.estimatedSize > 0)
+ assert(index.estimatedSize > empty.estimatedSize)
+ }
+
+ test("exclusive point bounds skip duplicated build keys") {
+ // 100 duplicate rows per key on both families. A strict partial predicate
+ // rejects the probe's own key, so the exclusive bound must skip it
+ // instead of surfacing 100 rows that all fail the condition.
+ val keyed = (1 to 5).flatMap(k => Seq.fill(100)((k: Any, row(k))))
+ val boxed = buildPoints(IntegerType, keyed)
+ val longPoints = RangeRelationBroadcastMode(
+ Seq(BoundReference(1, IntegerType, nullable = true)), PointIndexKind)
+ .transform(keyed.map(t => InternalRow(t._2.getInt(0), t._1)).toArray)
+ .asInstanceOf[LongPointIndex]
+
+ assert(ids(boxed.upTo(3)).size == 300)
+ assert(ids(boxed.upToExclusive(3)).size == 200)
+ assert(ids(boxed.from(3)).size == 300)
+ assert(ids(boxed.fromExclusive(3)).size == 200)
+ assert(ids(boxed.upToExclusive(1)).isEmpty)
+ assert(ids(boxed.fromExclusive(5)).isEmpty)
+
+ assert(ids(longPoints.upTo(3L)).size == 300)
+ assert(ids(longPoints.upToExclusive(3L)).size == 200)
+ assert(ids(longPoints.from(3L)).size == 300)
+ assert(ids(longPoints.fromExclusive(3L)).size == 200)
+ }
+
+ test("PointIndex.apply defends against a build-side iterator reusing its row
buffer") {
+ // Same hazard as the IntervalIndex case above: a point index built
straight off a
+ // reused row buffer must copy both the key and the row, not just
reference them.
+ val n = 50
+ val buildKeys = Seq(BoundReference(0, IntegerType, nullable = true))
+ val (shared, rows) = reusingRowIterator(fields = 2, n)
+ val index = PointIndex(IntegerType, buildKeys, rows, Some(n))
+
+ shared.update(0, -1)
+ shared.update(1, -1)
+
+ (0 until n).foreach { i =>
+ val payload = index.from(i).next().getInt(1)
+ assert(payload == i, s"point $i was corrupted by later reuse of the row
buffer")
+ }
+ }
+
+ test("PointIndex.apply defends against string keys sharing a reused byte
buffer") {
+ val n = 50
+ val buildKeys = Seq(BoundReference(0, StringType, nullable = true))
+ val (sharedBytes, rows) = reusingByteBufferStringIterator(n)
+ val index = PointIndex(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.from(key).take(1).toList
+ assert(matching.nonEmpty && matching.head.getInt(1) == i,
+ s"point $i was corrupted by shared buffer reuse")
+ }
+ }
+
+ test("point index orders NaN above +Infinity and -Infinity below every
finite value") {
+ // Total order: -Inf < -1.0 < 0.0 < 1.0 < +Inf < NaN, with NaN the unique
maximum.
+ val keyed = Array[(Any, InternalRow)](
+ (Double.NaN, row(90)),
+ (Double.PositiveInfinity, row(80)),
+ (1.0, row(1)),
+ (Double.NegativeInfinity, row(-80)),
+ (-1.0, row(-1)),
+ (0.0, row(0)))
+ val index = buildPoints(DoubleType, keyed.toSeq)
+ assert(ids(index.upTo(0.0)) == List(-80, -1, 0))
+ assert(ids(index.from(1.0)) == List(1, 80, 90))
+ // NaN is the unique maximum: only a NaN key is at or above it.
+ assert(ids(index.from(Double.NaN)) == List(90))
+ assert(ids(index.upTo(Double.NaN)) == List(-80, -1, 0, 1, 80, 90))
+ // +Infinity is below NaN but above every finite value.
+ assert(ids(index.upTo(Double.PositiveInfinity)) == List(-80, -1, 0, 1, 80))
+ assert(ids(index.from(Double.NegativeInfinity)) == List(-80, -1, 0, 1, 80,
90))
+ }
+
+ test("upTo/from matches a brute-force reference across randomized point
sets") {
+ // PointIndex's counterpart to "overlapping matches a brute-force reference
+ // across randomized interval sets": cross-checks the binary-search bound
+ // scan against an independent O(n) scan, across many random key sets and
+ // probe values, including duplicate keys.
+ val rand = new scala.util.Random(42)
+ val bound = 50
+ for (_ <- 0 until 30) {
+ val n = 1 + rand.nextInt(300)
+ val raw = (0 until n).map(i => (rand.nextInt(bound), i))
+ val index = buildPoints(IntegerType, raw.map { case (k, i) => (k,
row(i)) })
+
+ for (_ <- 0 until 20) {
+ val probe = rand.nextInt(bound)
+ val expectedUpTo = raw.collect { case (k, i) if k <= probe => i }.toSet
+ val expectedFrom = raw.collect { case (k, i) if k >= probe => i }.toSet
+ assert(ids(index.upTo(probe)).toSet == expectedUpTo,
+ s"n=$n probe=$probe expected=$expectedUpTo")
+ assert(ids(index.from(probe)).toSet == expectedFrom,
+ s"n=$n probe=$probe expected=$expectedFrom")
+ }
+ }
+ }
+
+ test("indexes round-trip through Java and strict Kryo serialization") {
+ val serializers = Seq(
+ "Java" -> new org.apache.spark.serializer.JavaSerializer(
+ new org.apache.spark.SparkConf()).newInstance(),
+ "Kryo" -> {
+ val conf = new org.apache.spark.SparkConf()
+
.set(org.apache.spark.internal.config.Kryo.KRYO_REGISTRATION_REQUIRED, true)
+ new org.apache.spark.serializer.KryoSerializer(conf).newInstance()
+ })
+
+ // (key type, lo, hi, probe) covering each supported key family: fixed
+ // width, floating point, decimal, binary, and collated strings. The
+ // collated probe mixes cases, so a restored index only answers it when
+ // its collation ordering survived the round trip. The key type itself
+ // never reaches a serializer: writes cross it as JSON, so this exercises
+ // the index registrations, not any DataType registration.
+ val cases: Seq[(DataType, Any, Any, Any)] = Seq(
+ (ByteType, 0.toByte, 10.toByte, 5.toByte),
+ (ShortType, 0.toShort, 10.toShort, 5.toShort),
+ (IntegerType, 0, 10, 5),
+ (LongType, 0L, 10L, 5L),
+ (FloatType, 0.0f, 10.0f, 5.0f),
+ (DoubleType, 0.0, 10.0, 5.0),
+ (DecimalType(10, 2), Decimal(0.0), Decimal(10.0), Decimal(5.0)),
+ (StringType("UTF8_LCASE"),
+ UTF8String.fromString("abc"), UTF8String.fromString("zzz"),
UTF8String.fromString("MMM")),
+ (CharType(3),
+ UTF8String.fromString("abc"), UTF8String.fromString("zzz"),
UTF8String.fromString("mmm")),
+ (VarcharType(10),
+ UTF8String.fromString("abc"), UTF8String.fromString("zzz"),
UTF8String.fromString("mmm")),
+ (BinaryType, Array[Byte](0), Array[Byte](10), Array[Byte](5)),
+ (DateType, 100, 200, 150),
+ (TimestampType, 1000L, 2000L, 1500L),
+ (TimestampNTZType, 1000L, 2000L, 1500L),
+ (BooleanType, false, true, true))
+
+ def roundTrip[A <: AnyRef: scala.reflect.ClassTag](
+ ser: org.apache.spark.serializer.SerializerInstance, index: A): A =
+ ser.deserialize[A](ser.serialize(index))
+
+ // One reader for the sealed family, mirroring
BroadcastRangeJoinExec.makeCandidateFn dispatch:
+ // the interval kinds answer a window, the point kinds answer a
+ // lower-bound scan, and every relation this test builds answers at
+ // row ordinal 0.
+ def answer(rel: RangeRelation, low: Any, high: Any): List[Int] = rel match
{
+ case r: LongIntervalIndex => ids(r.overlapping(r.toLong(low),
r.toLong(high)))
+ case r: IntervalIndex => ids(r.overlapping(low, high))
+ case r: LongPointIndex => ids(r.from(r.toLong(low)))
+ case r: PointIndex => ids(r.from(low))
+ }
+
+ cases.foreach { case (dt, lo, hi, probe) =>
+ // One straight span and one inverted span that normalizes to the same
+ // shape, plus the family the broadcast build routes the type to. Rows
+ // carry the asserted id at ordinal 0 and the keys at 1 and 2, so one
+ // reader answers every family.
+ val interval = buildIntervals(dt, Seq((lo, hi, row(5)), (hi, lo,
row(6))))
+ val point = buildPoints(dt, Seq((probe, row(7))))
+ val routedInterval = RangeRelationBroadcastMode(
+ Seq(BoundReference(1, dt, nullable = true), BoundReference(2, dt,
nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(5, lo, hi), InternalRow(6, hi, lo)))
+ val routedPoint = RangeRelationBroadcastMode(
+ Seq(BoundReference(1, dt, nullable = true)), PointIndexKind)
+ .transform(Array(InternalRow(7, probe)))
+
+ val intervalWindow = answer(interval, probe, probe)
+ val pointScan = answer(point, probe, probe)
+ assert(intervalWindow.sorted == List(5, 6), s"$dt window")
+ assert(pointScan.sorted == List(7), s"$dt bound")
+ // The routed family answers the same query on the same rows.
+ assert(answer(routedInterval, probe, probe).sorted ==
intervalWindow.sorted, s"$dt routed")
+ assert(answer(routedPoint, probe, probe).sorted == pointScan.sorted,
s"$dt routed point")
+
+ serializers.foreach { case (name, ser) =>
+ Seq(interval, routedInterval).foreach { rel =>
+ assert(answer(roundTrip(ser, rel), probe, probe).sorted ==
intervalWindow.sorted,
+ s"$dt interval via $name")
+ }
+ Seq(point, routedPoint).foreach { rel =>
+ assert(answer(roundTrip(ser, rel), probe, probe).sorted ==
pointScan.sorted,
+ s"$dt point via $name")
+ }
+ }
+ }
+ }
+
+ test("restored interval indexes answer windows that need the walk") {
+ // The round-trip fixtures above place every probe inside a common core, so
+ // the restored index answers from the row slice without reading the tree
+ // state. These intervals are disjoint: each probe must walk maxima, lows,
+ // and highs on the restored index, under every serializer.
+ val serializers = Seq(
+ "Java" -> new org.apache.spark.serializer.JavaSerializer(
+ new org.apache.spark.SparkConf()).newInstance(),
+ "Kryo" -> {
+ val conf = new org.apache.spark.SparkConf()
+
.set(org.apache.spark.internal.config.Kryo.KRYO_REGISTRATION_REQUIRED, true)
+ new org.apache.spark.serializer.KryoSerializer(conf).newInstance()
+ },
+ // Reference tracking off duplicates any aliased object on the wire, so
it
+ // also probes a restored graph that re-materialized every array entry.
+ "Kryo without references" -> {
+ val conf = new org.apache.spark.SparkConf()
+
.set(org.apache.spark.internal.config.Kryo.KRYO_REGISTRATION_REQUIRED, true)
+ .set(org.apache.spark.internal.config.Kryo.KRYO_REFERENCE_TRACKING,
false)
+ new org.apache.spark.serializer.KryoSerializer(conf).newInstance()
+ })
+
+ def roundTrip[A <: AnyRef: scala.reflect.ClassTag](
+ ser: org.apache.spark.serializer.SerializerInstance, index: A): A =
+ ser.deserialize[A](ser.serialize(index))
+
+ val windows = Seq((5.0, 5.0, List(0)), (25.0, 25.0, List(1)), (45.0, 45.0,
List(2)),
+ (5.0, 25.0, List(0, 1)), (25.0, 45.0, List(1, 2)))
+ val spans = Seq(span(0.0, 10.0, 0), span(20.0, 30.0, 1), span(40.0, 50.0,
2))
+ val boxed = buildIntervals(DoubleType, spans)
+ val longIdx = RangeRelationBroadcastMode(
+ Seq(BoundReference(1, IntegerType, nullable = true),
+ BoundReference(2, IntegerType, nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(0, 0, 10), InternalRow(1, 20, 30),
InternalRow(2, 40, 50)))
+ .asInstanceOf[LongIntervalIndex]
+
+ windows.foreach { case (low, high, expected) =>
+ assert(probeIds(boxed, low, high).sorted == expected, s"fresh boxed
$low-$high")
+ assert(longIdx.overlapping(low.toLong,
high.toLong).map(_.getInt(0)).toList.sorted ==
+ expected, s"fresh long $low-$high")
+ }
+
+ serializers.foreach { case (name, ser) =>
+ val restoredBoxed = roundTrip(ser, boxed)
+ val restoredLong = roundTrip(ser, longIdx)
+ windows.foreach { case (low, high, expected) =>
+ // A disjoint build has no all-covering shortcut for these windows, so
+ // the restored index must answer through a fresh walk iterator.
+ assert(restoredBoxed.overlapping(low,
high).isInstanceOf[OverlapIterator],
+ s"boxed $low-$high did not walk after $name")
+ assert(restoredLong.overlapping(low.toLong, high.toLong)
+ .isInstanceOf[LongOverlapIterator], s"long $low-$high did not walk
after $name")
+ assert(probeIds(restoredBoxed, low, high).sorted == expected,
+ s"boxed $low-$high via $name")
+ assert(restoredLong.overlapping(low.toLong, high.toLong)
+ .map(_.getInt(0)).toList.sorted == expected, s"long $low-$high via
$name")
+ }
+ }
+ }
+
+ test("restored point indexes answer multi-entry bound scans") {
+ // The round-trip fixtures above hold one point equal to the probe, so any
+ // restore that keeps that single entry answers it whatever ordering it
+ // re-derives. These indexes hold several entries: duplicated keys, keys on
+ // both sides of each probe, and - for the collated boxed family - keys
+ // that compare equal without being equal. Every bound scan the join uses
+ // answers through the restored ordering, so a restore that reverses,
+ // truncates, or desynchronizes keys and rows hands back the wrong ids.
+ // The long family has no distinct-but-equal keys; it gets the duplicates.
+ val serializers = Seq(
+ "Java" -> new org.apache.spark.serializer.JavaSerializer(
+ new org.apache.spark.SparkConf()).newInstance(),
+ "Kryo" -> {
+ val conf = new org.apache.spark.SparkConf()
+
.set(org.apache.spark.internal.config.Kryo.KRYO_REGISTRATION_REQUIRED, true)
+ new org.apache.spark.serializer.KryoSerializer(conf).newInstance()
+ },
+ // Reference tracking off duplicates any aliased object on the wire, so
it
+ // also probes a restored graph that re-materialized every array entry.
+ "Kryo without references" -> {
+ val conf = new org.apache.spark.SparkConf()
+
.set(org.apache.spark.internal.config.Kryo.KRYO_REGISTRATION_REQUIRED, true)
+ .set(org.apache.spark.internal.config.Kryo.KRYO_REFERENCE_TRACKING,
false)
+ new org.apache.spark.serializer.KryoSerializer(conf).newInstance()
+ })
+
+ def roundTrip[A <: AnyRef: scala.reflect.ClassTag](
+ ser: org.apache.spark.serializer.SerializerInstance, index: A): A =
+ ser.deserialize[A](ser.serialize(index))
+
+ // UNICODE_CI sorts "M" with "m", so rows 3, 4, and 5 stay in input order
as
+ // one collation class, and an exclusive scan skips all of them. Probes use
+ // both spellings and two off-key bounds, so the scans must consult the
+ // restored ordering rather than remembered positions.
+ val boxed = buildPoints(StringType("UNICODE_CI"), Seq(
+ (UTF8String.fromString("a"), row(1)),
+ (UTF8String.fromString("m"), row(3)),
+ (UTF8String.fromString("M"), row(4)),
+ (UTF8String.fromString("m"), row(5)),
+ (UTF8String.fromString("z"), row(9))))
+ // (probe, upTo, upToExclusive, from, fromExclusive)
+ val boxedBounds: Seq[(UTF8String, List[Int], List[Int], List[Int],
List[Int])] = Seq(
+ (UTF8String.fromString("m"), List(1, 3, 4, 5), List(1), List(3, 4, 5,
9), List(9)),
+ (UTF8String.fromString("M"), List(1, 3, 4, 5), List(1), List(3, 4, 5,
9), List(9)),
+ (UTF8String.fromString("n"), List(1, 3, 4, 5), List(1, 3, 4, 5),
List(9), List(9)),
+ (UTF8String.fromString("l"), List(1), List(1), List(3, 4, 5, 9), List(3,
4, 5, 9)))
+
+ val longPoints = RangeRelationBroadcastMode(
+ Seq(BoundReference(1, IntegerType, nullable = true)), PointIndexKind)
+ .transform(Array(
+ InternalRow(1, 1), InternalRow(3, 5), InternalRow(4, 5),
+ InternalRow(5, 5), InternalRow(9, 9)))
+ .asInstanceOf[LongPointIndex]
+ val longBounds: Seq[(Long, List[Int], List[Int], List[Int], List[Int])] =
Seq(
+ (5L, List(1, 3, 4, 5), List(1), List(3, 4, 5, 9), List(9)),
+ (4L, List(1), List(1), List(3, 4, 5, 9), List(3, 4, 5, 9)),
+ (6L, List(1, 3, 4, 5), List(1, 3, 4, 5), List(9), List(9)))
+
+ boxedBounds.foreach { case (probe, upTo, upToEx, from, fromEx) =>
+ assert(ids(boxed.upTo(probe)) == upTo, s"fresh boxed upTo($probe)")
+ assert(ids(boxed.upToExclusive(probe)) == upToEx, s"fresh boxed
upToExclusive($probe)")
+ assert(ids(boxed.from(probe)) == from, s"fresh boxed from($probe)")
+ assert(ids(boxed.fromExclusive(probe)) == fromEx, s"fresh boxed
fromExclusive($probe)")
+ }
+ longBounds.foreach { case (probe, upTo, upToEx, from, fromEx) =>
+ assert(ids(longPoints.upTo(probe)) == upTo, s"fresh long upTo($probe)")
+ assert(ids(longPoints.upToExclusive(probe)) == upToEx, s"fresh long
upToExclusive($probe)")
+ assert(ids(longPoints.from(probe)) == from, s"fresh long from($probe)")
+ assert(ids(longPoints.fromExclusive(probe)) == fromEx, s"fresh long
fromExclusive($probe)")
+ }
+
+ serializers.foreach { case (name, ser) =>
+ val restoredBoxed = roundTrip(ser, boxed)
+ val restoredLong = roundTrip(ser, longPoints)
+ boxedBounds.foreach { case (probe, upTo, upToEx, from, fromEx) =>
+ assert(ids(restoredBoxed.upTo(probe)) == upTo, s"boxed upTo($probe)
via $name")
+ assert(ids(restoredBoxed.upToExclusive(probe)) == upToEx,
+ s"boxed upToExclusive($probe) via $name")
+ assert(ids(restoredBoxed.from(probe)) == from, s"boxed from($probe)
via $name")
+ assert(ids(restoredBoxed.fromExclusive(probe)) == fromEx,
+ s"boxed fromExclusive($probe) via $name")
+ }
+ longBounds.foreach { case (probe, upTo, upToEx, from, fromEx) =>
+ assert(ids(restoredLong.upTo(probe)) == upTo, s"long upTo($probe) via
$name")
+ assert(ids(restoredLong.upToExclusive(probe)) == upToEx,
+ s"long upToExclusive($probe) via $name")
+ assert(ids(restoredLong.from(probe)) == from, s"long from($probe) via
$name")
+ assert(ids(restoredLong.fromExclusive(probe)) == fromEx,
+ s"long fromExclusive($probe) via $name")
+ }
+ }
+ }
+
+ test("overlapping uses the common-core short cut without changing the
answer") {
+ // [0, 10], [5, 15], and [8, 20] share the core [8, 10] (1-D Helly). Any
+ // window that reaches it overlaps every interval, so the index returns
+ // all rows directly; windows outside keep the walk's selective answer.
+ val index = buildIntervals(Seq(span(0, 10, 0), span(5, 15, 1), span(8, 20,
2)))
+ assert(probeIds(index, 9, 9).sorted == List(0, 1, 2))
+ assert(probeIds(index, 8, 10).sorted == List(0, 1, 2))
+ assert(probeIds(index, 6, 12).sorted == List(0, 1, 2))
+ // Endpoints are normalized first, so a reversed window reaches the core
too.
+ assert(probeIds(index, 12, 6).sorted == List(0, 1, 2))
+ // The short cut skips the walk's iterator for a core window, and only
there.
+ assert(!index.overlapping(9, 9).isInstanceOf[OverlapIterator])
+ assert(probeIds(index, 0, 2) == List(0))
+ assert(probeIds(index, 6, 7).sorted == List(0, 1))
+ assert(probeIds(index, 11, 12).sorted == List(1, 2))
+ assert(probeIds(index, 16, 30) == List(2))
+ }
+
+ test("overlapping short circuits all-covering windows over builds without a
common core") {
+ // Disjoint intervals and distinct points share no core, but a window
+ // spanning max-low and min-high still covers every interval, so the
+ // direct all-rows scan answers it and no walk iterator is built. A
+ // window covering some but not all intervals keeps the walk.
+ val disjoint = buildIntervals(Seq(span(0, 10, 0), span(20, 30, 1)))
+ assert(probeIds(disjoint, -5, 35).sorted == List(0, 1))
+ assert(!disjoint.overlapping(-5, 35).isInstanceOf[OverlapIterator])
+ assert(probeIds(disjoint, 5, 25).sorted == List(0, 1))
+ assert(!disjoint.overlapping(5, 25).isInstanceOf[OverlapIterator])
+ assert(probeIds(disjoint, 5, 15) == List(0))
+ assert(disjoint.overlapping(5, 15).isInstanceOf[OverlapIterator])
+
+ val points = buildIntervals(Seq(span(1, 1, 1), span(5, 5, 5)))
+ assert(probeIds(points, 0, 6).sorted == List(1, 5))
+ assert(!points.overlapping(0, 6).isInstanceOf[OverlapIterator])
+ assert(probeIds(points, 4, 5) == List(5))
+ assert(points.overlapping(4, 5).isInstanceOf[OverlapIterator])
+
+ val longIdx = RangeRelationBroadcastMode(
+ Seq(BoundReference(0, IntegerType, nullable = true),
+ BoundReference(1, IntegerType, nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(0, 10), InternalRow(20, 30)))
+ .asInstanceOf[LongIntervalIndex]
+ assert(longIdx.overlapping(-5L, 35L).map(_.getInt(0)).toSet == Set(0, 20))
+ assert(!longIdx.overlapping(-5L, 35L).isInstanceOf[LongOverlapIterator])
+ assert(longIdx.overlapping(5L, 15L).map(_.getInt(0)).toList == List(0))
+ assert(longIdx.overlapping(5L, 15L).isInstanceOf[LongOverlapIterator])
+ }
+
+ test("overlap iterator sizes its scratch stack to tree depth, not row
count") {
+ // The scratch stack is pre-sized off `lows.length` alone (tree depth),
independent of
+ // data distribution. Bounding against that same depth formula (rather
than a fixed
+ // magic number) still catches a regression to O(row count) sizing,
without breaking
+ // if the depth-to-capacity multiplier or the floor constant is tuned to a
different
+ // (but still O(log n)) value.
+ def depthOf(n: Int): Int = 32 - java.lang.Integer.numberOfLeadingZeros(n)
+
+ val oneRowIndex = buildIntervals(Seq(span(0, 10, 0)))
+ val oneRowIter = oneRowIndex.overlapping(15,
15).asInstanceOf[OverlapIterator]
+ assert(oneRowIter.stackCapacity <= math.max(8, depthOf(1) * 4),
+ s"a one-row index should need only a handful of scratch slots, " +
+ s"got ${oneRowIter.stackCapacity}")
+
+ val n = 10000
+ val largeIndex = buildIntervals((0 until n).map(i => span(i, i + 100, i)))
+ val largeIter = largeIndex.overlapping(50,
50).asInstanceOf[OverlapIterator]
+ // log2(10000) ~= 13.3: a handful of tree levels, nowhere near n.
+ assert(largeIter.stackCapacity <= depthOf(n) * 4,
+ s"a $n-row index should size its scratch stack to tree depth
(~${depthOf(n)}), " +
+ s"not row count; got ${largeIter.stackCapacity}")
+ }
+
+ test("overlapping traversal is lazy and supports early exit under high
overlap") {
+ // The pending-work stack tracks tree depth, not match count: draining a
handful
+ // of candidates out of ~100 matches among n=10,000 intervals must not
grow it
+ // toward n. Staggered spans also avoid the common-core short cut, which
would
+ // skip the walk entirely. Reading `pendingSpans` confirms work remains on
the
+ // stack after taking initial matches, proving the walk did not drain
eagerly.
+ val n = 10000
+ val overlappingSpans = (0 until n).map(i => span(i, i + 100, i))
+ val index = buildIntervals(overlappingSpans)
+ val iter = index.overlapping(5000, 5000).asInstanceOf[OverlapIterator]
+ val initialCapacity = iter.stackCapacity
+
+ assert(iter.take(3).size == 3)
+ assert(iter.pendingSpans > 0,
+ s"lazy traversal must leave work pending on the stack, but had
${iter.pendingSpans} spans")
+ assert(iter.stackCapacity == initialCapacity,
+ s"stack grew from $initialCapacity to ${iter.stackCapacity} after taking
3 of $n matches")
+ }
+
+ test("overlapping matches a brute-force reference across randomized interval
sets") {
+ // Cross-checks the tree-pruning search in overlapping() against an
independent
+ // O(n) scan, across many random interval sets and probe windows.
Hand-written
+ // cases above cover specific shapes; this covers combinations they would
miss.
+ val rand = new scala.util.Random(42)
+ val bound = 50
+ for (_ <- 0 until 30) {
+ val n = 1 + rand.nextInt(300)
+ // Endpoints are not pre-sorted, exercising apply()'s lo/hi-swap
normalization.
+ val raw = (0 until n).map(i => (rand.nextInt(bound),
rand.nextInt(bound), i))
+ val index = buildIntervals(raw.map { case (a, b, i) => span(a, b, i) })
+ val normalized = raw.map { case (a, b, i) => (math.min(a, b),
math.max(a, b), i) }
+
+ for (_ <- 0 until 20) {
+ val a = rand.nextInt(bound)
+ val b = rand.nextInt(bound)
+ val (low, high) = (math.min(a, b), math.max(a, b))
+ val expected = normalized.collect {
+ case (lo, hi, i) if lo <= high && low <= hi => i
+ }.toSet
+ assert(probeIds(index, low, high).toSet == expected,
+ s"n=$n low=$low high=$high expected=$expected")
+ }
+ }
+ }
+
+ test("broadcast modes route keys that map to one long to the long indexes") {
+ // Both build keys are long-convertible: the long interval index.
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, IntegerType, nullable = true),
+ BoundReference(1, IntegerType, nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(0, 10, 1))).isInstanceOf[LongIntervalIndex])
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, IntegerType, nullable = true)), PointIndexKind)
+ .transform(Array(InternalRow(5, 1))).isInstanceOf[LongPointIndex])
+ // Floating point has no bit-preserving SQL order, and strings are not one
long.
+ // Pinned for both kinds and both widths: signed zero and NaN need the
boxed
+ // ordering's special-cased comparator, not a bit-preserving long encoding.
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, FloatType, nullable = true)), PointIndexKind)
+ .transform(Array(InternalRow(1.0f, 1))).isInstanceOf[PointIndex])
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, DoubleType, nullable = true)), PointIndexKind)
+ .transform(Array(InternalRow(1.0, 1))).isInstanceOf[PointIndex])
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, FloatType, nullable = true),
+ BoundReference(1, FloatType, nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(1.0f, 2.0f,
1))).isInstanceOf[IntervalIndex])
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, DoubleType, nullable = true),
+ BoundReference(1, DoubleType, nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(1.0, 2.0, 1))).isInstanceOf[IntervalIndex])
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, StringType, nullable = true),
+ BoundReference(1, StringType, nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(
+ UTF8String.fromString("a"), UTF8String.fromString("b"),
1))).isInstanceOf[IntervalIndex])
+ // A compact decimal (precision <= 18) is representable as a scaled long.
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, DecimalType(10, 2), nullable = true)),
PointIndexKind)
+ .transform(Array(InternalRow(Decimal(1.0),
1))).isInstanceOf[LongPointIndex])
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, DecimalType(18, 2), nullable = true)),
PointIndexKind)
+ .transform(Array(InternalRow(Decimal(1.0),
1))).isInstanceOf[LongPointIndex])
+ // Precision above 18 cannot be read back as an unscaled long: stays boxed.
+ assert(RangeRelationBroadcastMode(
+ Seq(BoundReference(0, DecimalType(19, 2), nullable = true)),
PointIndexKind)
+ .transform(Array(InternalRow(Decimal(1.0), 1))).isInstanceOf[PointIndex])
+ }
+
+ test("the long interval index answers the same windows as the generic one") {
+ def longIndex(spans: Seq[(Any, Any, InternalRow)]): LongIntervalIndex =
+ RangeRelationBroadcastMode(
+ Seq(BoundReference(0, IntegerType, nullable = true),
+ BoundReference(1, IntegerType, nullable = true)),
+ IntervalIndexKind)
+ .transform(spans.map(t => InternalRow(t._1, t._2,
t._3.getInt(0))).toArray)
+ .asInstanceOf[LongIntervalIndex]
+ def assertSameWindow(
+ generic: IntervalIndex,
+ longIdx: LongIntervalIndex,
+ lo: Int,
+ hi: Int): Unit = {
+ val expected = probeIds(generic, lo, hi).sorted
+ val actual = longIdx.overlapping(lo.toLong,
hi.toLong).map(_.getInt(2)).toList.sorted
+ assert(actual == expected, s"window ($lo, $hi): expected $expected, got
$actual")
+ }
+
+ // Includes an inverted span and a null bound, and shares the core [8, 10].
+ val spans = Seq(span(0, 10, 0), span(5, 15, 1), (null, 4, row(9)),
+ span(8, 20, 2), span(15, 5, 3))
+ val generic = buildIntervals(spans)
+ val longIdx = longIndex(spans)
+ assert(longIdx.estimatedSize > 0)
Review Comment:
**Nit (P3):** `BroadcastExchangeExec` charges `RangeRelation.estimatedSize`
against `maxBroadcastTableSize`, and for int, long, date, timestamp, and
compact-decimal keys that is a `LongIntervalIndex` or `LongPointIndex`. Their
estimates are only asserted to be positive, here and at line 1112. Nothing
reads `LongPointIndex.estimatedSize`, and the SQL size test uses a 1-byte
limit. If either estimate stopped charging `rows`, the key arrays alone would
still pass every check, and long-key broadcasts holding all their rows would be
admitted under the size limit. Could you add assertions that both long
estimates grow with the row payload, for example the same keys with wider rows
estimating larger, and that they exceed what the key arrays alone take?
**Verification:**
- **Regression:** For both long index families, an index over the same keys
with a wider row payload reports a larger estimate, and every estimate exceeds
what the key arrays alone occupy, so an estimate that stops charging rows fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
