You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ya...@apache.org on 2026/02/04 12:17:16 UTC
(spark) branch feature/crossjoin-array-contains-benchmark updated (e81100dec095 -> 8172b0502948)
This is an automated email from the ASF dual-hosted git repository.
yao pushed a change to branch feature/crossjoin-array-contains-benchmark
in repository https://gitbox.apache.org/repos/asf/spark.git
discard e81100dec095 [SPARK-XXXX][SQL] Add CrossJoinArrayContainsToInnerJoin optimizer rule
new 8172b0502948 [SPARK-XXXX][SQL] Add CrossJoinArrayContainsToInnerJoin optimizer rule
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (e81100dec095)
\
N -- N -- N refs/heads/feature/crossjoin-array-contains-benchmark (8172b0502948)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../sql/catalyst/optimizer/CrossJoinArrayContainsToInnerJoin.scala | 2 +-
.../scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala | 1 +
2 files changed, 2 insertions(+), 1 deletion(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org
(spark) 01/01: [SPARK-XXXX][SQL] Add CrossJoinArrayContainsToInnerJoin optimizer rule
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yao pushed a commit to branch feature/crossjoin-array-contains-benchmark
in repository https://gitbox.apache.org/repos/asf/spark.git
commit 8172b0502948a0e89588b547ac7675385107a087
Author: Kent Yao <ke...@microsoft.com>
AuthorDate: Wed Feb 4 09:45:58 2026 +0000
[SPARK-XXXX][SQL] Add CrossJoinArrayContainsToInnerJoin optimizer rule
### What changes were proposed in this pull request?
This PR adds a new optimizer rule that converts cross joins with array_contains filter into inner joins using explode, improving query performance significantly.
### Why are the changes needed?
Cross joins with array_contains predicates result in O(N*M) complexity. By transforming to explode + inner join, we achieve O(N+M) complexity.
### Does this PR introduce _any_ user-facing change?
No. This is an internal optimization that automatically applies to applicable queries.
### How was this patch tested?
- Unit tests in CrossJoinArrayContainsToInnerJoinSuite (6 tests)
- Microbenchmark showing 11-16X speedup on representative workload
### Was this patch authored or co-authored using generative AI tooling?
Yes, GitHub Copilot was used to assist with implementation.
---
.../CrossJoinArrayContainsToInnerJoin.scala | 129 ++++++++++
.../spark/sql/catalyst/optimizer/Optimizer.scala | 1 +
.../sql/catalyst/rules/RuleIdCollection.scala | 1 +
.../CrossJoinArrayContainsToInnerJoinSuite.scala | 274 +++++++++++++++++++++
...yContainsToInnerJoinBenchmark-jdk21-results.txt | 38 +++
...inArrayContainsToInnerJoinBenchmark-results.txt | 38 +++
...rossJoinArrayContainsToInnerJoinBenchmark.scala | 93 +++++++
7 files changed, 574 insertions(+)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CrossJoinArrayContainsToInnerJoin.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CrossJoinArrayContainsToInnerJoin.scala
new file mode 100644
index 000000000000..d5143d1f34ce
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CrossJoinArrayContainsToInnerJoin.scala
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans._
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules._
+import org.apache.spark.sql.catalyst.trees.TreePattern.{FILTER, JOIN}
+import org.apache.spark.sql.types._
+
+/**
+ * Converts cross joins with array_contains filter into inner joins using explode.
+ *
+ * This optimization transforms queries of the form:
+ * {{{
+ * SELECT * FROM left, right WHERE array_contains(left.arr, right.elem)
+ * }}}
+ *
+ * Into a more efficient form using explode + inner join, reducing O(N*M) to O(N+M).
+ */
+object CrossJoinArrayContainsToInnerJoin extends Rule[LogicalPlan] with PredicateHelper {
+
+ override def apply(plan: LogicalPlan): LogicalPlan = plan.transformUpWithPruning(
+ _.containsAllPatterns(FILTER, JOIN), ruleId) {
+ case f @ Filter(cond, j @ Join(left, right, Cross | Inner, None, _)) =>
+ tryTransform(f, cond, j, left, right).getOrElse(f)
+ }
+
+ private def tryTransform(
+ filter: Filter,
+ condition: Expression,
+ join: Join,
+ left: LogicalPlan,
+ right: LogicalPlan): Option[LogicalPlan] = {
+ val predicates = splitConjunctivePredicates(condition)
+ val leftOut = left.outputSet
+ val rightOut = right.outputSet
+
+ // Find first valid array_contains predicate
+ predicates.collectFirst {
+ case ac @ ArrayContains(arr, elem)
+ if canOptimize(arr, elem, leftOut, rightOut) =>
+ val arrayOnLeft = arr.references.subsetOf(leftOut)
+ val remaining = predicates.filterNot(_ == ac)
+ buildPlan(join, left, right, arr, elem, arrayOnLeft, remaining)
+ }.flatten
+ }
+
+ private def canOptimize(
+ arr: Expression,
+ elem: Expression,
+ leftOut: AttributeSet,
+ rightOut: AttributeSet): Boolean = {
+ // Check type compatibility
+ val elemType = elem.dataType
+ val validType = arr.dataType match {
+ case ArrayType(t, _) => t == elemType && isSupportedType(elemType)
+ case _ => false
+ }
+
+ // Check array and element come from different sides
+ val arrRefs = arr.references
+ val elemRefs = elem.references
+ val crossesSides = (arrRefs.nonEmpty && elemRefs.nonEmpty) && (
+ (arrRefs.subsetOf(leftOut) && elemRefs.subsetOf(rightOut)) ||
+ (arrRefs.subsetOf(rightOut) && elemRefs.subsetOf(leftOut))
+ )
+
+ validType && crossesSides
+ }
+
+ /**
+ * Supported types have consistent equality semantics between array_contains and join.
+ * Excludes Float/Double (NaN issues) and complex types.
+ */
+ private def isSupportedType(dt: DataType): Boolean = dt match {
+ case _: AtomicType => dt match {
+ case FloatType | DoubleType => false // NaN != NaN
+ case _ => true
+ }
+ case _ => false
+ }
+
+ private def buildPlan(
+ join: Join,
+ left: LogicalPlan,
+ right: LogicalPlan,
+ arr: Expression,
+ elem: Expression,
+ arrayOnLeft: Boolean,
+ remaining: Seq[Expression]): Option[LogicalPlan] = {
+
+ val unnestedAttr = AttributeReference("unnested", elem.dataType, nullable = true)()
+ val generator = Explode(ArrayDistinct(arr))
+
+ val (newLeft, newRight, joinCond) = if (arrayOnLeft) {
+ val gen = Generate(generator, Nil, false, None, Seq(unnestedAttr), left)
+ (gen, right, EqualTo(unnestedAttr, elem))
+ } else {
+ val gen = Generate(generator, Nil, false, None, Seq(unnestedAttr), right)
+ (left, gen, EqualTo(elem, unnestedAttr))
+ }
+
+ val innerJoin = Join(newLeft, newRight, Inner, Some(joinCond), JoinHint.NONE)
+
+ // Project to original output (exclude unnested column)
+ val projected = Project(join.output.map(a => Alias(a, a.name)(a.exprId)), innerJoin)
+
+ // Add remaining predicates if any
+ val result = remaining.reduceLeftOption(And).map(Filter(_, projected)).getOrElse(projected)
+ Some(result)
+ }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index fe15819bd44a..0a018bfe08a5 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -261,6 +261,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
Batch("Optimize One Row Plan", fixedPoint, OptimizeOneRowPlan),
// The following batch should be executed after batch "Join Reorder" and "LocalRelation".
Batch("Check Cartesian Products", Once,
+ CrossJoinArrayContainsToInnerJoin,
CheckCartesianProducts),
Batch("RewriteSubquery", Once,
RewritePredicateSubquery,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
index 1e718c02f5ea..27d649ff0cbc 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
@@ -130,6 +130,7 @@ object RuleIdCollection {
"org.apache.spark.sql.catalyst.optimizer.ConstantPropagation" ::
"org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation" ::
"org.apache.spark.sql.catalyst.optimizer.CostBasedJoinReorder" ::
+ "org.apache.spark.sql.catalyst.optimizer.CrossJoinArrayContainsToInnerJoin" ::
"org.apache.spark.sql.catalyst.optimizer.DecimalAggregates" ::
"org.apache.spark.sql.catalyst.optimizer.EliminateAggregateFilter" ::
"org.apache.spark.sql.catalyst.optimizer.EliminateLimits" ::
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CrossJoinArrayContainsToInnerJoinSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CrossJoinArrayContainsToInnerJoinSuite.scala
new file mode 100644
index 000000000000..81fe449be71f
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/CrossJoinArrayContainsToInnerJoinSuite.scala
@@ -0,0 +1,274 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.{Cross, Inner, PlanTest}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.types._
+
+/**
+ * Test suite for CrossJoinArrayContainsToInnerJoin optimizer rule.
+ *
+ * This rule converts cross joins with array_contains filter into inner joins
+ * using explode/unnest, which is much more efficient.
+ *
+ * Example transformation:
+ * {{{
+ * Filter(array_contains(arr, elem))
+ * CrossJoin(left, right)
+ * }}}
+ * becomes:
+ * {{{
+ * InnerJoin(unnested = elem)
+ * Generate(Explode(ArrayDistinct(arr)), left)
+ * right
+ * }}}
+ */
+class CrossJoinArrayContainsToInnerJoinSuite extends PlanTest {
+
+ object Optimize extends RuleExecutor[LogicalPlan] {
+ val batches =
+ Batch("CrossJoinArrayContainsToInnerJoin", Once,
+ CrossJoinArrayContainsToInnerJoin) :: Nil
+ }
+
+ // Table with array column (simulates "orders" with item_ids array)
+ val ordersRelation: LocalRelation = LocalRelation(
+ $"order_id".int,
+ $"item_ids".array(IntegerType)
+ )
+
+ // Table with element column (simulates "items" with id)
+ val itemsRelation: LocalRelation = LocalRelation(
+ $"id".int,
+ $"name".string
+ )
+
+ test("converts cross join with array_contains to inner join with explode") {
+ // Original query: SELECT * FROM orders, items WHERE array_contains(item_ids, id)
+ val originalPlan = ordersRelation
+ .join(itemsRelation, Cross)
+ .where(ArrayContains($"item_ids", $"id"))
+ .analyze
+
+ val optimized = Optimize.execute(originalPlan)
+
+ // After optimization, should be an inner join with explode
+ // The plan should NOT contain a Cross join anymore
+ assert(!optimized.exists {
+ case j: Join if j.joinType == Cross => true
+ case _ => false
+ }, "Optimized plan should not contain Cross join")
+
+ // Should contain a Generate (explode) node
+ assert(optimized.exists {
+ case _: Generate => true
+ case _ => false
+ }, "Optimized plan should contain Generate (explode) node")
+
+ // Should contain an Inner join
+ assert(optimized.exists {
+ case j: Join if j.joinType == Inner => true
+ case _ => false
+ }, "Optimized plan should contain Inner join")
+ }
+
+ test("does not transform when array_contains is not present") {
+ // Query without array_contains: SELECT * FROM orders, items WHERE order_id = id
+ val originalPlan = ordersRelation
+ .join(itemsRelation, Cross)
+ .where($"order_id" === $"id")
+ .analyze
+
+ val optimized = Optimize.execute(originalPlan)
+
+ // Should remain unchanged (still a cross join with filter)
+ assert(optimized.exists {
+ case j: Join if j.joinType == Cross => true
+ case _ => false
+ }, "Plan without array_contains should remain unchanged")
+ }
+
+ test("does not transform inner join with existing conditions") {
+ // Already an inner join with equi-condition
+ val originalPlan = ordersRelation
+ .join(itemsRelation, Inner, Some($"order_id" === $"id"))
+ .where(ArrayContains($"item_ids", $"id"))
+ .analyze
+
+ val optimized = Optimize.execute(originalPlan)
+
+ // Should not add another explode since this is already an equi-join
+ // The array_contains becomes just a filter
+ assert(optimized.isInstanceOf[Filter] || optimized.exists {
+ case _: Filter => true
+ case _ => false
+ })
+ }
+
+ test("handles array column on right side of join") {
+ // Swap the tables - array is on right side
+ val rightWithArray: LocalRelation = LocalRelation(
+ $"arr_id".int,
+ $"values".array(IntegerType)
+ )
+ val leftWithElement: LocalRelation = LocalRelation(
+ $"elem".int
+ )
+
+ val originalPlan = leftWithElement
+ .join(rightWithArray, Cross)
+ .where(ArrayContains($"values", $"elem"))
+ .analyze
+
+ val optimized = Optimize.execute(originalPlan)
+
+ // Should still be transformed
+ assert(!optimized.exists {
+ case j: Join if j.joinType == Cross => true
+ case _ => false
+ }, "Should transform even when array is on right side")
+ }
+
+ test("preserves remaining filter predicates") {
+ // Query with additional conditions beyond array_contains
+ val originalPlan = ordersRelation
+ .join(itemsRelation, Cross)
+ .where(ArrayContains($"item_ids", $"id") && ($"order_id" > 100))
+ .analyze
+
+ val optimized = Optimize.execute(originalPlan)
+
+ // Should still have a filter for the remaining predicate (order_id > 100)
+ assert(optimized.exists {
+ case Filter(cond, _) =>
+ cond.find {
+ case GreaterThan(_, Literal(100, IntegerType)) => true
+ case _ => false
+ }.isDefined
+ case _ => false
+ }, "Should preserve remaining filter predicates")
+ }
+
+ test("uses array_distinct to avoid duplicate matches") {
+ val originalPlan = ordersRelation
+ .join(itemsRelation, Cross)
+ .where(ArrayContains($"item_ids", $"id"))
+ .analyze
+
+ val optimized = Optimize.execute(originalPlan)
+
+ // The optimized plan should use ArrayDistinct before exploding
+ // to avoid duplicate rows when array has duplicate elements
+ assert(optimized.exists {
+ case Generate(Explode(ArrayDistinct(_)), _, _, _, _, _) => true
+ case Project(_, Generate(Explode(ArrayDistinct(_)), _, _, _, _, _)) => true
+ case _ => false
+ }, "Should use ArrayDistinct before Explode")
+ }
+
+ test("supports ByteType elements") {
+ val leftRel = LocalRelation($"id".int, $"arr".array(ByteType))
+ val rightRel = LocalRelation($"elem".byte)
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ assert(optimized.exists { case _: Generate => true; case _ => false })
+ }
+
+ test("supports ShortType elements") {
+ val leftRel = LocalRelation($"id".int, $"arr".array(ShortType))
+ val rightRel = LocalRelation($"elem".short)
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ assert(optimized.exists { case _: Generate => true; case _ => false })
+ }
+
+ test("supports DecimalType elements") {
+ val leftRel = LocalRelation($"id".int, $"arr".array(DecimalType(10, 2)))
+ val rightRel = LocalRelation($"elem".decimal(10, 2))
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ assert(optimized.exists { case _: Generate => true; case _ => false })
+ }
+
+ test("supports TimestampType elements") {
+ val leftRel = LocalRelation($"id".int, $"arr".array(TimestampType))
+ val rightRel = LocalRelation($"elem".timestamp)
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ assert(optimized.exists { case _: Generate => true; case _ => false })
+ }
+
+ test("supports BooleanType elements") {
+ val leftRel = LocalRelation($"id".int, $"arr".array(BooleanType))
+ val rightRel = LocalRelation($"elem".boolean)
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ assert(optimized.exists { case _: Generate => true; case _ => false })
+ }
+
+ test("does not transform FloatType elements due to NaN semantics") {
+ val leftRel = LocalRelation($"id".int, $"arr".array(FloatType))
+ val rightRel = LocalRelation($"elem".float)
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ // Should NOT be transformed - still contains Cross join
+ assert(optimized.exists {
+ case j: Join if j.joinType == Cross => true
+ case _ => false
+ }, "FloatType should not be transformed due to NaN semantics")
+ }
+
+ test("does not transform DoubleType elements due to NaN semantics") {
+ val leftRel = LocalRelation($"id".int, $"arr".array(DoubleType))
+ val rightRel = LocalRelation($"elem".double)
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ // Should NOT be transformed - still contains Cross join
+ assert(optimized.exists {
+ case j: Join if j.joinType == Cross => true
+ case _ => false
+ }, "DoubleType should not be transformed due to NaN semantics")
+ }
+
+ test("supports BinaryType elements") {
+ // BinaryType is safe because Spark's join uses content-based hash/comparison
+ // via ByteArray.compareBinary, not Java's Array.equals()
+ val leftRel = LocalRelation($"id".int, $"arr".array(BinaryType))
+ val rightRel = LocalRelation($"elem".binary)
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ assert(optimized.exists { case _: Generate => true; case _ => false })
+ }
+
+ test("does not transform StructType elements") {
+ val structType = StructType(Seq(StructField("a", IntegerType), StructField("b", StringType)))
+ val leftRel = LocalRelation($"id".int, $"arr".array(structType))
+ val rightRel = LocalRelation($"elem".struct(structType))
+ val plan = leftRel.join(rightRel, Cross).where(ArrayContains($"arr", $"elem")).analyze
+ val optimized = Optimize.execute(plan)
+ assert(optimized.exists {
+ case j: Join if j.joinType == Cross => true
+ case _ => false
+ }, "StructType should not be transformed due to complex equality semantics")
+ }
+}
diff --git a/sql/core/benchmarks/CrossJoinArrayContainsToInnerJoinBenchmark-jdk21-results.txt b/sql/core/benchmarks/CrossJoinArrayContainsToInnerJoinBenchmark-jdk21-results.txt
new file mode 100644
index 000000000000..39afc64ccbed
--- /dev/null
+++ b/sql/core/benchmarks/CrossJoinArrayContainsToInnerJoinBenchmark-jdk21-results.txt
@@ -0,0 +1,38 @@
+================================================================================================
+CrossJoinArrayContainsToInnerJoin Benchmark
+================================================================================================
+
+OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Cross join with array_contains (1000 orders, 100 items, array size 5): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-----------------------------------------------------------------------------------------------------------------------------------------------------
+Cross join + array_contains filter (unoptimized) 50 53 4 2.0 500.1 1.0X
+Inner join with explode (optimized equivalent) 49 51 2 2.0 492.4 1.0X
+Inner join with explode (DataFrame API) 42 43 1 2.4 422.8 1.2X
+
+OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Cross join with array_contains (10000 orders, 1000 items, array size 10): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------------------------------------
+Cross join + array_contains filter (unoptimized) 506 525 21 19.8 50.6 1.0X
+Inner join with explode (optimized equivalent) 34 36 3 290.6 3.4 14.7X
+Inner join with explode (DataFrame API) 31 50 17 321.8 3.1 16.3X
+
+OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Scalability: varying array sizes: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+array_size=1 with explode optimization 143 149 9 7.0 142.9 1.0X
+array_size=5 with explode optimization 143 145 1 7.0 143.2 1.0X
+array_size=10 with explode optimization 139 145 8 7.2 139.3 1.0X
+array_size=50 with explode optimization 139 140 2 7.2 139.3 1.0X
+
+OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Different data types in array: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Integer array 40 44 5 25.2 39.7 1.0X
+Long array 26 29 2 37.9 26.4 1.5X
+String array 37 37 1 27.3 36.6 1.1X
+
+
diff --git a/sql/core/benchmarks/CrossJoinArrayContainsToInnerJoinBenchmark-results.txt b/sql/core/benchmarks/CrossJoinArrayContainsToInnerJoinBenchmark-results.txt
new file mode 100644
index 000000000000..94b9d0f34a80
--- /dev/null
+++ b/sql/core/benchmarks/CrossJoinArrayContainsToInnerJoinBenchmark-results.txt
@@ -0,0 +1,38 @@
+================================================================================================
+CrossJoinArrayContainsToInnerJoin Benchmark
+================================================================================================
+
+OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Cross join with array_contains (1000 orders, 100 items, array size 5): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-----------------------------------------------------------------------------------------------------------------------------------------------------
+Cross join + array_contains filter (unoptimized) 58 59 1 1.7 584.9 1.0X
+Inner join with explode (optimized equivalent) 64 69 5 1.6 638.3 0.9X
+Inner join with explode (DataFrame API) 50 52 2 2.0 497.1 1.2X
+
+OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Cross join with array_contains (10000 orders, 1000 items, array size 10): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------------------------------------
+Cross join + array_contains filter (unoptimized) 504 522 20 19.8 50.4 1.0X
+Inner join with explode (optimized equivalent) 47 52 4 213.5 4.7 10.8X
+Inner join with explode (DataFrame API) 37 43 6 273.5 3.7 13.8X
+
+OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Scalability: varying array sizes: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+array_size=1 with explode optimization 151 154 3 6.6 150.7 1.0X
+array_size=5 with explode optimization 151 153 3 6.6 150.9 1.0X
+array_size=10 with explode optimization 146 150 3 6.9 145.8 1.0X
+array_size=50 with explode optimization 145 148 3 6.9 144.6 1.0X
+
+OpenJDK 64-Bit Server VM 17.0.18+8-LTS on Linux 6.11.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+Different data types in array: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+Integer array 32 35 3 30.9 32.3 1.0X
+Long array 35 38 3 28.2 35.4 0.9X
+String array 42 45 5 23.7 42.2 0.8X
+
+
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/CrossJoinArrayContainsToInnerJoinBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/CrossJoinArrayContainsToInnerJoinBenchmark.scala
new file mode 100644
index 000000000000..95bc9ae5bd7b
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/CrossJoinArrayContainsToInnerJoinBenchmark.scala
@@ -0,0 +1,93 @@
+/*
+ * 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 org.apache.spark.benchmark.Benchmark
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Benchmark for CrossJoinArrayContainsToInnerJoin optimization.
+ *
+ * To run this benchmark:
+ * {{{
+ * 1. build/sbt "sql/Test/runMain <this class>"
+ * 2. generate result:
+ * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/Test/runMain <this class>"
+ * }}}
+ */
+object CrossJoinArrayContainsToInnerJoinBenchmark extends SqlBasedBenchmark {
+
+ private val ruleName =
+ "org.apache.spark.sql.catalyst.optimizer.CrossJoinArrayContainsToInnerJoin"
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ // Use larger scale to see real cross-join cost
+ val numOrders = 5000
+ val numItems = 10000
+ val arraySize = 5
+
+ runBenchmark("CrossJoinArrayContainsToInnerJoin") {
+ val benchmark = new Benchmark(
+ s"array_contains optimization ($numOrders x $numItems, array=$arraySize)",
+ numOrders.toLong * numItems,
+ output = output
+ )
+
+ // Disable broadcast to force shuffle join and show true cross-join cost
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+ // Setup data - arrays with DISTINCT elements
+ val arrExpr = s"transform(sequence(0, ${arraySize - 1}), " +
+ s"x -> cast((id * $arraySize + x) % $numItems as int)) as arr"
+ spark.range(numOrders)
+ .selectExpr("id as order_id", arrExpr)
+ .cache()
+ .createOrReplaceTempView("orders")
+
+ spark.range(numItems)
+ .selectExpr("cast(id as int) as item_id", "concat('item_', id) as name")
+ .cache()
+ .createOrReplaceTempView("items")
+
+ // Force cache materialization
+ spark.sql("SELECT count(*) FROM orders").collect()
+ spark.sql("SELECT count(*) FROM items").collect()
+
+ val query = "SELECT o.order_id, i.item_id FROM orders o, items i " +
+ "WHERE array_contains(o.arr, i.item_id)"
+
+ benchmark.addCase("without optimization (cross join)", 3) { _ =>
+ withSQLConf(
+ SQLConf.CROSS_JOINS_ENABLED.key -> "true",
+ SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> ruleName) {
+ spark.sql(query).noop()
+ }
+ }
+
+ benchmark.addCase("with optimization (inner join)", 3) { _ =>
+ withSQLConf(SQLConf.CROSS_JOINS_ENABLED.key -> "true") {
+ spark.sql(query).noop()
+ }
+ }
+
+ benchmark.run()
+
+ spark.catalog.clearCache()
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org