spark git commit: [SPARK-23301][SQL] data source column pruning should work for arbitrary expressions
Repository: spark Updated Branches: refs/heads/master b3a04283f -> 19c7c7ebd [SPARK-23301][SQL] data source column pruning should work for arbitrary expressions ## What changes were proposed in this pull request? This PR fixes a mistake in the `PushDownOperatorsToDataSource` rule, the column pruning logic is incorrect about `Project`. ## How was this patch tested? a new test case for column pruning with arbitrary expressions, and improve the existing tests to make sure the `PushDownOperatorsToDataSource` really works. Author: Wenchen Fan Closes #20476 from cloud-fan/push-down. Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/19c7c7eb Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/19c7c7eb Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/19c7c7eb Branch: refs/heads/master Commit: 19c7c7ebdef6c1c7a02ebac9af6a24f521b52c37 Parents: b3a0428 Author: Wenchen Fan Authored: Thu Feb 1 20:44:46 2018 -0800 Committer: gatorsmile Committed: Thu Feb 1 20:44:46 2018 -0800 -- .../v2/PushDownOperatorsToDataSource.scala | 53 + .../sources/v2/JavaAdvancedDataSourceV2.java| 29 - .../sql/sources/v2/DataSourceV2Suite.scala | 113 +-- 3 files changed, 155 insertions(+), 40 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/19c7c7eb/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala index df034ad..566a483 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala @@ -17,7 +17,7 @@ package org.apache.spark.sql.execution.datasources.v2 -import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeMap, Expression, NamedExpression, PredicateHelper} +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeMap, AttributeSet, Expression, NamedExpression, PredicateHelper} import org.apache.spark.sql.catalyst.optimizer.RemoveRedundantProject import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project} import org.apache.spark.sql.catalyst.rules.Rule @@ -81,35 +81,34 @@ object PushDownOperatorsToDataSource extends Rule[LogicalPlan] with PredicateHel // TODO: add more push down rules. -// TODO: nested fields pruning -def pushDownRequiredColumns(plan: LogicalPlan, requiredByParent: Seq[Attribute]): Unit = { - plan match { -case Project(projectList, child) => - val required = projectList.filter(requiredByParent.contains).flatMap(_.references) - pushDownRequiredColumns(child, required) - -case Filter(condition, child) => - val required = requiredByParent ++ condition.references - pushDownRequiredColumns(child, required) - -case DataSourceV2Relation(fullOutput, reader) => reader match { - case r: SupportsPushDownRequiredColumns => -// Match original case of attributes. -val attrMap = AttributeMap(fullOutput.zip(fullOutput)) -val requiredColumns = requiredByParent.map(attrMap) -r.pruneColumns(requiredColumns.toStructType) - case _ => -} +pushDownRequiredColumns(filterPushed, filterPushed.outputSet) +// After column pruning, we may have redundant PROJECT nodes in the query plan, remove them. +RemoveRedundantProject(filterPushed) + } + + // TODO: nested fields pruning + private def pushDownRequiredColumns(plan: LogicalPlan, requiredByParent: AttributeSet): Unit = { +plan match { + case Project(projectList, child) => +val required = projectList.flatMap(_.references) +pushDownRequiredColumns(child, AttributeSet(required)) + + case Filter(condition, child) => +val required = requiredByParent ++ condition.references +pushDownRequiredColumns(child, required) -// TODO: there may be more operators can be used to calculate required columns, we can add -// more and more in the future. -case _ => plan.children.foreach(child => pushDownRequiredColumns(child, child.output)) + case relation: DataSourceV2Relation => relation.reader match { +case reader: SupportsPushDownRequiredColumns => + val requiredColumns = relation.output.filter(requiredByParent.contains) + reader.pruneColumns(requiredColumns.toStr
spark git commit: [SPARK-23301][SQL] data source column pruning should work for arbitrary expressions
Repository: spark Updated Branches: refs/heads/branch-2.3 7baae3aef -> 2b07452ca [SPARK-23301][SQL] data source column pruning should work for arbitrary expressions This PR fixes a mistake in the `PushDownOperatorsToDataSource` rule, the column pruning logic is incorrect about `Project`. a new test case for column pruning with arbitrary expressions, and improve the existing tests to make sure the `PushDownOperatorsToDataSource` really works. Author: Wenchen Fan Closes #20476 from cloud-fan/push-down. (cherry picked from commit 19c7c7ebdef6c1c7a02ebac9af6a24f521b52c37) Signed-off-by: gatorsmile Project: http://git-wip-us.apache.org/repos/asf/spark/repo Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/2b07452c Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/2b07452c Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/2b07452c Branch: refs/heads/branch-2.3 Commit: 2b07452cacb4c226c815a216c4cfea2a04227700 Parents: 7baae3a Author: Wenchen Fan Authored: Thu Feb 1 20:44:46 2018 -0800 Committer: gatorsmile Committed: Thu Feb 1 20:46:09 2018 -0800 -- .../v2/PushDownOperatorsToDataSource.scala | 53 + .../sources/v2/JavaAdvancedDataSourceV2.java| 29 - .../sql/sources/v2/DataSourceV2Suite.scala | 111 +-- 3 files changed, 154 insertions(+), 39 deletions(-) -- http://git-wip-us.apache.org/repos/asf/spark/blob/2b07452c/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala -- diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala index df034ad..566a483 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala @@ -17,7 +17,7 @@ package org.apache.spark.sql.execution.datasources.v2 -import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeMap, Expression, NamedExpression, PredicateHelper} +import org.apache.spark.sql.catalyst.expressions.{And, Attribute, AttributeMap, AttributeSet, Expression, NamedExpression, PredicateHelper} import org.apache.spark.sql.catalyst.optimizer.RemoveRedundantProject import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project} import org.apache.spark.sql.catalyst.rules.Rule @@ -81,35 +81,34 @@ object PushDownOperatorsToDataSource extends Rule[LogicalPlan] with PredicateHel // TODO: add more push down rules. -// TODO: nested fields pruning -def pushDownRequiredColumns(plan: LogicalPlan, requiredByParent: Seq[Attribute]): Unit = { - plan match { -case Project(projectList, child) => - val required = projectList.filter(requiredByParent.contains).flatMap(_.references) - pushDownRequiredColumns(child, required) - -case Filter(condition, child) => - val required = requiredByParent ++ condition.references - pushDownRequiredColumns(child, required) - -case DataSourceV2Relation(fullOutput, reader) => reader match { - case r: SupportsPushDownRequiredColumns => -// Match original case of attributes. -val attrMap = AttributeMap(fullOutput.zip(fullOutput)) -val requiredColumns = requiredByParent.map(attrMap) -r.pruneColumns(requiredColumns.toStructType) - case _ => -} +pushDownRequiredColumns(filterPushed, filterPushed.outputSet) +// After column pruning, we may have redundant PROJECT nodes in the query plan, remove them. +RemoveRedundantProject(filterPushed) + } + + // TODO: nested fields pruning + private def pushDownRequiredColumns(plan: LogicalPlan, requiredByParent: AttributeSet): Unit = { +plan match { + case Project(projectList, child) => +val required = projectList.flatMap(_.references) +pushDownRequiredColumns(child, AttributeSet(required)) + + case Filter(condition, child) => +val required = requiredByParent ++ condition.references +pushDownRequiredColumns(child, required) -// TODO: there may be more operators can be used to calculate required columns, we can add -// more and more in the future. -case _ => plan.children.foreach(child => pushDownRequiredColumns(child, child.output)) + case relation: DataSourceV2Relation => relation.reader match { +case reader: SupportsPushDownRequiredColumns => + val requiredColumns = relation.output.filter(requiredByParent.contains) + reader.pruneColumns(