[SPARK-12476][SQL] Implement JdbcRelation#unhandledFilters for removing unnecessary Spark Filter

[maropu]

Input: SELECT * FROM jdbcTable WHERE col0 = 'xxx'

Current plan:

== Optimized Logical Plan ==
Project [col0#0,col1#1]
+- Filter (col0#0 = xxx)
+- Relation[col0#0,col1#1] JDBCRelation(jdbc:postgresql:postgres,testRel,[Lorg.apache.spark.Partition;@2ac7c683,{user=maropu, password=, driver=org.postgresql.Driver})
== Physical Plan ==
+- Filter (col0#0 = xxx)
+- Scan JDBCRelation(jdbc:postgresql:postgres,testRel,[Lorg.apache.spark.Partition;@2ac7c683,{user=maropu, password=, driver=org.postgresql.Driver})[col0#0,col1#1] PushedFilters: [EqualTo(col0,xxx)]

This patch enables a plan below;

== Optimized Logical Plan ==
Project [col0#0,col1#1]
+- Filter (col0#0 = xxx)
+- Relation[col0#0,col1#1] JDBCRelation(jdbc:postgresql:postgres,testRel,[Lorg.apache.spark.Partition;@2ac7c683,{user=maropu, password=, driver=org.postgresql.Driver})
== Physical Plan ==
Scan JDBCRelation(jdbc:postgresql:postgres,testRel,[Lorg.apache.spark.Partition;@2ac7c683,{user=maropu, password=, driver=org.postgresql.Driver})[col0#0,col1#1] PushedFilters: [EqualTo(col0,xxx)]

[SparkQA]

Test build #48159 has finished for PR 10427 at commit c129f28.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

BTW, a role of PrunedFilteredScan conflicts with that of BaseRelation#unhandledFilters, that is, BaseRelation#unhandledFilters does the same thing with PrunedFilteredScan.
I think we should rename PrunedFilteredScan to another name, e.g., DbScan.

[SparkQA]

Test build #48161 has finished for PR 10427 at commit c2c6e20.

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

[HyukjinKwon]

Should we maybe add some comments or messages to know what this assert mean?

[maropu]

Added the comment

[HyukjinKwon]

cc @liancheng@yhuai

[HyukjinKwon]

A nit. Please wrap with [[]] Scala classes (SparkPlan, Filterand JDBCRelation) in comments.

[liancheng]

Actually you don't need to do so. [[...]] notation is only used in ScalaDoc for adding links. It's not effective in inline comments.

[maropu]

We need the wrap for scaladoc though, we also need it just for comments?
There are the comments that have no wrap for spark classes;
https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/DataFrame.scala#L558

[liancheng]

Yeah, that's what I mean. Double brackets are not required for inline comments (i.e., here).

[HyukjinKwon]

I see. Thanks! That was my bad.

[HyukjinKwon]

@liancheng would you tell me what you think on this? I made some commits locally and want to be sure and make them structurally consistent with this PR becore creating new PRs for them.

[SparkQA]

Test build #48188 has finished for PR 10427 at commit a910a7e.

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

[liancheng]

We can make JDBCRDD.compileFiltersprivate[jdbc], and then simplify this one to compileFilter(filter) != null.

[maropu]

Oh, smart idea. Fixed.

[SparkQA]

Test build #48187 has finished for PR 10427 at commit 9ead74a.

• This patch passes all tests.
• This patch does not merge cleanly.
• This patch adds no public classes.

[liancheng]

Nit: It's a little bit weird to use double-negation here when filters.filter(JDBCRDD.compileFilter(_) == null) is perfectly OK.

[maropu]

Okay and I'll fix it.

[SparkQA]

Test build #48201 has finished for PR 10427 at commit c362c91.

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

[liancheng]

Overall LGTM. Would like to ask @yhuai to have a look, though.

[HyukjinKwon]

Let me leave one more comment. I tested some cases with this PR and looks generally working fine. But I would like to mention one thing that I am pretty sure you folks already know though.

Since now Filter is removed, for operations such as count() which does not require any columns, it works different internally with filtering. It looks okay with JDBC but it does not with others such as Parquet.

For example, in case of Parquet, I implemented this locally and tested and found it calculating a wrong result. That produced the plans below:

== Physical Plan ==
TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[count#10L])
+- TungstenExchange SinglePartition, None
+- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#14L])
+- Project
+- Scan ParquetRelation[] InputPaths: file:/private/var/folders/9j/gf_c342d7d150mwrxvkqnc180000gn/T/spark-bc91cad7-b7c6-4ef0-985d-abaad921572d/part=1, PushedFilters: [EqualTo(a,2)]

Here, I could find any requiredColumns not given but some filters are applied for some columns with Filter producing wrong results by Parquet.

I tried with the original code and it showed the plan below assigning requiredColumns properly.

== Physical Plan ==
TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[count#10L])
+- TungstenExchange SinglePartition, None
+- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#14L])
+- Project
+- Filter (a#8 = 2)
+- Scan ParquetRelation[a#8] InputPaths: file:/private/var/folders/9j/gf_c342d7d150mwrxvkqnc180000gn/T/spark-ef271ec6-95e1-43ae-9b3e-1d4dae6f69c3/part=1, PushedFilters: [EqualTo(a,2)]

The reason seems that it adds all the columns in projects and filters to requiredColumns but it does not add columns only for pushed filters here. So, it ends up with no columns with Project and without Filter. As result, Parquet tries to filter on non-requested column(s) internally. As you know, the columns in filters should be given to requestedSchema for Parquet to filter rows properly.

For this JDBCRelation, I tested and saw the count() working fine with filtering because it gives 1 when requiredColumns is empty and JDBCRelation can filter rows on non-required columns by constructing a WHERE clause. This produced the plan below:

== Physical Plan ==
TungstenAggregate(key=[], functions=[(count(1),mode=Final,isDistinct=false)], output=[count#38L])
+- TungstenExchange SinglePartition, None
+- TungstenAggregate(key=[], functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#42L])
+- Project
+- Scan JDBCRelation(jdbc:h2:mem:testdb0,TEST.PEOPLE,[Lorg.apache.spark.Partition;@3e906375,{user=testUser, password=testPass, url=jdbc:h2:mem:testdb0, dbtable=TEST.PEOPLE})[] PushedFilters: [EqualTo(NAME,fred)]

So please take this into account.

+Actually, I think it should include the columns in pushed filters for requiredColumns not exclude them. I will try to fix after asking if this should be fixed in this way.

[SparkQA]

Test build #48277 has finished for PR 10427 at commit 56d14cf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class JavaWordBlacklist\n * class JavaDroppedWordsCounter\n

[maropu]

@HyukjinKwon ISTM the root problem is that catalyst cannot fill required columns on-the-fly for data sources. The interface unhandleFilters is little related to the problem.

[HyukjinKwon]

@maropu I believe it might be about a Parquet stuff. AFAIK, the columns in filters should be set to requestedSchema for Parquet. But this exlcudes the columns for pushed filters which ends up with no columns but filters. As result, this produces a wrong result.

Actually, I am now talking with @yhuai to check if it is really an issue. For me I think we can simply include the columns for pushed filters as well just for safety.

[maropu]

If this patch merged, the comment below is not suitable for JDBCRelation.
https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/sources/interfaces.scala#L281

Is it okay to fix it?

[HyukjinKwon]

And you are right I think the comment I said is not related with this PR. Let's wait for their comments!

[maropu]

@HyukjinKwon Adding a special handing for Parquet is better? I think that required columns for push-down filters should be passed to each data sources, and this is a more natural way.

[HyukjinKwon]

@maropu I agree that can be another way! but I just think an interface should be inclusive not exclusive. Handing in ParquetRelation might mean other datasources might have to deal with similar problems (including external datasources).

[maropu]

@HyukjinKwon Anyway, I added a test by reflecting this discussion. thanks.

[maropu]

@yhuai ping

[yhuai]

Seems you want to use isEmpty?

[maropu]

@yhuai fixed.

[SparkQA]

Test build #50818 has finished for PR 10427 at commit b4dc961.

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

[yhuai]

Thank you for the update!

When we were using == null (I guess all filters were marked as unhandled, right?), all tests still passed. So, I am wondering if existing tests are sufficient?

[maropu]

Added tests though, ISTM that JDBCRDD.compileFilter does not return None for given Filter because the function can compile all the Filter implemented in sql.sources.filters.

[SparkQA]

Test build #50904 has finished for PR 10427 at commit 1ebacc2.

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

[yhuai]

Remove these two lines?

[yhuai]

Thank you for the update! Overall looks good. There are two lines of unnecessary changes. Let's remove them and we can merge it once it passes jenkins.

[SparkQA]

Test build #50968 has finished for PR 10427 at commit 7038bc0.

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

[SparkQA]

Test build #50976 has finished for PR 10427 at commit 7a7b9fa.

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

[maropu]

@yhuai okay, ready to commit.

[yhuai]

LGTM. Merging to master.