XiDuo You created SPARK-58996:
---------------------------------
Summary: Partially clustered storage-partitioned join duplicates
rows when EnsureRequirements re-runs
Key: SPARK-58996
URL: https://issues.apache.org/jira/browse/SPARK-58996
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: XiDuo You
`EnsureRequirements` is not idempotent for a storage-partitioned join that uses
a partially
clustered distribution. Re-running it on a plan it already produced stacks a
second
`GroupPartitionsExec` on top of the first, and the result duplicates rows.
AQE hands the whole plan back to `EnsureRequirements` after rewriting some
other join --
`ConvertSortMergeJoinToShuffledHashJoin` and `OptimizeSkewedJoin` both do this.
On that second
pass a join child is `SortExec(GroupPartitionsExec(...))` rather than a bare
scan. A partially
clustered `KeyedPartitioning` reports `isGrouped = false` by design, so the
distribution step
treats it as satisfied "only after grouping" and adds a plain
`GroupPartitionsExec` on top;
`applyGroupPartitions` then writes the join's `expectedPartitionKeys` and
`distributePartitions`
into that fresh outer node.
The alignment is therefore re-derived from an already-aligned layout. The inner
node replicates
an input partition across the expected partitions, and the outer node
concatenates those replicas
back into a single partition before replicating again, so every row of the
replicated side is
emitted twice.
Reproduction, against a DSv2 catalog that reports `KeyGroupedPartitioning`:
{code}
spark.sql.sources.v2.bucketing.enabled=true
spark.sql.sources.v2.bucketing.pushPartValues.enabled=true
spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled=true
spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold=100m
CREATE TABLE sp1 (id BIGINT, data STRING) PARTITIONED BY (id);
INSERT INTO sp1 VALUES (1, 'aa'), (2, 'bb');
ALTER TABLE sp1 ADD COLUMN extra STRING; -- opens a second split for id = 1
INSERT INTO sp1 VALUES (1, 'ab', 'x');
CREATE TABLE sp2 (id BIGINT, data STRING) PARTITIONED BY (id);
INSERT INTO sp2 VALUES (1, 'p'), (2, 'q');
CREATE TABLE np1 (id BIGINT, data STRING);
INSERT INTO np1 VALUES (7, 'x');
CREATE TABLE np2 (id BIGINT, data STRING);
INSERT INTO np2 VALUES (7, 'y');
SELECT /*+ MERGE(a, b) */ a.id AS k FROM sp1 a JOIN sp2 b ON a.id = b.id
UNION ALL
SELECT c.id AS k FROM np1 c JOIN np2 d ON c.id = d.id;
{code}
This returns {{(1, 1, 1, 1, 2, 7)}}; the correct answer is {{(1, 1, 2, 7)}}.
Two details are load-bearing in the reproduction:
* The {{np1}}/{{np2}} branch exists only to create a materialized shuffle
stage, which is what
makes `ConvertSortMergeJoinToShuffledHashJoin` fire and re-run
`EnsureRequirements` over the
whole plan. The storage-partitioned join has no shuffle of its own, so it
cannot trigger the
re-run by itself.
* {{id = 1}} must map to two input splits. In the in-memory test catalog the
`ALTER TABLE`
between the two inserts changes the write schema and opens the second split;
on a real
connector this corresponds to a partition value with more than one data file,
where the files
are not combined into a single task.
A single pass is self-consistent: within one `ensureDistributionAndOrdering`
call the
distribution step creates the `GroupPartitionsExec` and `applyGroupPartitions`
rewrites that same
node. The stacking only appears once the rule is applied to its own output.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
