cloud-fan commented on code in PR #59027: URL: https://github.com/apache/spark/pull/59027#discussion_r4420441251
########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/RangeRelation.scala: ########## @@ -0,0 +1,1222 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.joins + +import java.io.{Externalizable, ObjectInput, ObjectOutput} + +import scala.collection.mutable.ArrayBuffer + +import com.esotericsoftware.kryo.{Kryo, KryoSerializable} +import com.esotericsoftware.kryo.io.{Input, Output} + +import org.apache.spark.SparkException +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{BoundReference, Expression} +import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildSide} +import org.apache.spark.sql.catalyst.planning.{IntervalOverlapJoin, PartialRangeJoin, PointInRangeJoin, RangeJoinKind} +import org.apache.spark.sql.catalyst.plans.physical.BroadcastMode +import org.apache.spark.sql.catalyst.types.PhysicalDataType +import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DateType, Decimal, DecimalType, IntegerType, LongType, ShortType, TimestampNTZType, TimestampType} +import org.apache.spark.util.{KnownSizeEstimation, SizeEstimator, Utils} + +/** + * Broadcast payload of a range join. An index only names candidate rows. + * [[BroadcastRangeJoinExec]] evaluates the original join condition to filter + * candidates. Keys that map to one order-preserving long are stored raw and + * compared unboxed ([[LongIntervalIndex]], [[LongPointIndex]]); every other + * supported type is compared as boxed values ([[IntervalIndex]], + * [[PointIndex]]). [[RangeRelationBroadcastMode]] picks the family from the + * build keys' data types. + * + * - The interval indexes answer "which ranges overlap this window?" + * - The point indexes answer "which points lie on this side of a bound?" + */ +private[execution] sealed trait RangeRelation extends KnownSizeEstimation with Serializable { + def keyType: DataType +} + +/** + * Which [[RangeRelation]] a broadcast materializes. + * + * A point index has no direction, so one broadcast of a column serves both + * `<` and `>`. The join records which side holds the lower bound and scans + * accordingly. + */ +private[execution] sealed abstract class RangeIndexKind { + def matches(rangeJoinKind: RangeJoinKind): Boolean = RangeIndexKind(rangeJoinKind) == this +} + +private[execution] case object IntervalIndexKind extends RangeIndexKind + +private[execution] case object PointIndexKind extends RangeIndexKind + +private[execution] object RangeIndexKind { + def apply(rangeJoinKind: RangeJoinKind): RangeIndexKind = rangeJoinKind match { + case PointInRangeJoin | IntervalOverlapJoin => IntervalIndexKind + case _: PartialRangeJoin => PointIndexKind + } +} + +/** + * The boxed-key index family's shared ordering cache, derived from `keyType` + * on first comparison and kept out of the serialized form, so each executor + * re-derives its own instance instead of receiving one per broadcast. The + * long index family compares raw keys and has no ordering to cache. + */ +private[joins] sealed trait BoxedKeyOrdering extends RangeRelation { + @transient + @volatile private[this] var _ordering: Ordering[Any] = _ + + private[joins] def ordering: Ordering[Any] = { + var ord = _ordering + if (ord == null) { + ord = PhysicalDataType.ordering(keyType) + _ordering = ord + } + ord + } +} + +/** + * The long index family's shared key conversion, derived from `keyType` on + * first probe and kept out of the serialized form, so each executor derives + * its own converter after deserialization. The interpreted probe path calls + * [[toLong]] once per streamed key; generated probe code keeps its own + * conversion, emitted at plan time. + */ +private[joins] sealed trait LongKeyConversion extends RangeRelation { + @transient + @volatile private[this] var _toLong: Any => Long = _ + + /** + * One evaluated streamed key to its order-preserving long. `RangePredicate` + * gives both sides of a range predicate one type after analysis, so the Review Comment: **Nit (P3):** `RangePredicate` only reads `low.dataType`; it doesn't unify anything. Within one predicate, the analyzer's comparison coercion gives both operands one type. Across predicates it is `ExtractRangeJoinKeys`' guards that do it: `d1 == d2` in `pairJoin`, and the shape checks `l1.dataType == h2.dataType` and `l2.dataType == h1.dataType`. That cross-predicate part is what this comment and the one on `canMapKeysToLong` rely on, because an interval-overlap join's build keys `(l1, h2)` and streamed keys `(l2, h1)` come from two different predicates. If those guards were relaxed, a DATE predicate could pair with a TIMESTAMP one: both map to longs, so `canMapKeysToLong` would pick the long family from the build keys alone, and `toLong` would convert every streamed key with this index's single `keyType`. Could both comments name the analyzer's coercion and the extractor's type guards instead? **Verification:** - **Inspection:** The comments on the long-key conversion and the build-key long test name the analyzer's comparison type coercion and the extractor's same-type guards as what keeps build and streamed keys on one type, and they match the guards ExtractRangeJoinKeys applies for point-in-range, interval overlap, and single inequalities. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala: ########## @@ -1220,6 +1220,35 @@ object SQLConf { .bytesConf(ByteUnit.BYTE) .createWithDefaultString("10MB") + val BROADCAST_RANGE_JOIN_ENABLED = buildConf("spark.sql.join.broadcastRangeJoin.enabled") + .internal() + .doc("When true, joins with range predicates (such as point-in-range, interval-overlap, " + + "or single inequalities) are planned as broadcast range joins instead of broadcast " + + "nested loop joins if the build side is eligible for broadcast. Supported join types " + + "include inner, left outer, right outer, left semi, and left anti. Broadcast range joins " + + "build an index on the broadcast side to significantly accelerate selective range " + + "probes, but planning does not consult overlap, probe size, or index-size estimates, " + + "so three trade-offs apply uniformly. First, a build where some, but not all, " + + "intervals overlap each probe can run slower than a nested loop join, and how much " + + "overlap triggers that depends heavily on the key family: boxed keys " + + "(double, string, wide decimal) slow down at far lower overlap than long-specialized " + Review Comment: **Nit (P3):** `RangeIndex.longKey` routes `TimestampNTZType` to the long indexes (`case _: LongType | _: TimestampType | _: TimestampNTZType`) and returns `None` for `FloatType` and `BinaryType`, so those two stay boxed. None of the three appears in either list here. This doc is the only guidance on when to turn the flag on, so a user with TIMESTAMP_NTZ, FLOAT, or BINARY range keys can't tell which slowdown profile applies. Could timestamp_ntz join the long-specialized list, and float and binary the boxed one? This extends an [existing discussion](https://github.com/apache/spark/pull/59027#discussion_r4367132555). **Verification:** - **Inspection:** Each supported range-key type (boolean, byte, short, int, long, float, double, date, timestamp, timestamp_ntz, binary, string, and decimal at both precision ranges) is named in the config doc's family list that matches the index family RangeIndex.longKey selects for it. ########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastRangeJoinExec.scala: ########## @@ -0,0 +1,589 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.joins + +import org.apache.spark.SparkException +import org.apache.spark.broadcast.Broadcast +import org.apache.spark.rdd.RDD +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.expressions.codegen._ +import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide} +import org.apache.spark.sql.catalyst.planning.RangeJoinKind +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.physical.{BroadcastDistribution, Distribution, Partitioning, UnspecifiedDistribution} +import org.apache.spark.sql.execution.{CodegenSupport, ExplainUtils, SparkPlan} +import org.apache.spark.sql.execution.metric.SQLMetrics + +/** + * Performs a range join on two tables. Supported types are inner, left outer, right outer, + * left semi, and left anti. The preserved side is streamed: unmatched build rows are not + * emitted, because the broadcast is copied to every partition. See `ExtractRangeJoinKeys` + * for the exact set of join-condition shapes that are recognized as a range join. + * + * The build side is an index. Point-in-range and interval overlap both build an + * [[IntervalIndex]] and probe it with `overlapping`. A single inequality builds a + * [[PointIndex]] and scans the side of the bound the build side holds. Keys that + * map to one order-preserving long instead build [[LongIntervalIndex]] or + * [[LongPointIndex]], so their probes compare raw longs without boxing. Either + * probe returns a superset. A candidate is kept only when the original join + * condition is true, so inclusivity stays on the predicate. Heavy overlap can + * make one probe return almost every build row. + */ +case class BroadcastRangeJoinExec( + left: SparkPlan, + right: SparkPlan, + buildSide: BuildSide, + joinType: JoinType, + leftKeys: Seq[Expression], + rightKeys: Seq[Expression], + condition: Option[Expression], + rangeJoinKind: RangeJoinKind) + extends JoinCodegenSupport { + + if (condition.isEmpty) { + throw SparkException.internalError("BroadcastRangeJoinExec requires a join condition.") + } + + override lazy val metrics = Map( + "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output rows")) + + /** BuildRight means the right relation is the broadcast relation. */ + private val (streamed, broadcast, streamedKeys, buildKeys) = buildSide match { + case BuildRight => (left, right, leftKeys, rightKeys) + case BuildLeft => (right, left, rightKeys, leftKeys) + } + + override def simpleStringWithNodeId(): String = { + val opId = ExplainUtils.getOpId(this) + s"$nodeName $joinType $buildSide ($opId)".trim + } + + override def requiredChildDistribution: Seq[Distribution] = { + val mode = RangeRelationBroadcastMode( + BindReferences.bindReferences(buildKeys, broadcast.output), + RangeIndexKind(rangeJoinKind)) + buildSide match { + case BuildLeft => + BroadcastDistribution(mode) :: UnspecifiedDistribution :: Nil + case BuildRight => + UnspecifiedDistribution :: BroadcastDistribution(mode) :: Nil + } + } + + override def outputPartitioning: Partitioning = streamed.outputPartitioning Review Comment: **Nit (P3):** These overrides report the streamed child's partitioning and ordering unconditionally. That is correct today only because `supports()` admits just the stream-preserving `(joinType, buildSide)` pairs; `BroadcastNestedLoopJoinExec` makes the same claims only for those pairs and reports `UnknownPartitioning` and no ordering otherwise. No range-join test puts an aggregate, sort-merge join, or sort on the streamed key above the join. So if `supports()` were widened to a pair that also emits unmatched build rows, such as a left outer join that builds the left side, these claims would cover null-padded rows that don't follow the streamed partitioning, and the planner would skip a needed shuffle and return wrong groups. If an override instead stopped reporting the streamed side's properties, every such plan would gain a needless shuffle or sort. Neither change would fail a current test. Could a range-join counterpart of `JoinSuite`'s SPARK-34593 test, run for both build side s, pin both claims? **Verification:** - **Regression:** The assertions fail if the broadcast range join reports the build side's or the left child's partitioning or ordering instead of the streamed side's, for both the build-right and build-left cases. - **Behavior:** With the flag on and off, the queries over bucketed streamed inputs return identical answers, so preserving the streamed side's partitioning and ordering does not change results. ########## sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/RangeJoinBenchmark.scala: ########## @@ -0,0 +1,526 @@ +/* + * 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.benchmark + +import java.sql.Date + +import org.apache.spark.benchmark.Benchmark +import org.apache.spark.sql.{Column, DataFrame} +import org.apache.spark.sql.catalyst.planning.{GreaterPartialRangeJoin, IntervalOverlapJoin, LessPartialRangeJoin, PointInRangeJoin, RangeJoinKind} +import org.apache.spark.sql.execution.joins.{BroadcastNestedLoopJoinExec, BroadcastRangeJoinExec} +import org.apache.spark.sql.functions.{broadcast, col, date_add, lit, lpad, when} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.StringType + +/** + * Benchmark broadcast range join against broadcast nested loop join. + * + * Long keys cross every supported shape with every supported join type. + * Inner joins broadcast each side once. Left anti probes values that match + * nothing. A partial range whose bound sits at the median is not selective. + * Wide interval overlap is mostly selective: a probe overlaps a build + * interval only when |i - j| <= 20, so about 2% of the 100,000 probes hit + * up to 41 of the 2,000 build intervals each and the rest hit nothing. + * Point-in-range inner is repeated for date, timestamp, decimal, + * and UNICODE_CI strings. + * + * More cases probe the cost model directly: a streamed side of 1, 10, 100, + * 1000, 5000, or 10000 rows, where range join pays to sort the whole build + * side instead of one nested-loop scan, and a fully overlapping inner join, + * where every pair matches and the index prunes nothing. + * + * Staggered-overlap builds cover the regime between those: a fraction of build + * intervals -- from a small minority at 5%/10%/25% to half or more at 50%/90% -- + * overlap each probe window, miss intervals are interleaved so no subtree's + * max high falls under the probe, and the build has no common core, so the + * tree is walked without pruning and evaluated per candidate. Long and double + * (boxed) keys run at 5%, 10%, 25%, 50%, and 90% overlap. + * + * To run this benchmark: + * {{{ + * 1. build/sbt "sql/Test/runMain org.apache.spark.sql.execution.benchmark.RangeJoinBenchmark" + * 2. generate result: + * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt \ + * "sql/Test/runMain org.apache.spark.sql.execution.benchmark.RangeJoinBenchmark" + * Results will be written to "benchmarks/RangeJoinBenchmark-results.txt". + * }}} + */ +object RangeJoinBenchmark extends SqlBasedBenchmark { + + // An optional regex in the environment variable RANGE_BENCH_FILTER runs only the + // sections whose name matches, e.g. RANGE_BENCH_FILTER=".*tiny probe.*|.*full overlap.*". + // A workaround for sbt splitting runMain arguments on spaces. + private var nameFilter: Option[String] = Option(System.getenv("RANGE_BENCH_FILTER")) + + private def shouldRun(name: String): Boolean = nameFilter.forall(name.matches) + + private val buildRows = 4000 + private val probeRows = 400000 + private val width = 10 + // Wide overlap is measured over 100,000 probes and 2,000 build intervals. + private val heavyBuildRows = 2000 + private val heavyProbeRows = 100000 + private val halfBuildRows = 1000 + private val halfProbeRows = 10000 + private val wideSpan = 20 * width + + private val epoch = lit(Date.valueOf("1970-01-01")) + + // [0, buildRows * width). Each range covers `width` values and none overlap. + private lazy val ranges: DataFrame = cached(spark.range(buildRows).select( + (col("id") * width).as("lo"), + (col("id") * width + (width - 1)).as("hi"))) + + // One hit per point: the value falls in exactly one range above. + private lazy val points: DataFrame = cached(spark.range(probeRows).select( + (col("id") % (buildRows.toLong * width)).as("p"))) + + // Above every range, so a point-in-range probe matches nothing. + private lazy val misses: DataFrame = cached(spark.range(probeRows).select( + (lit(buildRows.toLong * width) + (col("id") % width)).as("p"))) + + // Short intervals. A probe overlaps the build intervals whose starts are near it. + private lazy val probeIntervals: DataFrame = + cached(intervals(probeRows, "a", width + width / 2)) + private lazy val buildIntervals: DataFrame = + cached(intervals(buildRows, "b", width + width / 2)) + // Starts after every build interval. + private lazy val missIntervals: DataFrame = cached(spark.range(probeRows).select( + (lit(buildRows.toLong * width + width) + col("id") * width).as("alo"), + (lit(buildRows.toLong * width + width * 2) + col("id") * width).as("ahi"))) + + // Mostly selective: a probe overlaps a build interval only when |i - j| <= 20, + // so about 2% of the 100,000 probes hit up to 41 of the 2,000 build intervals + // each and the rest hit nothing (see the class doc above). + private lazy val wideProbe: DataFrame = cached(intervals(heavyProbeRows, "a", wideSpan)) + private lazy val wideBuild: DataFrame = cached(intervals(heavyBuildRows, "b", wideSpan)) + + // Build keys 0 .. buildRows - 1. High probes sit at the top, low probes at the bottom, + // so a selective inequality returns only a few candidates. + private lazy val bounds: DataFrame = cached(spark.range(buildRows).select(col("id").as("b"))) + private lazy val highProbes: DataFrame = cached(spark.range(probeRows).select( + (lit(buildRows - 1) - (col("id") % width)).as("a"))) + private lazy val lowProbes: DataFrame = cached(spark.range(probeRows).select( + (col("id") % width).as("a"))) + // Strictly above every bound, so `a < b` matches nothing. + private lazy val lessMisses: DataFrame = cached(spark.range(probeRows).select( + lit(buildRows.toLong).as("a"))) + // `a > b` matches nothing when every bound is non-negative. + private lazy val greaterMisses: DataFrame = cached(spark.range(probeRows).select(lit(0L).as("a"))) + + // `a` is the median bound, so each inequality matches about half the build side. + private lazy val halfBounds: DataFrame = + cached(spark.range(halfBuildRows).select(col("id").as("b"))) + private lazy val halfProbes: DataFrame = cached(spark.range(halfProbeRows).select( + lit(halfBuildRows.toLong / 2).as("a"))) + + // M approx 1, 10, 100, 1000, 5000, and 10000. The build side still has to be + // sorted to answer a few streamed probes, so this weighs that one-time O(N log N) + // cost against one O(N) nested-loop scan of the same build side, and shows how + // fast that cost amortizes + // as M grows just past one. + private val tinyProbeRowCounts = Seq(1, 10, 100, 1000, 5000, 10000) + private lazy val tinyPoints: Map[Int, DataFrame] = tinyProbeRowCounts.map { n => + n -> cached(spark.range(n).select((col("id") % (buildRows.toLong * width)).as("p"))) + }.toMap + private lazy val tinyBounds: Map[Int, DataFrame] = tinyProbeRowCounts.map { n => + n -> cached(spark.range(n).select(lit(buildRows.toLong / 2).as("a"))) + }.toMap + + // K approx N. Every build range covers the whole probe domain, so every probe + // matches every build row: an inner join emits all N * M pairs and the index + // prunes nothing, only adding its build cost on top of the join itself. + private val fullBuildRows = 1000 + private val fullProbeRows = 2000 + private lazy val fullRanges: DataFrame = cached(spark.range(fullBuildRows).select( + lit(0L).as("lo"), + lit(fullBuildRows.toLong * width).as("hi"))) + private lazy val fullPoints: DataFrame = cached(spark.range(fullProbeRows).select( + (col("id") % (fullBuildRows.toLong * width)).as("p"))) + + // Sweep the whole broadcast budget for cache behavior: the walk touches + // three long arrays far beyond any cache, and the sort and copy dominate + // the build side. + private val largeBuildRows = 1000000 + private val largeProbeRows = 10000 + private lazy val largeRanges: DataFrame = cached(spark.range(largeBuildRows).select( + (col("id") * width).as("lo"), + (col("id") * width + (width - 1)).as("hi"))) + private lazy val largePoints: DataFrame = cached(spark.range(largeProbeRows).select( + ((col("id") * width + width / 2) % lit(largeBuildRows.toLong * width)).as("p"))) + + // Staggered high-overlap builds. Every miss interval ends before the probe + // window starts and every keep interval spans it, but the two interleave in + // low order, so no subtree's max high falls under the probe and the walk Review Comment: **Nit (P3):** This held when the only staggered builds were 50% and 90%, where every miss sits alone between spans. The new 5%, 10%, and 25% builds put spans on every 20th, 10th, and 4th id, so misses come in runs of 19, 9, and 3. Every miss ends by 28,990, below each probe's low of 30,000 to 30,002, so `advance()` drops an all-miss subtree at its root; at 5% the subtree over ids [8, 15) tops out at 9,140. In the 2,000-row build's midpoint tree that happens to 231, 279, and 124 multi-node subtrees at 5%, 10%, and 25%, and to none at 50% or 90%. The rows are still a realistic interleaving, but a reader takes them, and the config-doc boundaries they back, as timing a walk with no pruning. The class doc's "walked without pruning" has the same problem. Could both comments say what each overlap level actually prunes? **Verification:** - **Inspection:** For each staggered build (5%, 10%, 25%, 50%, and 90%), the benchmark comments agree with which subtrees the interval walk prunes for the committed generators: subtrees inside runs of miss intervals at 5%, 10%, and 25%, and only isolated miss leaves at 50% and 90%. -- 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]
