sunchao opened a new pull request, #57437: URL: https://github.com/apache/spark/pull/57437
### What changes were proposed in this pull request? Tracks [SPARK-58265](https://issues.apache.org/jira/browse/SPARK-58265). Extend dynamic partition pruning so it can reuse values from the complete rows of an existing broadcast hash relation, not only the hash keys used to build that relation. Spark still tries ordinary broadcast-key reuse first. If that cannot satisfy a broadcast-only pruning expression, it can follow a deterministic projection back to an earlier inner equi-join and check whether the exact source, complete ordered hash keys, and standard hash mode match a broadcast that the query already needs. If they match, a separate physical subquery projects the required expression from the broadcast's stored rows, removes null and duplicate values, and supplies the resulting domain to the existing pruning predicate. The optimization does not build another broadcast, rerun the filtering plan, or change the constructors and identities of Spark's existing execution nodes. Broadcast rows that do not survive the earlier join can contribute extra pruning values. This is intentional: the projected domain is a safe superset, and the original joins still decide which rows appear in the result. Source statistics, input-row counts, and projected-value sizes bound the work. If the source cannot be verified, an expression is unsupported, statistics are unavailable, a limit is exceeded, or optional projection fails, Spark discards the entire domain and continues without the extra filter. In particular, an unavailable domain is not treated as an empty domain, including Spark's iterative V2 partition-filter pushdown. The change covers both adaptive and nonadaptive planning and supports Spark's default ANSI mode. It is deliberately limited to dynamic partition pruning; broadcast-backed Bloom filters remain separate work under SPARK-40909. ### Why are the changes needed? The existing reuse path knows how to extract a broadcast's join keys. A query can nevertheless already hold the exact information needed to prune a later partitioned scan in another column of the same broadcast. For example, suppose `wide_history` is partitioned by `ds`: ```sql WITH cohorts AS ( SELECT /*+ BROADCAST(d) */ d.cohort_date, u.user_id FROM dates d JOIN users u ON d.cohort_ds = u.cohort_ds AND d.region = u.region WHERE u.active ) SELECT c.user_id, h.payload FROM cohorts c LEFT JOIN wide_history h ON h.user_id = c.user_id AND h.ds = date_format(date_add(c.cohort_date, 6), 'yyyy-MM-dd') ``` The first join broadcasts `dates` under `(cohort_ds, region)`. The later join needs history partitions derived from `cohort_date`, which is present in every broadcast value row but is not a hash key. For cohort dates `2024-01-01` and `2024-01-08`, the required history partitions are `2024-01-07` and `2024-01-14`. A third, unmatched broadcast date `2024-01-29` can also include partition `2024-02-04`. Reading that extra partition is safe; dropping either required partition would not be. Today Spark either runs the filtering-side plan again to produce pruning values or, when `spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=true`, drops the pruning filter. Reusing the existing broadcast value rows avoids rerunning that plan and reading every target partition. ### Does this PR introduce _any_ user-facing change? Yes. It adds an internal, opt-in dynamic partition pruning optimization: ```properties spark.sql.optimizer.dynamicPartitionPruning.broadcastProjection.enabled=true ``` The optimization is disabled by default. When enabled, eligible queries can scan fewer partitions without changing SQL syntax, query results, or existing direct broadcast reuse. ### How was this patch tested? Regression tests exercise the motivating derived-date query with ANSI mode both enabled and disabled; composite and single-column broadcast keys; dense and sparse long-key hash relations; duplicate and null broadcast rows; unmatched broadcast rows and safe-superset pruning; feature-disabled behavior; resource-limit fail-open; preservation of Catalyst case-class identity and metadata across rewrites; and the first and iterative second passes of V2 partition filtering. The shared DPP regressions run against V1, V2, and runtime-filtering V2 sources with adaptive execution both enabled and disabled. Separate enhanced V2 connector tests isolate the iterative partition-filtering pass with adaptive execution disabled and verify both successful pruning and unavailable-domain fallback. The DPP matrix, ordinary subquery suite, runtime-filter injection suite, and existing Bloom filter suite passed locally: 390 executed tests succeeded, with five unrelated existing tests ignored. ```bash SERIAL_SBT_TESTS=1 ./build/sbt \ 'catalyst/testOnly org.apache.spark.sql.catalyst.expressions.DynamicPruningSubquerySuite' \ 'sql/testOnly org.apache.spark.sql.DynamicPartitionPruningV1SuiteAEOff org.apache.spark.sql.DynamicPartitionPruningV1SuiteAEOn org.apache.spark.sql.DynamicPartitionPruningV2SuiteAEOff org.apache.spark.sql.DynamicPartitionPruningV2SuiteAEOn org.apache.spark.sql.DynamicPartitionPruningV2FilterSuiteAEOff org.apache.spark.sql.DynamicPartitionPruningV2FilterSuiteAEOn org.apache.spark.sql.connector.DataSourceV2EnhancedRuntimePartitionFilterSuite' SERIAL_SBT_TESTS=1 ./build/sbt \ 'sql/testOnly org.apache.spark.sql.SubquerySuite org.apache.spark.sql.InjectRuntimeFilterSuite org.apache.spark.sql.BloomFilterAggregateQuerySuite' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-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]
