Robert Joseph Evans created SPARK-45243:
-------------------------------------------
Summary: RADIX sort is not stable and can produce different
results for first/collect_list aggs
Key: SPARK-45243
URL: https://issues.apache.org/jira/browse/SPARK-45243
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 3.3.0
Reporter: Robert Joseph Evans
I just set the version as 3.3.0. I think it is all versions, but I have not
tested it to be sure.
I simple query like
{code:java}
spark.read.parquet("/tmp/TEST").groupBy("a").agg(first(col("b"))).show()
{code}
can produce different results depending on if RADIX sort is enabled or not. In
this case "a" is a LONG and "b" is a STRING. The STRING forces the aggregation
to be a sort based aggregation and b being a long causes the sort to default to
a RADIX sort.
In practice first, last, and collect_list are not deterministic because the
order that a shuffle arrives to a task is a race so the order that the rows
arrive after the second stage of the aggregation is totally up in the air.
first and last might just be called pick_one. But in this case the data is
small enough that there is a single partition so it should be deterministic.
But it is not.
{code:java}
scala> spark.read.parquet("/tmp/TEST").show(100, false)
+--------------------+----+
|a |b |
+--------------------+----+
|4080731634774120135 |HH |
|-7996385019137306797|AA |
|4891386765580059730 |BI |
|-2578026341565473682|DE |
|-7264635988756013877|CH |
|5656737394922367923 |AG |
|-6183011807271780569|BG |
|109827782918242415 |CD |
|-4058328039203991995|FA |
|null |FG |
|4080731634774120135 |ID |
|-7996385019137306797|GG |
|4891386765580059730 |AC |
|-2578026341565473682|null|
|-7264635988756013877|HF |
|5656737394922367923 |II |
|-6183011807271780569|FC |
|109827782918242415 |DI |
|-4058328039203991995|IH |
|null |FE |
|4080731634774120135 |HA |
|-7996385019137306797|ID |
|4891386765580059730 |GI |
|-2578026341565473682|GB |
|-7264635988756013877|EC |
|5656737394922367923 |DA |
|-6183011807271780569|BB |
|109827782918242415 |AE |
|-4058328039203991995|FE |
|null |AE |
|4080731634774120135 |BC |
|-7996385019137306797|HF |
+--------------------+----+
scala> spark.read.parquet("/tmp/TEST").groupBy("a").agg(first(col("b"))).show()
+--------------------+--------+
| a|first(b)|
+--------------------+--------+
| null| FG|
|-7996385019137306797| GG|
|-7264635988756013877| CH|
|-6183011807271780569| BG|
|-4058328039203991995| FA|
|-2578026341565473682| DE|
| 109827782918242415| CD|
| 4080731634774120135| HH|
| 4891386765580059730| AC|
| 5656737394922367923| AG|
+--------------------+--------+
scala> spark.conf.set("spark.sql.sort.enableRadixSort", false)
scala> spark.read.parquet("/tmp/TEST").groupBy("a").agg(first(col("b"))).show()
+--------------------+--------+
| a|first(b)|
+--------------------+--------+
| null| FG|
|-7996385019137306797| AA|
|-7264635988756013877| CH|
|-6183011807271780569| BG|
|-4058328039203991995| FA|
|-2578026341565473682| DE|
| 109827782918242415| CD|
| 4080731634774120135| HH|
| 4891386765580059730| BI|
| 5656737394922367923| AG|
+--------------------+--------+
{code}
Here the values for -7996385019137306797 changed from GG with radix sort on to
AA with it off. AA is technially correct, because it appears on line 2 of the
input where as GG shows up on line 12.
I honestly don't know if Spark expects the sort to be stable or not. Looking at
the code SortExec and UnsafeExternalSorter do not make any claims about being
stable and https://issues.apache.org/jira/browse/SPARK-23973 indicates that
sort is not stable, so this might just works as designed. I just find it odd
that in most cases it is stable, so I guess that was just by accident.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
