[SPARK-36351][SQL] Refactor filter push down in file source v2

[huaxingao]

What changes were proposed in this pull request?

Currently in V2ScanRelationPushDown, we push the filters (partition filters + data filters) to file source, and then pass all the filters (partition filters + data filters) as post scan filters to v2 Scan, and later in PruneFileSourcePartitions, we separate partition filters and data filters, set them in the format of Expression to file source.

Changes in this PR:
When we push filters to file sources in V2ScanRelationPushDown, since we already have the information about partition column , we want to separate partition filter and data filter there.

The benefit of doing this:

• we can handle all the filter related work for v2 file source at one place instead of two (V2ScanRelationPushDown and PruneFileSourcePartitions), so the code will be cleaner and easier to maintain.
• we actually have to separate partition filters and data filters at V2ScanRelationPushDown, otherwise, there is no way to find out which filters are partition filters, and we can't push down aggregate for parquet even if we only have partition filter.
• By separating the filters early at V2ScanRelationPushDown, we only needs to check data filters to find out which one needs to be converted to data source filters (e.g. Parquet predicates, ORC predicates) and pushed down to file source, right now we are checking all the filters (both partition filters and data filters)
• Similarly, we can only pass data filters as post scan filters to v2 Scan, because partition filters are used for partition pruning only, no need to pass them as post scan filters.

In order to do this, we will have the following changes

• add pushFilters in file source v2. In this method:
  • push both Expression partition filter and Expression data filter to file source. Have to use Expression filters because we need these for partition pruning.
  • data filters are used for filter push down. If file source needs to push down data filters, it translates the data filters from Expression to Sources.Filer, and then decides which filters to push down.
  • partition filters are used for partition pruning.
• file source v2 no need to implement SupportsPushdownFilters any more, because when we separating the two types of filters, we have already set them on file data sources. It's redundant to use SupportsPushdownFilters to set the filters again on file data sources.

Why are the changes needed?

see section one

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing tests

[SparkQA]

Test build #142077 has finished for PR 33650 at commit 4876db0.

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

[SparkQA]

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46588/

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46616/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46616/

[SparkQA]

Test build #142104 has finished for PR 33650 at commit df5d9cc.

• This patch fails from timeout after a configured wait of 500m.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46700/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46700/

[SparkQA]

Test build #142188 has finished for PR 33650 at commit 4d2c4e1.

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

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46702/

[SparkQA]

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46702/

[SparkQA]

Test build #142190 has finished for PR 33650 at commit 1d3a3ef.

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

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46751/

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46751/

[SparkQA]

Test build #142244 has finished for PR 33650 at commit 9ca01b0.

• This patch fails from timeout after a configured wait of 500m.
• This patch merges cleanly.
• This patch adds no public classes.

[huaxingao]

This method returns pushed down sources.Filters and post scan Filters Expression. In the returned post scan Filters Expressions, we want the partition Filters already have been removed so we don't need a second rule (PruneFileSourcePartitions) to prune off the partition Filters.
We will separate the two types of filters for FileScanBuilder, and only pass the data Filter to ScanBuilder.pushFilters. The separated partition filters are set on FileScanBuilder in the format of Expression and are used for partition pruning in https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScan.scala#L138

[GithubZhitao]

How to get the postScan Filters in DS V2 classes(implemented ScanBuilder).
I tried to with trait SupportsPushDownFilters or SupportsPushDownV2Filters, neither of them can push postScan Filters down . Is this by design ?
If users can have all the filters , it would be great to do more optimization to our own business. It depends to us to choose how to use theses conditions? Is this better ?

[huaxingao]

The postScan filers are those that can't be pushed down to data source. Basically, we put the un-pushable filters in postScan filers so these filters will be applied by Spark.

[GithubZhitao]

