Skip to content

[SPARK-22211][SQL] Remove incorrect FOJ limit pushdown #19647

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
henryr wants to merge 1 commit into from
Closed

[SPARK-22211][SQL] Remove incorrect FOJ limit pushdown #19647

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
24 changes: 5 additions & 19 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 @@ -332,32 +332,18 @@ object LimitPushDown extends Rule[LogicalPlan] {
// pushdown Limit.
case LocalLimit(exp, Union(children)) =>
LocalLimit(exp, Union(children.map(maybePushLocalLimit(exp, _))))
// Add extra limits below OUTER JOIN. For LEFT OUTER and FULL OUTER JOIN we push limits to the
// left and right sides, respectively. For FULL OUTER JOIN, we can only push limits to one side
// because we need to ensure that rows from the limited side still have an opportunity to match
// against all candidates from the non-limited side. We also need to ensure that this limit
// pushdown rule will not eventually introduce limits on both sides if it is applied multiple
// times. Therefore:
// Add extra limits below OUTER JOIN. For LEFT OUTER and RIGHT OUTER JOIN we push limits to
// the left and right sides, respectively. It's not safe to push limits below FULL OUTER
// JOIN in the general case without a more invasive rewrite.
// We also need to ensure that this limit pushdown rule will not eventually introduce limits
// on both sides if it is applied multiple times. Therefore:
// - If one side is already limited, stack another limit on top if the new limit is smaller.
// The redundant limit will be collapsed by the CombineLimits rule.
// - If neither side is limited, limit the side that is estimated to be bigger.
case LocalLimit(exp, join @ Join(left, right, joinType, _)) =>
val newJoin = joinType match {
case RightOuter => join.copy(right = maybePushLocalLimit(exp, right))
case LeftOuter => join.copy(left = maybePushLocalLimit(exp, left))
case FullOuter =>
Copy link
Member

@gatorsmile gatorsmile Nov 3, 2017

Choose a reason for hiding this comment

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

Thanks for working on it! We should still keep it.

Let me fix it based on my original PR: #10454

Copy link
Member

@gatorsmile gatorsmile Nov 3, 2017

Choose a reason for hiding this comment

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

We decide to disable it and merge it to master and the previous release

(left.maxRows, right.maxRows) match {
case (None, None) =>
if (left.stats.sizeInBytes >= right.stats.sizeInBytes) {
join.copy(left = maybePushLocalLimit(exp, left))
} else {
join.copy(right = maybePushLocalLimit(exp, right))
}
case (Some(_), Some(_)) => join
case (Some(_), None) => join.copy(left = maybePushLocalLimit(exp, left))
case (None, Some(_)) => join.copy(right = maybePushLocalLimit(exp, right))

}
case _ => join
}
LocalLimit(exp, newJoin)
Expand Down
33 changes: 16 additions & 17 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/LimitPushdownSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,35 +113,34 @@ class LimitPushdownSuite extends PlanTest {

test("full outer join where neither side is limited and both sides have same statistics") {
Copy link
Member

@dongjoon-hyun dongjoon-hyun Nov 9, 2017

Choose a reason for hiding this comment

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

Hi, @gatorsmile and @henryr .
This breaks branch-2.2.

[info] - full outer join where neither side is limited and both sides have same statistics *** FAILED *** (43 milliseconds)
[info] - full outer join where neither side is limited and left side has larger statistics *** FAILED *** (12 milliseconds)
[info] - full outer join where neither side is limited and right side has larger statistics *** FAILED *** (9 milliseconds)

cc @felixcheung for 2.2.1.

Copy link
Member

@gatorsmile gatorsmile Nov 9, 2017

Choose a reason for hiding this comment

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

Could you submit a fix to resolve the test failure?

Copy link
Contributor Author

@henryr henryr Nov 9, 2017

Choose a reason for hiding this comment

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

Sure, I'm taking a look now.

Copy link
Member

@dongjoon-hyun dongjoon-hyun Nov 9, 2017

Choose a reason for hiding this comment

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

Thank you, all!

Copy link
Contributor Author

@henryr henryr Nov 9, 2017

Choose a reason for hiding this comment

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

See #19701. I think something got dropped in the backport to branch-2.2, so thankfully it's an easy fix.

assert(x.stats.sizeInBytes === y.stats.sizeInBytes)
val originalQuery = x.join(y, FullOuter).limit(1)
val optimized = Optimize.execute(originalQuery.analyze)
val correctAnswer = Limit(1, LocalLimit(1, x).join(y, FullOuter)).analyze
comparePlans(optimized, correctAnswer)
val originalQuery = x.join(y, FullOuter).limit(1).analyze
val optimized = Optimize.execute(originalQuery)
// No pushdown for FULL OUTER JOINS.
comparePlans(optimized, originalQuery)
}

test("full outer join where neither side is limited and left side has larger statistics") {
val xBig = testRelation.copy(data = Seq.fill(2)(null)).subquery('x)
assert(xBig.stats.sizeInBytes > y.stats.sizeInBytes)
val originalQuery = xBig.join(y, FullOuter).limit(1)
val optimized = Optimize.execute(originalQuery.analyze)
val correctAnswer = Limit(1, LocalLimit(1, xBig).join(y, FullOuter)).analyze
comparePlans(optimized, correctAnswer)
val originalQuery = xBig.join(y, FullOuter).limit(1).analyze
val optimized = Optimize.execute(originalQuery)
// No pushdown for FULL OUTER JOINS.
comparePlans(optimized, originalQuery)
}

test("full outer join where neither side is limited and right side has larger statistics") {
val yBig = testRelation.copy(data = Seq.fill(2)(null)).subquery('y)
assert(x.stats.sizeInBytes < yBig.stats.sizeInBytes)
val originalQuery = x.join(yBig, FullOuter).limit(1)
val optimized = Optimize.execute(originalQuery.analyze)
val correctAnswer = Limit(1, x.join(LocalLimit(1, yBig), FullOuter)).analyze
comparePlans(optimized, correctAnswer)
val originalQuery = x.join(yBig, FullOuter).limit(1).analyze
val optimized = Optimize.execute(originalQuery)
// No pushdown for FULL OUTER JOINS.
comparePlans(optimized, originalQuery)
}

test("full outer join where both sides are limited") {
val originalQuery = x.limit(2).join(y.limit(2), FullOuter).limit(1)
val optimized = Optimize.execute(originalQuery.analyze)
val correctAnswer = Limit(1, Limit(2, x).join(Limit(2, y), FullOuter)).analyze
comparePlans(optimized, correctAnswer)
val originalQuery = x.limit(2).join(y.limit(2), FullOuter).limit(1).analyze
val optimized = Optimize.execute(originalQuery)
// No pushdown for FULL OUTER JOINS.
comparePlans(optimized, originalQuery)
}
}