Skip to content

[SPARK-24172][SQL] we should not apply operator pushdown to data source v2 many times #21230

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

[SPARK-24172][SQL] we should not apply operator pushdown to data source v2 many times #21230

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f73440c
we should not apply operator pushdown to data source v2 many times
cloud-fan May 3, 2018
953cd7a
fix a typo
cloud-fan May 11, 2018
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
15 changes: 5 additions & 10 deletions ...n/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,17 +23,10 @@ import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan, Project
import org.apache.spark.sql.catalyst.rules.Rule

object PushDownOperatorsToDataSource extends Rule[LogicalPlan] {
override def apply(
plan: LogicalPlan): LogicalPlan = plan transformUp {
override def apply(plan: LogicalPlan): LogicalPlan = plan match {
// PhysicalOperation guarantees that filters are deterministic; no need to check
case PhysicalOperation(project, newFilters, relation : DataSourceV2Relation) =>
// merge the filters
val filters = relation.filters match {
case Some(existing) =>
existing ++ newFilters
case _ =>
newFilters
}
case PhysicalOperation(project, filters, relation: DataSourceV2Relation) =>
assert(relation.filters.isEmpty, "data source v2 should do push down only once.")

val projectAttrs = project.map(_.toAttribute)
val projectSet = AttributeSet(project.flatMap(_.references))
Expand Down Expand Up @@ -67,5 +60,7 @@ object PushDownOperatorsToDataSource extends Rule[LogicalPlan] {
} else {
filtered
}

case other => other.mapChildren(apply)
}
}