[SPARK-22197][SQL] push down operators to data source before planning #19424

cloud-fan · 2017-10-04T06:52:55Z

What changes were proposed in this pull request?

As we discussed in #19136 (comment) , we should push down operators to data source before planning, so that data source can report statistics more accurate.

This PR also includes some cleanup for the read path.

How was this patch tested?

existing tests.

cloud-fan · 2017-10-04T06:53:13Z

cc @rdblue @gatorsmile

SparkQA · 2017-10-04T07:04:43Z

Test build #82444 has finished for PR 19424 at commit 75457f6.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-10-04T11:12:02Z

Test build #82447 has finished for PR 19424 at commit 75457f6.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2017-10-04T21:42:14Z

...main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala

+ * won't change if we switch the operators within a layer(e.g. we can switch the order of predicates
+ * and required columns). The operators in layer N can only be pushed down if operators in layer N-1
+ * that above the data source relation are all pushed down. As an example, you can't push down limit
+ * if a filter below limit is not pushed down.


As an example, given a LIMIT has a FILTER child, you can't push down LIMIT if FILTER is not completely pushed down. When both are pushed down, the data source should execute FILTER before LIMIT.

gatorsmile · 2017-10-04T21:46:02Z

...core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExec.scala

@@ -32,13 +32,12 @@ import org.apache.spark.sql.types.StructType
 case class DataSourceV2ScanExec(


/** * Physical plan node for scanning data from a data source. */

gatorsmile · 2017-10-04T22:13:02Z

...main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala

+ *   layer 1: predicates, required columns.
+ */
+object PushDownOperatorsToDataSource extends Rule[LogicalPlan] {
+  override def apply(plan: LogicalPlan): LogicalPlan = {


This is an optimizer rule? The input is a LogicalPlan and the output is still a LogicalPlan?

yea it's an optimizer rule run before planner

Can we add a test suite for the unit test cases of this rule?

SparkQA · 2017-10-05T00:24:22Z

Test build #82461 has finished for PR 19424 at commit a9f2c57.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-10-05T04:46:51Z

Test build #82464 has finished for PR 19424 at commit df1d1c5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-10-06T17:04:14Z

Test build #82513 has finished for PR 19424 at commit e880328.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait DataSourceReaderHolder

SparkQA · 2017-10-06T17:40:08Z

Test build #82514 has finished for PR 19424 at commit 24f1a75.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait DataSourceReaderHolder

cloud-fan · 2017-10-09T08:22:32Z

retest this please

SparkQA · 2017-10-09T11:01:07Z

Test build #82547 has finished for PR 19424 at commit 24f1a75.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait DataSourceReaderHolder

rdblue

A few comments, but mostly it looks about ready to me.

rdblue · 2017-10-09T22:27:25Z

...main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala

+object PushDownOperatorsToDataSource extends Rule[LogicalPlan] with PredicateHelper {
+  override def apply(plan: LogicalPlan): LogicalPlan = {
+    // make sure filters are at very bottom.
+    val prepared = PushDownPredicate(plan)


Why apply this rule one more time? Is there reason to suspect that predicates won't already be pushed and that one more run would be worth it?

rdblue · 2017-10-09T22:41:06Z

...main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala

+    val afterPushDown = prepared transformUp {
+      case Filter(condition, r @ DataSourceV2Relation(_, reader)) =>
+        val (candidates, containingNonDeterministic) =
+          splitConjunctivePredicates(condition).span(_.deterministic)


It isn't immediately clear why you would use span here instead of partition. I think it is because span will produce all deterministic predicates that would be run before the first non-deterministic predicate in an in-order traversal of teh condition, right? If so, then a comment would be really useful to make this clear. I'd also like to see a comment about why deterministic predicates "after" the first non-deterministic predicate shouldn't be pushed down. An example would really help, too.

rdblue · 2017-10-09T23:03:36Z

...main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala

+          case r: SupportsPushDownRequiredColumns =>
+            val attrMap = AttributeMap(fullOutput.zip(fullOutput))
+            val requiredColumns = requiredByParent.map(attrMap)
+            // Match original case of attributes.


Shouldn't this comment be on line 90? That's the purpose of looking up the required attributes in the set of attrs that was produced from the data source's schema right? That lookup happens by exprId, which was generated by the DataSourceV2Relation so we know we have a copy of the original attribute and case.

rdblue · 2017-10-09T23:04:20Z

...main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownOperatorsToDataSource.scala

+          case _ =>
+        }
+
+        case _ => plan.children.foreach(child => pushDownRequiredColumns(child, child.output))


How do we know that there aren't more cases that need to be supported?

rdblue · 2017-10-10T17:00:59Z

What are the guarantees made by the previous batches in the optimizer? The work done by FilterAndProject seems redundant to me because the optimizer should already push filters below projection. Is that not guaranteed by the time this runs?

SparkQA · 2017-10-10T17:42:48Z

Test build #82586 has finished for PR 19424 at commit e8e8fee.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
trait DataSourceReaderHolder

cloud-fan · 2017-10-11T00:47:19Z

@rdblue we assume the previous batches should push down operators as close to data source relations as possible. One special case is column pruning. The ColumnPruning rule tries to generate a new Project node under every node that has more input columns than it references, which means it will also generate a Project under Filter, and this conflicts with PushDownPredicate rule. Currently the conflict is resolved by a hacky way so there are cases Project under Filter.

gatorsmile · 2017-10-11T19:17:03Z

...re/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceReaderHolder.scala

+import org.apache.spark.sql.sources.v2.reader._
+
+/**
+ * A base class for data source reader holder and defines equals/hashCode methods.


defines -> with customized

gatorsmile · 2017-10-11T19:25:22Z

...re/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceReaderHolder.scala

+        case (l: SupportsPushDownCatalystFilters, r: SupportsPushDownCatalystFilters) =>
+          l.pushedCatalystFilters().toSeq == r.pushedCatalystFilters().toSeq
+        case (l: SupportsPushDownFilters, r: SupportsPushDownFilters) =>
+          l.pushedFilters().toSeq == r.pushedFilters().toSeq


The evaluation order of these filters must be the same? If the orders are different, they are still the same, right?

good catch!

gatorsmile · 2017-10-12T07:22:21Z

I think the current PR is good enough to merge. Later, we still continue improving the Data source API v2.

LGTM pending Jenkins.

SparkQA · 2017-10-12T07:23:24Z

Test build #82662 has finished for PR 19424 at commit 98dbd70.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-10-12T09:12:32Z

...re/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceReaderHolder.scala

+  }
+
+  lazy val output: Seq[Attribute] = reader.readSchema().map(_.name).map { name =>
+    fullOutput.find(_.name == name).get


Shall we use resolver instead of string comparison?

These names should already be normalized before reaching here.

viirya · 2017-10-12T09:15:59Z

sql/core/src/main/java/org/apache/spark/sql/sources/v2/reader/SupportsPushDownFilters.java

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.sources.v2.reader;

 import org.apache.spark.annotation.InterfaceStability;
+import org.apache.spark.sql.catalyst.expressions.Expression;


Seems we don't use Expression here?

good catch. I'll remove it in my following PRs.

viirya · 2017-10-12T09:19:10Z

late LGTM with minor comments.

SparkQA · 2017-10-12T10:21:43Z

Test build #82665 has finished for PR 19424 at commit 200cd20.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-10-12T12:34:47Z

thanks for review, merging to master!

As we discussed in apache#19136 (comment) , we should push down operators to data source before planning, so that data source can report statistics more accurate. This PR also includes some cleanup for the read path. existing tests. Author: Wenchen Fan <[email protected]> Closes apache#19424 from cloud-fan/follow.

rafaelkyrdan · 2019-10-09T15:42:17Z

Could you confirm that limit now optimized and pushed to PushedFilters? In the physical plan I see that it is not. I'm using spark 2.4.4

cloud-fan · 2019-10-10T00:06:16Z

@rafaelkyrdan limit pushdown has not been supported yet.

rafaelkyrdan · 2019-10-10T16:28:11Z

@cloud-fan
I'm wondering whether it is not a priority or it wasn't easy to implement?
If I want to implement it what class/interface should I extend?
Is it possible to do it in a custom JDBC source?
Do you have any suggestions?
The use case looks like this: there are 50M records in the DB I wanna select only 1M why I should read all of them?

rafaelkyrdan · 2019-10-10T17:23:21Z

I think I found what I have to implement:
trait TableScan { def buildScan(): RDD[Row] }
am I right?

cloud-fan · 2019-10-11T03:53:19Z

Usually a data source scans its data incrementally. So when the query has a limit, Spark stops consuming the iterator from data source, and data source won't scan all the data.

But limit pushdown does have use cases. For now the only way is to write a catalyst rule, to catch the limit + scan query plan, and convert it to a custom query plan with limit pushed down.

lokkju · 2020-01-29T09:29:34Z

It seems the (very abandoned) issue for adding LIMIT is https://issues.apache.org/jira/browse/SPARK-22388

This was a nasty surprise when running a df.take(10) on a large postgres dataset, considering limit type pushdowns are implemented in most other connectors.

gatorsmile reviewed Oct 4, 2017

View reviewed changes

cloud-fan force-pushed the follow branch from 75457f6 to a9f2c57 Compare October 5, 2017 00:04

cloud-fan force-pushed the follow branch from a9f2c57 to df1d1c5 Compare October 5, 2017 02:58

cloud-fan force-pushed the follow branch 2 times, most recently from e880328 to 24f1a75 Compare October 6, 2017 14:57

rdblue requested changes Oct 9, 2017

View reviewed changes

cloud-fan added 2 commits October 10, 2017 21:19

improve documents and minor clean up

a556ee6

push down operators to data source before planning

e8e8fee

cloud-fan force-pushed the follow branch from 24f1a75 to e8e8fee Compare October 10, 2017 15:04

gatorsmile reviewed Oct 11, 2017

View reviewed changes

address comments

200cd20

cloud-fan force-pushed the follow branch from 98dbd70 to 200cd20 Compare October 12, 2017 07:40

viirya reviewed Oct 12, 2017

View reviewed changes

asfgit closed this in 73d80ec Oct 12, 2017

gatorsmile mentioned this pull request Feb 1, 2018

[SPARK-23301][SQL] data source column pruning should work for arbitrary expressions #20476

Closed

gengliangwang mentioned this pull request Sep 7, 2020

[SPARK-32708] Query optimization fails to reuse exchange with DataSourceV2 #29564

Closed

		@@ -32,13 +32,12 @@ import org.apache.spark.sql.types.StructType
		case class DataSourceV2ScanExec(

[SPARK-22197][SQL] push down operators to data source before planning #19424

[SPARK-22197][SQL] push down operators to data source before planning #19424

Conversation

cloud-fan commented Oct 4, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

cloud-fan commented Oct 4, 2017

SparkQA commented Oct 4, 2017

SparkQA commented Oct 4, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Oct 5, 2017

SparkQA commented Oct 5, 2017

SparkQA commented Oct 6, 2017

SparkQA commented Oct 6, 2017

cloud-fan commented Oct 9, 2017

SparkQA commented Oct 9, 2017

rdblue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rdblue commented Oct 10, 2017

SparkQA commented Oct 10, 2017

cloud-fan commented Oct 11, 2017

gatorsmile Oct 11, 2017 • edited Loading

Choose a reason for hiding this comment

gatorsmile Oct 11, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gatorsmile commented Oct 12, 2017

SparkQA commented Oct 12, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya commented Oct 12, 2017 • edited Loading

SparkQA commented Oct 12, 2017

cloud-fan commented Oct 12, 2017

rafaelkyrdan commented Oct 9, 2019 • edited Loading

cloud-fan commented Oct 10, 2019

rafaelkyrdan commented Oct 10, 2019

rafaelkyrdan commented Oct 10, 2019

cloud-fan commented Oct 11, 2019

lokkju commented Jan 29, 2020

cloud-fan commented Oct 4, 2017 •

edited

Loading

gatorsmile Oct 11, 2017 •

edited

Loading

gatorsmile Oct 11, 2017 •

edited

Loading

viirya commented Oct 12, 2017 •

edited

Loading

rafaelkyrdan commented Oct 9, 2019 •

edited

Loading