[
https://issues.apache.org/jira/browse/SPARK-58996?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-58996:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> `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]
