Skip to content
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

[SPARK-12085] [SQL] The join condition hidden in DNF can't be pushed down to join operator #10087

Closed
wants to merge 2 commits into from
Closed

[SPARK-12085] [SQL] The join condition hidden in DNF can't be pushed down to join operator #10087

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d09d905
Extract the common equality conditions that can be used as a join con…
Dec 2, 2015
08a76ae
bug fix for ColumnPruning rule to deal with Project <- Filter <- Join…
Dec 3, 2015
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
50 changes: 50 additions & 0 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 @@ -40,6 +40,8 @@ object DefaultOptimizer extends Optimizer {
Batch("Aggregate", FixedPoint(100),
ReplaceDistinctWithAggregate,
RemoveLiteralFromGroupExpressions) ::
Batch("CNF factorization", FixedPoint(100),
ExtractEqualJoinCondition) ::
Batch("Operator Optimizations", FixedPoint(100),
// Operator push down
SetOperationPushDown,
Expand Down Expand Up @@ -911,3 +913,51 @@ object RemoveLiteralFromGroupExpressions extends Rule[LogicalPlan] {
a.copy(groupingExpressions = newGrouping)
}
}

/**
* Extracts the equal-join condition if any, so that query planner avoids generating cartsian
* product which cause out of memory exception, and performance issues
Copy link
Contributor

Choose a reason for hiding this comment

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

Can we make this rule simpler and more general? It seems the rule is aptly named, "CNF factorization", and should just do a transformation on expressions (instead of transforming the plan tree and specific to joins). The rest of the rules should then properly fix the join i think.

*/
object ExtractEqualJoinCondition extends Rule[LogicalPlan] with PredicateHelper{
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case f @ Join(left, right, joinType, joinCondition) =>
joinCondition match {
case Some(e) if isDNF(e) => {
val disjConditions = splitDisjunctivePredicates(e)
val exprMatrix = disjConditions.map(splitConjunctivePredicates)
if(exprMatrix.length <= 1) f
else {
val pattern = exprMatrix(0)
val comExprs: Seq[Expression] = pattern.filter(p => isCommonExpr(p, exprMatrix, 1))
val newExprMatrix = exprMatrix.map(_.diff(comExprs))
val newJoinCond = (comExprs :+ newExprMatrix.map(_.reduceLeft(And)).reduceLeft(Or))
.reduceLeftOption(And)
Join(left, right, joinType, newJoinCond)
}
}
case _ => f
}
}

def isCommonExpr(pattern: Expression, matrix: Seq[Seq[Expression]], startIndex: Int) : Boolean = {
val duplicatedCount = matrix.drop(startIndex).count(arr => arr.contains(pattern))
return duplicatedCount == matrix.length - startIndex
}

def isDNF(condition: Expression) : Boolean = {
condition match {
case Or(left, right) => isDNF(left) && isDNF(right)
case And(left, right) => isCNF(left) && isCNF(right)
case _ => true
}
}

def isCNF(condition: Expression): Boolean = {
condition match {
case And(left, right) => isCNF(left) && isCNF(right)
case Or(left, right) => false
case _ => true
}
}
}