Peter Toth created SPARK-60579:
----------------------------------
Summary: Fix ORDER BY returning rows in the wrong order when a
storage-partitioned layout cannot be reordered
Key: SPARK-60579
URL: https://issues.apache.org/jira/browse/SPARK-60579
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.1.3, 4.0.4
Reporter: Peter Toth
On 4.0 and 4.1, with {{spark.sql.sources.v2.bucketing.sorting.enabled=true}},
an {{ORDER BY}} that a key-grouped partitioning satisfies can return its rows
in the wrong order. {{EnsureRequirements.ensureOrdering}} accepts the
partitioning as ordered, so no range shuffle is added. In two cases it then
does not reorder the partitions:
* The child reports a {{PartitioningCollection}}, for example an inner
storage-partitioned join that outputs both sides' key columns.
{{ensureOrdering}} handles only a bare {{KeyGroupedPartitioning}}, so it
returns the plan unchanged.
* The partitioning comes from a leaf that the push of the sorted partition
values does not reach. A cached table ({{InMemoryTableScanExec}}) is one. A
checkpointed scan ({{RDDScanExec}}) should be another, by reading the code.
{code:sql}
-- xa, xb PARTITIONED BY (id): 1, 2, 3
SELECT /*+ MERGE(xa, xb) */ xa.id, xb.id
FROM xa JOIN xb ON xa.id = xb.id
ORDER BY xa.id DESC
-- ct PARTITIONED BY (id): 1, 2, 3, then spark.table("ct").cache()
SELECT id FROM ct ORDER BY id DESC
{code}
Measured on branch-4.0 (076b9d674fb) and branch-4.1 (627c1fb1758), AQE on and
off:
* Both queries return the ids as 1, 2, 3 instead of 3, 2, 1.
* With {{sorting.enabled=false}}, both return 3, 2, 1, with one range shuffle.
* The same happens with a one-side shuffle. {{sa (store, dept, data)}} is
{{PARTITIONED BY (store, dept)}} and holds {{(1, 20, 'a1')}} and {{(2, 10,
'a2')}}. {{sb}} has the same columns, is not partitioned, and holds {{(0, 10,
'b10')}} and {{(0, 20, 'b20')}}. With {{shuffle.enabled}},
{{allowJoinKeysSubsetOfPartitionKeys.enabled}} and
{{requireAllClusterKeysForCoPartition=false}}, {{SELECT sa.dept, sb.dept,
sa.data, sb.data, sa.store FROM sa JOIN sb ON sa.dept = sb.dept ORDER BY
sa.dept}} returns dept 20 before dept 10.
branch-4.2 (aef8fb7426a) and master (972b507214f) return the right order for
the two queries above. They group partitions with {{GroupPartitionsExec}}
(SPARK-55535), which also handles a {{PartitioningCollection}}.
Found while reviewing apache/spark#59599 (SPARK-60232), which also changes
{{ensureOrdering}}. SPARK-60552 is a different bug in the same function: there
the push reaches a scan below a shuffle, and rows are lost.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
