Skip to content

[SPARK-11390] [SQL] Query plan with/without filterPushdown indistinguishable #9679

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

[SPARK-11390] [SQL] Query plan with/without filterPushdown indistinguishable #9679

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
6 changes: 4 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,9 @@ private[sql] object PhysicalRDD {
def createFromDataSource(
output: Seq[Attribute],
rdd: RDD[InternalRow],
relation: BaseRelation): PhysicalRDD = {
PhysicalRDD(output, rdd, relation.toString, relation.isInstanceOf[HadoopFsRelation])
relation: BaseRelation,
extraInformation: String = ""): PhysicalRDD = {
PhysicalRDD(output, rdd, relation.toString + extraInformation,
relation.isInstanceOf[HadoopFsRelation])
}
}
6 changes: 4 additions & 2 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,8 @@ private[sql] object DataSourceStrategy extends Strategy with Logging {
// `Filter`s or cannot be handled by `relation`.
val filterCondition = unhandledPredicates.reduceLeftOption(expressions.And)

val pushedFiltersString = pushedFilters.mkString(" PushedFilter: [", ",", "] ")

if (projects.map(_.toAttribute) == projects &&
projectSet.size == projects.size &&
filterSet.subsetOf(projectSet)) {
Expand All @@ -332,7 +334,7 @@ private[sql] object DataSourceStrategy extends Strategy with Logging {
val scan = execution.PhysicalRDD.createFromDataSource(
projects.map(_.toAttribute),
scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
relation.relation)
relation.relation, pushedFiltersString)
filterCondition.map(execution.Filter(_, scan)).getOrElse(scan)
} else {
// Don't request columns that are only referenced by pushed filters.
Expand All @@ -342,7 +344,7 @@ private[sql] object DataSourceStrategy extends Strategy with Logging {
val scan = execution.PhysicalRDD.createFromDataSource(
requestedColumns,
scanBuilder(requestedColumns, candidatePredicates, pushedFilters),
relation.relation)
relation.relation, pushedFiltersString)
execution.Project(
projects, filterCondition.map(execution.Filter(_, scan)).getOrElse(scan))
}
Expand Down
14 changes: 14 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,20 @@ class PlannerSuite extends SharedSQLContext {
}
}

test("SPARK-11390 explain should print PushedFilters of PhysicalRDD") {
withTempPath { file =>
val path = file.getCanonicalPath
testData.write.parquet(path)
val df = sqlContext.read.parquet(path)
sqlContext.registerDataFrameAsTable(df, "testPushed")

withTempTable("testPushed") {
val exp = sql("select * from testPushed where key = 15").queryExecution.executedPlan
assert(exp.toString.contains("PushedFilter: [EqualTo(key,15)]"))
}
}
}

test("efficient limit -> project -> sort") {
{
val query =
Expand Down