Thanks for your reply. But how can I get the all filters or how to define "un-pushable" ? I run some cases and found it weird some conditions were classified to post-scan Filters, wherever I extends SupportsPushDownFilters or SupportsPushDownV2Filters. Only some isNotNull(#v) in the pushFilters function . Here is our case:
SELECT * FROM iteblog_tab1 t1 LEFT SEMI JOIN iteblog_tab2 t2 ON t1.k= t2.k AND t2._c0 < 2
In the case we partition iteblog_tab1 and iteblog_tab2 by k. Following is our calling exlain result . Why the t2._c0 < 2 are not happened in the Scan but in the post-Scan. Code version is updated to newest master branch.

== Parsed Logical Plan ==
'Project [*]
+- 'Join LeftSemi, (('t1.k = 't2.k) AND ('t2._c0 < 2))
:- 'SubqueryAlias t1
: +- 'UnresolvedRelation [iteblog_tab1], [], false
+- 'SubqueryAlias t2
+- 'UnresolvedRelation [iteblog_tab2], [], false

== Analyzed Logical Plan ==
_c0: string, k: int
Project [_c0#38, k#39]
+- Join LeftSemi, ((k#39 = k#59) AND (cast(_c0#58 as int) < 2))
:- SubqueryAlias t1
: +- SubqueryAlias iteblog_tab1
: +- View (iteblog_tab1, [_c0#38,k#39])
: +- RelationV2[_c0#38, k#39] /home/user/spark/spark-warehouse/iteblog_tab1
+- SubqueryAlias t2
+- SubqueryAlias iteblog_tab2
+- View (iteblog_tab2, [_c0#58,k#59])
+- RelationV2[_c0#58, k#59] /home/user/spark/spark-warehouse/iteblog_tab2

== Optimized Logical Plan ==
Join LeftSemi, (k#39 = k#59)
:- Filter dynamicpruning#72 [k#39]
: : +- Project [k#59]
: : +- Filter (cast(_c0#58 as int) < 2)
: : +- RelationV2[_c0#58, k#59] /home/user/spark/spark-warehouse/iteblog_tab2
: +- RelationV2[_c0#38, k#39] /home/user/spark/spark-warehouse/iteblog_tab1
+- Project [k#59]
+- Filter (cast(_c0#58 as int) < 2)
+- RelationV2[_c0#58, k#59] /home/user/spark/spark-warehouse/iteblog_tab2

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- BroadcastHashJoin [k#39], [k#59], LeftSemi, BuildRight, false
:- Project [_c0#38, k#39]
: +- BatchScan[_c0#38, k#39] CSVScan DataFilters: [], Format: csv, Location: InMemoryFileIndex(1 paths)[file:/home/user/spark/spark-warehouse/iteblog_tab1], PartitionFilters: [], ReadSchema: struct<_c0:string>, PushedFilters---------: [] RuntimeFilters: [dynamicpruningexpression(k#39 IN dynamicpruning#72)]
: +- SubqueryAdaptiveBroadcast dynamicpruning#72, 0, false, Project [k#59], [k#59]
: +- AdaptiveSparkPlan isFinalPlan=false
: +- Project [k#59]
: +- Filter (cast(_c0#58 as int) < 2)
: +- BatchScan[_c0#58, k#59] CSVScan DataFilters: [], Format: csv, Location: InMemoryFileIndex(1 paths)[file:/home/user/spark/spark-warehouse/iteblog_tab2], PartitionFilters: [], ReadSchema: struct<_c0:string>, PushedFilters---------: [] RuntimeFilters: []
+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [id=#105]
+- Project [k#59]
+- Filter (cast(_c0#58 as int) < 2)
+- BatchScan[_c0#58, k#59] CSVScan DataFilters: [], Format: csv, Location: InMemoryFileIndex(1 paths)[file:/home/i-bozhitao/spark/spark-warehouse/iteblog_tab2], PartitionFilters: [], ReadSchema: struct<_c0:string>, PushedFilters---------: [] RuntimeFilters: []

[GithubZhitao]

But it works for the sub-classes of the FileScanBuilder, such as CSVScanBuilder, which extends the class SupportsPushDownCatalyst. Is this by design ?
BTW.
SupportsPushDownFilters
SupportsPushDownFiltersV2
SupportsPushDownCatalyst
We really need some descriptions or debug works to distinguish the difference between them. Hope some merge or clarify works for this.

[cloud-fan]

Post-scan filters mean Spark should evaluate them. Pushed filters mean the data source should handle them. These two can overlap. For example, data filters are evaluated by both parquet reader and Spark (because parquet does row group filtering, not per-row).

[huaxingao]

This is not relevant to the partition filters change. Just want to add a test to cover the partially filters push down case for JDBC.

[cloud-fan]

I'd like to discuss the high-level idea first. Ideally, the process of ds v2 filter pushdown should be

1. the rule V2ScanRelationPushDown pushes down filters to the v2 source
2. the v2 source looks at the pushed filters, and returns a list of filters that Spark should evaluate. These "post scan" filters can either be filters that are not supported by this v2 source, or the v2 source can't fully finish the filtering and need Spark to do the filter again.
3. for file source, we need Spark to evaluate data filters, as statistics-based filtering can't 100% filter out unneeded data. We don't need Spark to evaluate the partition filters again as file source can filter out partitions exactly.
4. then spark continues the query compiling, applies CBO and finally submits spark jobs.

I think the key problem we should fix is: file source v2 should not return partition filters as the "post scan" filters, in the implementation of SupportsPushDownFilters.pushFilters

[huaxingao]

@cloud-fan

I think the key problem we should fix is: file source v2 should not return partition filters as the "post scan" filters, in the implementation of SupportsPushDownFilters.pushFilters

Agree.

Seems to me there are two ways to fix this:

1. in SupportsPushDownFilters.pushFilters, separate the partition filter and data filter. Something like this:

def pushFilters = {
   separate the partition filter and data filter
   set both filter on ScanBuilder so these can be passed to V2Scan at construction time
   return data filter as post scan filter
}

The problem with this approach is that not all data source implement SupportsPushDownFilters. For the data source that doesn't have pushFilters, we will need a different path to separate the partition filter and data filter.

2. separate the partition filter and data filter before calling SupportsPushDownFilters.pushFilters, and set these two types of filters on ScanBuilder. The filter passed to SupportsPushDownFilters.pushFilters only contains data filter.

I tried the first approach, and then changed to the second one.

BTW, I had a PR to push down only data filter to ORCScan #33680

[cloud-fan]

The problem with this approach is that not all data source implement SupportsPushDownFilters. For the data source that doesn't have pushFilters, we will need a different path to separate the partition filter and data filter.

Which file source doesn't implement SupportsPushDownFilters?

[cloud-fan]

I think all v2 file sources need to implement SupportsPushDownFilters, as they need to do partition pruning.

[huaxingao]

TextScanBuilder doesn't implement SupportsPushDownFilters

[cloud-fan]

By design, all the filter pushdown should happen in SupportsPushDownFilters.pushFilters, including partition pruning. We may special-case file source before, but that's not right. All v2 file sources need to implement SupportsPushDownFilters, to support partition pruning at least.

[huaxingao]

The reason that partition pruning works OK for TextScan is because we currently prune the partitions in PruneFileSourcePartitions.
Should we make FileScanBuilder implement SupportsPushDownFilters too and do partition pruning in pushFilters?

[cloud-fan]

Yes please, that's how external v2 sources should do as well.

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/46958/

[SparkQA]

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/47371/

[SparkQA]

Test build #142868 has finished for PR 33650 at commit 68ace26.

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

[cloud-fan]

Now we have an interface, I think it's better to make the API more general

def pushCatalystFilters(filter: Seq[Expression]): Seq[Expression]

The file source should split partition and date filters.

[huaxingao]

Done :)

[cloud-fan]

Nit: this change is not needed now

[cloud-fan]

the input filters are already normalized. We can simply compare the names instead of considering the case sensitivity.

[cloud-fan]

It seems we can simplify the caller side if this method takes partitionSchema: StructType

[viirya]

pushFilters(Filter[])? pushFilters(Expression[])?

[viirya]

BTW, this only returns pushed data filters, right? Could you mention it too?

[huaxingao]

Fixed.

[viirya]

Looks okay. A few minor comments.

[viirya]

Pending CI

[viirya]

Thanks. Merging to master!

[huaxingao]

Thank you all!