Skip to content

[SPARK-17357][SQL] Fix current predicate pushdown #14912

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 3 commits into from
Closed

[SPARK-17357][SQL] Fix current predicate pushdown #14912

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9e1c315
Simplified predicates should be pushdown.
viirya Sep 1, 2016
8f6f91d
Address comment. Consider more general case.
viirya Sep 5, 2016
f69473f
Focus on the first problem of predicate pushdown.
viirya Sep 13, 2016
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
6 changes: 5 additions & 1 deletion 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 @@ -588,13 +588,17 @@ object CombineUnions extends Rule[LogicalPlan] {
object CombineFilters extends Rule[LogicalPlan] with PredicateHelper {
def apply(plan: LogicalPlan): LogicalPlan = plan transform {
case Filter(fc, nf @ Filter(nc, grandChild)) =>
(ExpressionSet(splitConjunctivePredicates(fc)) --
val combinedFilter = (ExpressionSet(splitConjunctivePredicates(fc)) --
ExpressionSet(splitConjunctivePredicates(nc))).reduceOption(And) match {
case Some(ac) =>
Filter(And(nc, ac), grandChild)
case None =>
nf
}
// [[Filter]] can't pushdown through another [[Filter]]. Once they are combined,
// [[BooleanSimplification]] rule will possibly simplify the predicate to the form that
// will not be able to pushdown. So we pushdown the combined [[Filter]] immediately.
PushDownPredicate(combinedFilter)
}
}

Expand Down
45 changes: 44 additions & 1 deletion ...catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,8 @@ class FilterPushdownSuite extends PlanTest {
Batch("Subqueries", Once,
EliminateSubqueryAliases) ::
Batch("Filter Pushdown", FixedPoint(10),
CombineFilters,
PushDownPredicate,
CombineFilters,
BooleanSimplification,
PushPredicateThroughJoin,
CollapseProject) :: Nil
Expand Down Expand Up @@ -171,6 +171,49 @@ class FilterPushdownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}

test("push down filters that are combined") {
// The following predicate ('a === 2 || 'a === 3) && ('c > 10 || 'a === 2)
// will be simplified as ('a == 2) || ('c > 10 && 'a == 3).
// ('a === 2 || 'a === 3) can be pushed down. But the simplified one can't.
Copy link
Contributor

@srinathshankar srinathshankar Sep 2, 2016

Choose a reason for hiding this comment

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

So what happens if I just have the predicate
(a = 2) || (c > 10 && a = 3)
Will anything will be pushed down ? Have you considered instead modifying the boolean simplification logic.
Another approach that will catch these cases is as follows:
1.a Convert filters to conjunctive normal form
1.b combine filters
1.c Push filters
1.a, b and c will be run in a batch until fixed point.
Follow this batch by BooleanSimplification -- this can find and extract common factors for efficiency.
Overall, cnf may maximize the potential for filter push down

Copy link
Member Author

@viirya viirya Sep 3, 2016 •
edited
Loading

Choose a reason for hiding this comment

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

yeah, as I mentioned in the description, this is currently a simplest to prevent the predicates which can be pushed down becoming not pushed down.

Your case is not pushed down at the beginning. This patch currently doesn't help it.

Because the optimization rules are independent, boolean simplification logic is just a general rule to simplify predicates, and doesn't be aware of the pushdown logic. Basically boolean simplification now looks good and it makes sense to do (a > 10 || a < 100) && (a > 10 || b == 5) => (a > 10) || (a < 100 && b == 5), however, it causes the pushdown issue.

Your another approach makes sense to me. I have thought about this, just don't know if it is necessary to come out it for this corner case, because it needs more code changes.

If it is acceptable, I will implement it. Thank you.

Copy link
Member Author

@viirya viirya Sep 5, 2016

Choose a reason for hiding this comment

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

Considering how the Optimizer works, we can't extract CombineFilters and PushDownPredicates as a new batch, as we should also respect the interaction between them and other rules. I do an alternative approach to convert predicates of filters to cnf during combining filters, and then perform additional predicate pushdown immediately. So the following BooleanSimplification will not affect the predicate pushdown.

Copy link
Contributor

@srinathshankar srinathshankar Sep 7, 2016

Choose a reason for hiding this comment

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

I agree with you that we should respect the interaction between CombineFilters, PushDownPredicates and other rules. I do think it's important that cnf conversion run before any of the push-down / reordering rules. And the simplification rules should run afterwards.
My concern with rolling this into CombineFilters is that it doesn't get triggered unless there are adjoining Filter nodes. In the example you have:
val originalQuery = testRelation
.select('a, 'b, ('c + 1) as 'cc)
.groupBy('a)('a, count('cc) as 'c)
.where('c > 10)
.where(('a === 2) || ('c > 10 && 'a === 3))

I think that (a == 2 || a==3) should get pushed down even if you don't have ".where (c > 10)",
but I'm not sure that it will be since toCNF is in CombineFilters. Could you confirm ?
My suggestion is that toCNF warrants a separate rule -- for example when you're doing joins, and you have
select * from A inner join C on (A.a1 = C.c1) where A.a2 = 2 || (C.c2 = 10 && A.a2 = 3),
you want (A.a2 = 2 || A.a2 = 3) pushed down into A

Copy link
Member Author

@viirya viirya Sep 7, 2016

Choose a reason for hiding this comment

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

You are right. It is only triggered when adjoining Filters are there. So in above example, the predicate (a == 2 || a==3) will not be pushed down when there is no .where(c > 10).

val originalQuery = testRelation
.select('a, 'b, ('c + 1) as 'cc)
.groupBy('a)('a, count('cc) as 'c)
.where('c > 10) // this predicate can't be pushed down.
.where(('a === 2 || 'a === 3) && ('c > 10 || 'a === 2))

val optimized = Optimize.execute(originalQuery.analyze)
val correctAnswer =
testRelation
.where('a === 2 || 'a === 3)
.select('a, 'b, ('c + 1) as 'cc)
.groupBy('a)('a, count('cc) as 'c)
.where('c > 10).analyze

comparePlans(optimized, correctAnswer)
}

// TODO: currently predicate pushdown doesn't convert predicates to pushdown-able form.
// We should do it to push down more predicates.
ignore("predicates which are able to pushdown should be pushed down after converted") {
// (('a === 2) || ('c > 10 || 'a === 3)) can't be pushdown due to the disjunctive form.
// However, its conjunctive normal form can be pushdown.
val originalQuery = testRelation
.select('a, 'b, ('c + 1) as 'cc)
.groupBy('a)('a, count('cc) as 'c)
.where('c > 10)
.where(('a === 2) || ('c > 10 && 'a === 3))

val optimized = Optimize.execute(originalQuery.analyze)
val correctAnswer =
testRelation
.where('a === 2 || 'a === 3)
.select('a, 'b, ('c + 1) as 'cc)
.groupBy('a)('a, count('cc) as 'c)
.where('c > 10).analyze

comparePlans(optimized, correctAnswer)
}

test("joins: push to either side") {
val x = testRelation.subquery('x)
val y = testRelation.subquery('y)
Expand Down