From 164ee00f90866a217a90be2afe2630781445fd6c Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Fri, 25 Sep 2026 07:35:38 -0600
Subject: [PATCH] test: accept CometHashAggregateExec in the Spark 4.0
 CollationSuite hash agg check

#6195 made Comet check the shuffle manager that SparkEnv holds instead of
spark.shuffle.manager in the session's SQLConf. Spark 4.0's
DatasourceV2SQLBase clears the whole SQLConf after each test, which used to
drop spark.shuffle.manager and silently turn Comet shuffle off for the rest
of CollationSuite. "hash agg is not used for non binary collations" relied
on that: with Comet shuffle on, the UTF8_BINARY query's final aggregate is a
CometHashAggregateExec, and the assertion finds no Spark HashAggregateExec.

Accept CometHashAggregateExec there, as the 4.1.3 diff already does. Spark
4.1's DatasourceV2SQLBase unsets only two catalog configs, so 4.1 always ran
this query with Comet shuffle.

Regenerating the diff also corrects the ParquetRowIndexSuite index line left
stale by the hand edit in #6046. That file's content is unchanged.
---
 dev/diffs/4.0.4.diff | 23 ++++++++++++++++-------
 1 file changed, 16 insertions(+), 7 deletions(-)

diff --git a/dev/diffs/4.0.4.diff b/dev/diffs/4.0.4.diff
index 1d9978f4186..a28ccf00ea8 100644
--- a/dev/diffs/4.0.4.diff
+++ b/dev/diffs/4.0.4.diff
@@ -1415,18 +1415,19 @@ index a40e34d94d0..abc1f035d15 100644
        }
      }
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
-index 11e9547dfc5..637411056ae 100644
+index 11e9547dfc5..ba340c4ebcf 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
-@@ -24,6 +24,7 @@ import org.apache.spark.sql.{AnalysisException, Row}
+@@ -24,6 +24,8 @@ import org.apache.spark.sql.{AnalysisException, Row}
  import org.apache.spark.sql.catalyst.ExtendedAnalysisException
  import org.apache.spark.sql.catalyst.expressions._
  import org.apache.spark.sql.catalyst.util.CollationFactory
 +import org.apache.spark.sql.comet.{CometBroadcastHashJoinExec, CometHashJoinExec, CometSortMergeJoinExec}
++import org.apache.spark.sql.comet.CometHashAggregateExec
  import org.apache.spark.sql.connector.{DatasourceV2SQLBase, FakeV2ProviderWithCustomSchema}
  import org.apache.spark.sql.connector.catalog.{Identifier, InMemoryTable}
  import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.CatalogHelper
-@@ -55,7 +56,9 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
+@@ -55,7 +57,9 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
      assert(
        collectFirst(queryPlan) {
          case _: SortMergeJoinExec => assert(isSortMergeForced)
@@ -1436,7 +1437,15 @@ index 11e9547dfc5..637411056ae 100644
        }.nonEmpty
      )
    }
-@@ -1547,10 +1550,14 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
+@@ -463,6 +467,7 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
+         val dfBinary = sql(s"SELECT COUNT(*), c FROM $tableNameBinary GROUP BY c")
+         assert(collectFirst(dfBinary.queryExecution.executedPlan) {
+           case _: HashAggregateExec | _: ObjectHashAggregateExec => ()
++          case _: CometHashAggregateExec => ()
+         }.nonEmpty)
+       }
+     }
+@@ -1547,10 +1552,14 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
              if (!CollationFactory.fetchCollation(t.collation).supportsBinaryEquality) {
                assert(collectFirst(queryPlan) {
                  case b: HashJoin => b.leftKeys.head
@@ -1451,7 +1460,7 @@ index 11e9547dfc5..637411056ae 100644
                }.head.isInstanceOf[CollationKey])
              }
            }
-@@ -1606,11 +1613,13 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
+@@ -1606,11 +1615,13 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
              if (!CollationFactory.fetchCollation(t.collation).supportsBinaryEquality) {
                assert(collectFirst(queryPlan) {
                  case b: BroadcastHashJoinExec => b.leftKeys.head
@@ -1465,7 +1474,7 @@ index 11e9547dfc5..637411056ae 100644
                }.head.isInstanceOf[ArrayTransform])
              }
            }
-@@ -1676,6 +1685,7 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
+@@ -1676,6 +1687,7 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
              } else {
                assert(!collectFirst(queryPlan) {
                  case b: BroadcastHashJoinExec => b.leftKeys.head
@@ -3030,7 +3039,7 @@ index 30503af0fab..1491f4bc2d5 100644
  
    import testImplicits._
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala
-index 08fd8a9ecb5..06967aec8e1 100644
+index 08fd8a9ecb5..e0b8cada307 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowIndexSuite.scala
 @@ -27,6 +27,7 @@ import org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE