[
https://issues.apache.org/jira/browse/SPARK-60579?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-60579:
-------------------------------
Affects Version/s: 4.0.5
4.1.4
(was: 4.1.3)
(was: 4.0.4)
> 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.0.5, 4.1.4
> Reporter: Peter Toth
> Priority: Major
>
> 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]
