peter-toth opened a new pull request, #59916: URL: https://github.com/apache/spark/pull/59916
### What changes were proposed in this pull request? `InjectRuntimeFilter` adds no bloom filter on a join key that already has a DPP filter. An `onlyInBroadcast` DPP filter runs only if physical planning finds a broadcast to reuse. This PR counts such a filter only when that is likely, which needs both: - exchange reuse is on; - the join is likely to broadcast the filter's build side, or the filter has a broadcast value projection (SPARK-58265). The rule guesses the broadcast side as `JoinSelection` picks it. When such a filter does not count, physical planning usually replaces it with `true`, and the key can now get a bloom filter. ### Why are the changes needed? With the default `spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly`, `PartitionPruning` adds an `onlyInBroadcast` DPP filter on a join key that the scan offers for runtime filtering, such as a partition column. `InjectRuntimeFilter` then skips that key. In a shuffle join or a storage-partitioned join, physical planning replaces the DPP filter with `true`. So the key ends up with no runtime filter at all. For example: - a V1 table partitioned by the join key, in a shuffle join; - a V2 scan that offers the source column of a `bucket(N, id)` partition transform for runtime filtering, as Iceberg's does, in a shuffle join and in a storage-partitioned join. The same happens in a broadcast join that broadcasts the pruned side, and with exchange reuse off. Discussed on #58635: https://github.com/apache/spark/pull/58635#issuecomment-6468432320 ### Does this PR introduce _any_ user-facing change? Yes, a plan change. Such a join key can now get a runtime bloom filter, where it had no runtime filter before. The results stay the same. ### How was this patch tested? `InjectRuntimeFilterSuite`: - The two existing DPP tests used a shuffle join with an `onlyInBroadcast` DPP filter, which is the case this PR changes. They now turn `reuseBroadcastOnly` off, so their DPP filter runs as its own subquery and still keeps the bloom filter off its key. - New tests: - a shuffle join gets the bloom filter, and physical planning replaces the DPP filter with `true`, with AQE on and off; - a broadcast join that broadcasts the pruned side gets the bloom filter, and the DPP filter is `true`; - with exchange reuse off, a broadcast join gets the bloom filter, and the DPP filter is `true`; - no bloom filter when the DPP filter reuses the broadcast of a broadcast join that has a shuffle below it; - with a broadcast value projection and AQE off, no bloom filter, and the DPP filter reads the projected broadcast values. With the projection off, the bloom filter. `StorageFilterPushDownSuite`: two tests turned DPP off, so that DPP would not take the bloom filter's place. They now run with DPP on. A scratch run, not part of this PR, with AQE on and off: - a V1 table partitioned by the join key, in a shuffle join; - a V2 scan that offers the source column of `bucket(4, id)` for runtime filtering, in a shuffle join and in a storage-partitioned join. With DPP on, each gets the bloom filter, and the storage-partitioned join still plans no shuffle. The TPC-DS plan stability suites pass without golden file changes. Known limits: - The rule guesses the broadcast side from the statistics and hints the optimizer sees. When physical planning picks another side, or another join in the plan broadcasts the same build side, the key can get both filters, or neither as before this PR. Both filters cost the bloom filter's subquery, and the results stay the same. - A DPP filter with a broadcast value projection (`spark.sql.optimizer.dynamicPartitionPruning.broadcastProjection.enabled`, off by default) still keeps the bloom filter out. Physical planning still replaces it with `true` when it finds no broadcast to project from: - the inner join that the projection reads from is not a broadcast join; - under AQE, the plan under that broadcast has no shuffle, see SPARK-60703. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code with Claude Opus 5.5 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
