naveenp2708 opened a new pull request, #60009:
URL: https://github.com/apache/spark/pull/60009
***What changes were proposed in this pull request?***
FileSourceScanExec.bucketedScan read spark.sql.sources.bucketing.enabled
from the live session config every time it was evaluated, and
outputPartitioning is derived from it. When PlanAdaptiveDynamicPruningFilters
rebuilds the scan at AQE stage creation (the makeCopy triggered when the
dynamic-pruning filter is planted into partitionFilters), the lazy val
re-evaluated on the copy and re-read the config.
This PR freezes the bucketing decision at planning. FileSourceScanLike gains
a bucketingEnabled member; FileSourceScanExec captures it as a constructor
parameter (defaulting to SQLConf.get.bucketingEnabled, read once when the
planner builds the scan), so the value is carried through every makeCopy.
bucketedScan, the bucketing line in the explain metadata, the bucket-pruning
guard in createReadRDD, and doCanonicalize all use the frozen value instead of
re-reading the config.
***Why are the changes needed?***
If spark.sql.sources.bucketing.enabled is turned off between planning and
execution, the rebuilt scan reported a different outputPartitioning
(UnknownPartitioning) than the plan was built with (HashPartitioning). A
bucketed sort-merge join planned without a shuffle then tried to zip RDDs with
unequal numbers of partitions and failed:
java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of
partitions: List(2, 8)
This is the V1 file-source analogue of SPARK-60260 — the same class of
defect where a node rebuilt after planning re-reads a config and drops the
layout it was planned with.
***Does this PR introduce any user-facing change?***
Yes. A query with a bucketed sort-merge join and dynamic partition pruning,
run with AQE on and spark.sql.sources.bucketing.enabled changed from true at
planning to false before execution, previously failed:
Can't zip RDDs with unequal numbers of partitions: List(2, 8)
It now runs and returns the correct result.
***How was this patch tested?***
New test in BucketedReadSuite ("SPARK-60610: bucketed scan keeps its planned
output partitioning when spark.sql.sources.bucketing.enabled is disabled after
planning"): a bucketed, shuffle-free sort-merge join with DPP is planned with
bucketing on and executed with it off under AQE. It asserts the query returns
the correct rows and that the DPP-rebuilt scan keeps its HashPartitioning. The
test fails on master with the "Can't zip RDDs" error and passes with this
change (verified by reverting the bucketedScan read). Also ran SameResultSuite,
CoalesceBucketsInJoinSuite, and DynamicPartitionPruningV1SuiteAEOn; scalastyle
clean on main and test.
### 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]
