cloud-fan commented on code in PR #59027: URL: https://github.com/apache/spark/pull/59027#discussion_r4249393630
########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeIndex.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.types.PhysicalDataType +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 + def keyType: DataType +} + +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( + keyType: DataType, + intervals: Array[(Any, Any, InternalRow)]): IntervalIndex = { + val ordering = PhysicalDataType.ordering(keyType) + val valid = intervals.iterator + .collect { + case (lo, hi, r) if lo != null && hi != null => + if (ordering.lteq(lo, hi)) (lo, hi, r) else (hi, lo, r) + } + .toArray + 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(keyType, valid, maxHighs) + } +} + +private[execution] class IntervalIndex private[joins] ( + val keyType: DataType, + private[this] val intervals: Array[(Any, Any, InternalRow)], + private[this] val maxHighs: Array[Any]) + extends RangeRelation { + + @transient + private[this] lazy val ordering: Ordering[Any] = PhysicalDataType.ordering(keyType) + + 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) + + new Iterator[InternalRow] { + private[this] var stackStart = new Array[Int](128) Review Comment: **Non-blocking (P2):** Each `overlapping` call allocates two `Int[128]` stacks before it can yield a candidate, even for a one-row index. Both interpreted and generated joins open this iterator once per non-null streamed row, so a million selective probes create roughly a gigabyte of short-lived primitive-array payload despite needing only a few stack entries. **Verification:** - **Regression:** Distinguish the scratch allocated for a one-row index from the former fixed 128-by-two capacity and verify bounded growth for larger balanced trees. - **Behavior:** Compare overlap candidates with brute force across empty, tiny, nested, duplicate, inverted, and highly overlapping inputs, including simultaneous iterators. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeIndex.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.types.PhysicalDataType +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 + def keyType: DataType +} + +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( + keyType: DataType, + intervals: Array[(Any, Any, InternalRow)]): IntervalIndex = { + val ordering = PhysicalDataType.ordering(keyType) + val valid = intervals.iterator + .collect { + case (lo, hi, r) if lo != null && hi != null => + if (ordering.lteq(lo, hi)) (lo, hi, r) else (hi, lo, r) + } + .toArray + 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(keyType, valid, maxHighs) + } +} + +private[execution] class IntervalIndex private[joins] ( + val keyType: DataType, + private[this] val intervals: Array[(Any, Any, InternalRow)], + private[this] val maxHighs: Array[Any]) + extends RangeRelation { + + @transient + private[this] lazy val ordering: Ordering[Any] = PhysicalDataType.ordering(keyType) + + override def estimatedSize(): Long = + SizeEstimator.estimate(intervals) + SizeEstimator.estimate(maxHighs) Review Comment: **Non-blocking (P2):** `maxHighs` contains aliases of high-key objects already reachable through `intervals`, but `estimatedSize` starts a separate `SizeEstimator` traversal for each array. The second traversal therefore charges shared variable-width keys again, so `maxBroadcastTableSize` can reject a retained object graph that actually fits; the existing relative-size assertions remain green under that recount. **Verification:** - **Regression:** Use shared wide high-key objects to distinguish one-graph accounting from two independent deep traversals and show the assertion fails under the current implementation. - **Behavior:** Exercise an interval payload near the broadcast limit and verify admission follows the retained object graph while a genuinely oversized payload is still rejected. - **Inspection:** Confirm identity-aware accounting remains correct after the finalized range-index builder representation is selected. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeIndex.scala: ########## @@ -0,0 +1,210 @@ +/* + * 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.types.PhysicalDataType +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 + def keyType: DataType +} + +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( + keyType: DataType, + intervals: Array[(Any, Any, InternalRow)]): IntervalIndex = { + val ordering = PhysicalDataType.ordering(keyType) + val valid = intervals.iterator + .collect { + case (lo, hi, r) if lo != null && hi != null => + if (ordering.lteq(lo, hi)) (lo, hi, r) else (hi, lo, r) + } + .toArray + 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(keyType, valid, maxHighs) + } +} + +private[execution] class IntervalIndex private[joins] ( + val keyType: DataType, + private[this] val intervals: Array[(Any, Any, InternalRow)], + private[this] val maxHighs: Array[Any]) + extends RangeRelation { + + @transient Review Comment: **Blocking (P1):** These new relation roots are passed through `TorrentBroadcast`, which serializes with `SparkEnv.get.serializer`. With `KryoSerializer` and `spark.kryo.registrationRequired=true`, neither `IntervalIndex` nor `PointIndex` is registered by Spark, so every enabled broadcast range join fails during serialization unless users somehow register Spark-private implementation classes themselves. **Verification:** - **Compatibility:** With strict registration enabled, round-trip and probe interval and point payloads for representative numeric, decimal, string-collation, binary, date, timestamp, and boolean keys. - **Regression:** Execute enabled broadcast range joins through the configured Spark serializer and prove they complete without class-not-registered failures. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeBroadcastMode.scala: ########## @@ -0,0 +1,114 @@ +/* + * 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 + +/** + * 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 projection = new InterpretedProjection(buildKeys) + val lowOf = RangeIndex.getValue(buildKeys.head.dataType, 0) + + indexKind match { + case IntervalIndexKind => + val highOf = + if (buildKeys.length == 1) lowOf else RangeIndex.getValue(buildKeys(1).dataType, 1) + val intervals = rows.map { r => + val p = projection(r) + (lowOf(p), highOf(p), r) Review Comment: **Blocking (P1):** The projection result is treated as owned state here, but a deterministic typed ScalaUDF can expose a `UTF8String` view backed by its encoder's reused `UnsafeRow`; `InterpretedProjection` wraps that field without copying it. A later build-row evaluation can therefore overwrite an earlier retained key, after which sorting and bound scans may omit a row whose freshly evaluated original predicate is true. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala: ########## @@ -1169,6 +1169,23 @@ 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, provided that a " + + "supported build side qualifies for broadcast by size or by a BROADCAST hint. " + + "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. " + Review Comment: **Nit (P3):** This overstates the fallback boundary. The mixed-condition test with `p >= randBound AND p <= rHi` rejects the nondeterministic candidate but still recognizes the deterministic second inequality as a partial range join, so a nondeterministic expression is not itself used as a key while another eligible conjunct may still select `BroadcastRangeJoinExec`. **Verification:** - **Inspection:** Compare every nondeterministic-fallback statement with a mixed condition where a deterministic sibling inequality remains eligible. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/joins/RangeJoinSuite.scala: ########## @@ -0,0 +1,253 @@ +/* + * 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 checkAnswerBothModes( + left: DataFrame, + right: DataFrame, + plan: (SparkPlan, SparkPlan) => SparkPlan, + expected: Seq[Row]): Unit = { + Seq(true, false).foreach { codegen => + withSQLConf( + org.apache.spark.sql.internal.SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString, + org.apache.spark.sql.internal.SQLConf.CODEGEN_FALLBACK.key -> "false") { + checkAnswer2(left, right, plan, expected) + } + } + } + + private def checkPointJoin( + left: DataFrame, + right: DataFrame, + pointOnLeft: Boolean, + buildSide: BuildSide, + joinType: JoinType, + lowInclusive: Boolean, + highInclusive: Boolean, + expected: Seq[Row], + highFirst: Boolean = false): Unit = { + checkAnswerBothModes(left, right, (leftPlan: SparkPlan, rightPlan: SparkPlan) => { + val pointPlan = if (pointOnLeft) leftPlan else rightPlan + val boundPlan = if (pointOnLeft) rightPlan else leftPlan + val point = pointPlan.output.head + val bounds = boundPlan.output + val pointKeys = point :: Nil + val buildIsPoint = buildSide match { + case BuildRight => !pointOnLeft + case BuildLeft => pointOnLeft + } + val (buildKeys, streamedKeys): (Seq[Expression], Seq[Expression]) = + if (buildIsPoint) (pointKeys, bounds) else (bounds, pointKeys) + rangeJoinExec( + leftPlan, rightPlan, buildSide, joinType, buildKeys, streamedKeys, + pointCondition(point, bounds(0), bounds(1), lowInclusive, highInclusive, highFirst)) + }, expected) + } + + test("point-in-range keeps boundary inclusivity on the condition") { + val lowInclHighExcl = Seq((0, 2, 1), (1, 5, 1), (1, 5, 3)).map(Row.fromTuple) + val bothExcl = Seq((0, 2, 1), (1, 5, 3)).map(Row.fromTuple) + val bothIncl = Seq((1, 0, 1), (1, 0, 2), (1, 1, 5), (3, 1, 5)).map(Row.fromTuple) + + Seq(false, true).foreach { highFirst => + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildRight, Inner, + lowInclusive = true, highInclusive = false, lowInclHighExcl, highFirst) + } + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildRight, Inner, + lowInclusive = false, highInclusive = false, bothExcl) + checkPointJoin( + points, intervals1, pointOnLeft = true, BuildRight, Inner, + lowInclusive = true, highInclusive = true, bothIncl) + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildLeft, Inner, + lowInclusive = false, highInclusive = false, bothExcl) + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildLeft, Inner, + lowInclusive = true, highInclusive = false, lowInclHighExcl) + } + + test("outer, semi, and anti preserve unmatched streamed rows") { + // low <= point && point < high. (-1, 0) and (0, 1) contain no point. + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildRight, LeftOuter, + lowInclusive = true, highInclusive = false, + Seq((-1, 0, null), (0, 1, null), (0, 2, 1), (1, 5, 1), (1, 5, 3)).map(Row.fromTuple)) + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildLeft, RightOuter, + lowInclusive = true, highInclusive = false, + Seq((null, null, -3), (0, 2, 1), (1, 5, 1), (1, 5, 3), (null, null, 6)).map(Row.fromTuple)) + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildRight, LeftSemi, + lowInclusive = true, highInclusive = false, + Seq((0, 2), (1, 5)).map(Row.fromTuple)) + checkPointJoin( + intervals1, points, pointOnLeft = false, BuildRight, LeftAnti, + lowInclusive = true, highInclusive = false, + Seq((-1, 0), (0, 1)).map(Row.fromTuple)) + } + + test("a null streamed bound is not a match") { + val nullInterval = spark.createDataFrame( + sparkContext.parallelize(Seq(Row(null, 5))), + new StructType().add("low1", IntegerType).add("high1", IntegerType)) + checkPointJoin( + nullInterval, points, pointOnLeft = false, BuildRight, LeftOuter, + lowInclusive = true, highInclusive = false, Seq(Row(null, 5, null))) + checkPointJoin( + nullInterval, points, pointOnLeft = false, BuildRight, LeftSemi, + lowInclusive = true, highInclusive = false, Seq.empty) + checkPointJoin( + nullInterval, points, pointOnLeft = false, BuildRight, LeftAnti, + lowInclusive = true, highInclusive = false, Seq(Row(null, 5))) + } + + test("partial range scans upTo or from according to the lower bound") { + // a.lo < b.hi builds the high side and scans from the stream key. + // a.lo > b.hi builds the low side and scans up to the stream key. + def check(rangeJoin: RangeJoin, condition: (SparkPlan, SparkPlan) => Expression, + expected: Seq[Row]): Unit = { + checkAnswerBothModes(intervals1, intervals2, (left: SparkPlan, right: SparkPlan) => { + rangeJoinExec( + left, right, BuildRight, Inner, Review Comment: **Nit (P3):** Both cases in this helper hard-code `BuildRight`, and the SQL suite's only BuildLeft partial-range case uses `LessPartialRangeJoin`. The supported `GreaterPartialRangeJoin` plus `BuildLeft` combination selects the opposite point-index scan arm, so swapping `from` and `upTo` there can produce wrong rows without any generated or interpreted result assertion failing. **Verification:** - **Regression:** Use asymmetric point keys and bounds so selecting from instead of upTo, or vice versa, changes the asserted result for GreaterPartialRangeJoin with BuildLeft. ########## sql/core/src/test/scala/org/apache/spark/sql/RangeJoinSQLSuite.scala: ########## @@ -0,0 +1,560 @@ +/* + * 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) + } + + private def setupDateAndTimestampViews(): Unit = { + sql( + """ + |CREATE OR REPLACE TEMP VIEW dt_ranges(lo, hi) AS VALUES + | (DATE'2024-01-01', DATE'2024-01-31'), + | (DATE'2024-06-01', DATE'2024-06-30') + |""".stripMargin) + sql( + """ + |CREATE OR REPLACE TEMP VIEW dt_points(d) AS VALUES + | (DATE'2024-01-15'), + | (DATE'2024-05-01'), + | (DATE'2024-06-15') + |""".stripMargin) + sql( + """ + |CREATE OR REPLACE TEMP VIEW ts_ranges(lo, hi) AS VALUES + | (TIMESTAMP'2024-01-01 00:00:00', TIMESTAMP'2024-01-01 12:00:00') + |""".stripMargin) + sql( + """ + |CREATE OR REPLACE TEMP VIEW ts_points(t) AS VALUES + | (TIMESTAMP'2024-01-01 06:00:00'), + | (TIMESTAMP'2024-01-01 18:00:00') + |""".stripMargin) + } + + test("range join is disabled by default and falls back to nested loop join") { + setupIpLookupViews() + val df = sql(s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON $pointCond") + assert(findRangeJoin(df.queryExecution.executedPlan).isEmpty) + assert(findBroadcastNestedLoopJoin(df.queryExecution.executedPlan).size == 1) + } + + test("point-in-range matches the nested loop result") { + setupIpLookupViews() + val query = s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON $pointCond" + val expected = sql(query).collect() + withRangeJoin() { + val df = sql(query) + assert(findRangeJoin(df.queryExecution.executedPlan).size == 1) + checkAnswer(df, expected) + } + } + + test("BETWEEN is planned as a point-in-range join") { + setupIpLookupViews() + val query = + "SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l " + + "ON s.ip_add_int BETWEEN l.BEGIN_IP2LONG AND l.END_IP2LONG" + val expected = sql(query).collect() + withRangeJoin(SQLConf.ALWAYS_INLINE_COMMON_EXPR.key -> "false") { + val df = sql(query) + val planned = findRangeJoin(df.queryExecution.executedPlan) + assert(planned.size == 1) + // BETWEEN lowers to one point expression. The other side keeps (low, high). + val keyCounts = Set(planned.head.leftKeys.length, planned.head.rightKeys.length) + assert(keyCounts == Set(1, 2)) + checkAnswer(df, expected) + } + } + + test("hints select range join, nested loop, or cartesian") { + setupIpLookupViews() + setupIntervalViews() + val expected = sql( + s"SELECT s.id, l.zip_id FROM signin_ip s JOIN ip_lookup l ON $pointCond").collect() + withRangeJoin() { + val broadcast = sql( Review Comment: **Nit (P3):** This suite never enters the distinct `hintOnly=true` path with both sides carrying `BROADCAST`. A regression can therefore choose the wrong side when both qualify, or fail to discard a preserved side that `BroadcastRangeJoinExec` cannot build, while the one-sided-hint and unhinted-size assertions still pass. **Verification:** - **Regression:** Assert range-plan presence and exact build side for asymmetric dual-hinted inner and stream-preserving joins, with fixture sizes chosen to distinguish the branches. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeBroadcastMode.scala: ########## @@ -0,0 +1,114 @@ +/* + * 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 + +/** + * 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 projection = new InterpretedProjection(buildKeys) + val lowOf = RangeIndex.getValue(buildKeys.head.dataType, 0) + + indexKind match { + case IntervalIndexKind => + val highOf = + if (buildKeys.length == 1) lowOf else RangeIndex.getValue(buildKeys(1).dataType, 1) + val intervals = rows.map { r => Review Comment: **Non-blocking (P2):** Both index branches first consume the build iterator into a full projected tuple array, then immediately create a second full retained array during filtering or normalization; `sizeHint` is unused. For a large hinted build this keeps two O(B) arrays and replacement tuples live at the broadcast-construction boundary, adding peak heap and GC pressure beyond the final relation and configured size gate. See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala: ########## @@ -421,6 +422,236 @@ 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 the point once and `(low, high)`. The index probes + * that one value as both ends of the window. 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)) + val rangePreds = predicates.collect { + case RangePredicate(dt, low, high) if RangePredicate.supportedType(dt) => + (dt, low, high) + } + if (rangePreds.isEmpty) { + None + } else { + pointInRange(rangePreds, left, right, joinType) + .orElse(intervalOverlap(rangePreds, left, right, joinType)) + .orElse(partialRange(rangePreds, 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( + rangePreds: Seq[(DataType, Expression, Expression)], + left: LogicalPlan, + right: LogicalPlan, + joinType: JoinType): Option[ReturnType] = { + rangePreds.combinations(2).flatMap { + case Seq((d1, l1, h1), (d2, l2, h2)) if d1 == d2 => + val pointCandidate = + if (h1.semanticEquals(l2) && !l1.semanticEquals(h2) && l1.dataType == h2.dataType) { + place(Seq(l2), Seq(l1, h2), left, right) + } else if (l1.semanticEquals(h2) && !h1.semanticEquals(l2) && + l2.dataType == h1.dataType) { + place(Seq(l1), Seq(l2, h1), left, right) + } else { + None + } + pointCandidate.collect { + case (leftKeys, rightKeys) if supportedKeys(leftKeys, rightKeys) => + (left, right, leftKeys, rightKeys, joinType, PointInRangeJoin) + } + case _ => None + }.nextOption() + } + + /** + * Puts `leftExprs` on the left child when that child can evaluate all of them + * and the right child can evaluate `rightExprs`. Otherwise swaps, when the + * opposite assignment fits. A key is a column of one input, so at most one + * assignment succeeds. + */ + private def place( + leftExprs: Seq[Expression], + rightExprs: Seq[Expression], + left: LogicalPlan, + right: LogicalPlan): Option[(Seq[Expression], Seq[Expression])] = { + if (leftExprs.forall(canEvaluate(_, left)) && rightExprs.forall(canEvaluate(_, right))) { + Some((leftExprs, rightExprs)) + } else if (leftExprs.forall(canEvaluate(_, right)) && + rightExprs.forall(canEvaluate(_, left))) { + Some((rightExprs, leftExprs)) + } else { + None + } + } + + private def intervalOverlap( + rangePreds: Seq[(DataType, Expression, Expression)], + left: LogicalPlan, + right: LogicalPlan, + joinType: JoinType): Option[ReturnType] = { + rangePreds.combinations(2).flatMap { + case Seq((d1, l1, h1), (d2, l2, h2)) if d1 == d2 => + // `(l1, h2)` is one interval and `(l2, h1)` is the other. + val overlapCandidate = place(Seq(l1, h2), Seq(l2, h1), left, right) + overlapCandidate.collect { + case (leftKeys, rightKeys) + if !leftKeys.head.semanticEquals(leftKeys(1)) && + !rightKeys.head.semanticEquals(rightKeys(1)) && + leftKeys.head.dataType == leftKeys(1).dataType && + rightKeys.head.dataType == rightKeys(1).dataType && + supportedKeys(leftKeys, rightKeys) => + (left, right, leftKeys, rightKeys, joinType, IntervalOverlapJoin) + } + case _ => None + }.nextOption() + } + + private def supportedKeys(keys: Seq[Expression]*): Boolean = { + keys.flatten.forall { key => + key.references.nonEmpty && key.deterministic && RangePredicate.supportedType(key.dataType) + } + } + + private def partialRange( + rangePreds: Seq[(DataType, Expression, Expression)], + left: LogicalPlan, + right: LogicalPlan, + joinType: JoinType): Option[ReturnType] = { + rangePreds.iterator.flatMap { case (_, low, high) => + if (low.semanticEquals(high) || low.dataType != high.dataType || + !supportedKeys(Seq(low, high))) { + None + } else { + // Unswapped, the left child holds `low`. Swapped, it holds `high`. + place(Seq(low), Seq(high), left, right).map { case (leftKeys, rightKeys) => + val rangeJoin = if (canEvaluate(low, left)) { + LessPartialRangeJoin + } else { + GreaterPartialRangeJoin + } + (left, right, leftKeys, rightKeys, joinType, rangeJoin) + } + } + }.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. + * Other orderable types (arrays, intervals, timestamp-with-nanos, UDTs) stay a + * nested loop join: the index getter does not produce the value codegen compares. Review Comment: **Nit (P3):** The named types are rejected by the explicit `supportedType` allowlist, but the stated getter-versus-codegen representation mismatch is not the actual boundary: their `InternalRow` accessors and generated values use the corresponding Catalyst physical representations. This rationale can misdirect later work on which types are deliberately supported. **Verification:** - **Inspection:** Check the rationale against InternalRow accessors, generated physical values, PhysicalDataType ordering, and every explicitly named excluded family. -- 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]
