Anil K created SPARK-60782:
------------------------------
Summary: Pandas and Arrow cogroups pair groups wrongly when the
two sides' grouping keys have different types
Key: SPARK-60782
URL: https://issues.apache.org/jira/browse/SPARK-60782
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Anil K
A cogroup in PySpark checks only that both sides group by the same number of
keys (SPARK-40601). It does not check or coerce their types, so one side can
group by an {{INT}} and the other by a {{BIGINT}}:
{code:python}
left = spark.range(10).selectExpr("CAST(id AS INT) AS id", "'l' AS v") # INT
key
right = spark.range(10).selectExpr("id", "'r' AS w") #
BIGINT key
def count(l, r):
key = l["id"][0] if len(l) else r["id"][0]
return pd.DataFrame({"id": [int(key)], "l": [len(l)], "r": [len(r)]})
left.groupby("id").cogroup(right.groupby("id")) \
.applyInPandas(count, "id long, l long, r long").collect()
{code}
Every id should form one group with one row from each side, i.e. 10 groups.
Measured on PySpark 4.3.0 in local mode, with
{{spark.sql.shuffle.partitions=200}}:
||{{spark.sql.adaptive.enabled}}||{{spark.sql.adaptive.coalescePartitions.enabled}}||result||
|false|-|20 groups, none matched: each id is a left-only and a right-only group|
|true|false|20 groups, none matched|
|true|true|10 groups, all matched, since AQE coalesces the 10 rows into one
partition|
So it returns wrong results with no error whenever AQE does not put both sides
of a key into one partition, which is the case for data of any size.
{{FlatMapCoGroupsInPandasExec}} and {{FlatMapCoGroupsInArrowExec}} require each
side to be clustered on its own keys, and {{EnsureRequirements}}
hash-partitions each side on them. {{Murmur3Hash}} hashes an {{INT}} and a
{{BIGINT}} differently: with 200 partitions, 5 :: INT goes to partition 66 and
5 :: BIGINT to partition 68. So the two groups for one key value land in
different partitions, and the function is called once with the left group and
an empty right one, and once the other way round.
With {{spark.sql.sources.v2.bucketing.shuffle.enabled}} and a
storage-partitioned side, the other side can instead be shuffled onto that
side's layout, rebuilt over its own key. Then the query can fail, e.g. with a
{{ClassCastException}} when the layout's {{bucket}} declares an {{INT}} and
gets a {{BIGINT}}.
A join does not have this problem, since the analyzer coerces the two sides of
an equi-join condition to one type.
A fix could coerce the two sides' grouping keys to a common type in
{{flatMapCoGroupsInPandas}} and {{flatMapCoGroupsInArrow}}, as a join does, or
reject keys of different types with a clear error, as SPARK-40601 does for a
different number of keys.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
