Native Iceberg scan returns incorrect results for storage-partitioned joins

[lyne7-sc]

Describe the bug

Native Iceberg scans do not preserve Spark's key-grouped input partitions. This can cause incorrect results in storage-partitioned joins.

To Reproduce

Create two Iceberg tables partitioned by the join key:

CREATE TABLE local.db.t_left (id INT, p INT)
USING iceberg
PARTITIONED BY (p);

INSERT INTO local.db.t_left VALUES (0, 0), (1, 1);

CREATE TABLE local.db.t_right (value INT, p INT)
USING iceberg
PARTITIONED BY (p);

INSERT INTO local.db.t_right VALUES (10, 0), (11, 0), (12, 1);

Enable storage-partitioned joins:

spark.sql.sources.v2.bucketing.enabled=true
spark.sql.iceberg.planning.preserve-data-grouping=true

Run a sort-merge join:

SELECT /*+ MERGE(l, r) */ l.id, l.p, r.value
FROM local.db.t_left l
JOIN local.db.t_right r ON l.p = r.p;

Expected behavior

The query returns:

(0, 0, 10)
(0, 0, 11)
(1, 1, 12)

Actual result

The query returns only one row:

(0, 0, 11)

The expected rows (0, 0, 10) and (1, 1, 12) are missing.

Screenshots

Additional context