Copilot commented on code in PR #12482:
URL: https://github.com/apache/gluten/pull/12482#discussion_r3549259006
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala:
##########
@@ -2034,4 +2034,55 @@ class GlutenKeyGroupedPartitioningSuite
}
}
+ testGluten("SPARK-54439: KeyGroupedPartitioning and join key size mismatch")
{
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ // `time` and `item_id` in the required `ClusteredDistribution` for
`purchases`, but `item` is
+ // storage partitioned only by `id`
+ val df = createJoinTestDF(Seq("arrive_time" -> "time", "id" ->
"item_id"))
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.size == 1, "only shuffle one side not report
partitioning")
+
Review Comment:
These new SPARK-54439 tests only assert the number of shuffles, but they
don't verify the intended Gluten behavior described in the PR
(KeyGroupedPartitioning shuffle must fall back to a vanilla ShuffleExchangeExec
and must not be offloaded to ColumnarShuffleExchangeExec). As written, the
tests could pass even if a KeyGroupedPartitioning shuffle is incorrectly
handled or not exercised.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/connector/GlutenKeyGroupedPartitioningSuite.scala:
##########
@@ -2034,4 +2034,55 @@ class GlutenKeyGroupedPartitioningSuite
}
}
+ testGluten("SPARK-54439: KeyGroupedPartitioning and join key size mismatch")
{
+ val items_partitions = Array(identity("id"))
+ createTable(items, itemsColumns, items_partitions)
+
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 'bb', 10.0, cast('2020-01-01' as timestamp)), " +
+ "(4, 'cc', 15.5, cast('2020-02-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2020-02-01' as timestamp))")
+
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ // `time` and `item_id` in the required `ClusteredDistribution` for
`purchases`, but `item` is
+ // storage partitioned only by `id`
+ val df = createJoinTestDF(Seq("arrive_time" -> "time", "id" ->
"item_id"))
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.size == 1, "only shuffle one side not report
partitioning")
+
+ checkAnswer(df, Seq(Row(1, "aa", 40.0, 42.0)))
+ }
+ }
+
+ testGluten("SPARK-54439: KeyGroupedPartitioning with transform and join key
size mismatch") {
+ // Do not use `bucket()` in "one side partition" tests as its
implementation in
+ // `InMemoryBaseTable` conflicts with `BucketFunction`
+ val items_partitions = Array(years("arrive_time"))
+ createTable(items, itemsColumns, items_partitions)
+
+ sql(s"INSERT INTO testcat.ns.$items VALUES " +
+ "(1, 'aa', 40.0, cast('2020-01-01' as timestamp)), " +
+ "(1, 'bb', 10.0, cast('2021-01-01' as timestamp)), " +
+ "(4, 'cc', 15.5, cast('2021-02-01' as timestamp))")
+
+ createTable(purchases, purchasesColumns, Array.empty)
+ sql(s"INSERT INTO testcat.ns.$purchases VALUES " +
+ "(1, 42.0, cast('2020-01-01' as timestamp)), " +
+ "(3, 19.5, cast('2021-02-01' as timestamp))")
+
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ // `item_id` and `time` in the required `ClusteredDistribution` for
`purchases`, but `item` is
+ // storage partitioned only by `year(arrive_time)`
+ val df = createJoinTestDF(Seq("id" -> "item_id", "arrive_time" ->
"time"))
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.size == 1, "only shuffle one side not report
partitioning")
+
Review Comment:
Same issue here: the test asserts shuffle count but doesn't validate that
any KeyGroupedPartitioning shuffle falls back to a vanilla ShuffleExchangeExec
(and is not offloaded to ColumnarShuffleExchangeExec). Adding explicit
assertions like in the existing GLUTEN-10992 test makes this SPARK-54439
coverage meaningful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
