You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by viirya <gi...@git.apache.org> on 2015/04/05 15:53:12 UTC
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
GitHub user viirya opened a pull request:
https://github.com/apache/spark/pull/5369
[SPARK-6715][SQL] Eliminate duplicate filters from pushdown predicates
Now in `DataSourceStrategy`, the pushdown predicates are duplicate of original `Filter` conditions. Thus, some predicates are performed both by data source relation and `Filter` plan. I think it is a duplicate loading. Once the predicates are pushed down and performed by a data source relation, it
can be eliminated from outside `Filter` plan's condition.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/viirya/spark-1 de_dup_filter
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/5369.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5369
----
commit 86162521d71462559cbd5886de491b57020b1518
Author: Liang-Chi Hsieh <vi...@gmail.com>
Date: 2015-04-05T13:45:22Z
Eliminate duplicate filters from pushdown predicates.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5369#discussion_r27808307
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala ---
@@ -233,6 +258,73 @@ class FilteredScanSuite extends DataSourceTest {
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%eE%'", 1)
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%Ee%'", 0)
+
+ // Filter should not be duplicate with pushdown predicate
+ // This query should not have Filter node and have 1 row
+ testNotFilter("SELECT a, b FROM oneToTenFiltered WHERE a = 1", 1)
+ // This query have Filter node which filters 10 rows to 1 row
+ testFilter("SELECT a, b FROM oneToTenFiltered WHERE b = 2", 10, 1)
+
+ def testNotFilter(sqlString: String, count: Int): Unit = {
+ test(s"Without Filter: $sqlString") {
+ val queryExecution = sql(sqlString).queryExecution
+ val filterPlan = queryExecution.executedPlan.collect {
+ case f: execution.Filter => f
+ } match {
+ case Seq(f) => fail(s"Shouldn't find Filter\n$queryExecution")
+ case _ =>
+ }
+
+ val physicalRDDPlan = queryExecution.executedPlan.collect {
+ case p: execution.PhysicalRDD => p
+ } match {
+ case Seq(p) => p
+ case _ => fail(s"Can't find PhysicalRDD\n$queryExecution")
+ }
--- End diff --
Oh I see, you're checking whether the filter is pushed down here. Good point.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5369#discussion_r27807620
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala ---
@@ -162,6 +162,7 @@ trait PrunedScan {
@DeveloperApi
trait PrunedFilteredScan {
def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row]
+ def supportPredicate(predicate: Expression): Boolean
--- End diff --
Would be better if we rename this method to `supportFilter` since we're using the term "filter" instead of "predicate" through out the data sources API.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5369#discussion_r27808662
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala ---
@@ -233,6 +258,73 @@ class FilteredScanSuite extends DataSourceTest {
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%eE%'", 1)
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%Ee%'", 0)
+
+ // Filter should not be duplicate with pushdown predicate
+ // This query should not have Filter node and have 1 row
+ testNotFilter("SELECT a, b FROM oneToTenFiltered WHERE a = 1", 1)
+ // This query have Filter node which filters 10 rows to 1 row
+ testFilter("SELECT a, b FROM oneToTenFiltered WHERE b = 2", 10, 1)
+
+ def testNotFilter(sqlString: String, count: Int): Unit = {
+ test(s"Without Filter: $sqlString") {
+ val queryExecution = sql(sqlString).queryExecution
+ val filterPlan = queryExecution.executedPlan.collect {
+ case f: execution.Filter => f
+ } match {
+ case Seq(f) => fail(s"Shouldn't find Filter\n$queryExecution")
+ case _ =>
+ }
+
+ val physicalRDDPlan = queryExecution.executedPlan.collect {
+ case p: execution.PhysicalRDD => p
+ } match {
+ case Seq(p) => p
+ case _ => fail(s"Can't find PhysicalRDD\n$queryExecution")
+ }
--- End diff --
The above code can be simplified as:
```scala
val queryExecution = sql(sqlString).queryExecution
val Seq(physicalRDDPlan) = queryExecution.executedPlan.collect {
case _: execution.Filter => fail(s"Filter node shouldn't appear.\n$queryExecution")
case plan: execution.PhysicalRDD => plan
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5369#discussion_r27807612
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/sources/DataSourceStrategy.scala ---
@@ -85,8 +85,11 @@ private[sql] object DataSourceStrategy extends Strategy {
scanBuilder: (Seq[Attribute], Seq[Expression]) => RDD[Row]) = {
val projectSet = AttributeSet(projectList.flatMap(_.references))
- val filterSet = AttributeSet(filterPredicates.flatMap(_.references))
- val filterCondition = filterPredicates.reduceLeftOption(expressions.And)
+
+ // eliminate the filters which are supported in the BaseRelation
+ val filteredPredicates = eliminateFilters(filterPredicates, relation)
--- End diff --
How about renaming this to `unrecognizedFilters`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5369#discussion_r27807736
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala ---
@@ -233,6 +258,73 @@ class FilteredScanSuite extends DataSourceTest {
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%eE%'", 1)
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%Ee%'", 0)
+
+ // Filter should not be duplicate with pushdown predicate
+ // This query should not have Filter node and have 1 row
+ testNotFilter("SELECT a, b FROM oneToTenFiltered WHERE a = 1", 1)
+ // This query have Filter node which filters 10 rows to 1 row
+ testFilter("SELECT a, b FROM oneToTenFiltered WHERE b = 2", 10, 1)
+
+ def testNotFilter(sqlString: String, count: Int): Unit = {
+ test(s"Without Filter: $sqlString") {
+ val queryExecution = sql(sqlString).queryExecution
+ val filterPlan = queryExecution.executedPlan.collect {
+ case f: execution.Filter => f
+ } match {
+ case Seq(f) => fail(s"Shouldn't find Filter\n$queryExecution")
+ case _ =>
+ }
+
+ val physicalRDDPlan = queryExecution.executedPlan.collect {
+ case p: execution.PhysicalRDD => p
+ } match {
+ case Seq(p) => p
+ case _ => fail(s"Can't find PhysicalRDD\n$queryExecution")
+ }
--- End diff --
Why do we want to assert for `PhysicalRDD` here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the pull request:
https://github.com/apache/spark/pull/5369#issuecomment-90160565
Good suggestion for me. I will address these styling first and then wait for your design doc. Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on the pull request:
https://github.com/apache/spark/pull/5369#issuecomment-90156494
Thanks for working on this! Left several commented about styling. However, I just realized that adding a new method `supportPredicate` in `PrunedFilteredScan` breaks backwards compatibility of the data sources API. This makes hesitant to have this.
Incidentally, I just wrote a design doc for improving data sources API and adding partitioning support to it. This very issue is also covered by adding a new trait rather than a new method to an existing trait. This doc should probably be made public very soon. Would you mind me revisiting this later? I'll keep you updated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5369#discussion_r27808698
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala ---
@@ -233,6 +258,73 @@ class FilteredScanSuite extends DataSourceTest {
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%eE%'", 1)
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%Ee%'", 0)
+
+ // Filter should not be duplicate with pushdown predicate
+ // This query should not have Filter node and have 1 row
+ testNotFilter("SELECT a, b FROM oneToTenFiltered WHERE a = 1", 1)
+ // This query have Filter node which filters 10 rows to 1 row
+ testFilter("SELECT a, b FROM oneToTenFiltered WHERE b = 2", 10, 1)
+
+ def testNotFilter(sqlString: String, count: Int): Unit = {
+ test(s"Without Filter: $sqlString") {
+ val queryExecution = sql(sqlString).queryExecution
+ val filterPlan = queryExecution.executedPlan.collect {
+ case f: execution.Filter => f
+ } match {
+ case Seq(f) => fail(s"Shouldn't find Filter\n$queryExecution")
+ case _ =>
+ }
+
+ val physicalRDDPlan = queryExecution.executedPlan.collect {
+ case p: execution.PhysicalRDD => p
+ } match {
+ case Seq(p) => p
+ case _ => fail(s"Can't find PhysicalRDD\n$queryExecution")
+ }
+
+ val rawCount = physicalRDDPlan.execute().count()
+
+ if (rawCount != count) {
+ fail(
+ s"Wrong # of results for filter. Got $rawCount, Expected $count\n" +
+ queryExecution)
+ }
+ }
+ }
+
+ def testFilter(sqlString: String, countRDD: Int, countFilter: Int): Unit = {
+ test(s"Filter: $sqlString") {
+ val queryExecution = sql(sqlString).queryExecution
+ val filterPlan = queryExecution.executedPlan.collect {
+ case f: execution.Filter => f
+ } match {
+ case Seq(f) => Some(f)
+ case _ => fail(s"Can't find Filter\n$queryExecution")
+ }
+
+ val physicalRDDPlan = queryExecution.executedPlan.collect {
+ case p: execution.PhysicalRDD => p
+ } match {
+ case Seq(p) => p
+ case _ => fail(s"Can't find PhysicalRDD\n$queryExecution")
+ }
--- End diff --
Similar simplification can be made as mentioned above.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by viirya <gi...@git.apache.org>.
Github user viirya closed the pull request at:
https://github.com/apache/spark/pull/5369
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the pull request:
https://github.com/apache/spark/pull/5369#issuecomment-91129184
Okay. However, I think it is still important to make the query plan simpler by removing redundant parts even those parts are fast operations.
E.g., for this pr, the redundant filters will cause problems. First, because there are duplicate predicates in pushdown parts and `Filter` node, you may not know the filtering is done by which one. Previously, we have noticed such problem and don't find the pushdown predicates don't working.
Second, if the data source has different logic on a specific predicate than `Filter`, duplicate predicates will make something wrong.
The time spent on filtering depends on how many predicates are used and how these predicates are complicated. Also the data size. Redundant filtering will still cause somewhat performance cost.
As @liancheng I think it would be good if we can revisit this after new api is coming out.
Thank you.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by liancheng <gi...@git.apache.org>.
Github user liancheng commented on a diff in the pull request:
https://github.com/apache/spark/pull/5369#discussion_r27808713
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/sources/FilteredScanSuite.scala ---
@@ -233,6 +258,73 @@ class FilteredScanSuite extends DataSourceTest {
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%eE%'", 1)
testPushDown("SELECT a, b, c FROM oneToTenFiltered WHERE c like '%Ee%'", 0)
+
+ // Filter should not be duplicate with pushdown predicate
+ // This query should not have Filter node and have 1 row
+ testNotFilter("SELECT a, b FROM oneToTenFiltered WHERE a = 1", 1)
+ // This query have Filter node which filters 10 rows to 1 row
+ testFilter("SELECT a, b FROM oneToTenFiltered WHERE b = 2", 10, 1)
+
+ def testNotFilter(sqlString: String, count: Int): Unit = {
+ test(s"Without Filter: $sqlString") {
+ val queryExecution = sql(sqlString).queryExecution
+ val filterPlan = queryExecution.executedPlan.collect {
+ case f: execution.Filter => f
+ } match {
+ case Seq(f) => fail(s"Shouldn't find Filter\n$queryExecution")
+ case _ =>
+ }
+
+ val physicalRDDPlan = queryExecution.executedPlan.collect {
+ case p: execution.PhysicalRDD => p
+ } match {
+ case Seq(p) => p
+ case _ => fail(s"Can't find PhysicalRDD\n$queryExecution")
+ }
+
+ val rawCount = physicalRDDPlan.execute().count()
+
+ if (rawCount != count) {
+ fail(
+ s"Wrong # of results for filter. Got $rawCount, Expected $count\n" +
+ queryExecution)
+ }
+ }
+ }
+
+ def testFilter(sqlString: String, countRDD: Int, countFilter: Int): Unit = {
+ test(s"Filter: $sqlString") {
+ val queryExecution = sql(sqlString).queryExecution
+ val filterPlan = queryExecution.executedPlan.collect {
+ case f: execution.Filter => f
+ } match {
+ case Seq(f) => Some(f)
+ case _ => fail(s"Can't find Filter\n$queryExecution")
+ }
+
+ val physicalRDDPlan = queryExecution.executedPlan.collect {
+ case p: execution.PhysicalRDD => p
+ } match {
+ case Seq(p) => p
+ case _ => fail(s"Can't find PhysicalRDD\n$queryExecution")
+ }
+
+ val rawCount = filterPlan.get.execute().count()
+ val rawCountRDD = physicalRDDPlan.execute().count()
+
+ if (rawCount != countFilter) {
+ fail(
+ s"Wrong # of results for Filter Plan. Got $rawCount, Expected $countFilter\n" +
+ queryExecution)
+ }
+ if (rawCountRDD != countRDD) {
+ fail(
+ s"Wrong # of results for PhysicalRDD Plan. Got $rawCountRDD, Expected $countRDD\n" +
+ queryExecution)
+ }
+
--- End diff --
Remove new line here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-6715][SQL] Eliminate duplicate filters ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/5369#issuecomment-89790053
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/29727/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org