Skip to content

[SPARK-39851][SQL] Improve join stats estimation if one side can keep uniqueness #37267

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 4 commits into from
Closed

[SPARK-39851][SQL] Improve join stats estimation if one side can keep uniqueness #37267

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
242502c
Improve join stats estimation if one side can keep uniqueness
wangyum Jul 24, 2022
57c9d3a
fix
wangyum Jul 25, 2022
690d5fb
Fix
wangyum Jul 25, 2022
21179bf
Fix
wangyum Aug 24, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 16 additions & 6 deletions ...in/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/JoinEstimation.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, Expression}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap, AttributeReference, Expression, ExpressionSet}
import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, Histogram, Join, Statistics}
Expand Down Expand Up @@ -56,10 +56,13 @@ case class JoinEstimation(join: Join) extends Logging {
case _ if !rowCountsExist(join.left, join.right) =>
None

case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _, _, _, _) =>
case ExtractEquiJoinKeys(joinType, leftKeys, rightKeys, _, _, left, right, _) =>
// 1. Compute join selectivity
val joinKeyPairs = extractJoinKeysWithColStats(leftKeys, rightKeys)
val (numInnerJoinedRows, keyStatsAfterJoin) = computeCardinalityAndStats(joinKeyPairs)
val leftUniqueness = left.distinctKeys.exists(_.subsetOf(ExpressionSet(leftKeys)))
val rightUniqueness = right.distinctKeys.exists(_.subsetOf(ExpressionSet(rightKeys)))
val (numInnerJoinedRows, keyStatsAfterJoin) =
computeCardinalityAndStats(joinKeyPairs, leftUniqueness, rightUniqueness)

// 2. Estimate the number of output rows
val leftRows = leftStats.rowCount.get
Expand Down Expand Up @@ -177,10 +180,17 @@ case class JoinEstimation(join: Join) extends Logging {
* @return join cardinality, and column stats for join keys after the join
*/
// scalastyle:on
private def computeCardinalityAndStats(keyPairs: Seq[(AttributeReference, AttributeReference)])
: (BigInt, AttributeMap[ColumnStat]) = {
private def computeCardinalityAndStats(
keyPairs: Seq[(AttributeReference, AttributeReference)],
leftUniqueness: Boolean,
rightUniqueness: Boolean): (BigInt, AttributeMap[ColumnStat]) = {
// If there's no column stats available for join keys, estimate as cartesian product.
var joinCard: BigInt = leftStats.rowCount.get * rightStats.rowCount.get
var joinCard: BigInt = (leftUniqueness, rightUniqueness) match {
case (true, true) => leftStats.rowCount.get.min(rightStats.rowCount.get)
case (true, false) => rightStats.rowCount.get
case (false, true) => leftStats.rowCount.get
case _ => leftStats.rowCount.get * rightStats.rowCount.get
}
val keyStatsAfterJoin = new mutable.HashMap[Attribute, ColumnStat]()
var i = 0
while(i < keyPairs.length && joinCard != 0) {
Expand Down
15 changes: 13 additions & 2 deletions ...he/spark/sql/catalyst/plans/logical/statsEstimation/SizeInBytesOnlyStatsPlanVisitor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,9 @@

package org.apache.spark.sql.catalyst.plans.logical.statsEstimation

import org.apache.spark.sql.catalyst.expressions.AttributeMap
import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
import org.apache.spark.sql.catalyst.expressions.{AttributeMap, ExpressionSet}
import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
import org.apache.spark.sql.catalyst.plans._
import org.apache.spark.sql.catalyst.plans.logical._

/**
Expand Down Expand Up @@ -112,6 +113,16 @@ object SizeInBytesOnlyStatsPlanVisitor extends LogicalPlanVisitor[Statistics] {
case LeftAnti | LeftSemi =>
// LeftSemi and LeftAnti won't ever be bigger than left
p.left.stats
case Inner | LeftOuter | RightOuter | FullOuter =>
p match {
case ExtractEquiJoinKeys(_, leftKeys, rightKeys, _, _, left, right, _)
if left.distinctKeys.exists(_.subsetOf(ExpressionSet(leftKeys))) ||
right.distinctKeys.exists(_.subsetOf(ExpressionSet(rightKeys))) =>
// The sizeInBytes should be > 1 because sizeInBytes * 1 != sizeInBytes + 1.
Statistics(sizeInBytes = p.children.map(_.stats.sizeInBytes).filter(_ > 1L).sum)
case _ =>
default(p)
}
case _ =>
default(p)
}
Expand Down
33 changes: 32 additions & 1 deletion .../test/scala/org/apache/spark/sql/catalyst/statsEstimation/BasicStatsEstimationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.analysis.ResolvedNamespace
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, AttributeMap, AttributeReference, Literal, SortOrder}
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.connector.catalog.SupportsNamespaces
import org.apache.spark.sql.internal.SQLConf
Expand Down Expand Up @@ -328,6 +328,37 @@ class BasicStatsEstimationSuite extends PlanTest with StatsEstimationTestBase {
expectedStatsCboOff = Statistics(sizeInBytes = sizeInBytes))
}

test("SPARK-39851: Improve join stats estimation if one side can keep uniqueness") {
val brandId = attr("brand_id")
val classId = attr("class_id")
val aliasedBrandId = brandId.as("new_brand_id")
val aliasedClassId = classId.as("new_class_id")

val tableSize = 4059900
val tableRowCnt = 202995

val tbl = StatsTestPlan(
outputList = Seq(brandId, classId),
size = Some(tableSize),
rowCount = tableRowCnt,
attributeStats =
AttributeMap(Seq(
brandId -> ColumnStat(Some(858), Some(101001), Some(1016017), Some(0), Some(4), Some(4)),
classId -> ColumnStat(Some(16), Some(1), Some(16), Some(0), Some(4), Some(4)))))

val join = Join(
tbl,
tbl.groupBy(brandId, classId)(aliasedBrandId, aliasedClassId),
Inner,
Some(brandId === aliasedBrandId.toAttribute && classId === aliasedClassId.toAttribute),
JoinHint.NONE)

checkStats(
join,
expectedStatsCboOn = Statistics(4871880, Some(tableRowCnt), join.stats.attributeStats),
expectedStatsCboOff = Statistics(sizeInBytes = 4059900 * 2))
}

test("row size and column stats estimation for sort") {
val columnInfo = AttributeMap(
Seq(
Expand Down
Loading