Skip to content

[SPARK-31705][SQL][FOLLOWUP] Avoid the unnecessary CNF computation for full-outer joins #28810

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
maropu wants to merge 4 commits into from
Closed

[SPARK-31705][SQL][FOLLOWUP] Avoid the unnecessary CNF computation for full-outer joins #28810

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6cad6c6
Fix
maropu Jun 12, 2020
7850c2b
review
maropu Jun 12, 2020
1251b68
review
maropu Jun 12, 2020
6dbff0c
fix
maropu Jun 16, 2020
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: 14 additions & 8 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1290,11 +1290,17 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
(leftEvaluateCondition, rightEvaluateCondition, commonCondition ++ nonDeterministic)
}

private def canPushThrough(joinType: JoinType): Boolean = joinType match {
case _: InnerLike | LeftSemi | RightOuter | LeftOuter | LeftAnti | ExistenceJoin(_) => true
case _ => false
}

def apply(plan: LogicalPlan): LogicalPlan = plan transform applyLocally

val applyLocally: PartialFunction[LogicalPlan, LogicalPlan] = {
// push the where condition down into join filter
case f @ Filter(filterCondition, Join(left, right, joinType, joinCondition, hint)) =>
case f @ Filter(filterCondition, Join(left, right, joinType, joinCondition, hint))
if canPushThrough(joinType) =>
val (leftFilterConditions, rightFilterConditions, commonFilterCondition) =
split(splitConjunctivePredicates(filterCondition), left, right)
joinType match {
Expand Down Expand Up @@ -1334,13 +1340,13 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {

(rightFilterConditions ++ commonFilterCondition).
reduceLeftOption(And).map(Filter(_, newJoin)).getOrElse(newJoin)
case FullOuter => f // DO Nothing for Full Outer Join
case NaturalJoin(_) => sys.error("Untransformed NaturalJoin node")
case UsingJoin(_, _) => sys.error("Untransformed Using join node")

case jt =>

@gengliangwang gengliangwang Jun 16, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: jt => other

sys.error(s"Unexpected join type: $jt")

@cloud-fan cloud-fan Jun 12, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

since we are here, can we throw an exception instead? sys.error will exit the JVM IIRC

@maropu maropu Jun 12, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about IllegalStateException? Throwing an exception here looks okay to me, but I personally think we need consistent handling for unexpected code pathes. I've checed how to handle this unexpected behaivour in the other optimzier rules, then I found there are some rules to use sys.error;

$ grep -nr "sys.error" .
./Optimizer.scala:1345:          sys.error(s"Unexpected join type: $jt")
./Optimizer.scala:1381:          sys.error(s"Unexpected join type: $jt")
./PushCNFPredicateThroughJoin.scala:64:            sys.error(s"Unexpected join type: $jt")
./subquery.scala:458:        sys.error(s"Unexpected operator in scalar subquery: $lp")
./subquery.scala:496:          sys.error(s"Correlated subquery has unexpected operator $op below filter")
./subquery.scala:498:        case op @ _ => sys.error(s"Unexpected operator $op in correlated subquery")
./subquery.scala:502:    sys.error("This line should be unreachable")
./subquery.scala:564:              case op => sys.error(s"Unexpected operator $op in corelated subquery")

Is it better to change them, too?

@gengliangwang gengliangwang Jun 12, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about AnalysisException?

@maropu maropu Jun 12, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

AnalysisException is mainly used for an analyzing phase? https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala#L24

yea, I know some rules in the optimizer throw AnalysisException though...

(base) maropu@~/Repositories/spark/spark-master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer: (SPARK-31705 $)$grep -nr "throw" .
./joins.scala:187:        // `requires attributes from more than one child`, we throw firstly here for better
./joins.scala:189:        throw new AnalysisException("Using PythonUDF in join condition of join type" +
./joins.scala:206:          throw new AnalysisException("Using PythonUDF in join condition of join type" +
./NormalizeFloatingNumbers.scala:100:      throw new IllegalStateException("grouping/join/window partition keys cannot be map type.")
./NormalizeFloatingNumbers.scala:134:    case _ => throw new IllegalStateException(s"fail to normalize $expr")
./objects.scala:250:        throw new IllegalStateException("LambdaVariable should never has 0 as its ID.")
./objects.scala:255:          throw new IllegalStateException(
./objects.scala:264:          throw new IllegalStateException(
./Optimizer.scala:1345:          throw new IllegalStateException(s"Unexpected join type: $jt")
./Optimizer.scala:1381:          throw new IllegalStateException(s"Unexpected join type: $jt")
./Optimizer.scala:1436:          throw new AnalysisException(
./PushCNFPredicateThroughJoin.scala:64:            throw new IllegalStateException(s"Unexpected join type: $jt")
./ReplaceExceptWithFilter.scala:106:    throw new IllegalStateException("Leaf node is expected")
./ReplaceNullWithFalseInPredicate.scala:105:        throw new IllegalArgumentException(message)
./subquery.scala:79:            throw new AnalysisException("Found conflicting attributes " +

}

// push down the join filter into sub query scanning if applicable
case j @ Join(left, right, joinType, joinCondition, hint) =>
case j @ Join(left, right, joinType, joinCondition, hint) if canPushThrough(joinType) =>
val (leftJoinConditions, rightJoinConditions, commonJoinCondition) =
split(joinCondition.map(splitConjunctivePredicates).getOrElse(Nil), left, right)

Expand Down Expand Up @@ -1370,9 +1376,9 @@ object PushPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {
val newJoinCond = (leftJoinConditions ++ commonJoinCondition).reduceLeftOption(And)

Join(newLeft, newRight, joinType, newJoinCond, hint)
case FullOuter => j
case NaturalJoin(_) => sys.error("Untransformed NaturalJoin node")
case UsingJoin(_, _) => sys.error("Untransformed Using join node")

case jt =>
sys.error(s"Unexpected join type: $jt")
}
}
}
Expand Down
14 changes: 10 additions & 4 deletions .../src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushCNFPredicateThroughJoin.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,15 @@ import org.apache.spark.sql.catalyst.rules.Rule
* when predicate pushdown happens.
*/
object PushCNFPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelper {

private def canPushThrough(joinType: JoinType): Boolean = joinType match {
case _: InnerLike | LeftSemi | RightOuter | LeftOuter | LeftAnti | ExistenceJoin(_) => true
case _ => false
}

def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case j @ Join(left, right, joinType, Some(joinCondition), hint) =>
case j @ Join(left, right, joinType, Some(joinCondition), hint)
if canPushThrough(joinType) =>
val predicates = conjunctiveNormalForm(joinCondition)
if (predicates.isEmpty) {
j
Expand All @@ -53,9 +60,8 @@ object PushCNFPredicateThroughJoin extends Rule[LogicalPlan] with PredicateHelpe
Join(newLeft, right, RightOuter, Some(joinCondition), hint)
case LeftOuter | LeftAnti | ExistenceJoin(_) =>
Join(left, newRight, joinType, Some(joinCondition), hint)
case FullOuter => j
case NaturalJoin(_) => sys.error("Untransformed NaturalJoin node")
case UsingJoin(_, _) => sys.error("Untransformed Using join node")
case jt =>

@gengliangwang gengliangwang Jun 16, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto

sys.error(s"Unexpected join type: $jt")
}
}
}
Expand Down