[
https://issues.apache.org/jira/browse/SPARK-60782?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Anil K updated SPARK-60782:
---------------------------
Affects Version/s: 4.3.0
(was: 5.0.0)
> 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: 4.3.0
> Reporter: Anil K
> Priority: Major
>
> 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]
