naveenp2708 opened a new pull request, #59788: URL: https://github.com/apache/spark/pull/59788
### What changes were proposed in this pull request? `BatchScanExec` now grabs `spark.sql.sources.v2.bucketing.enabled` when the scan is planned and uses that value, instead of reading the config again in `DataSourceV2ScanExecBase.reportedKeyedPartitioning`. The flag is in a second parameter list so it doesn't change the node's extractor arity, which keeps all the existing `case BatchScanExec(...)` matches working. It's carried through `otherCopyArgs` and `doCanonicalize` so a rebuilt node keeps it. Scans that don't set it (streaming, continuous) still read the live config. ### Why are the changes needed? A storage-partitioned join can return wrong results, or fail, when `spark.sql.sources.v2.bucketing.enabled` is turned off between planning and execution. It needs dynamic partition pruning on the keyed scan of a join that has no `GroupPartitionsExec` above it, so two sources that already report one split per key. `PlanAdaptiveDynamicPruningFilters` builds a new `BatchScanExec` when the stage is created. The new node reads the config again, now reports no `KeyedPartitioning`, and `PushDownUtils.replanWithRuntimeFilters` hands back the splits in the connector's order without the empty partitions for the pruned keys. The other side still has the sorted, padded layout the join was planned with, so the sort-merge join lines up the wrong partitions. With `p.price > 0.0` it returns 2 rows instead of 7. When a runtime filter prunes splits (`p.price > 5.0`) it fails with `Can't zip RDDs with unequal numbers of partitions`. This is the third one of these (SPARK-59279, SPARK-60044): a config read again after planning changes what a planned node does. I kept the fix the same shape as those two, freezing the one config in the one node. ### Does this PR introduce _any_ user-facing change? No. It only changes the case where the config flips between planning and execution, which was giving wrong results before. ### How was this patch tested? Added a test in `KeyGroupedPartitioningSuite`. It builds the seven-key identity-partitioned join with DPP on the keyed scan, asserts the planned shape is actually the storage-partitioned join (sort-merge join, no shuffle, no `GroupPartitionsExec`, DPP on the scan), then turns bucketing off and runs it. It covers both symptoms from the description: the 2-instead-of-7 case and the split-pruning failure. On master the test fails; with the fix it passes. Also ran the SPJ and `BatchScanExec` identity suites (`KeyGroupedPartitioning*`, `GroupPartitionsExecSuite`, `SparkPlanSuite`, `SameResultSuite`, `EnsureRequirementsSuite`, `DataSourceV2OptionSuite`, `SQLAppStatusListenerSuite`) and scalastyle. ### Was this patch authored or co-authored using generative AI tooling? NO -- 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]
