[SPARK-32444][SQL] Infer filters from DPP

[wangyum]

What changes were proposed in this pull request?

This pr add support infer filters from DPP.

For this test suite.
Table fact_stats is partitioned by store_id and table code_stats is partitioned by store_id.
DPP add a new predicate: fact_stats.store_id IN dynamicpruning#2723.
We can infer code_stats.store_id IN dynamicpruning#2723 base on t1.store_id = t2.store_id.

Why are the changes needed?

Improve query performance.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Unit test.

[SparkQA]

Test build #126568 has finished for PR 29243 at commit 69be17b.

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

[SparkQA]

Test build #126574 has finished for PR 29243 at commit f001b27.

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

[SparkQA]

Test build #126641 has finished for PR 29243 at commit bcc81be.

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

[bart-samwel]

The partition pruning already prunes all partitions that don't satisfy that predicate, so all remaining rows will satisfy that predicate, right? Where is the performance gain then?

[wangyum]

Before this pr. We can only prune on fact_stats.store_id column:

== Physical Plan ==
*(3) Project [store_id#2705, code#2706, product_id#2708L]
+- *(3) BroadcastHashJoin [cast(store_id#2705 as bigint)], [store_id#2709L], Inner, BuildRight, false
:- *(3) Project [store_id#2705, code#2706]
: +- *(3) BroadcastHashJoin [store_id#2705], [store_id#2707], Inner, BuildRight, false
: :- *(3) ColumnarToRow
: : +- FileScan parquet default.fact_stats[store_id#2705] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [isnotnull(store_id#2705), dynamicpruningexpression(cast(store_id#2705 as bigint) IN subquery#2716)], PushedFilters: [], ReadSchema: struct<>
: : +- Subquery subquery#2716, [id=#245]
: : +- *(2) HashAggregate(keys=[store_id#2709L#2715L], functions=[])
: : +- Exchange hashpartitioning(store_id#2709L#2715L, 5), true, [id=#241]
: : +- *(1) HashAggregate(keys=[store_id#2709L AS store_id#2709L#2715L], functions=[])
: : +- *(1) Filter ((isnotnull(product_id#2708L) AND (product_id#2708L < 3)) AND isnotnull(store_id#2709L))
: : +- *(1) ColumnarToRow
: : +- FileScan parquet default.product[product_id#2708L,store_id#2709L] Batched: true, DataFilters: [isnotnull(product_id#2708L), (product_id#2708L < 3), isnotnull(store_id#2709L)], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(product_id), LessThan(product_id,3), IsNotNull(store_id)], ReadSchema: struct<product_id:bigint,store_id:bigint>
: +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[1, int, true] as bigint)),false), [id=#275]
: +- *(1) ColumnarToRow
: +- FileScan parquet default.code_stats[code#2706,store_id#2707] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [isnotnull(store_id#2707)], PushedFilters: [], ReadSchema: struct<code:int>
+- BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#283]
+- *(2) Filter ((isnotnull(product_id#2708L) AND (product_id#2708L < 3)) AND isnotnull(store_id#2709L))
+- *(2) ColumnarToRow
+- FileScan parquet default.product[product_id#2708L,store_id#2709L] Batched: true, DataFilters: [isnotnull(product_id#2708L), (product_id#2708L < 3), isnotnull(store_id#2709L)], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(product_id), LessThan(product_id,3), IsNotNull(store_id)], ReadSchema: struct<product_id:bigint,store_id:bigint>

After this pr. We can also prune on code_stats.store_id column:

== Physical Plan ==
*(3) Project [store_id#2705, code#2706, product_id#2708L]
+- *(3) BroadcastHashJoin [cast(store_id#2705 as bigint)], [store_id#2709L], Inner, BuildRight, false
:- *(3) Project [store_id#2705, code#2706]
: +- *(3) BroadcastHashJoin [store_id#2705], [store_id#2707], Inner, BuildRight, false
: :- *(3) ColumnarToRow
: : +- FileScan parquet default.fact_stats[store_id#2705] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [isnotnull(store_id#2705), dynamicpruningexpression(cast(store_id#2705 as bigint) IN subquery#2716)], PushedFilters: [], ReadSchema: struct<>
: : +- Subquery subquery#2716, [id=#250]
: : +- *(2) HashAggregate(keys=[store_id#2709L#2715L], functions=[])
: : +- Exchange hashpartitioning(store_id#2709L#2715L, 5), true, [id=#246]
: : +- *(1) HashAggregate(keys=[store_id#2709L AS store_id#2709L#2715L], functions=[])
: : +- *(1) Filter ((isnotnull(product_id#2708L) AND (product_id#2708L < 3)) AND isnotnull(store_id#2709L))
: : +- *(1) ColumnarToRow
: : +- FileScan parquet default.product[product_id#2708L,store_id#2709L] Batched: true, DataFilters: [isnotnull(product_id#2708L), (product_id#2708L < 3), isnotnull(store_id#2709L)], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(product_id), LessThan(product_id,3), IsNotNull(store_id)], ReadSchema: struct<product_id:bigint,store_id:bigint>
: +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[1, int, true] as bigint)),false), [id=#309]
: +- *(1) ColumnarToRow
: +- FileScan parquet default.code_stats[code#2706,store_id#2707] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [isnotnull(store_id#2707), dynamicpruningexpression(cast(store_id#2707 as bigint) IN subquery#2718)], PushedFilters: [], ReadSchema: struct<code:int>
: +- Subquery subquery#2718, [id=#279]
: +- *(2) HashAggregate(keys=[store_id#2709L#2717L], functions=[])
: +- Exchange hashpartitioning(store_id#2709L#2717L, 5), true, [id=#275]
: +- *(1) HashAggregate(keys=[store_id#2709L AS store_id#2709L#2717L], functions=[])
: +- *(1) Filter ((isnotnull(product_id#2708L) AND (product_id#2708L < 3)) AND isnotnull(store_id#2709L))
: +- *(1) ColumnarToRow
: +- FileScan parquet default.product[product_id#2708L,store_id#2709L] Batched: true, DataFilters: [isnotnull(product_id#2708L), (product_id#2708L < 3), isnotnull(store_id#2709L)], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(product_id), LessThan(product_id,3), IsNotNull(store_id)], ReadSchema: struct<product_id:bigint,store_id:bigint>
+- BroadcastExchange HashedRelationBroadcastMode(List(input[1, bigint, false]),false), [id=#317]
+- *(2) Filter ((isnotnull(product_id#2708L) AND (product_id#2708L < 3)) AND isnotnull(store_id#2709L))
+- *(2) ColumnarToRow
+- FileScan parquet default.product[product_id#2708L,store_id#2709L] Batched: true, DataFilters: [isnotnull(product_id#2708L), (product_id#2708L < 3), isnotnull(store_id#2709L)], Format: Parquet, Location: InMemoryFileIndex[file:/Users/yumwang/spark/SPARK-27227/sql/core/spark-warehouse/org.apache.spark..., PartitionFilters: [], PushedFilters: [IsNotNull(product_id), LessThan(product_id,3), IsNotNull(store_id)], ReadSchema: struct<product_id:bigint,store_id:bigint>

[bart-samwel]

Ah, that makes sense. Thanks for the example!

[maryannxue]

We don't just apply DPP to any column, right? Shouldn't we check that the inferred filter is applied on partition columns?

In that sense we should fix the DPP rule for it to be apply the same dimension table filter on partition columns from different tables. In other words, the DPP rule itself should generate a filter on fact_stats.store_id and another on code_stats.store_id.

[wangyum]

Yes. We should only infer the filter to partition column.

The DPP rule itself could not generate a filter on code_stats.store_id because it should be after PushDownPredicates.

I add logic to CleanupDynamicPruningFilters to remove all filters with DynamicPruning that are not filtered on the partition column.

[SparkQA]

Test build #127734 has finished for PR 29243 at commit 98f7275.

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

[maryannxue]

I add logic to CleanupDynamicPruningFilters to remove all filters with DynamicPruning that are not filtered on the partition column.

It is better now.
Still, the pruningHasBenefit flag carried over from the original DPP filter might not be correct for the new filter.

[SparkQA]

Test build #127835 has finished for PR 29243 at commit c436bc4.

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

[wangyum]

I have fixed pruningHasBenefit for the new filter, and this change will infer more filters for subquery.

[SparkQA]

Test build #127987 has finished for PR 29243 at commit b5bb9a2.

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

[wangyum]

This change is similar to #22778.

[wangyum]

retest this please

[SparkQA]

Test build #127990 has finished for PR 29243 at commit b5bb9a2.

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

[github-actions]

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

[SparkQA]

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

[SparkQA]

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

[SparkQA]

Test build #135118 has finished for PR 29243 at commit b5bb9a2.

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

[SparkQA]

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

[SparkQA]

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

[SparkQA]

Test build #135125 has finished for PR 29243 at commit 9e9a633.

• 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/39717/

[SparkQA]

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

[SparkQA]

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

[SparkQA]

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

[SparkQA]

Test build #135136 has finished for PR 29243 at commit ba71ead.

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

[SparkQA]

Test build #135137 has finished for PR 29243 at commit c63b162.

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