Skip to content
Snippets Groups Projects
Commit c6f8135e authored by Yijie Shen's avatar Yijie Shen Committed by Yin Huai
Browse files

[SPARK-10539] [SQL] Project should not be pushed down through Intersect or Except #8742

Intersect and Except are both set operators and they use the all the columns to compare equality between rows. When pushing their Project parent down, the relations they based on would change, therefore not an equivalent transformation.

JIRA: https://issues.apache.org/jira/browse/SPARK-10539

I added some comments based on the fix of https://github.com/apache/spark/pull/8742.

Author: Yijie Shen <henry.yijieshen@gmail.com>
Author: Yin Huai <yhuai@databricks.com>

Closes #8823 from yhuai/fix_set_optimization.
parent 00a2911c
No related branches found
No related tags found
No related merge requests found
...@@ -85,7 +85,22 @@ object SamplePushDown extends Rule[LogicalPlan] { ...@@ -85,7 +85,22 @@ object SamplePushDown extends Rule[LogicalPlan] {
} }
/** /**
* Pushes operations to either side of a Union, Intersect or Except. * Pushes certain operations to both sides of a Union, Intersect or Except operator.
* Operations that are safe to pushdown are listed as follows.
* Union:
* Right now, Union means UNION ALL, which does not de-duplicate rows. So, it is
* safe to pushdown Filters and Projections through it. Once we add UNION DISTINCT,
* we will not be able to pushdown Projections.
*
* Intersect:
* It is not safe to pushdown Projections through it because we need to get the
* intersect of rows by comparing the entire rows. It is fine to pushdown Filters
* because we will not have non-deterministic expressions.
*
* Except:
* It is not safe to pushdown Projections through it because we need to get the
* intersect of rows by comparing the entire rows. It is fine to pushdown Filters
* because we will not have non-deterministic expressions.
*/ */
object SetOperationPushDown extends Rule[LogicalPlan] { object SetOperationPushDown extends Rule[LogicalPlan] {
...@@ -122,40 +137,26 @@ object SetOperationPushDown extends Rule[LogicalPlan] { ...@@ -122,40 +137,26 @@ object SetOperationPushDown extends Rule[LogicalPlan] {
Filter(condition, left), Filter(condition, left),
Filter(pushToRight(condition, rewrites), right)) Filter(pushToRight(condition, rewrites), right))
// Push down projection into union // Push down projection through UNION ALL
case Project(projectList, u @ Union(left, right)) => case Project(projectList, u @ Union(left, right)) =>
val rewrites = buildRewrites(u) val rewrites = buildRewrites(u)
Union( Union(
Project(projectList, left), Project(projectList, left),
Project(projectList.map(pushToRight(_, rewrites)), right)) Project(projectList.map(pushToRight(_, rewrites)), right))
// Push down filter into intersect // Push down filter through INTERSECT
case Filter(condition, i @ Intersect(left, right)) => case Filter(condition, i @ Intersect(left, right)) =>
val rewrites = buildRewrites(i) val rewrites = buildRewrites(i)
Intersect( Intersect(
Filter(condition, left), Filter(condition, left),
Filter(pushToRight(condition, rewrites), right)) Filter(pushToRight(condition, rewrites), right))
// Push down projection into intersect // Push down filter through EXCEPT
case Project(projectList, i @ Intersect(left, right)) =>
val rewrites = buildRewrites(i)
Intersect(
Project(projectList, left),
Project(projectList.map(pushToRight(_, rewrites)), right))
// Push down filter into except
case Filter(condition, e @ Except(left, right)) => case Filter(condition, e @ Except(left, right)) =>
val rewrites = buildRewrites(e) val rewrites = buildRewrites(e)
Except( Except(
Filter(condition, left), Filter(condition, left),
Filter(pushToRight(condition, rewrites), right)) Filter(pushToRight(condition, rewrites), right))
// Push down projection into except
case Project(projectList, e @ Except(left, right)) =>
val rewrites = buildRewrites(e)
Except(
Project(projectList, left),
Project(projectList.map(pushToRight(_, rewrites)), right))
} }
} }
......
...@@ -60,23 +60,22 @@ class SetOperationPushDownSuite extends PlanTest { ...@@ -60,23 +60,22 @@ class SetOperationPushDownSuite extends PlanTest {
comparePlans(exceptOptimized, exceptCorrectAnswer) comparePlans(exceptOptimized, exceptCorrectAnswer)
} }
test("union/intersect/except: project to each side") { test("union: project to each side") {
val unionQuery = testUnion.select('a) val unionQuery = testUnion.select('a)
val unionOptimized = Optimize.execute(unionQuery.analyze)
val unionCorrectAnswer =
Union(testRelation.select('a), testRelation2.select('d)).analyze
comparePlans(unionOptimized, unionCorrectAnswer)
}
test("SPARK-10539: Project should not be pushed down through Intersect or Except") {
val intersectQuery = testIntersect.select('b, 'c) val intersectQuery = testIntersect.select('b, 'c)
val exceptQuery = testExcept.select('a, 'b, 'c) val exceptQuery = testExcept.select('a, 'b, 'c)
val unionOptimized = Optimize.execute(unionQuery.analyze)
val intersectOptimized = Optimize.execute(intersectQuery.analyze) val intersectOptimized = Optimize.execute(intersectQuery.analyze)
val exceptOptimized = Optimize.execute(exceptQuery.analyze) val exceptOptimized = Optimize.execute(exceptQuery.analyze)
val unionCorrectAnswer = comparePlans(intersectOptimized, intersectQuery.analyze)
Union(testRelation.select('a), testRelation2.select('d)).analyze comparePlans(exceptOptimized, exceptQuery.analyze)
val intersectCorrectAnswer = }
Intersect(testRelation.select('b, 'c), testRelation2.select('e, 'f)).analyze
val exceptCorrectAnswer =
Except(testRelation.select('a, 'b, 'c), testRelation2.select('d, 'e, 'f)).analyze
comparePlans(unionOptimized, unionCorrectAnswer)
comparePlans(intersectOptimized, intersectCorrectAnswer)
comparePlans(exceptOptimized, exceptCorrectAnswer) }
} }
...@@ -907,4 +907,13 @@ class DataFrameSuite extends QueryTest with SharedSQLContext { ...@@ -907,4 +907,13 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {
assert(row.getDouble(1) - row.getDouble(3) === 0.0 +- 0.001) assert(row.getDouble(1) - row.getDouble(3) === 0.0 +- 0.001)
} }
} }
test("SPARK-10539: Project should not be pushed down through Intersect or Except") {
val df1 = (1 to 100).map(Tuple1.apply).toDF("i")
val df2 = (1 to 30).map(Tuple1.apply).toDF("i")
val intersect = df1.intersect(df2)
val except = df1.except(df2)
assert(intersect.count() === 30)
assert(except.count() === 70)
}
} }
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment