diff options
author | Wenchen Fan <cloud0fan@outlook.com> | 2015-05-12 11:51:55 -0700 |
---|---|---|
committer | Michael Armbrust <michael@databricks.com> | 2015-05-12 11:51:55 -0700 |
commit | 4e290522c2a6310636317c54589dc35c91d95486 (patch) | |
tree | 0cdf8a549ba2f082bdb4511d96d3bd5e47e0c20f /sql/catalyst | |
parent | 65697bbeafe507dda066e2dc14ca5183f278dfe9 (diff) | |
download | spark-4e290522c2a6310636317c54589dc35c91d95486.tar.gz spark-4e290522c2a6310636317c54589dc35c91d95486.tar.bz2 spark-4e290522c2a6310636317c54589dc35c91d95486.zip |
[SPARK-7276] [DATAFRAME] speed up DataFrame.select by collapsing Project
Author: Wenchen Fan <cloud0fan@outlook.com>
Closes #5831 from cloud-fan/7276 and squashes the following commits:
ee4a1e1 [Wenchen Fan] fix rebase mistake
a3b565d [Wenchen Fan] refactor
99deb5d [Wenchen Fan] add test
f1f67ad [Wenchen Fan] fix 7276
Diffstat (limited to 'sql/catalyst')
2 files changed, 26 insertions, 17 deletions
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index 1ee5fb245f..b163707cc9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -43,6 +43,7 @@ object DefaultOptimizer extends Optimizer { PushPredicateThroughJoin, PushPredicateThroughGenerate, ColumnPruning, + ProjectCollapsing, CombineLimits) :: Batch("ConstantFolding", FixedPoint(100), NullPropagation, @@ -114,7 +115,7 @@ object UnionPushdown extends Rule[LogicalPlan] { * - Aggregate * - Project <- Join * - LeftSemiJoin - * - Collapse adjacent projections, performing alias substitution. + * - Performing alias substitution. */ object ColumnPruning extends Rule[LogicalPlan] { def apply(plan: LogicalPlan): LogicalPlan = plan transform { @@ -152,7 +153,28 @@ object ColumnPruning extends Rule[LogicalPlan] { Join(left, prunedChild(right, allReferences), LeftSemi, condition) - // Combine adjacent Projects. + case Project(projectList, Limit(exp, child)) => + Limit(exp, Project(projectList, child)) + + // Eliminate no-op Projects + case Project(projectList, child) if child.output == projectList => child + } + + /** Applies a projection only when the child is producing unnecessary attributes */ + private def prunedChild(c: LogicalPlan, allReferences: AttributeSet) = + if ((c.outputSet -- allReferences.filter(c.outputSet.contains)).nonEmpty) { + Project(allReferences.filter(c.outputSet.contains).toSeq, c) + } else { + c + } +} + +/** + * Combines two adjacent [[Project]] operators into one, merging the + * expressions into one single expression. + */ +object ProjectCollapsing extends Rule[LogicalPlan] { + def apply(plan: LogicalPlan): LogicalPlan = plan transformUp { case Project(projectList1, Project(projectList2, child)) => // Create a map of Aliases to their values from the child projection. // e.g., 'SELECT ... FROM (SELECT a + b AS c, d ...)' produces Map(c -> Alias(a + b, c)). @@ -169,21 +191,7 @@ object ColumnPruning extends Rule[LogicalPlan] { }).asInstanceOf[Seq[NamedExpression]] Project(substitutedProjection, child) - - case Project(projectList, Limit(exp, child)) => - Limit(exp, Project(projectList, child)) - - // Eliminate no-op Projects - case Project(projectList, child) if child.output == projectList => child } - - /** Applies a projection only when the child is producing unnecessary attributes */ - private def prunedChild(c: LogicalPlan, allReferences: AttributeSet) = - if ((c.outputSet -- allReferences.filter(c.outputSet.contains)).nonEmpty) { - Project(allReferences.filter(c.outputSet.contains).toSeq, c) - } else { - c - } } /** diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala index 58d415d901..0c428f7231 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala @@ -38,7 +38,8 @@ class FilterPushdownSuite extends PlanTest { PushPredicateThroughProject, PushPredicateThroughJoin, PushPredicateThroughGenerate, - ColumnPruning) :: Nil + ColumnPruning, + ProjectCollapsing) :: Nil } val testRelation = LocalRelation('a.int, 'b.int, 'c.int) |