wangyum commented on code in PR #59027: URL: https://github.com/apache/spark/pull/59027#discussion_r4429515336
########## 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: Fixed: both comments now credit analyzer comparison-type coercion and ExtractRangeJoinKeys' own same-type guards instead of RangePredicate. ########## 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: Fixed: the comments now say 5%/10%/25% prune 231/279/124 multi-node subtrees respectively while 50%/90% prune none. ########## 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: Added RangeJoinSQLSuite tests pinning that outputPartitioning/outputOrdering follow the streamed side for all 6 (joinType, buildSide) pairs supports() admits. -- 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]
