[jira] [Created] (SPARK-58996) Partially clustered storage-partitioned join duplicates rows when EnsureRequirements re-runs

["XiDuo You (Jira)" <ji...@apache.org>]

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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org