[SPARK-32628][SQL] Use bloom filter to improve dynamic partition pruning

[wangyum]

What changes were proposed in this pull request?

This PR enhances DPP to use bloom filters if spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly is disabled and build plan can't build broadcast by size and can reuse the existing shuffle exchanges.

Why are the changes needed?

Avoid job fail if spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly is disabled:

select catalog_sales.* from  catalog_sales join catalog_returns  where cr_order_number = cs_sold_date_sk and cr_returned_time_sk < 40000;

20/08/16 06:44:42 ERROR TaskSetManager: Total size of serialized results of 494 tasks (1225.3 MiB) is bigger than spark.driver.maxResultSize (1024.0 MiB)

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Unit test.

[wangyum]

cc @cloud-fan @sigmod @aokolnychyi @dongjoon-hyun @huaxingao @viirya

[dongjoon-hyun]

Thank you for pinging me, @wangyum .

[cloud-fan]

this can be over estimated. The final plan has an Aggregate which may dramatically reduce the data size.

[cloud-fan]

This is another tricky part: is reusing shuffle always better than starting a new query with column pruning?

[cloud-fan]

I agree with using bloom filters, as the size estimation can be wrong and the build size can be too large that InSubquery can't work. However, this PR contains another optimization that forces shuffle reuse when building the subquery to build bloom filter. Can we do it later with more discussions? This is a general optimization that can apply in other places as well: InSubquery DPP, bloom filter join.

[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!