peter-toth opened a new pull request, #59234:
URL: https://github.com/apache/spark/pull/59234
### Differences from #53098
This is the branch-4.0 backport of #53098. It reached master and branch-4.1
(#53132), but not branch-4.0. The description below is the original one. On
this branch:
- `KeyGroupedPartitionedScan` does not exist here.
`canPushDownSPJParamsToScan` asks for a `BatchScanExec` instead, the node
`populateCommonPartitionInfo` pushes the SPJ parameters into. On branch-4.1,
`BatchScanExec` is the only implementer of that trait.
- The tests are the branch-4.1 ones. They conflict only at the end of
`KeyGroupedPartitioningSuite` and `EnsureRequirementsSuite`.
`DummySparkPlanWithBatchScanChild` is already on this branch, copied in by the
SPARK-59080 backport, so its comment that pointed at this ticket is dropped.
- Three SPARK-59050 tests now match branch-4.1:
- Without this change, this branch storage-partitions a keyed side above a
one-side-shuffled join of a parquet table. When the partition counts differ,
the query fails with `Can't zip RDDs with unequal numbers of partitions` or
with the `PartitioningCollection` error below. So the SPARK-59050 backport
(#58572) gave two of these tests equal key sets, and kept the third at 1
shuffle.
- With this change, the first join's output reads a parquet table, so the
second join shuffles one side instead, as on branch-4.1. The rows are the same.
- Measured without this change: two of them fail with those two errors,
and the third gets 1 shuffle instead of 2.
- As on branch-4.1, every leaf under a join side must now be a
`BatchScanExec`. So some joins that storage-partitioned here before now shuffle:
- a keyed side that also reads another kind of source, for example through
a broadcast join with a parquet build side. Measured with every conf at its
default, such a three-table join goes from 0 to 2 shuffles;
- a cached or checkpointed keyed side, even when its partition values
match the other side's.
- Master and branch-4.2 dropped this check later, with the SPARK-55535
refactor.
- The bug reproduces on branch-4.0 without this change, with
`spark.sql.sources.v2.bucketing.enabled` on:
- both sides checkpointed: the join returns 1 of 2 rows;
- one side checkpointed: planning fails with `PartitioningCollection
requires all of its partitionings have the same numPartitions`.
- New tests: three of the four fail on branch-4.0 without this change.
"checkpointed scans avoid shuffles for aggregates" passes, since it needs no
storage-partitioned join. "checkpointed scans can't shuffle other children on
SPJ" fails on the shuffle count only, 2 instead of 3.
- Each half was removed on its own. Without the `candidateSpecs` filter,
"checkpointed scans can't shuffle other children on SPJ" fails. Without the
check in `checkKeyGroupCompatible`, "checkpointed scans aren't used for SPJ"
and "checkpointed scans can be shuffled by children on SPJ" fail.
- Ran the `KeyGroupedPartitioning*` suites, `EnsureRequirementsSuite`,
`PlannerSuite`, `WriteDistributionAndOrderingSuite`,
`BucketedReadWithoutHiveSupportSuite`, the plan stability suites, the
`*JoinSuite` suites, `AdaptiveQueryExecSuite`, `DataSourceV2Suite` and
`CheckpointSuite` on branch-4.0, 1313 tests in all, plus `dev/lint-scala`.
- The backport was prepared with Claude Code. Generated-by: Claude Code
(Claude Opus 5.5)
### What changes were proposed in this pull request?
When a KeyGroupedShuffleSpec is used to shuffle another child of a JOIN, we
must be able to push down JOIN keys or partition values to be able to ensure
that both children have matching partitioning. If one child reports a
KeyGroupedPartitioning but we can't push down these values (for example, if the
child was a key-grouped scan that was checkpointed), then this information
cannot be pushed down to the child scan and we should avoid using this shuffle
spec to shuffle other children.
### Why are the changes needed?
Prevents potential correctness issue when key-grouped partitioning is used
on a checkpointed RDD.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
See test changes.
### 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]
