HyukjinKwon commented on code in PR #59027:
URL: https://github.com/apache/spark/pull/59027#discussion_r4120683855
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -179,6 +180,12 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
* Only supported for equi-joins and the join keys have to be sortable.
* Supported for all join types.
*
+ * - Broadcast range join (BRJ):
+ * Supports a point-in-range predicate or one cross-side inequality.
Review Comment:
**Nit (P3):** Nit: this lists only point-in-range and one cross-side
inequality, but `ExtractRangeJoinKeys` also recognizes interval overlap (`a.lo
< b.hi AND b.lo < a.hi`), and that is planned as `BroadcastRangeJoinExec` too.
Could this mention all three shapes? The empty line a few lines below is also
missing its leading ` *` inside the scaladoc block.
**Verification:**
- **Inspection:** The BRJ entry lists interval overlap, and no line in that
scaladoc block is missing its leading asterisk.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala:
##########
@@ -421,6 +422,270 @@ object ExtractSingleColumnNullAwareAntiJoin extends
JoinSelectionHelper with Pre
}
}
+/**
+ * Finds joins whose condition is a range predicate usable by a broadcast
range join.
+ *
+ * Recognizes point-in-range (`a.ip` against `[b.lo, b.hi]`), interval overlap
+ * (`a.lo < b.hi AND b.lo < a.hi`), and a cross-side inequality (`a.start <
b.end`).
+ * Point-in-range keys are `(point, point)` and `(low, high)`. Overlap keys are
+ * `(low, high)` on each side. A partial range carries the one column from
each side.
+ * Extra conjuncts stay on the original condition and are evaluated per
candidate.
+ * A point-in-range pair wins over overlap, and overlap wins over a partial
range.
+ * [[RangePredicate]] supplies the canonical `low <= high` form; inclusivity
stays
+ * on the original condition. Keys whose type the range index cannot compare,
and
+ * keys that are not deterministic, stay unrecognized, so the join stays a
nested
+ * loop. The index stores one evaluation of each key and the original condition
+ * evaluates it again; a second `rand()` would drop rows the condition accepts.
+ * A non-deterministic conjunct that is not a key stays on that condition.
+ * A deterministic [[With]] (`BETWEEN` lowers to one) is inlined first, so the
+ * keys are the common expressions. A non-deterministic definition is left in
+ * place and is not used as a range key.
+ */
+object ExtractRangeJoinKeys extends PredicateHelper {
+ type ReturnType = (LogicalPlan, LogicalPlan, Seq[Expression],
Seq[Expression],
+ JoinType, RangeJoin)
+
+ def unapply(plan: LogicalPlan): Option[ReturnType] = plan match {
+ case Join(left, right, joinType, Some(condition), _) =>
+ val predicates =
splitConjunctivePredicates(inlineCommonExpressions(condition))
+ pointInRange(predicates, left, right, joinType)
+ .orElse(intervalOverlap(predicates, left, right, joinType))
+ .orElse(partialRange(predicates, left, right, joinType))
+ case _ => None
+ }
+
+ /**
+ * Replaces each deterministic [[With]] with its child, substituting every
+ * [[CommonExpressionRef]] by the common expression. A ref has no attributes,
+ * so `canEvaluate` would accept it on either side. A cycle or a ref with no
+ * definition in this [[With]] is left as a ref, and [[RangePredicate]] then
+ * rejects the predicate.
+ */
+ private def inlineCommonExpressions(expression: Expression): Expression = {
+ expression.transformUp {
+ case w: With if w.defs.forall(_.deterministic) =>
+ val defsById = w.defs.iterator.map(d => d.id -> d.child).toMap
+ def substitute(e: Expression, stack: Set[Long]): Expression = {
+ e.transform {
+ case r: CommonExpressionRef =>
+ defsById.get(r.id) match {
+ case Some(definition) if !stack.contains(r.id.id) =>
+ substitute(definition, stack + r.id.id)
+ case _ => r
+ }
+ }
+ }
+ substitute(w.child, Set.empty)
+ }
+ }
+
+ private def pointInRange(
+ predicates: Seq[Expression],
+ left: LogicalPlan,
+ right: LogicalPlan,
+ joinType: JoinType): Option[ReturnType] = {
+ predicates.combinations(2).flatMap {
+ case Seq(RangePredicate(d1, l1, h1), RangePredicate(d2, l2, h2))
+ if d1 == d2 && RangePredicate.supportedType(d1) =>
+ pointInRangeKeys(l1, h1, l2, h2, left, right).iterator.collect {
+ case (leftKeys, rightKeys) if supportedKeys(leftKeys, rightKeys) =>
+ (left, right, leftKeys, rightKeys, joinType, PointInRangeJoin)
+ }
+ case _ => Iterator.empty
+ }.nextOption()
+ }
+
+ /**
+ * `(l1 <= h1) AND (l2 <= h2)` is a point-in-range when the two predicates
share
+ * one expression and the other two are the bounds, on the opposite side.
+ * Returned keys are `(point, point)` and `(low, high)`.
+ */
+ private def pointInRangeKeys(
+ l1: Expression,
+ h1: Expression,
+ l2: Expression,
+ h2: Expression,
+ left: LogicalPlan,
+ right: LogicalPlan): Option[(Seq[Expression], Seq[Expression])] = {
+ if (h1.semanticEquals(l2) && !l1.semanticEquals(h2) && l1.dataType ==
h2.dataType) {
+ // l1 <= point <= h2
+ assignPoint(Seq(l2, h1), l1, h2, left, right)
+ } else if (l1.semanticEquals(h2) && !h1.semanticEquals(l2) &&
+ l2.dataType == h1.dataType) {
+ // l2 <= point <= h1
+ assignPoint(Seq(l1, h2), l2, h1, left, right)
+ } else {
+ None
+ }
+ }
+
+ private def assignPoint(
+ pointKeys: Seq[Expression],
+ low: Expression,
+ high: Expression,
+ left: LogicalPlan,
+ right: LogicalPlan): Option[(Seq[Expression], Seq[Expression])] = {
+ def keys(pointOnLeft: Boolean): Option[(Seq[Expression], Seq[Expression])]
= {
+ val pointSide = if (pointOnLeft) left else right
+ val boundSide = if (pointOnLeft) right else left
+ if (pointKeys.forall(canEvaluate(_, pointSide)) &&
+ canEvaluate(low, boundSide) && canEvaluate(high, boundSide)) {
+ if (pointOnLeft) Some((pointKeys, Seq(low, high)))
+ else Some((Seq(low, high), pointKeys))
+ } else {
+ None
+ }
+ }
+ keys(pointOnLeft = true).orElse(keys(pointOnLeft = false))
+ }
+
+ /**
+ * `(l1 <= h1) AND (l2 <= h2)` is an interval overlap when each side holds
one
+ * low and one high, and the two ends on a side are different expressions.
+ * Keys are `(low, high)` on each side and are probed as a window. A
degenerate
+ * side (`low` equals `high`) is a point-in-range and is left to
[[pointInRange]].
+ */
+ private def intervalOverlap(
+ predicates: Seq[Expression],
+ left: LogicalPlan,
+ right: LogicalPlan,
+ joinType: JoinType): Option[ReturnType] = {
+ predicates.combinations(2).flatMap {
+ case Seq(RangePredicate(d1, l1, h1), RangePredicate(d2, l2, h2))
+ if d1 == d2 && RangePredicate.supportedType(d1) =>
+ overlapKeys(l1, h1, l2, h2, left, right, joinType).iterator
+ case _ => Iterator.empty
+ }.nextOption()
+ }
+
+ private def overlapKeys(
+ l1: Expression,
+ h1: Expression,
+ l2: Expression,
+ h2: Expression,
+ left: LogicalPlan,
+ right: LogicalPlan,
+ joinType: JoinType): Option[ReturnType] = {
+ def assign(
+ leftLow: Expression,
+ leftHigh: Expression,
+ rightLow: Expression,
+ rightHigh: Expression): Option[ReturnType] = {
+ val leftKeys = Seq(leftLow, leftHigh)
+ val rightKeys = Seq(rightLow, rightHigh)
+ if (leftLow.semanticEquals(leftHigh) ||
rightLow.semanticEquals(rightHigh)) {
+ None
+ } else if (leftLow.dataType != leftHigh.dataType ||
+ rightLow.dataType != rightHigh.dataType ||
+ !supportedKeys(leftKeys, rightKeys)) {
+ None
+ } else if (canEvaluate(leftLow, left) && canEvaluate(leftHigh, left) &&
+ canEvaluate(rightLow, right) && canEvaluate(rightHigh, right)) {
+ Some((left, right, leftKeys, rightKeys, joinType, IntervalOverlapJoin))
+ } else {
+ None
+ }
+ }
+ // l1 <= h1 and l2 <= h2, with the low of each interval on the opposite
side
+ // from its high.
+ assign(l1, h2, l2, h1).orElse(assign(l2, h1, l1, h2))
+ }
+
+ private def supportedKeys(keys: Seq[Expression]*): Boolean = {
+ keys.flatten.forall { key =>
+ key.references.nonEmpty && key.deterministic &&
RangePredicate.supportedType(key.dataType)
+ }
+ }
+
+ private def partialRange(
+ predicates: Seq[Expression],
+ left: LogicalPlan,
+ right: LogicalPlan,
+ joinType: JoinType): Option[ReturnType] = {
+ predicates.iterator.flatMap {
+ case RangePredicate(_, low, high)
+ if !low.semanticEquals(high) && low.dataType == high.dataType &&
+ supportedKeys(Seq(low, high)) =>
+ if (canEvaluate(low, left) && canEvaluate(high, right)) {
+ Iterator((left, right, low :: Nil, high :: Nil, joinType,
LessPartialRangeJoin))
+ } else if (canEvaluate(low, right) && canEvaluate(high, left)) {
+ Iterator((left, right, high :: Nil, low :: Nil, joinType,
GreaterPartialRangeJoin))
+ } else {
+ Iterator.empty
+ }
+ case _ => Iterator.empty
+ }.nextOption()
+ }
+}
+
+/**
+ * Normalizes `<`, `<=`, `>`, `>=` to `(low.dataType, low, high)`.
+ * Inclusivity stays on the original predicate, which execution evaluates
as-is.
+ */
+object RangePredicate {
+ /**
+ * Types the range index orders with the same comparison as the join
predicate.
+ * A UDT is ordered as its sql type, which is how the row stores the value.
Review Comment:
**Nit (P3):** This says a UDT is ordered as its SQL type, and the PR
description lists UDTs as recognized, but `supportedType` below has no
`UserDefinedType` case, so a UDT key falls through to `case _ => false` and the
join stays a nested loop join. If UDTs should be supported, I think this needs
`case udt: UserDefinedType[_] => supportedType(udt.sqlType)`, plus unwrapping
the ordering in `RangeBroadcastMode.keyOrdering` (`PhysicalDataType(udt)`
resolves to `UninitializedPhysicalType`), and a test. Otherwise could we drop
the UDT claim here and in the description?
**Verification:**
- **Behavior:** Key extraction recognizes a UDT-over-long predicate, and an
end-to-end join over that UDT matches the conf-off result on both build sides.
- **Compatibility:** Types outside the supported list, including a UDT over
an array, remain unrecognized.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/RangeIndexSuite.scala:
##########
@@ -0,0 +1,334 @@
+/*
+ * 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, Projection}
+import org.apache.spark.sql.catalyst.types.PhysicalDataType
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * Unit tests for the two range-join indexes. [[IntervalIndex]] is the sweep
used by
+ * point-in-range. [[PointIndex]] is the sorted array used by a single
inequality.
+ * Operator-level behavior is covered by `RangeJoinSuite`.
+ */
+class RangeIndexSuite extends SparkFunSuite {
+
+ private val intOrdering: Ordering[Any] =
PhysicalDataType.ordering(IntegerType)
+ private val doubleOrdering: Ordering[Any] =
PhysicalDataType.ordering(DoubleType)
+
+ private def row(id: Int): InternalRow = InternalRow(id)
+
+ private def interval(lo: Int, hi: Int, id: Int): Seq[RangeIndex.RangeEvent]
= {
+ val stored = row(id)
+ RangeIndex.RangeEvent(lo, RangeIndex.RangeEvent.Start, stored, id) ::
+ RangeIndex.RangeEvent(hi, RangeIndex.RangeEvent.End, stored, id) :: Nil
+ }
+
+ private def point(v: Int, id: Int): Seq[RangeIndex.RangeEvent] =
+ RangeIndex.RangeEvent(v, RangeIndex.RangeEvent.Point, row(id), id) :: Nil
+
+ private def buildIntervals(events: Seq[RangeIndex.RangeEvent]):
IntervalIndex =
+ IntervalIndex.build(intOrdering, events.toArray)
+
+ 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))
+
+ 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("toRangeEvents: point, interval, reversed, and null") {
+ val getters = List(RangeIndex.getValue(IntegerType, 0),
RangeIndex.getValue(IntegerType, 1))
+ val identity = new Projection { override def apply(input: InternalRow):
InternalRow = input }
+ val eventifier = RangeIndex.toRangeEvents(getters, identity, intOrdering)
+
+ assert(eventifier(InternalRow(5, 5), 0) ==
+ Seq(RangeIndex.RangeEvent(5, RangeIndex.RangeEvent.Point, InternalRow(5,
5), 0)))
+ assert(eventifier(InternalRow(1, 3), 1) ==
+ Seq(
+ RangeIndex.RangeEvent(1, RangeIndex.RangeEvent.Start, InternalRow(1,
3), 1),
+ RangeIndex.RangeEvent(3, RangeIndex.RangeEvent.End, InternalRow(1, 3),
1)))
+ // Normalized, not dropped: the join condition still accepts a pair whose
sweep
+ // events would otherwise never exist.
+ assert(eventifier(InternalRow(3, 1), 2) ==
+ Seq(
+ RangeIndex.RangeEvent(1, RangeIndex.RangeEvent.Start, InternalRow(3,
1), 2),
+ RangeIndex.RangeEvent(3, RangeIndex.RangeEvent.End, InternalRow(3, 1),
2)))
+ assert(eventifier(InternalRow(null, 1), 3).isEmpty)
+ }
+
+ test("an inverted build interval reaches the windows the condition accepts")
{
+ // (3, 1) is normalized to [1, 3]. A window with low < 1 and high > 3
accepts it,
+ // and dropping the row would lose the pair the join condition then
rejects.
Review Comment:
**Nit (P3):** Nit: I think this should be "accepts". For build `(3, 1)` and
window `(0, 4)`, `a.lo < b.hi AND b.lo < a.hi` holds, which is why dropping the
row would lose a real result (matching the `toRangeEvents` doc).
**Verification:**
- **Inspection:** The test comment matches the toRangeEvents rationale: the
pair is accepted by the condition.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeIndex.scala:
##########
@@ -0,0 +1,350 @@
+/*
+ * 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 scala.collection.immutable.HashMap
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.types.DataType
+
+/**
+ * Broadcast payload of a range join. An index only names candidate rows.
+ * [[BroadcastRangeJoinExec]] keeps a candidate when the original join
condition
+ * is true, so inclusivity never lives in the index.
+ *
+ * - [[IntervalIndex]] answers "which ranges overlap this window?"
+ * - [[PointIndex]] answers "which points lie on this side of a bound?"
+ */
+private[execution] sealed trait RangeRelation extends Serializable {
+ def sizeInBytes(): Long
+}
+
+private[execution] object RangeIndex {
+
+ /** In-memory size of one indexed row. Non-unsafe rows are a test-only
estimate. */
+ def rowSize(row: InternalRow): Long = row match {
+ case u: UnsafeRow => u.getSizeInBytes.toLong
+ case _ => row.numFields * 8L
+ }
+
+ /**
+ * One sweep-line event used to build an [[IntervalIndex]].
+ *
+ * @param key the bound value this event occurs at
+ * @param flow [[RangeEvent.Start]], [[RangeEvent.Point]], or
[[RangeEvent.End]]
+ * @param row the build-side row that produced this event
+ * @param index a unique id for `row` so deactivation is O(1)
+ */
+ case class RangeEvent(key: Any, flow: Int, row: InternalRow, index: Int)
+
+ object RangeEvent {
+ /** Interval start. Sorts after Point at the same key. */
+ val Start = 1
+ /** Degenerate interval (low == high). */
+ val Point = 0
+ /** Interval end. Sorts before Point at the same key. */
+ val End = -1
+ }
+
+ /**
+ * A type-specialized accessor that reads a single field from an
`InternalRow` as a boxed
+ * value (or `null` if the field is null). Shared by the exchange
(build-side key extraction)
+ * and the join (stream-side key extraction) so the dispatch table lives in
one place.
+ */
+ def getValue(dt: DataType, ordinal: Int): InternalRow => Any = {
+ val accessor = InternalRow.getAccessor(dt)
+ (input: InternalRow) => accessor(input, ordinal)
+ }
+
+ /**
+ * First index in `keys` whose key compares greater than `value` under
`cmp`, or, when
+ * `orEqual` is set, greater than or equal to `value`. Shared by the
lowerBound/upperBound
+ * pair on both [[IntervalIndex]] and [[PointIndex]]; `cmp` is where they
differ, since
+ * `IntervalIndex` treats a leading null key as sorting before every value.
+ */
+ def searchBound(keys: Array[Any], value: Any, orEqual: Boolean, cmp: (Any,
Any) => Int): Int = {
+ var lo = 0
+ var hi = keys.length
+ while (lo < hi) {
+ val mid = lo + ((hi - lo) >>> 1)
+ val c = cmp(keys(mid), value)
+ if (if (orEqual) c <= 0 else c < 0) lo = mid + 1 else hi = mid
+ }
+ lo
+ }
+
+ /**
+ * Turn a projected `(low, high)` row into sweep events. A null bound emits
none.
+ *
+ * An inverted interval (`low > high`) is normalized to `(high, low)` rather
than dropped.
+ * The join condition still accepts a row the predicate accepts, so dropping
would lose it:
+ * with build `(3, 1)` and stream `(0, 5)`, `a.lo < b.hi AND b.lo < a.hi`
holds and the row
+ * has to be a candidate. Normalizing only widens the window. `start <= end`
also keeps the
+ * sweep's invariant, so `IntervalIndex` never sees a reversed pair.
+ */
+ def toRangeEvent(
+ buildSideKeyValueGetter: List[InternalRow => Any],
+ lowHighExtr: Projection,
+ cmp: Ordering[Any]): (InternalRow, Int) => Seq[RangeEvent] = {
+ (row: InternalRow, index: Int) => {
+ val lowHigh: InternalRow = lowHighExtr(row)
+ val low = buildSideKeyValueGetter(0)(lowHigh)
+ val high = buildSideKeyValueGetter(1)(lowHigh)
+ if (low != null && high != null) {
+ val result = cmp.compare(low, high)
+ if (result == 0) {
+ RangeEvent(low, RangeEvent.Point, row, index) :: Nil
+ } else {
+ val (start, end) = if (result < 0) (low, high) else (high, low)
+ RangeEvent(start, RangeEvent.Start, row, index) ::
+ RangeEvent(end, RangeEvent.End, row, index) :: Nil
+ }
+ } else {
+ Nil
+ }
+ }
+ }
+}
+
+/**
+ * Sweep-line index of build-side intervals. `overlapping` returns every row
whose
+ * interval may overlap `[low, high]`. The window is a superset: the low
search is
+ * exclusive and the high search is inclusive, and the caller drops false
positives.
+ */
+private[execution] object IntervalIndex {
+
+ /** Build an interval index from unsorted events. */
+ def build(ordering: Ordering[Any], events: Array[RangeIndex.RangeEvent]):
IntervalIndex = {
+ val eventComparator = new java.util.Comparator[RangeIndex.RangeEvent] {
+ override def compare(a: RangeIndex.RangeEvent, b:
RangeIndex.RangeEvent): Int = {
+ val keyCmp = ordering.compare(a.key, b.key)
+ if (keyCmp != 0) keyCmp else Integer.compare(a.flow, b.flow)
+ }
+ }
+ java.util.Arrays.sort(events, eventComparator)
+ buildFromSorted(ordering, events)
+ }
+
+ private def buildFromSorted(
+ ordering: Ordering[Any],
+ events: Array[RangeIndex.RangeEvent]): IntervalIndex = {
+ // A leading null key makes the binary search uniform, including an empty
index.
+ val keys = mutable.Buffer[Any](null)
+ val offsets = mutable.Buffer[Int](0)
+ val activatedRows = mutable.Buffer.empty[InternalRow]
+ val activeOld = mutable.Buffer.empty[HashMap[Int, InternalRow]]
+ val activeAll = mutable.Buffer.empty[HashMap[Int, InternalRow]]
+
+ // Keyed by the event id so a row leaves the active set in O(1). Two
snapshots
+ // are recorded per distinct key: `activeOld` (before this key's own Start
+ // events) and `activeAll` (after them); `overlapping` below explains why
both
+ // are needed. A build side of heavily-overlapping intervals (e.g. n
intervals
+ // all spanning the same window) has O(n) distinct keys each with an O(n)
+ // active set, so a plain map copied at every key would make the snapshots
+ // alone O(n^2). `HashMap` is persistent: `updated`/`removed` share
structure
+ // with the previous version instead of copying it, so recording a
snapshot is
+ // O(log n) and the total cost across all keys stays O(n log n). It is also
+ // `Serializable` for the broadcast, unlike the order-preserving persistent
+ // maps in the same package (`TreeSeqMap`, `VectorMap`) -- iteration order
+ // does not matter here since old/all are tracked as separate snapshots.
+ var currentKey: Any = null
+ var currentActiveRows = HashMap.empty[Int, InternalRow]
+ var currentOldActiveRows = currentActiveRows
+ var oldActiveRowsCaptured = false
+
+ def writeActiveRows(): Unit = {
+ activeOld += (if (oldActiveRowsCaptured) currentOldActiveRows else
currentActiveRows)
+ activeAll += currentActiveRows
+ }
+
+ events.foreach { event =>
+ // Group by the index ordering. Array equality is identity, and
+ // UTF8String equality is binary, so `!=` would split one sweep key.
+ if (currentKey == null || ordering.compare(currentKey, event.key) != 0) {
+ writeActiveRows()
+ currentKey = event.key
+ oldActiveRowsCaptured = false
+ keys += event.key
+ offsets += activatedRows.size
+ }
+ // The active set before any row starts at this key. Ends at this key
have
+ // already been removed by the time the first Point/Start event is seen.
+ if (event.flow >= RangeIndex.RangeEvent.Point && !oldActiveRowsCaptured)
{
+ currentOldActiveRows = currentActiveRows
+ oldActiveRowsCaptured = true
+ }
+ event.flow match {
+ case RangeIndex.RangeEvent.Start =>
+ activatedRows += event.row
+ currentActiveRows = currentActiveRows.updated(event.index, event.row)
+ case RangeIndex.RangeEvent.Point =>
+ activatedRows += event.row
+ case RangeIndex.RangeEvent.End =>
+ currentActiveRows = currentActiveRows.removed(event.index)
+ }
+ }
+ writeActiveRows()
+
+ new IntervalIndex(ordering, keys.toArray, offsets.toArray,
activeOld.toArray,
+ activeAll.toArray, activatedRows.toArray)
+ }
+}
+
+private[execution] class IntervalIndex(
+ private[this] val ordering: Ordering[Any],
+ private[this] val keys: Array[Any],
+ private[this] val offsets: Array[Int],
+ private[this] val activeOld: Array[HashMap[Int, InternalRow]],
+ private[this] val activeAll: Array[HashMap[Int, InternalRow]],
+ private[this] val activated: Array[InternalRow])
+ extends RangeRelation {
+
+ override def sizeInBytes(): Long = activated.map(RangeIndex.rowSize).sum
Review Comment:
Thanks, rebuilding the snapshots lazily does fix the serialized size, and
the round-trip test pins that down. The executor-side half still seems open,
though. Each executor now builds the per-key IntMap snapshots on first probe,
and they are still O(events x trie depth). I measured about 55 MB for 100k and
244 MB for 400k fully overlapping intervals, roughly 275-305 bytes per sweep
event, while `sizeInBytes` charges 8 bytes per event plus the rows. So
`spark.sql.maxBroadcastTableSize` still doesn't bound what the executor holds.
Could `sizeInBytes` include an estimate for the snapshots (for example, events
x log2(active) x node size), or could the probe avoid materializing a snapshot
per key (for example, an interval tree or a periodic checkpoint plus a delta
scan)?
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4118191678","thread_id":"inline:4118191678","verdict_sha256":"1244643e174c88ee6b828a54906352e7e83f63b77f996c18fa44ec11c38637f9"}
-->
--
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]
