[
https://issues.apache.org/jira/browse/SPARK-60782?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18137497#comment-18137497
]
Anil K edited comment on SPARK-60782 at 11/3/26, 6:31 PM:
----------------------------------------------------------
A key's collation is part of a string's type, so the same applies to it: a
Pandas cogroup of a {{STRING}} key with a {{STRING COLLATE UTF8_LCASE}} key
isn't coerced either.
Measured by [~ptoth] on master (fce7d51a376) withÂ
{{{}FlatMapCoGroupsInPandasExec{}}}, see
[https://github.com/apache/spark/pull/59760#discussion_r4457220509:]
- The cogroup sorts the right side by {{{}UTF8_LCASE{}}}, and
{{CoGroupedIterator}} then merges the two sides by the left key's binary
ordering.
- With keys {{B}} and {{a}} on both sides, {{B}} came out twice, once with
only its left rows and once with only its right rows. Lowercase keys paired
correctly.
On the plain hash path, each side also hashes its key by its own collation, so
equal keys can land in different partitions, as with {{INT}} and {{{}BIGINT{}}}.
So a fix should coerce, or reject, keys whose collations differ too, not only
keys of different types. The typed Scala cogroup isn't affected, since both
sides group by one key type.
With SPARK-60027 ([https://github.com/apache/spark/pull/59760], open), a
storage-partitioned layout that would have to evaluate a partition transform
over such a key fails while planning with
{{{}STORAGE_PARTITION_JOIN_KEY_TYPE_MISMATCH{}}}. That covers only the path
where a side is shuffled onto a keyed layout; the hash path still returns wrong
groups.
was (Author: metanil):
A key's collation is part of a string's type, so the same applies to it: a
Pandas cogroup of a {{STRING}} key with a {{STRING COLLATE UTF8_LCASE}} key
isn't coerced either.
Measured by [~petertoth] on master (fce7d51a376) withÂ
{{{}FlatMapCoGroupsInPandasExec{}}}, see
[https://github.com/apache/spark/pull/59760#discussion_r4457220509:]
- The cogroup sorts the right side by {{{}UTF8_LCASE{}}}, and
{{CoGroupedIterator}} then merges the two sides by the left key's binary
ordering.
- With keys {{B}} and {{a}} on both sides, {{B}} came out twice, once with
only its left rows and once with only its right rows. Lowercase keys paired
correctly.
On the plain hash path, each side also hashes its key by its own collation, so
equal keys can land in different partitions, as with {{INT}} and {{{}BIGINT{}}}.
So a fix should coerce, or reject, keys whose collations differ too, not only
keys of different types. The typed Scala cogroup isn't affected, since both
sides group by one key type.
With SPARK-60027 ([https://github.com/apache/spark/pull/59760], open), a
storage-partitioned layout that would have to evaluate a partition transform
over such a key fails while planning with
{{{}STORAGE_PARTITION_JOIN_KEY_TYPE_MISMATCH{}}}. That covers only the path
where a side is shuffled onto a keyed layout; the hash path still returns wrong
groups.
> 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]
