cloud-fan commented on code in PR #59027:
URL: https://github.com/apache/spark/pull/59027#discussion_r4237663576
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1169,6 +1169,22 @@ object SQLConf {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("10MB")
+ val BROADCAST_RANGE_JOIN_ENABLED =
buildConf("spark.sql.join.broadcastRangeJoin.enabled")
+ .doc("When true, a join whose condition is a range predicate is planned as
a " +
Review Comment:
**Nit (P3):** Enablement alone does not guarantee this plan: JoinSelection
also needs a supported side selected by a BROADCAST hint or by
canBroadcastBySize. Without stating that prerequisite, the public description
promises a range join for queries that still take another physical plan.
**Verification:**
- **Inspection:** Trace the enabled-description conditions against every
successful and fallback branch of broadcast range join selection.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1169,6 +1169,22 @@ object SQLConf {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("10MB")
+ val BROADCAST_RANGE_JOIN_ENABLED =
buildConf("spark.sql.join.broadcastRangeJoin.enabled")
+ .doc("When true, a join whose condition is a range predicate is planned as
a " +
+ "broadcast range join instead of a broadcast nested loop join. Supported
join " +
+ "types are inner, left outer, right outer, left semi, and left anti. The
" +
+ "preserved side is streamed. Full outer stays a broadcast nested loop
join. " +
+ "Point-in-range and interval overlap (a.lo < b.hi AND b.lo < a.hi)
broadcast " +
+ "an interval index. A single inequality broadcasts a sorted point index.
" +
+ "Only types the index compares with the join predicate are recognized. "
+
+ "A non-deterministic key stays a nested loop join. " +
+ "Overlapping intervals make the interval index larger than the build
table. " +
Review Comment:
**Nit (P3):** The flat index remains linear in the retained build-row count
regardless of overlap depth. Heavy overlap affects candidate evaluation and the
current per-probe buffer, so this warning attributes the resource risk to the
wrong part of the implementation.
**Verification:**
- **Inspection:** Compare every resource-scaling claim in the config
description with the final interval arrays and overlap traversal, including the
K approaching B case.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/planning/ExtractRangeJoinKeysSuite.scala:
##########
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.planning
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest}
+import org.apache.spark.sql.catalyst.plans.logical.{Join, JoinHint,
LocalRelation}
+import org.apache.spark.sql.types.{DayTimeIntervalType, DoubleType,
IntegerType}
+
+/**
+ * Plan-level coverage for [[ExtractRangeJoinKeys]]: which condition shapes are
+ * recognized, which `RangeJoin` tag is produced, and what residual remains.
+ */
+class ExtractRangeJoinKeysSuite extends PlanTest {
+
+ private val points = LocalRelation($"p".int)
+ private val ranges = LocalRelation($"lo".int, $"hi".int)
+ private val intervalsA = LocalRelation($"lo".int, $"hi".int)
+ private val intervalsB = LocalRelation($"lo".int, $"hi".int)
+
+ private val p = points.output.head
+ private val rLo = ranges.output(0)
+ private val rHi = ranges.output(1)
+ private val aLo = intervalsA.output(0)
+ private val aHi = intervalsA.output(1)
+ private val bLo = intervalsB.output(0)
+ private val bHi = intervalsB.output(1)
+
+ private def join(left: LocalRelation, right: LocalRelation, cond:
Expression): Join =
+ Join(left, right, Inner, Some(cond), JoinHint.NONE)
+
+ private def assertKeys(
+ left: LocalRelation,
+ right: LocalRelation,
+ cond: Expression,
+ expected: RangeJoin,
+ leftKeys: Seq[Expression],
+ rightKeys: Seq[Expression]): Unit = {
+ join(left, right, cond) match {
+ case ExtractRangeJoinKeys(_, _, actualLeft, actualRight, _, rangeJoin) =>
+ assert(rangeJoin === expected)
+ assert(actualLeft === leftKeys)
+ assert(actualRight === rightKeys)
+ case other => fail(s"expected $expected, got $other")
+ }
+ }
+
+ private def assertUnrecognized(plan: Join): Unit = {
+ assert(ExtractRangeJoinKeys.unapply(plan).isEmpty)
+ }
+
+ test("point-in-range keys are the point and the two bounds") {
+ // Inclusivity and conjunct order stay on the condition. BETWEEN is a With
+ // of the point when common expressions are not inlined; the ref has no
Review Comment:
**Nit (P3):** BETWEEN is recognized because deterministic With definitions
are substituted before range matching; an unexpanded CommonExpressionRef is
explicitly rejected. The current explanation attributes success to the
reference itself and can mislead future changes to the inlining step.
**Verification:**
- **Inspection:** Check the explanation against the order and rejection
branches of common-expression inlining and range-predicate extraction.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeBroadcastMode.scala:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.planning.{IntervalOverlapJoin,
PartialRangeJoin, PointInRangeJoin, RangeJoin}
+import org.apache.spark.sql.catalyst.plans.physical.BroadcastMode
+import org.apache.spark.sql.catalyst.types.PhysicalDataType
+
+/**
+ * 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(rangeJoin: RangeJoin): Boolean = RangeIndexKind(rangeJoin) ==
this
+}
+
+private[execution] case object IntervalIndexKind extends RangeIndexKind
+
+private[execution] case object PointIndexKind extends RangeIndexKind
+
+private[execution] object RangeIndexKind {
+ def apply(rangeJoin: RangeJoin): RangeIndexKind = rangeJoin match {
+ case PointInRangeJoin | IntervalOverlapJoin => IntervalIndexKind
+ case _: PartialRangeJoin => PointIndexKind
+ }
+}
+
+/**
+ * A [[BroadcastMode]] that materializes the build side into a
[[RangeRelation]].
+ *
+ * Symmetric with `HashedRelationBroadcastMode`. `buildKeys` are bound to the
+ * build row, so the index does not depend on attribute ids. Which join input
is
+ * broadcast, and the original join condition, live on
[[BroadcastRangeJoinExec]].
+ * Under AQE, `LogicalQueryStageStrategy` checks that this mode's bound keys
are
+ * the keys `ExtractRangeJoinKeys` finds.
+ *
+ * Bound keys are stored with nullability forced on. AQE can rewrite a stage's
+ * nullability, and
[[org.apache.spark.sql.catalyst.plans.physical.BroadcastPartitioning]]
+ * decides reuse with `==`. Forcing it here keeps that check aligned with the
+ * strategy, which compares modes the same way. The index already treats a null
+ * key as non-matching.
+ */
+private[execution] case class RangeBroadcastMode private(
+ buildKeys: Seq[Expression],
+ indexKind: RangeIndexKind)
+ extends BroadcastMode with Serializable {
+
+ override def transform(rows: Array[InternalRow]): RangeRelation =
+ transform(rows.iterator, Some(rows.length))
+
+ override def transform(
+ rows: Iterator[InternalRow],
+ sizeHint: Option[Long]): RangeRelation = {
+ indexKind match {
+ case IntervalIndexKind =>
+ require(buildKeys.nonEmpty && buildKeys.length <= 2,
+ "An interval index expects (low, high), or one point used as both
ends.")
+ case PointIndexKind =>
+ require(buildKeys.length == 1, "A point index expects one build key.")
+ }
+
+ val ordering = PhysicalDataType.ordering(buildKeys.head.dataType)
Review Comment:
**Blocking (P1):** For StringType this ordering closes over
Collation.comparator, whose Java-lambda target is not serializable. The
RangeRelation retains that closure and TorrentBroadcast serializes it with
JavaSerializer by default, so supported collated-string range joins can fail
before execution.
**Recommended change:** Store serializable key-type or collation metadata in
range relations and reconstruct a transient/lazy Ordering on the executor
instead of serializing the captured Ordering object.
**Why this works:** Serialize the DataType or collation identity needed to
recreate PhysicalDataType.ordering, mark the runtime ordering as derived rather
than payload state, and initialize it deterministically after deserialization
for every comparison entry point.
**Scope:** Move non-serializable comparison objects out of the broadcast
payload while preserving supported ordering semantics.
**Compatibility:** Supported data types, planner selection, index shape, and
range comparison results remain unchanged.
**Risks:** Reconstruction from incomplete metadata can silently use binary
ordering for a collated string. A non-thread-safe lazy initialization could
race when a broadcast relation is shared by executor tasks.
**Constraints:** Preserve binary and non-binary collation semantics exactly.
Cover both IntervalIndexKind and PointIndexKind. Keep broadcast size accounting
representative of retained serialized state.
**Success:** Default Java serialization round-trips every supported string
range relation. Deserialized comparisons match build-side binary and collated
string ordering. Integer and other already-serializable key behavior remains
unchanged.
**Verification:**
- **Regression:** Round-trip point and interval relations for binary,
case-insensitive, and ICU-backed string collations through the default
serializer and execute comparisons after deserialization.
- **Behavior:** Run supported string range predicates with either build side
and verify results match the nested-loop path for collation-sensitive values.
- **Compatibility:** Verify non-string point and interval indexes retain
their existing comparison and size behavior.
##########
sql/core/src/test/scala/org/apache/spark/sql/RangeJoinSQLSuite.scala:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.planning.{IntervalOverlapJoin,
LessPartialRangeJoin}
+import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftOuter,
LeftSemi}
+import org.apache.spark.sql.execution.{
+ InputAdapter, ReusedSubqueryExec, ScalarSubquery, SparkPlan, SubqueryExec,
+ WholeStageCodegenExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.joins.{
+ BroadcastNestedLoopJoinExec, BroadcastRangeJoinExec, CartesianProductExec}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Planner and end-to-end coverage for broadcast range join.
+ *
+ * Which condition is recognized: `ExtractRangeJoinKeysSuite`.
+ * What the index returns: `RangeIndexSuite`.
+ * What the operator returns: `RangeJoinSuite`.
+ * This suite checks that the planner chooses the operator, and that the SQL
+ * result matches broadcast nested loop join where that comparison is the
point.
+ */
+class RangeJoinSQLSuite extends QueryTest with SharedSparkSession with
AdaptiveSparkPlanHelper {
+
+ private val pointCond =
+ "s.ip_add_int >= l.BEGIN_IP2LONG AND s.ip_add_int <= l.END_IP2LONG"
+
+ private def findRangeJoin(plan: SparkPlan): Seq[BroadcastRangeJoinExec] = {
+ collect(plan) { case j: BroadcastRangeJoinExec => j }
+ }
+
+ /** `n` range joins sit in one whole-stage, not in nested stages or input
adapters. */
+ private def oneCodegenStageOwns(plan: SparkPlan, n: Int): Boolean = {
+ exists(plan) {
+ case w: WholeStageCodegenExec =>
+ def count(p: SparkPlan): Int = p match {
+ case _: InputAdapter | _: WholeStageCodegenExec => 0
+ case j: BroadcastRangeJoinExec => 1 + j.children.map(count).sum
+ case other => other.children.map(count).sum
+ }
+ count(w.child) == n
+ case _ => false
+ }
+ }
+
+ /** The range join is compiled into a whole-stage, not hidden under an input
adapter. */
+ private def rangeJoinIsCodegen(plan: SparkPlan): Boolean = {
+ exists(plan) {
+ case w: WholeStageCodegenExec =>
+ def contains(p: SparkPlan): Boolean = p match {
+ case _: InputAdapter => false
+ case _: BroadcastRangeJoinExec => true
+ case other => other.children.exists(contains)
+ }
+ contains(w.child)
+ case _ => false
+ }
+ }
+
+ private def findBroadcastNestedLoopJoin(plan: SparkPlan):
Seq[BroadcastNestedLoopJoinExec] = {
+ collect(plan) { case j: BroadcastNestedLoopJoinExec => j }
+ }
+
+ private def findCartesianProduct(plan: SparkPlan): Seq[CartesianProductExec]
= {
+ collect(plan) { case c: CartesianProductExec => c }
+ }
+
+ private def withRangeJoin(extra: (String, String)*)(f: => Unit): Unit = {
+ withSQLConf((Seq(SQLConf.BROADCAST_RANGE_JOIN_ENABLED.key -> "true") ++
extra): _*)(f)
+ }
+
+ private def setupIpLookupViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW signin_ip(id, ip_add_int) AS VALUES
+ | ('1', 1467630666L),
+ | ('2', 1186929717L),
+ | ('3', 3501544352L),
+ | ('31', 3501544320L),
+ | ('32', 3501544383L),
+ | ('33', 3501544330L),
+ | ('34', 3501544390L),
+ | ('4', 1626243295L),
+ | ('41', 1626243072L),
+ | ('42', 1626243327L),
+ | ('5', 1206355995L),
+ | ('5', 1206355995L),
+ | ('6', CAST(NULL AS BIGINT)),
+ | ('7', 0L)
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW ip_lookup(
+ | BEGIN_IP2LONG, END_IP2LONG, TWO_CHAR_CNTRY_CD, zip_id) AS VALUES
+ | (1467630592L, 1467630847L, 'de', 2380520L),
+ | (1186929664L, 1186930175L, 'us', 6723936L),
+ | (1186929664L, 1186930175L, 'us', 6723936L),
+ | (3501544320L, 3501544383L, 'ca', 12804631L),
+ | (3501544330L, 3501544390L, 'ca', 12804631001L),
+ | (1626243072L, 1626243327L, 'us', 8849753L),
+ | (1206355968L, 1206356223L, 'us', 10138801L),
+ | (CAST(NULL AS BIGINT), 1206356223L, 'us', 10138801L),
+ | (1206355968L, CAST(NULL AS BIGINT), 'us', 10138801L),
+ | (CAST(NULL AS BIGINT), CAST(NULL AS BIGINT), 'us', 10138801L)
+ |""".stripMargin)
+ }
+
+ private def setupIntervalViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW intervals_a(lo, hi) AS VALUES
+ | (-1, 0), (0, 1), (0, 2), (1, 5)
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW intervals_b(lo, hi) AS VALUES
+ | (-2, -1), (-4, -2), (1, 3), (5, 7)
+ |""".stripMargin)
+ }
+
+ private def setupDecimalViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW dec_ranges(lo, hi) AS VALUES
+ | (CAST(0.00 AS DECIMAL(10, 2)), CAST(2.50 AS DECIMAL(10, 2))),
+ | (CAST(3.00 AS DECIMAL(10, 2)), CAST(5.00 AS DECIMAL(10, 2)))
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW dec_points(p) AS VALUES
+ | (CAST(1.25 AS DECIMAL(10, 2))),
+ | (CAST(3.00 AS DECIMAL(10, 2))),
+ | (CAST(9.00 AS DECIMAL(10, 2)))
+ |""".stripMargin)
+ }
+
+ private def setupCollatedStrings(): Unit = {
+ // 'M' and 'm' compare equal under UNICODE_CI and differ as bytes.
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW str_ranges(lo, hi) AS VALUES
+ | (CAST('a' AS STRING COLLATE UNICODE_CI), CAST('a' AS STRING COLLATE
UNICODE_CI)),
+ | (CAST('b' AS STRING COLLATE UNICODE_CI), CAST('b' AS STRING COLLATE
UNICODE_CI)),
+ | (CAST('c' AS STRING COLLATE UNICODE_CI), CAST('M' AS STRING COLLATE
UNICODE_CI)),
+ | (CAST('m' AS STRING COLLATE UNICODE_CI), CAST('z' AS STRING COLLATE
UNICODE_CI))
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW str_points(s) AS VALUES
+ | (CAST('M' AS STRING COLLATE UNICODE_CI)),
+ | (CAST('m' AS STRING COLLATE UNICODE_CI))
+ |""".stripMargin)
+ }
+
+ test("range join is disabled by default and falls back to nested loop join")
{
+ setupIpLookupViews()
+ val df = sql(s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON
$pointCond")
+ assert(findRangeJoin(df.queryExecution.executedPlan).isEmpty)
+ assert(findBroadcastNestedLoopJoin(df.queryExecution.executedPlan).size ==
1)
+ }
+
+ test("point-in-range matches the nested loop result") {
+ setupIpLookupViews()
+ val query = s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON
$pointCond"
+ val expected = sql(query).collect()
+ withRangeJoin() {
+ val df = sql(query)
+ assert(findRangeJoin(df.queryExecution.executedPlan).size == 1)
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("BETWEEN is planned as a point-in-range join") {
+ setupIpLookupViews()
+ val query =
+ "SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l " +
+ "ON s.ip_add_int BETWEEN l.BEGIN_IP2LONG AND l.END_IP2LONG"
+ val expected = sql(query).collect()
+ withRangeJoin(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> "false") {
+ val df = sql(query)
+ val planned = findRangeJoin(df.queryExecution.executedPlan)
+ assert(planned.size == 1)
+ // BETWEEN lowers to one point expression. The other side keeps (low,
high).
+ val keyCounts = Set(planned.head.leftKeys.length,
planned.head.rightKeys.length)
+ assert(keyCounts == Set(1, 2))
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("hints select range join, nested loop, or cartesian") {
+ setupIpLookupViews()
+ setupIntervalViews()
+ val expected = sql(
+ s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON
$pointCond").collect()
+ withRangeJoin() {
+ val broadcast = sql(
+ "SELECT /*+ BROADCAST(l) */ s.id, l.zip_id FROM signin_ip s JOIN
ip_lookup l " +
+ s"ON $pointCond")
+ val joins = findRangeJoin(broadcast.queryExecution.executedPlan)
+ assert(joins.size == 1 && joins.head.buildSide == BuildRight)
+ checkAnswer(broadcast, expected)
+
+ val unsupported = sql(
+ "SELECT /*+ BROADCAST(a) */ a.lo, b.hi FROM intervals_a a LEFT JOIN
intervals_b b " +
+ "ON a.lo < b.hi")
+ val loops =
findBroadcastNestedLoopJoin(unsupported.queryExecution.executedPlan)
+ assert(findRangeJoin(unsupported.queryExecution.executedPlan).isEmpty)
+ assert(loops.size == 1 && loops.head.buildSide == BuildLeft)
+
+ val cartesian = sql(
+ "SELECT /*+ SHUFFLE_REPLICATE_NL(s) */ s.id, l.zip_id " +
+ s"FROM signin_ip s JOIN ip_lookup l ON $pointCond")
+ val plan = cartesian.queryExecution.executedPlan
+ assert(findRangeJoin(plan).isEmpty)
+ assert(findCartesianProduct(plan).size == 1)
+ }
+ }
+
+ test("supported join types match nested loop; full outer does not") {
+ setupIpLookupViews()
+ Seq(
+ s"SELECT s.id, l.zip_id FROM signin_ip s LEFT JOIN ip_lookup l ON
$pointCond",
+ s"SELECT s.id, l.zip_id FROM signin_ip s RIGHT JOIN ip_lookup l ON
$pointCond",
+ s"SELECT s.id FROM signin_ip s LEFT SEMI JOIN ip_lookup l ON $pointCond",
+ s"SELECT s.id FROM signin_ip s LEFT ANTI JOIN ip_lookup l ON $pointCond"
+ ).foreach { query =>
+ val expected = sql(query).collect()
+ withRangeJoin() {
+ val df = sql(query)
+ assert(findRangeJoin(df.queryExecution.executedPlan).size == 1, query)
+ checkAnswer(df, expected)
+ }
+ }
+ val fullOuter =
+ s"SELECT s.id, l.zip_id FROM signin_ip s FULL OUTER JOIN ip_lookup l ON
$pointCond"
+ withRangeJoin() {
+ assert(findRangeJoin(sql(fullOuter).queryExecution.executedPlan).isEmpty)
+ }
+ }
+
+ test("partial range left outer broadcasts the right side") {
+ setupIntervalViews()
+ // (1, 5) is below every b.lo, so the left row is kept with nulls.
Review Comment:
**Nit (P3):** For `a.hi < b.lo`, this row is null-padded because no `b.lo`
is strictly greater than 5. Saying `(1, 5)` is below every lower bound reverses
the predicate and does not match the fixture values.
**Verification:**
- **Inspection:** Evaluate the stated ordering against every b.lo fixture
value and the strict a.hi < b.lo predicate.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/joins/RangeJoinSuite.scala:
##########
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{
+ And, Expression, GreaterThan, GreaterThanOrEqual, LessThan, LessThanOrEqual}
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight,
BuildSide}
+import org.apache.spark.sql.catalyst.planning.{GreaterPartialRangeJoin,
IntervalOverlapJoin, LessPartialRangeJoin, PointInRangeJoin, RangeJoin}
+import org.apache.spark.sql.catalyst.plans.{
+ Inner, JoinType, LeftAnti, LeftOuter, LeftSemi, RightOuter}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, StructType}
+
+/**
+ * Operator results for [[BroadcastRangeJoinExec]]. The planner is covered by
+ * `RangeJoinSQLSuite`; recognition and the index are covered below that.
+ */
+class RangeJoinSuite extends QueryTest with SharedSparkSession {
+ private lazy val intervals1: DataFrame = spark.createDataFrame(
+ sparkContext.parallelize(Seq(
+ Row(-1, 0),
+ Row(0, 1),
+ Row(0, 2),
+ Row(1, 5)
+ )), new StructType().add("low1", IntegerType).add("high1", IntegerType))
+
+ private lazy val intervals2: DataFrame = spark.createDataFrame(
+ sparkContext.parallelize(Seq(
+ Row(-2, -1),
+ Row(1, 3),
+ Row(5, 7)
+ )), new StructType().add("low2", IntegerType).add("high2", IntegerType))
+
+ private lazy val points: DataFrame = spark.createDataFrame(
+ sparkContext.parallelize(Seq(
+ Row(-3),
+ Row(1),
+ Row(3),
+ Row(6)
+ )), new StructType().add("point", IntegerType))
+
+ /**
+ * Operator-level helper. Keys are the original expressions from each child.
+ * `condition` is the original join predicate and the only accept/reject
check.
+ */
+ private def rangeJoinExec(
+ left: SparkPlan,
+ right: SparkPlan,
+ buildSide: BuildSide,
+ joinType: JoinType,
+ buildKeys: Seq[Expression],
+ streamedKeys: Seq[Expression],
+ condition: Expression,
+ rangeJoin: RangeJoin = PointInRangeJoin): BroadcastRangeJoinExec = {
+ val (leftKeys, rightKeys) = buildSide match {
+ case BuildLeft => (buildKeys, streamedKeys)
+ case BuildRight => (streamedKeys, buildKeys)
+ }
+ BroadcastRangeJoinExec(
+ left, right, buildSide, joinType, leftKeys, rightKeys, Some(condition),
rangeJoin)
+ }
+
+ private def pointCondition(
+ point: Expression,
+ low: Expression,
+ high: Expression,
+ lowInclusive: Boolean,
+ highInclusive: Boolean,
+ highFirst: Boolean = false): Expression = {
+ val lowPred = if (lowInclusive) GreaterThanOrEqual(point, low) else
GreaterThan(point, low)
+ val highPred = if (highInclusive) LessThanOrEqual(point, high) else
LessThan(point, high)
+ if (highFirst) And(highPred, lowPred) else And(lowPred, highPred)
+ }
+
+ private def checkPointJoin(
+ left: DataFrame,
+ right: DataFrame,
+ pointOnLeft: Boolean,
+ buildSide: BuildSide,
+ joinType: JoinType,
+ lowInclusive: Boolean,
+ highInclusive: Boolean,
+ expected: Seq[Row],
+ highFirst: Boolean = false): Unit = {
+ checkAnswer2(left, right, (leftPlan: SparkPlan, rightPlan: SparkPlan) => {
Review Comment:
**Non-blocking (P2):** These result checks run through the default
whole-stage-enabled preparation, while the explicit mode coverage elsewhere
forces codegen on. The separate `doExecute` / `streamPreservingJoin` loop has
no whole-stage-disabled result assertion, so regressions confined to the
interpreted path can remain green.
**Verification:**
- **Regression:** Establish a failure signal for changes confined to
streamPreservingJoin while generated execution remains correct.
- **Behavior:** Compare generated and interpreted results across supported
join families and unmatched-row behavior.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeIndex.scala:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.mutable.ArrayBuffer
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.util.SizeEstimator
+
+/**
+ * Broadcast payload of a range join. An index only names candidate rows.
+ * [[BroadcastRangeJoinExec]] evaluates the original join condition to filter
candidates.
+ *
+ * - [[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 estimatedSize(): Long
+}
+
+private[execution] object RangeIndex {
+
+ /** Accessor reading a single field from `InternalRow` as a boxed value (or
null). */
+ def getValue(dt: DataType, ordinal: Int): InternalRow => Any = {
+ val accessor = InternalRow.getAccessor(dt)
+ (input: InternalRow) => accessor(input, ordinal)
+ }
+}
+
+/**
+ * Augmented interval tree of build-side intervals, stored as flat arrays.
+ *
+ * Intervals are sorted by low bound. For any span `[start, end)`, its midpoint
+ * is the subtree root, and `maxHighs(mid)` records the maximum high bound in
+ * that span. Subtrees whose max high is below the probe window, or whose low
+ * starts after it, are pruned during search.
+ */
+private[execution] object IntervalIndex {
+
+ def build(
+ ordering: Ordering[Any],
+ intervals: Array[(Any, Any, InternalRow)]): IntervalIndex = {
+ val valid = intervals
+ .filter(t => t._1 != null && t._2 != null)
+ .map { case (lo, hi, r) =>
+ if (ordering.lteq(lo, hi)) (lo, hi, r) else (hi, lo, r)
+ }
+ java.util.Arrays.sort(valid, Ordering.by[(Any, Any, InternalRow),
Any](_._1)(ordering))
+
+ val length = valid.length
+ val maxHighs = new Array[Any](length)
+
+ def fillMaxHigh(start: Int, end: Int): Any = {
+ if (start >= end) return null
+ val mid = (start + end) >>> 1
+ var max = valid(mid)._2
+ val left = fillMaxHigh(start, mid)
+ val right = fillMaxHigh(mid + 1, end)
+ if (left != null) max = ordering.max(max, left)
+ if (right != null) max = ordering.max(max, right)
+ maxHighs(mid) = max
+ max
+ }
+
+ fillMaxHigh(0, length)
+ new IntervalIndex(ordering, valid, maxHighs)
+ }
+}
+
+private[execution] class IntervalIndex private[joins] (
+ private[this] val ordering: Ordering[Any],
+ private[this] val intervals: Array[(Any, Any, InternalRow)],
+ private[this] val maxHighs: Array[Any])
+ extends RangeRelation {
+
+ override def estimatedSize(): Long =
+ SizeEstimator.estimate(intervals) + SizeEstimator.estimate(maxHighs)
+
+ def overlapping(low: Any, high: Any): Iterator[InternalRow] = {
+ if (intervals.isEmpty || low == null || high == null) return Iterator.empty
+ val (probeLow, probeHigh) = if (ordering.lteq(low, high)) (low, high) else
(high, low)
+
+ val buffer = ArrayBuffer.empty[InternalRow]
+ def search(start: Int, end: Int): Unit = {
+ if (start >= end) return
+ val mid = (start + end) >>> 1
+ if (ordering.gteq(maxHighs(mid), probeLow)) {
+ search(start, mid)
+ val (lo, hi, row) = intervals(mid)
Review Comment:
**Non-blocking (P2):** This search materializes every matching build row
before the consumer can inspect the first candidate. For heavily overlapping
intervals that adds O(K) temporary references per probe and prevents semi/anti
joins from short-circuiting traversal after an early condition match.
**Recommended change:** Replace eager recursive collection with a lazy
stack-backed pruned iterator and add high-overlap coverage that proves bounded
probe state and early termination.
**Why this works:** Maintain explicit pending subtree/span state, advance
only until the next qualifying interval, and defer later subtree traversal
until the consumer calls hasNext or next again.
**Scope:** Make interval candidate discovery incremental across all join
consumers.
**Compatibility:** The flat balanced index, pruning comparisons, candidate
semantics, and join output remain unchanged.
**Risks:** An incorrect stack transition can skip or duplicate a subtree.
Iterator reuse of mutable rows or traversal state can corrupt output across
next calls.
**Constraints:** Preserve inclusive overlap semantics, inverted-bound
normalization, and subtree pruning. Do not retain O(K) candidate references
inside the iterator. Keep generated and interpreted consumers on the same
RangeRelation iterator contract.
**Success:** Opening an overlap probe does not traverse or retain all K
matches. Semi and anti joins can stop interval-tree traversal after the first
residual-condition match. Inner and outer joins return the same candidate and
final row sets for every supported bound shape.
**Verification:**
- **Behavior:** Compare lazy overlap results with brute-force overlap across
empty, null, inverted, duplicate, inclusive, and heavily overlapping intervals.
- **Regression:** Show that constructing and partially consuming a K
approximately B probe does not allocate or advance through all K candidates,
including an existence join that succeeds on an early candidate.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2874,6 +2874,83 @@ class AdaptiveQueryExecSuite
assert(LogicalQueryStageStrategy(nullAwareAntiJoinWithRegularStage).isEmpty)
}
+ test("LogicalQueryStageStrategy reuses a range broadcast only for the same
keys") {
+ val leftAttr = AttributeReference("a", IntegerType)()
+ val lo = AttributeReference("lo", IntegerType)()
+ val hi = AttributeReference("hi", IntegerType)()
+ val left = LocalRelation(leftAttr)
+ val right = LocalRelation(lo, hi)
+ val cond = LessThan(leftAttr, lo)
+
+ def stage(mode: RangeBroadcastMode): LogicalQueryStage = {
+ val scan = LocalTableScanExec(right.output, Nil, None)
+ val exchange = BroadcastExchangeExec(mode, scan)
+ LogicalQueryStage(right, BroadcastQueryStageExec(0, exchange, exchange))
+ }
+
+ def modeFor(buildKey: AttributeReference): RangeBroadcastMode = {
+ val bound: Expression = BindReferences.bindReference(buildKey,
right.output)
+ RangeBroadcastMode(Seq(bound), PointIndexKind)
Review Comment:
**Non-blocking (P2):** This fixture varies key identity while keeping
PointIndexKind constant, so it cannot fail if the separate index-kind
compatibility check is removed. A same-key stage with the wrong kind is a
supported, materially different AQE reuse case.
**Verification:**
- **Regression:** Demonstrate that deleting or bypassing the RangeIndexKind
compatibility check makes the cross-kind negative assertion fail.
##########
sql/core/src/test/scala/org/apache/spark/sql/RangeJoinSQLSuite.scala:
##########
@@ -0,0 +1,520 @@
+/*
+ * 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
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+import org.apache.spark.sql.catalyst.planning.{IntervalOverlapJoin,
LessPartialRangeJoin}
+import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti, LeftOuter,
LeftSemi}
+import org.apache.spark.sql.execution.{
+ InputAdapter, ReusedSubqueryExec, ScalarSubquery, SparkPlan, SubqueryExec,
+ WholeStageCodegenExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.joins.{
+ BroadcastNestedLoopJoinExec, BroadcastRangeJoinExec, CartesianProductExec}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+
+/**
+ * Planner and end-to-end coverage for broadcast range join.
+ *
+ * Which condition is recognized: `ExtractRangeJoinKeysSuite`.
+ * What the index returns: `RangeIndexSuite`.
+ * What the operator returns: `RangeJoinSuite`.
+ * This suite checks that the planner chooses the operator, and that the SQL
+ * result matches broadcast nested loop join where that comparison is the
point.
+ */
+class RangeJoinSQLSuite extends QueryTest with SharedSparkSession with
AdaptiveSparkPlanHelper {
+
+ private val pointCond =
+ "s.ip_add_int >= l.BEGIN_IP2LONG AND s.ip_add_int <= l.END_IP2LONG"
+
+ private def findRangeJoin(plan: SparkPlan): Seq[BroadcastRangeJoinExec] = {
+ collect(plan) { case j: BroadcastRangeJoinExec => j }
+ }
+
+ /** `n` range joins sit in one whole-stage, not in nested stages or input
adapters. */
+ private def oneCodegenStageOwns(plan: SparkPlan, n: Int): Boolean = {
+ exists(plan) {
+ case w: WholeStageCodegenExec =>
+ def count(p: SparkPlan): Int = p match {
+ case _: InputAdapter | _: WholeStageCodegenExec => 0
+ case j: BroadcastRangeJoinExec => 1 + j.children.map(count).sum
+ case other => other.children.map(count).sum
+ }
+ count(w.child) == n
+ case _ => false
+ }
+ }
+
+ /** The range join is compiled into a whole-stage, not hidden under an input
adapter. */
+ private def rangeJoinIsCodegen(plan: SparkPlan): Boolean = {
+ exists(plan) {
+ case w: WholeStageCodegenExec =>
+ def contains(p: SparkPlan): Boolean = p match {
+ case _: InputAdapter => false
+ case _: BroadcastRangeJoinExec => true
+ case other => other.children.exists(contains)
+ }
+ contains(w.child)
+ case _ => false
+ }
+ }
+
+ private def findBroadcastNestedLoopJoin(plan: SparkPlan):
Seq[BroadcastNestedLoopJoinExec] = {
+ collect(plan) { case j: BroadcastNestedLoopJoinExec => j }
+ }
+
+ private def findCartesianProduct(plan: SparkPlan): Seq[CartesianProductExec]
= {
+ collect(plan) { case c: CartesianProductExec => c }
+ }
+
+ private def withRangeJoin(extra: (String, String)*)(f: => Unit): Unit = {
+ withSQLConf((Seq(SQLConf.BROADCAST_RANGE_JOIN_ENABLED.key -> "true") ++
extra): _*)(f)
+ }
+
+ private def setupIpLookupViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW signin_ip(id, ip_add_int) AS VALUES
+ | ('1', 1467630666L),
+ | ('2', 1186929717L),
+ | ('3', 3501544352L),
+ | ('31', 3501544320L),
+ | ('32', 3501544383L),
+ | ('33', 3501544330L),
+ | ('34', 3501544390L),
+ | ('4', 1626243295L),
+ | ('41', 1626243072L),
+ | ('42', 1626243327L),
+ | ('5', 1206355995L),
+ | ('5', 1206355995L),
+ | ('6', CAST(NULL AS BIGINT)),
+ | ('7', 0L)
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW ip_lookup(
+ | BEGIN_IP2LONG, END_IP2LONG, TWO_CHAR_CNTRY_CD, zip_id) AS VALUES
+ | (1467630592L, 1467630847L, 'de', 2380520L),
+ | (1186929664L, 1186930175L, 'us', 6723936L),
+ | (1186929664L, 1186930175L, 'us', 6723936L),
+ | (3501544320L, 3501544383L, 'ca', 12804631L),
+ | (3501544330L, 3501544390L, 'ca', 12804631001L),
+ | (1626243072L, 1626243327L, 'us', 8849753L),
+ | (1206355968L, 1206356223L, 'us', 10138801L),
+ | (CAST(NULL AS BIGINT), 1206356223L, 'us', 10138801L),
+ | (1206355968L, CAST(NULL AS BIGINT), 'us', 10138801L),
+ | (CAST(NULL AS BIGINT), CAST(NULL AS BIGINT), 'us', 10138801L)
+ |""".stripMargin)
+ }
+
+ private def setupIntervalViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW intervals_a(lo, hi) AS VALUES
+ | (-1, 0), (0, 1), (0, 2), (1, 5)
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW intervals_b(lo, hi) AS VALUES
+ | (-2, -1), (-4, -2), (1, 3), (5, 7)
+ |""".stripMargin)
+ }
+
+ private def setupDecimalViews(): Unit = {
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW dec_ranges(lo, hi) AS VALUES
+ | (CAST(0.00 AS DECIMAL(10, 2)), CAST(2.50 AS DECIMAL(10, 2))),
+ | (CAST(3.00 AS DECIMAL(10, 2)), CAST(5.00 AS DECIMAL(10, 2)))
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW dec_points(p) AS VALUES
+ | (CAST(1.25 AS DECIMAL(10, 2))),
+ | (CAST(3.00 AS DECIMAL(10, 2))),
+ | (CAST(9.00 AS DECIMAL(10, 2)))
+ |""".stripMargin)
+ }
+
+ private def setupCollatedStrings(): Unit = {
+ // 'M' and 'm' compare equal under UNICODE_CI and differ as bytes.
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW str_ranges(lo, hi) AS VALUES
+ | (CAST('a' AS STRING COLLATE UNICODE_CI), CAST('a' AS STRING COLLATE
UNICODE_CI)),
+ | (CAST('b' AS STRING COLLATE UNICODE_CI), CAST('b' AS STRING COLLATE
UNICODE_CI)),
+ | (CAST('c' AS STRING COLLATE UNICODE_CI), CAST('M' AS STRING COLLATE
UNICODE_CI)),
+ | (CAST('m' AS STRING COLLATE UNICODE_CI), CAST('z' AS STRING COLLATE
UNICODE_CI))
+ |""".stripMargin)
+ sql(
+ """
+ |CREATE OR REPLACE TEMP VIEW str_points(s) AS VALUES
+ | (CAST('M' AS STRING COLLATE UNICODE_CI)),
+ | (CAST('m' AS STRING COLLATE UNICODE_CI))
+ |""".stripMargin)
+ }
+
+ test("range join is disabled by default and falls back to nested loop join")
{
+ setupIpLookupViews()
+ val df = sql(s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON
$pointCond")
+ assert(findRangeJoin(df.queryExecution.executedPlan).isEmpty)
+ assert(findBroadcastNestedLoopJoin(df.queryExecution.executedPlan).size ==
1)
+ }
+
+ test("point-in-range matches the nested loop result") {
+ setupIpLookupViews()
+ val query = s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON
$pointCond"
+ val expected = sql(query).collect()
+ withRangeJoin() {
+ val df = sql(query)
+ assert(findRangeJoin(df.queryExecution.executedPlan).size == 1)
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("BETWEEN is planned as a point-in-range join") {
+ setupIpLookupViews()
+ val query =
+ "SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l " +
+ "ON s.ip_add_int BETWEEN l.BEGIN_IP2LONG AND l.END_IP2LONG"
+ val expected = sql(query).collect()
+ withRangeJoin(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> "false") {
+ val df = sql(query)
+ val planned = findRangeJoin(df.queryExecution.executedPlan)
+ assert(planned.size == 1)
+ // BETWEEN lowers to one point expression. The other side keeps (low,
high).
+ val keyCounts = Set(planned.head.leftKeys.length,
planned.head.rightKeys.length)
+ assert(keyCounts == Set(1, 2))
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("hints select range join, nested loop, or cartesian") {
+ setupIpLookupViews()
+ setupIntervalViews()
+ val expected = sql(
+ s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON
$pointCond").collect()
+ withRangeJoin() {
+ val broadcast = sql(
+ "SELECT /*+ BROADCAST(l) */ s.id, l.zip_id FROM signin_ip s JOIN
ip_lookup l " +
+ s"ON $pointCond")
+ val joins = findRangeJoin(broadcast.queryExecution.executedPlan)
+ assert(joins.size == 1 && joins.head.buildSide == BuildRight)
+ checkAnswer(broadcast, expected)
+
+ val unsupported = sql(
+ "SELECT /*+ BROADCAST(a) */ a.lo, b.hi FROM intervals_a a LEFT JOIN
intervals_b b " +
+ "ON a.lo < b.hi")
+ val loops =
findBroadcastNestedLoopJoin(unsupported.queryExecution.executedPlan)
+ assert(findRangeJoin(unsupported.queryExecution.executedPlan).isEmpty)
+ assert(loops.size == 1 && loops.head.buildSide == BuildLeft)
+
+ val cartesian = sql(
+ "SELECT /*+ SHUFFLE_REPLICATE_NL(s) */ s.id, l.zip_id " +
+ s"FROM signin_ip s JOIN ip_lookup l ON $pointCond")
+ val plan = cartesian.queryExecution.executedPlan
+ assert(findRangeJoin(plan).isEmpty)
+ assert(findCartesianProduct(plan).size == 1)
+ }
+ }
+
+ test("supported join types match nested loop; full outer does not") {
+ setupIpLookupViews()
+ Seq(
+ s"SELECT s.id, l.zip_id FROM signin_ip s LEFT JOIN ip_lookup l ON
$pointCond",
+ s"SELECT s.id, l.zip_id FROM signin_ip s RIGHT JOIN ip_lookup l ON
$pointCond",
+ s"SELECT s.id FROM signin_ip s LEFT SEMI JOIN ip_lookup l ON $pointCond",
+ s"SELECT s.id FROM signin_ip s LEFT ANTI JOIN ip_lookup l ON $pointCond"
+ ).foreach { query =>
+ val expected = sql(query).collect()
+ withRangeJoin() {
+ val df = sql(query)
+ assert(findRangeJoin(df.queryExecution.executedPlan).size == 1, query)
+ checkAnswer(df, expected)
+ }
+ }
+ val fullOuter =
+ s"SELECT s.id, l.zip_id FROM signin_ip s FULL OUTER JOIN ip_lookup l ON
$pointCond"
+ withRangeJoin() {
+ assert(findRangeJoin(sql(fullOuter).queryExecution.executedPlan).isEmpty)
+ }
+ }
+
+ test("partial range left outer broadcasts the right side") {
+ setupIntervalViews()
+ // (1, 5) is below every b.lo, so the left row is kept with nulls.
+ val query =
+ "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a LEFT JOIN intervals_b
b " +
+ "ON a.hi < b.lo"
+ val expected = sql(query).collect()
+ assert(expected.exists(_.anyNull))
+ withRangeJoin() {
+ val df = sql(query)
+ val planned = findRangeJoin(df.queryExecution.executedPlan)
+ assert(planned.size == 1)
+ assert(planned.head.joinType == LeftOuter)
+ assert(planned.head.buildSide == BuildRight)
+ assert(planned.head.rangeJoin == LessPartialRangeJoin)
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("a residual predicate is still evaluated") {
+ sql("CREATE OR REPLACE TEMP VIEW complex_a(id, lo, hi) AS VALUES (1, 0,
10), (2, 0, 10)")
+ sql("CREATE OR REPLACE TEMP VIEW complex_b(id, point) AS VALUES (1, 5),
(2, 5)")
+ // Both ranges cover both points, so the range predicate matches all 4
pairs.
+ // a.id <> b.id is not an equi-join key and filters that down to 2.
+ val query =
+ "SELECT a.id, b.id FROM complex_a a JOIN complex_b b " +
+ "ON b.point >= a.lo AND b.point <= a.hi AND a.id <> b.id"
+ val expected = sql(query).collect()
+ assert(expected.length == 2)
+ withRangeJoin() {
+ val df = sql(query)
+ assert(findRangeJoin(df.queryExecution.executedPlan).size == 1)
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("interval overlap is planned with two keys on each side") {
+ setupIntervalViews()
+ val query =
+ "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a JOIN intervals_b b " +
+ "ON a.lo < b.hi AND b.lo < a.hi"
+ val expected = sql(query).collect()
+ withRangeJoin() {
+ val df = sql(query)
+ val planned = findRangeJoin(df.queryExecution.executedPlan)
+ assert(planned.size == 1)
+ assert(planned.head.rangeJoin == IntervalOverlapJoin)
+ assert(planned.head.leftKeys.length == 2)
+ assert(planned.head.rightKeys.length == 2)
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("interval overlap with an inverted interval matches the nested loop
result") {
+ sql("CREATE OR REPLACE TEMP VIEW inverted_a(lo, hi) AS VALUES (0, 5), (1,
2), (-4, 10)")
+ sql("CREATE OR REPLACE TEMP VIEW inverted_b(lo, hi) AS VALUES (3, 1), (7,
9)")
+ val query =
+ "SELECT a.lo, a.hi, b.lo, b.hi FROM inverted_a a " +
+ "JOIN inverted_b b ON a.lo < b.hi AND b.lo < a.hi"
+ // (3, 1) accepts rows with a.lo < 1 and a.hi > 3, so it has to reach the
result.
+ val expected = sql(query).collect()
+ assert(expected.exists(r => r.getInt(2) == 3 && r.getInt(3) == 1),
expected.toSeq)
+ assert(expected.length == 3, expected.toSeq)
+ // Broadcast each side in turn so the inverted interval lands on the index
once.
+ Seq("/*+ BROADCAST(a) */", "/*+ BROADCAST(b) */").foreach { hint =>
+ val hinted = query.replace("SELECT ", s"SELECT $hint ")
+ withRangeJoin() {
+ val df = sql(hinted)
+ assert(findRangeJoin(df.queryExecution.executedPlan).size == 1, hinted)
+ checkAnswer(df, expected)
+ }
+ }
+ }
+
+ test("a scalar subquery in the condition is planned") {
+ setupIpLookupViews()
+ // The extra conjunct names both sides, so it stays on the join. The
subquery
+ // has to be planned from that condition.
+ val query =
+ "SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l " +
+ s"ON $pointCond " +
+ "AND s.ip_add_int + l.zip_id > (SELECT min(ip_add_int) FROM signin_ip)"
+ val expected = sql(query).collect()
+ withRangeJoin() {
+ val df = sql(query)
+ val planned = findRangeJoin(df.queryExecution.executedPlan)
+ assert(planned.size == 1)
+ val cond = planned.head.condition.get
+ val subquery = cond.collectFirst { case s: ScalarSubquery => s.plan }
+ assert(subquery.exists(plan =>
+ plan.isInstanceOf[SubqueryExec] ||
plan.isInstanceOf[ReusedSubqueryExec]), cond)
+ checkAnswer(df, expected)
+ }
+ }
+
+ test("collated strings match the nested loop result on either build side") {
+ setupCollatedStrings()
+ val cond = "p.s >= r.lo AND p.s <= r.hi"
+ Seq(
+ (s"SELECT /*+ BROADCAST(r) */ p.s, r.lo, r.hi FROM str_points p JOIN
str_ranges r ON $cond",
+ BuildRight),
+ (s"SELECT /*+ BROADCAST(p) */ p.s, r.lo, r.hi FROM str_points p JOIN
str_ranges r ON $cond",
+ BuildLeft)
+ ).foreach { case (query, buildSide) =>
+ val expected = sql(query).collect()
+ // Each point matches the interval ending at 'M' and the one starting at
'm'.
+ assert(expected.length == 4, query)
+ withRangeJoin() {
+ val df = sql(query)
+ val joins = findRangeJoin(df.queryExecution.executedPlan)
+ assert(joins.size == 1 && joins.head.buildSide == buildSide, query)
+ checkAnswer(df, expected)
+ }
+ }
+ }
+
+ test("whole-stage codegen covers each range-join probe") {
+ setupIpLookupViews()
+ setupDecimalViews()
+ setupCollatedStrings()
+ setupIntervalViews()
+ val queries = Seq(
+ s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON $pointCond",
+ s"SELECT s.id, l.zip_id FROM signin_ip s LEFT JOIN ip_lookup l ON
$pointCond",
+ s"SELECT s.id, l.zip_id FROM signin_ip s RIGHT JOIN ip_lookup l ON
$pointCond",
+ s"SELECT s.id FROM signin_ip s LEFT SEMI JOIN ip_lookup l ON $pointCond",
+ s"SELECT s.id FROM signin_ip s LEFT ANTI JOIN ip_lookup l ON $pointCond",
+ "SELECT p.p, r.lo, r.hi FROM dec_points p JOIN dec_ranges r ON p.p >=
r.lo AND p.p <= r.hi",
+ "SELECT /*+ BROADCAST(r) */ p.s, r.lo, r.hi FROM str_points p JOIN
str_ranges r " +
+ "ON p.s >= r.lo AND p.s <= r.hi",
+ "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a JOIN intervals_b b " +
+ "ON a.lo < b.hi AND b.lo < a.hi",
+ "SELECT a.lo, a.hi, b.lo, b.hi FROM intervals_a a LEFT JOIN intervals_b
b ON a.hi < b.lo",
+ "SELECT /*+ BROADCAST(a) */ a.lo, b.hi FROM intervals_a a JOIN
intervals_b b ON a.lo < b.hi")
+ queries.foreach { query =>
+ val expected = sql(query).collect()
+ withRangeJoin(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.CODEGEN_FALLBACK.key -> "false") {
+ val df = sql(query)
+ checkAnswer(df, expected)
+ assert(rangeJoinIsCodegen(df.queryExecution.executedPlan), query)
+ }
+ }
+ }
+
+ test("two interval range joins in one whole-stage match nested loop") {
+ sql("CREATE OR REPLACE TEMP VIEW chain_a(id, lo, hi) AS VALUES (1, 0, 10),
(2, 20, 21)")
+ sql(
+ "CREATE OR REPLACE TEMP VIEW chain_b(id, lo, hi) AS VALUES " +
+ "(10, 1, 4), (11, 2, 8), (12, 30, 40)")
+ sql(
+ "CREATE OR REPLACE TEMP VIEW chain_c(id, lo, hi) AS VALUES " +
+ "(20, 0, 5), (21, 3, 9), (22, 100, 110)")
+ val query =
+ """
+ |SELECT /*+ BROADCAST(b), BROADCAST(c) */ a.id, b.id, c.id
+ |FROM chain_a a
+ |JOIN chain_b b ON a.lo < b.hi AND b.lo < a.hi
+ |JOIN chain_c c ON b.lo < c.hi AND c.lo < b.hi
+ """.stripMargin
+ val expected = sql(query).collect()
+ // Row 1 of a overlaps two rows of b, so the outer probe is still live when
+ // the inner probe runs.
+ assert(expected.length > 2)
+ withRangeJoin(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true",
+ SQLConf.CODEGEN_FALLBACK.key -> "false",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ val df = sql(query)
+ checkAnswer(df, expected)
+ val plan = df.queryExecution.executedPlan
+ val joins = findRangeJoin(plan)
+ assert(joins.size == 2 && joins.forall(_.rangeJoin ==
IntervalOverlapJoin))
+ assert(oneCodegenStageOwns(plan, 2))
+ }
+ }
+
+ test("semi and anti range joins fused outside an interval join match nested
loop") {
+ sql("CREATE OR REPLACE TEMP VIEW exist_a(id, lo, hi) AS VALUES (1, 0,
100)")
+ sql(
+ "CREATE OR REPLACE TEMP VIEW exist_b(id, lo, hi) AS VALUES " +
+ "(10, 1, 4), (11, 2, 8), (12, 50, 60)")
+ sql(
+ "CREATE OR REPLACE TEMP VIEW exist_c(id, lo, hi) AS VALUES " +
+ "(20, 0, 5), (21, 3, 9), (22, 100, 110)")
+ // b10 and b11 each overlap two rows of c, so semi and anti stop before
that
+ // probe is drained. b12 is still waiting on the outer probe.
+ // a.lo < c.hi references both legs of (a join b), so the semi/anti stays
Review Comment:
**Nit (P3):** `a.lo < c.hi` references only `a` and `c`; it contains no `b`
attribute. The compound-left dependency comes from the complete condition, so
the current placement rationale assigns the full expression's references to the
wrong conjunct.
**Verification:**
- **Inspection:** Resolve the attributes of each conjunct against a, b, and
c and compare the resulting dependency with the asserted plan placement.
##########
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
+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
interval tree
+ * used by point-in-range and interval overlap. [[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 span(lo: Any, hi: Any, id: Int): (Any, Any, InternalRow) = (lo,
hi, row(id))
+
+ private def buildIntervals(spans: Seq[(Any, Any, InternalRow)]):
IntervalIndex =
+ IntervalIndex.build(intOrdering, spans.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("build keeps a point, orders an inverted interval, and drops a null
bound") {
+ val index = IntervalIndex.build(intOrdering, Array(
+ (5, 5, InternalRow(5, 5)),
+ (1, 3, InternalRow(1, 3)),
+ (3, 1, InternalRow(3, 1)),
+ (null, 1, InternalRow(null, 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("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 accepts.
+ val index = RangeBroadcastMode(
+ Seq(BoundReference(0, IntegerType, nullable = true),
+ BoundReference(1, IntegerType, nullable = true)),
+ IntervalIndexKind)
+ .transform(Array(InternalRow(3, 1), InternalRow(0, 5)))
+ .asInstanceOf[IntervalIndex]
+ assert(ids(index.overlapping(0, 4)).toSet == Set(3, 0))
+ assert(ids(index.overlapping(4, 9)) == List(0))
+ }
+
+ test("interval tree returns each overlapping row once") {
+ // Spans arrive out of low order: [10, 20] then [0, 5].
+ val unsorted = buildIntervals(Seq(span(10, 20, 0), span(0, 5, 1)))
+ assert(probeIds(unsorted, 3, 3) == List(1))
+ assert(probeIds(unsorted, 15, 15) == List(0))
+
+ // Probe 50 is a build key. Row 0 spans it; row 1 starts there.
+ val spanning = buildIntervals(Seq(span(0, 100, 0), span(50, 200, 1)))
+ assert(probeIds(spanning, 50, 50).toSet == Set(0, 1))
+ assert(probeIds(spanning, 25, 25) == List(0))
+ assert(probeIds(spanning, 150, 150) == List(1))
+
+ val duplicated = buildIntervals(Seq(span(0, 100, 0), span(0, 100, 1)))
+ assert(probeIds(duplicated, 50, 50).sorted == List(0, 1))
+
+ val nested = buildIntervals(Seq(
+ span(0, 100, 0), span(10, 20, 1), span(10, 50, 2), span(90, 110, 3)))
+ def assertOnce(low: Int, expected: Set[Int]): Unit = {
+ val actual = probeIds(nested, low, low)
+ assert(actual.toSet == expected, s"probe $low -> $actual")
+ assert(actual.size == expected.size, s"probe $low duplicated $actual")
+ }
+ assertOnce(0, Set(0))
+ assertOnce(10, Set(0, 1, 2))
+ assertOnce(15, Set(0, 1, 2))
+ assertOnce(30, Set(0, 2))
+ assertOnce(95, Set(0, 3))
+ assertOnce(105, Set(3))
+ assertOnce(200, Set.empty)
+
+ assert(probeIds(buildIntervals(Seq.empty), 1, 1).isEmpty)
+
+ val withNull = buildIntervals(Seq(span(0, 10, 0), span(5, 5, 1), (null, 4,
row(9))))
+ assert(probeIds(withNull, null, 5).isEmpty)
+ assert(probeIds(withNull, 5, null).isEmpty)
+ // The null bound is dropped, so two intervals remain and the estimate
covers both.
+ val one = buildIntervals(Seq(span(0, 10, 0)))
+ assert(withNull.estimatedSize() > one.estimatedSize())
+
+ // At an interior endpoint, the range that ends and the one that starts
+ // are each a candidate once.
+ val chained = buildIntervals(Seq(span(0, 1, 0), span(1, 2, 1), span(2, 3,
2)))
+ assert(probeIds(chained, 0, 0) == List(0))
+ assert(probeIds(chained, 1, 1).sorted == List(0, 1))
+ assert(probeIds(chained, 2, 2).sorted == List(1, 2))
+ assert(probeIds(chained, 3, 3) == List(2))
+
+ // A probe whose endpoints are reversed is the closed window between them.
+ // [10, 25] meets [0, 30], [15, 40], and the point at 15.
+ val inverted = buildIntervals(Seq(span(0, 30, 0), span(15, 40, 1),
span(15, 15, 2)))
+ assert(probeIds(inverted, 25, 10).sorted == List(0, 1, 2))
+ assert(probeIds(inverted, 10, 20).sorted == List(0, 1, 2))
+ }
+
+ test("interval tree is correct under heavy build-side overlap") {
+ // Every interval nests inside the one before it, so a point in the middle
+ // hits about half of the rows. The tree still stores one node per row.
+ 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 = IntervalIndex.build(
+ PhysicalDataType.ordering(StringType),
+ Array((wide, wide, row(0))))
+ assert(strings.estimatedSize() > ints.estimatedSize())
+ }
+
+ test("interval tree treats NaN as its own point and Infinity as an open
bound") {
+ // PhysicalDoubleType's ordering (SQLOrderingUtil.compareDoubles) totally
orders
+ // -Inf < ... < +Inf < NaN, with NaN comparing equal only to itself. So
[1.0, +Inf)
Review Comment:
**Nit (P3):** The interval is closed at +Infinity here: the inclusive
comparison and the assertion below both include a probe at positive infinity.
The half-open notation and 'open bound' wording contradict the behavior this
test pins down.
**Verification:**
- **Inspection:** Compare both endpoint symbols and prose with the inclusive
comparisons and expected +Infinity match.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeIndex.scala:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.mutable.ArrayBuffer
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.util.SizeEstimator
+
+/**
+ * Broadcast payload of a range join. An index only names candidate rows.
+ * [[BroadcastRangeJoinExec]] evaluates the original join condition to filter
candidates.
+ *
+ * - [[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 estimatedSize(): Long
+}
+
+private[execution] object RangeIndex {
+
+ /** Accessor reading a single field from `InternalRow` as a boxed value (or
null). */
+ def getValue(dt: DataType, ordinal: Int): InternalRow => Any = {
+ val accessor = InternalRow.getAccessor(dt)
+ (input: InternalRow) => accessor(input, ordinal)
+ }
+}
+
+/**
+ * Augmented interval tree of build-side intervals, stored as flat arrays.
+ *
+ * Intervals are sorted by low bound. For any span `[start, end)`, its midpoint
+ * is the subtree root, and `maxHighs(mid)` records the maximum high bound in
+ * that span. Subtrees whose max high is below the probe window, or whose low
+ * starts after it, are pruned during search.
+ */
+private[execution] object IntervalIndex {
+
+ def build(
+ ordering: Ordering[Any],
+ intervals: Array[(Any, Any, InternalRow)]): IntervalIndex = {
+ val valid = intervals
Review Comment:
**Nit (P3):** The strict filter allocates a full filtered array and the
following map allocates the final normalized array. For a large valid build
this creates an avoidable O(B) temporary at the broadcast-construction boundary.
**Verification:**
- **Behavior:** Verify mixed valid, null-bound, ordinary, and inverted
intervals build the same normalized sorted index and overlap results.
- **Inspection:** Confirm the build path contains no separate
full-cardinality filtered array before the final normalized array.
--
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]
