sunchao commented on code in PR #4950:
URL: https://github.com/apache/datafusion-comet/pull/4950#discussion_r3889723520
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala:
##########
@@ -188,6 +188,15 @@ object CometWindowExec extends
CometOperatorSerde[WindowExec] {
val aggregateExpressions: Array[AggregateExpression] = windowExpr.flatMap
{ expr =>
expr match {
+ // Spark 4.2 allows FILTER (WHERE ...) on a window aggregate.
DataFusion window
+ // expressions have no filter, and the aggregate proto's filter field
is only honored by
+ // the native aggregate operator, so serializing this window
expression would silently
+ // evaluate the aggregate over every row of the frame and produce
wrong results.
+ case agg: AggregateExpression if agg.filter.isDefined =>
Review Comment:
[P2] Keep the fallback reason on the original decimal window expression
For low-precision decimal `SUM` and `AVG`, `extractWindowExpression`
rewrites the `MakeDecimal` or `Cast(Divide(...))` wrapper into a copied
`WindowExpression`. This guard tags that detached copy, while
`rollUpFallbackReasons` only traverses the original `WindowExec` expressions. I
reproduced this with a Spark 4.2 `DECIMAL(8,2)` `SUM ... FILTER` query: strict
mode threw `Comet did not convert Window but recorded no fallback reason`; the
integer fixture passes because it retains the original expression. Please carry
or tag the original expression (or propagate the reason to the operator) and
add decimal `SUM` and `AVG` FILTER coverage.
##########
dev/diffs/4.2.0.diff:
##########
@@ -0,0 +1,4334 @@
+diff --git
a/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
b/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
+index 6df8bc85b51..dabb75e2b75 100644
+--- a/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
++++ b/core/src/test/scala/org/apache/spark/storage/FallbackStorageSuite.scala
+@@ -268,6 +268,11 @@ class FallbackStorageSuite extends SparkFunSuite with
LocalSparkContext {
+ }
+
+ test("Upload from all decommissioned executors") {
++ // Comet replaces Spark's shuffle with its own native shuffle, which is
incompatible with
++ // the fallback storage migration path used by BlockManagerDecommissioner.
++ val cometEnv = System.getenv("ENABLE_COMET")
++ assume(cometEnv == null || cometEnv == "0" || cometEnv == "false",
++ "Skipped when Comet is enabled: incompatible with Comet native shuffle
storage")
+ sc = new SparkContext(getSparkConf(2, 2))
+ withSpark(sc) { sc =>
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+@@ -298,6 +303,11 @@ class FallbackStorageSuite extends SparkFunSuite with
LocalSparkContext {
+ }
+
+ test("Upload multi stages") {
++ // Comet replaces Spark's shuffle with its own native shuffle, which is
incompatible with
++ // the fallback storage migration path used by BlockManagerDecommissioner.
++ val cometEnv = System.getenv("ENABLE_COMET")
++ assume(cometEnv == null || cometEnv == "0" || cometEnv == "false",
++ "Skipped when Comet is enabled: incompatible with Comet native shuffle
storage")
+ sc = new SparkContext(getSparkConf())
+ withSpark(sc) { sc =>
+ TestUtils.waitUntilExecutorsUp(sc, 1, 60000)
+@@ -332,6 +342,11 @@ class FallbackStorageSuite extends SparkFunSuite with
LocalSparkContext {
+
+ CompressionCodec.shortCompressionCodecNames.keys.foreach { codec =>
+ test(s"$codec - Newly added executors should access old data from remote
storage") {
++ // Comet replaces Spark's shuffle with its own native shuffle, which is
incompatible with
++ // the fallback storage migration path used by
BlockManagerDecommissioner.
++ val cometEnv = System.getenv("ENABLE_COMET")
++ assume(cometEnv == null || cometEnv == "0" || cometEnv == "false",
++ "Skipped when Comet is enabled: incompatible with Comet native
shuffle storage")
+ sc = new SparkContext(getSparkConf(2, 0).set(IO_COMPRESSION_CODEC,
codec))
+ withSpark(sc) { sc =>
+ TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+diff --git a/pom.xml b/pom.xml
+index 46558134f41..862c9a6eb9e 100644
+--- a/pom.xml
++++ b/pom.xml
+@@ -154,6 +154,8 @@
+ <kryo.version>4.0.3</kryo.version>
+ <ivy.version>2.5.3</ivy.version>
+ <oro.version>2.0.8</oro.version>
++ <spark.version.short>4.2</spark.version.short>
++ <comet.version>1.1.0-SNAPSHOT</comet.version>
+ <!--
+ If you change codahale.metrics.version, you also need to change
+ the link to metrics.dropwizard.io in docs/monitoring.md.
+@@ -2646,6 +2648,25 @@
+ <artifactId>arpack</artifactId>
+ <version>${netlib.ludovic.dev.version}</version>
+ </dependency>
++ <dependency>
++ <groupId>org.apache.datafusion</groupId>
++
<artifactId>comet-spark-spark${spark.version.short}_${scala.binary.version}</artifactId>
++ <version>${comet.version}</version>
++ <exclusions>
++ <exclusion>
++ <groupId>org.apache.spark</groupId>
++ <artifactId>spark-sql_${scala.binary.version}</artifactId>
++ </exclusion>
++ <exclusion>
++ <groupId>org.apache.spark</groupId>
++ <artifactId>spark-core_${scala.binary.version}</artifactId>
++ </exclusion>
++ <exclusion>
++ <groupId>org.apache.spark</groupId>
++ <artifactId>spark-catalyst_${scala.binary.version}</artifactId>
++ </exclusion>
++ </exclusions>
++ </dependency>
+ <!-- SPARK-16484 add `datasketches-java` for support Datasketches
HllSketch -->
+ <dependency>
+ <groupId>org.apache.datasketches</groupId>
+diff --git a/sql/core/pom.xml b/sql/core/pom.xml
+index 82810f181ac..21a83831188 100644
+--- a/sql/core/pom.xml
++++ b/sql/core/pom.xml
+@@ -97,6 +97,10 @@
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-tags_${scala.binary.version}</artifactId>
+ </dependency>
++ <dependency>
++ <groupId>org.apache.datafusion</groupId>
++
<artifactId>comet-spark-spark${spark.version.short}_${scala.binary.version}</artifactId>
++ </dependency>
+
+ <!--
+ This spark-tags test-dep is needed even though it isn't used in this
module, otherwise testing-cmds that exclude
+diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
b/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
+index f03b4796314..d32d2c49ce3 100644
+--- a/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
++++ b/sql/core/src/main/scala/org/apache/spark/sql/classic/SparkSession.scala
+@@ -1248,6 +1248,23 @@ object SparkSession extends SparkSessionCompanion with
Logging {
+ extensions
+ }
+
++ /**
++ * Whether Comet extension is enabled
++ */
++ def isCometEnabled: Boolean = {
++ val v = System.getenv("ENABLE_COMET")
++ v == null || v == "1" || v.toBoolean
++ }
++
++
++ private def loadCometExtension(sparkContext: SparkContext): Seq[String] = {
++ if (sparkContext.getConf.getBoolean("spark.comet.enabled",
isCometEnabled)) {
++ Seq("org.apache.comet.CometSparkSessionExtensions")
++ } else {
++ Seq.empty
++ }
++ }
++
+ /**
+ * Initialize extensions specified in [[StaticSQLConf]]. The classes will
be applied to the
+ * extensions passed into this function.
+@@ -1257,7 +1274,8 @@ object SparkSession extends SparkSessionCompanion with
Logging {
+ extensions: SparkSessionExtensions): SparkSessionExtensions = {
+ val extensionConfClassNames =
sparkContext.conf.get(StaticSQLConf.SPARK_SESSION_EXTENSIONS)
+ .getOrElse(Seq.empty)
+- extensionConfClassNames.foreach { extensionConfClassName =>
++ val extensionClassNames = extensionConfClassNames ++
loadCometExtension(sparkContext)
++ extensionClassNames.foreach { extensionConfClassName =>
+ try {
+ val extensionConfClass = Utils.classForName(extensionConfClassName)
+ val extensionConf = extensionConfClass.getConstructor().newInstance()
+diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanInfo.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanInfo.scala
+index 4410fe50912..43bcce2a038 100644
+---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanInfo.scala
++++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlanInfo.scala
+@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution
+
+ import org.apache.spark.annotation.DeveloperApi
+ import org.apache.spark.sql.catalyst.plans.logical.{EmptyRelation,
LogicalPlan}
++import org.apache.spark.sql.comet.CometScanExec
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec,
QueryStageExec}
+ import org.apache.spark.sql.execution.adaptive.LogicalQueryStage
+ import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
+@@ -84,6 +85,7 @@ private[execution] object SparkPlanInfo {
+ // dump the file scan metadata (e.g file path) to event log
+ val metadata = plan match {
+ case fileScan: FileSourceScanLike => fileScan.metadata
++ case cometScan: CometScanExec => cometScan.metadata
+ case _ => Map[String, String]()
+ }
+ val childrenInfo = children.flatMap {
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
b/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
+index 13bbd9d81b7..541cdfb1e04 100644
+---
a/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
++++
b/sql/core/src/test/resources/sql-tests/inputs/decimalArithmeticOperations.sql
+@@ -15,6 +15,12 @@
+ -- limitations under the License.
+ --
+
++-- TODO: Disabled due to one of the test failed for Spark4.0
++-- TODO: https://github.com/apache/datafusion-comet/issues/1948
++-- The following query failed
++-- select /*+ COALESCE(1) */ id, a+b, a-b, a*b, a/b from decimals_test order
by id
++--SET spark.comet.enabled = false
++
+ CREATE TEMPORARY VIEW t AS SELECT 1.0 as a, 0.0 as b;
+
+ -- division, remainder and pmod by 0 return NULL
+diff --git a/sql/core/src/test/resources/sql-tests/inputs/explain-aqe.sql
b/sql/core/src/test/resources/sql-tests/inputs/explain-aqe.sql
+index 7aef901da4f..f3d6e18926d 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/explain-aqe.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/explain-aqe.sql
+@@ -2,3 +2,4 @@
+
+ --SET spark.sql.adaptive.enabled=true
+ --SET spark.sql.maxMetadataStringLength = 500
++--SET spark.comet.enabled = false
+diff --git a/sql/core/src/test/resources/sql-tests/inputs/explain-cbo.sql
b/sql/core/src/test/resources/sql-tests/inputs/explain-cbo.sql
+index eeb2180f7a5..afd1b5ec289 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/explain-cbo.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/explain-cbo.sql
+@@ -1,5 +1,6 @@
+ --SET spark.sql.cbo.enabled=true
+ --SET spark.sql.maxMetadataStringLength = 500
++--SET spark.comet.enabled = false
+
+ CREATE TABLE explain_temp1(a INT, b INT) USING PARQUET;
+ CREATE TABLE explain_temp2(c INT, d INT) USING PARQUET;
+diff --git a/sql/core/src/test/resources/sql-tests/inputs/explain.sql
b/sql/core/src/test/resources/sql-tests/inputs/explain.sql
+index 96dddafd82a..efedbbb3c0e 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/explain.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/explain.sql
+@@ -1,6 +1,7 @@
+ --SET spark.sql.codegen.wholeStage = true
+ --SET spark.sql.adaptive.enabled = false
+ --SET spark.sql.maxMetadataStringLength = 500
++--SET spark.comet.enabled = false
+
+ -- Test tables
+ CREATE table explain_temp1 (key int, val int) USING PARQUET;
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
b/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
+index de3f5c8cc43..ff6b284ff4d 100644
+---
a/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
++++
b/sql/core/src/test/resources/sql-tests/inputs/having-and-order-by-recursive-type-name-resolution.sql
+@@ -1,3 +1,7 @@
++-- TODO(https://github.com/apache/datafusion-comet/issues/4123)
++-- Comet native sort lacks row-format support for Struct(Map(...)) sort keys
++--SET spark.comet.enabled = false
++
+ -- This test file contains queries that test recursive types name resolution
in ORDER BY and HAVING clauses.
+
+ -- Alias type: String, Table column type: Struct
+diff --git a/sql/core/src/test/resources/sql-tests/inputs/hll.sql
b/sql/core/src/test/resources/sql-tests/inputs/hll.sql
+index 35128da97fd..25b873ae859 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/hll.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/hll.sql
+@@ -1,3 +1,8 @@
++-- TODO(https://github.com/apache/datafusion-comet/issues/4121)
++-- Comet's native scan rejects invalid UTF-8 byte sequences inserted into the
++-- string test table, which Spark allows.
++--SET spark.comet.enabled = false
++
+ -- Positive test cases
+ -- Create a table with some testing data.
+ DROP TABLE IF EXISTS t1;
+diff --git a/sql/core/src/test/resources/sql-tests/inputs/join-nearest-by.sql
b/sql/core/src/test/resources/sql-tests/inputs/join-nearest-by.sql
+index 40cfa87c4cd..4e1d5407100 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/join-nearest-by.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/join-nearest-by.sql
+@@ -1,5 +1,10 @@
+ -- Test cases for NEAREST BY top-K ranking join.
+
++-- The EXPLAIN queries in this file record Spark's operator names in the
golden file, which
++-- Comet replaces with its own operators, so run the whole file without Comet
(same approach as
++-- explain.sql / explain-aqe.sql / explain-cbo.sql).
++--SET spark.comet.enabled = false
++
+ CREATE VIEW users(user_id, score) AS VALUES (1, 10.0), (2, 20.0), (3, 30.0);
+ CREATE VIEW products(product, pscore) AS VALUES ('A', 11.0), ('B', 22.0),
('C', 5.0);
+
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part3.sql
b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part3.sql
+index 41fd4de2a09..162d5a817b6 100644
+---
a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part3.sql
++++
b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part3.sql
+@@ -6,6 +6,10 @@
+ --
https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/aggregates.sql#L352-L605
+
+ -- Test aggregate operator with codegen on and off.
++
++-- Floating-point precision difference between DataFusion and JVM for FILTER
aggregates
++--SET spark.comet.enabled = false
++
+ --CONFIG_DIM1 spark.sql.codegen.wholeStage=true
+ --CONFIG_DIM1
spark.sql.codegen.wholeStage=false,spark.sql.codegen.factoryMode=CODEGEN_ONLY
+ --CONFIG_DIM1
spark.sql.codegen.wholeStage=false,spark.sql.codegen.factoryMode=NO_CODEGEN
+diff --git a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int4.sql
b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int4.sql
+index 3a409eea348..26e9aaf215c 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int4.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int4.sql
+@@ -6,6 +6,9 @@
+ --
https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/int4.sql
+ --
+
++-- TODO: https://github.com/apache/datafusion-comet/issues/551
++--SET spark.comet.enabled = false
++
+ CREATE TABLE INT4_TBL(f1 int) USING parquet;
+
+ -- [SPARK-28023] Trim the string when cast string type to other types
+diff --git a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int8.sql
b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int8.sql
+index fac23b4a26f..98b12ae5ccc 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int8.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/int8.sql
+@@ -6,6 +6,10 @@
+ -- Test int8 64-bit integers.
+ --
https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/int8.sql
+ --
++
++-- TODO: https://github.com/apache/datafusion-comet/issues/551
++--SET spark.comet.enabled = false
++
+ CREATE TABLE INT8_TBL(q1 bigint, q2 bigint) USING parquet;
+
+ -- PostgreSQL implicitly casts string literals to data with integral types,
but
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/select_having.sql
b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/select_having.sql
+index 0efe0877e9b..f9df0400c99 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/select_having.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/select_having.sql
+@@ -6,6 +6,9 @@
+ --
https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/select_having.sql
+ --
+
++-- TODO: https://github.com/apache/datafusion-comet/issues/551
++--SET spark.comet.enabled = false
++
+ -- load test data
+ CREATE TABLE test_having (a int, b int, c string, d string) USING parquet;
+ INSERT INTO test_having VALUES (0, 1, 'XXXX', 'A');
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-limit.sql
b/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-limit.sql
+index 7c816d8a416..b1551a2b296 100644
+---
a/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-limit.sql
++++
b/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-limit.sql
+@@ -1,6 +1,23 @@
+ -- A test suite for IN LIMIT in parent side, subquery, and both predicate
subquery
+ -- It includes correlated cases.
+
++-- TODO: Disabled due to one of the test failed for Spark4.0
++-- TODO: https://github.com/apache/datafusion-comet/issues/1948
++-- The following query failed
++-- SELECT Count(DISTINCT( t1a )),
++-- t1b
++-- FROM t1
++-- WHERE t1d NOT IN (SELECT t2d
++-- FROM t2
++-- WHERE t2b > t1b
++-- ORDER BY t2b DESC nulls first, t2d
++-- LIMIT 1
++-- OFFSET 1)
++-- GROUP BY t1b
++-- ORDER BY t1b NULLS last
++-- LIMIT 1
++-- OFFSET 1;
++--SET spark.comet.enabled = false
+ --CONFIG_DIM1 spark.sql.optimizeNullAwareAntiJoin=true
+ --CONFIG_DIM1 spark.sql.optimizeNullAwareAntiJoin=false
+
+@@ -61,6 +78,7 @@ WHERE t1a IN (SELECT t2a
+ WHERE t1d = t2d)
+ LIMIT 2;
+
++--SET spark.sql.cbo.enabled=true
+ -- correlated IN subquery
+ -- LIMIT on both parent and subquery sides
+ SELECT *
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-order-by.sql
b/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-order-by.sql
+index 7fbb1c12924..a612d40813a 100644
+---
a/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-order-by.sql
++++
b/sql/core/src/test/resources/sql-tests/inputs/subquery/in-subquery/in-order-by.sql
+@@ -1,5 +1,11 @@
+ -- A test suite for ORDER BY in parent side, subquery, and both predicate
subquery
+ -- It includes correlated cases.
++
++-- Comet: the `... and t1c IN (SELECT t2c FROM t2 ORDER BY t2b DESC nulls
last) ORDER BY t1c DESC
++-- nulls last` query has ties on the sort key (two rows with t1c = 12), so
the relative order of
++-- the tied rows is implementation defined and Comet's sort does not
reproduce the order recorded
++-- in the golden file.
++--SET spark.comet.enabled = false
+ --SET spark.sql.autoBroadcastJoinThreshold=-1
+
+ -- Test sort operator with codegen on and off.
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/view-schema-binding-config.sql
b/sql/core/src/test/resources/sql-tests/inputs/view-schema-binding-config.sql
+index e803254ea64..74db78aee38 100644
+---
a/sql/core/src/test/resources/sql-tests/inputs/view-schema-binding-config.sql
++++
b/sql/core/src/test/resources/sql-tests/inputs/view-schema-binding-config.sql
+@@ -1,6 +1,9 @@
+ -- This test suits check the spark.sql.viewSchemaBindingMode configuration.
+ -- It can be DISABLED and COMPENSATION
+
++-- TODO: https://github.com/apache/datafusion-comet/issues/551
++--SET spark.comet.enabled = false
++
+ -- Verify the default binding is true
+ SET spark.sql.legacy.viewSchemaBindingMode;
+
+diff --git
a/sql/core/src/test/resources/sql-tests/inputs/view-schema-compensation.sql
b/sql/core/src/test/resources/sql-tests/inputs/view-schema-compensation.sql
+index 21a3ce1e122..f4762ab98f0 100644
+--- a/sql/core/src/test/resources/sql-tests/inputs/view-schema-compensation.sql
++++ b/sql/core/src/test/resources/sql-tests/inputs/view-schema-compensation.sql
+@@ -1,5 +1,9 @@
+ -- This test suite checks the WITH SCHEMA COMPENSATION clause
+ -- Disable ANSI mode to ensure we are forcing it explicitly in the CASTS
++
++-- TODO: https://github.com/apache/datafusion-comet/issues/551
++--SET spark.comet.enabled = false
++
+ SET spark.sql.ansi.enabled = false;
+
+ -- In COMPENSATION views get invalidated if the type can't cast
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+index 085dbcd8046..3090d321b6c 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+@@ -49,7 +49,7 @@ import
org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, AQEProp
+ import org.apache.spark.sql.execution.columnar._
+ import org.apache.spark.sql.execution.command.CommandUtils
+ import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+-import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
++import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
+ import
org.apache.spark.sql.execution.ui.SparkListenerSQLAdaptiveExecutionUpdate
+ import org.apache.spark.sql.functions._
+ import org.apache.spark.sql.internal.SQLConf
+@@ -564,7 +564,8 @@ class CachedTableSuite extends SharedSparkSession
+ df.collect()
+ }
+ assert(
+- collect(df.queryExecution.executedPlan) { case e: ShuffleExchangeExec
=> e }.size == expected)
++ collect(df.queryExecution.executedPlan) {
++ case _: ShuffleExchangeLike => 1 }.size == expected)
+ }
+
+ test("A cached table preserves the partitioning and ordering of its cached
SparkPlan") {
+@@ -1703,9 +1704,18 @@ class CachedTableSuite extends SharedSparkSession
+ _.nodeName.contains("TableCacheQueryStage"))
+ val aqeNode = findNodeInSparkPlanInfo(inMemoryScanNode.get,
+ _.nodeName.contains("AdaptiveSparkPlan"))
+- val aqePlanRoot = findNodeInSparkPlanInfo(inMemoryScanNode.get,
+- _.nodeName.contains("ResultQueryStage"))
+- aqePlanRoot.get.children.head.nodeName == "AQEShuffleRead"
++ // Spark 4.0 wraps results in ResultQueryStage. The coalescing
indicator is AQEShuffleRead
++ // as the direct child of InputAdapter.
++ // AdaptiveSparkPlan -> ResultQueryStage -> WholestageCodegen ->
++ // CometColumnarToRow -> InputAdapter -> AQEShuffleRead (if
coalesced)
++ val resultStage = aqeNode.get.children.head // ResultQueryStage
++ val wsc = resultStage.children.head // WholeStageCodegen
++ val c2r = wsc.children.head // ColumnarToRow or
CometColumnarToRow
++ val inputAdapter = c2r.children.head // InputAdapter
++ resultStage.nodeName == "ResultQueryStage" &&
++ wsc.nodeName.startsWith("WholeStageCodegen") && // could be
"WholeStageCodegen (1)"
++ (c2r.nodeName == "CometColumnarToRow" || c2r.nodeName ==
"ColumnarToRow") &&
++ inputAdapter.children.head.nodeName == "AQEShuffleRead"
+ }
+
+ withTempView("t0", "t1", "t2") {
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+index 5b8154d2900..f01366b66bc 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala
+@@ -30,8 +30,9 @@ import
org.apache.spark.sql.catalyst.util.AUTO_GENERATED_ALIAS
+ import org.apache.spark.sql.errors.DataTypeErrors.toSQLId
+ import org.apache.spark.sql.execution.WholeStageCodegenExec
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
++import org.apache.spark.sql.comet.CometHashAggregateExec
+ import org.apache.spark.sql.execution.aggregate.{HashAggregateExec,
ObjectHashAggregateExec, SortAggregateExec}
+-import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
++import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
+ import org.apache.spark.sql.expressions.Window
+ import org.apache.spark.sql.functions._
+ import org.apache.spark.sql.internal.SQLConf
+@@ -667,7 +668,9 @@ class DataFrameAggregateSuite extends SharedSparkSession
+ df.selectExpr("sort_array(collect_set(b) RESPECT NULLS)"),
Seq(Row(Seq(null, 2))))
+ }
+
+- test("SPARK-57298: collect_set normalizes NaN and -0.0 for floating-point
types") {
++ test("SPARK-57298: collect_set normalizes NaN and -0.0 for floating-point
types",
++ // Comet's native collect_set does not yet normalize NaN / -0.0.
++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4966";)) {
+ checkAnswer(
+ sql("SELECT collect_set(v) FROM VALUES (double('NaN')), (double('NaN'))
AS t(v)"),
+ Row(Seq(Double.NaN)))
+@@ -932,7 +935,9 @@ class DataFrameAggregateSuite extends SharedSparkSession
+ case _ => false
+ }.isDefined)
+ } else {
+- assert(stripAQEPlan(hashAggPlan).isInstanceOf[HashAggregateExec])
++ val strippedPlan = stripAQEPlan(hashAggPlan)
++ assert(strippedPlan.isInstanceOf[HashAggregateExec] ||
++ strippedPlan.exists(_.isInstanceOf[CometHashAggregateExec]))
+ }
+
+ // test case for ObjectHashAggregate and SortAggregate
+@@ -991,12 +996,12 @@ class DataFrameAggregateSuite extends SharedSparkSession
+ assert(sortAggPlans.isEmpty)
+
+ val objHashAggPlans = collect(aggPlan) {
+- case objHashAgg: ObjectHashAggregateExec => objHashAgg
++ case objHashAgg @ (_: ObjectHashAggregateExec | _:
CometHashAggregateExec) => objHashAgg
+ }
+ assert(objHashAggPlans.nonEmpty)
+
+ val exchangePlans = collect(aggPlan) {
+- case shuffle: ShuffleExchangeExec => shuffle
++ case shuffle: ShuffleExchangeLike => shuffle
+ }
+ assert(exchangePlans.length == 1)
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
+index 9733d51a91c..395a108abc8 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
+@@ -434,7 +434,9 @@ class DataFrameJoinSuite extends SharedSparkSession
+
+ withTempDatabase { dbName =>
+ withTable(table1Name, table2Name) {
+- withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
++ withSQLConf(
++ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
++ "spark.comet.enabled" -> "false") {
+ spark.range(50).write.saveAsTable(s"$dbName.$table1Name")
+ spark.range(100).write.saveAsTable(s"$dbName.$table2Name")
+
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala
+index d838ba4c234..cb0573d56d0 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSetOperationsSuite.scala
+@@ -23,10 +23,11 @@ import java.util.Locale
+ import org.apache.spark.sql.catalyst.optimizer.RemoveNoopUnion
+ import org.apache.spark.sql.catalyst.plans.logical.Union
+ import org.apache.spark.sql.catalyst.plans.physical.UnknownPartitioning
++import org.apache.spark.sql.comet.CometUnionExec
+ import org.apache.spark.sql.execution.{SparkPlan, UnionExec}
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+ import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
+-import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec,
ShuffleExchangeExec}
++import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec,
ShuffleExchangeLike}
+ import org.apache.spark.sql.functions._
+ import org.apache.spark.sql.internal.SQLConf
+ import org.apache.spark.sql.test.{ExamplePoint, ExamplePointUDT,
SharedSparkSession, SQLTestData}
+@@ -1518,11 +1519,12 @@ class DataFrameSetOperationsSuite extends
SharedSparkSession with AdaptiveSparkP
+ val union = df1.repartition($"a").union(df2.repartition($"a"))
+ val unionExec = union.queryExecution.executedPlan.collect {
+ case u: UnionExec => u
++ case u: CometUnionExec => u
+ }
+ assert(unionExec.size == 1)
+
+ val shuffle = df1.repartition($"a").queryExecution.executedPlan.collect
{
+- case s: ShuffleExchangeExec => s
++ case s: ShuffleExchangeLike => s
+ }
+ assert(shuffle.size == 1)
+
+@@ -1553,11 +1555,12 @@ class DataFrameSetOperationsSuite extends
SharedSparkSession with AdaptiveSparkP
+ val union = df1.repartition($"a").union(df2.repartition($"d"))
+ val unionExec = union.queryExecution.executedPlan.collect {
+ case u: UnionExec => u
++ case u: CometUnionExec => u
+ }
+ assert(unionExec.size == 1)
+
+ val shuffle =
df1.repartition($"a").queryExecution.executedPlan.collect {
+- case s: ShuffleExchangeExec => s
++ case s: ShuffleExchangeLike => s
+ }
+ assert(shuffle.size == 1)
+
+@@ -1572,10 +1575,10 @@ class DataFrameSetOperationsSuite extends
SharedSparkSession with AdaptiveSparkP
+ // Avoid unnecessary shuffle if union output partitioning is enabled
+ val shuffledUnion = union.repartition($"a")
+ val shuffleNumBefore = union.queryExecution.executedPlan.collect {
+- case s: ShuffleExchangeExec => s
++ case s: ShuffleExchangeLike => s
+ }
+ val shuffleNumAfter =
shuffledUnion.queryExecution.executedPlan.collect {
+- case s: ShuffleExchangeExec => s
++ case s: ShuffleExchangeLike => s
+ }
+
+ if (enabled) {
+@@ -1604,6 +1607,7 @@ class DataFrameSetOperationsSuite extends
SharedSparkSession with AdaptiveSparkP
+ val union =
df1.repartitionByRange($"a").union(df2.repartitionByRange($"d"))
+ val unionExec = union.queryExecution.executedPlan.collect {
+ case u: UnionExec => u
++ case u: CometUnionExec => u
+ }
+ assert(unionExec.size == 1)
+
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
+index 577ae025f1d..a0c95263931 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
+@@ -36,11 +36,12 @@ import
org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference,
+ import org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelation
+ import org.apache.spark.sql.catalyst.parser.ParseException
+ import org.apache.spark.sql.catalyst.plans.logical.{Filter, LeafNode,
LocalRelation, LogicalPlan, OneRowRelation}
++import org.apache.spark.sql.comet.CometBroadcastExchangeExec
+ import org.apache.spark.sql.connector.FakeV2Provider
+ import org.apache.spark.sql.execution.{FilterExec, LogicalRDD,
QueryExecution, SortExec, WholeStageCodegenExec}
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+ import org.apache.spark.sql.execution.aggregate.HashAggregateExec
+-import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike}
++import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
ReusedExchangeExec, ShuffleExchangeLike}
+ import org.apache.spark.sql.expressions.{Aggregator, Window}
+ import org.apache.spark.sql.functions._
+ import org.apache.spark.sql.internal.SQLConf
+@@ -1492,7 +1493,7 @@ class DataFrameSuite extends SharedSparkSession
+ fail("Should not have back to back Aggregates")
+ }
+ atFirstAgg = true
+- case e: ShuffleExchangeExec => atFirstAgg = false
++ case e: ShuffleExchangeLike => atFirstAgg = false
+ case _ =>
+ }
+ }
+@@ -1682,7 +1683,7 @@ class DataFrameSuite extends SharedSparkSession
+ checkAnswer(join, df)
+ assert(
+ collect(join.queryExecution.executedPlan) {
+- case e: ShuffleExchangeExec => true }.size === 1)
++ case _: ShuffleExchangeLike => true }.size === 1)
+ assert(
+ collect(join.queryExecution.executedPlan) { case e:
ReusedExchangeExec => true }.size === 1)
+ val broadcasted = broadcast(join)
+@@ -1690,10 +1691,12 @@ class DataFrameSuite extends SharedSparkSession
+ checkAnswer(join2, df)
+ assert(
+ collect(join2.queryExecution.executedPlan) {
+- case e: ShuffleExchangeExec => true }.size == 1)
++ case _: ShuffleExchangeLike => true }.size == 1)
+ assert(
+ collect(join2.queryExecution.executedPlan) {
+- case e: BroadcastExchangeExec => true }.size === 1)
++ case e: BroadcastExchangeExec => true
++ case _: CometBroadcastExchangeExec => true
++ }.size === 1)
+ assert(
+ collect(join2.queryExecution.executedPlan) { case e:
ReusedExchangeExec => true }.size == 4)
+ }
+@@ -2091,7 +2094,7 @@ class DataFrameSuite extends SharedSparkSession
+
+ // Assert that no extra shuffle introduced by cogroup.
+ val exchanges = collect(df3.queryExecution.executedPlan) {
+- case h: ShuffleExchangeExec => h
++ case h: ShuffleExchangeLike => h
+ }
+ assert(exchanges.size == 2)
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
+index f79824de8ff..5432984960f 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
+@@ -25,8 +25,10 @@ import
org.apache.spark.sql.catalyst.optimizer.TransposeWindow
+ import org.apache.spark.sql.catalyst.plans.logical.{Window => LogicalWindow}
+ import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
+ import org.apache.spark.sql.catalyst.trees.UnaryLike
++import org.apache.spark.sql.comet.CometWindowExec
++import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+-import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS,
Exchange, ShuffleExchangeExec}
++import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS,
Exchange, ShuffleExchangeExec, ShuffleExchangeLike}
+ import org.apache.spark.sql.execution.window.WindowExec
+ import org.apache.spark.sql.expressions.{Aggregator,
MutableAggregationBuffer, UserDefinedAggregateFunction, Window}
+ import org.apache.spark.sql.functions._
+@@ -975,7 +977,8 @@ class DataFrameWindowFunctionsSuite extends
SharedSparkSession
+ }
+ }
+
+- test("Window spill with more than the inMemoryThreshold and
spillThreshold") {
++ test("Window spill with more than the inMemoryThreshold and spillThreshold",
++ IgnoreComet("Comet does not support spilling")) {
+ val df = Seq((1, "1"), (2, "2"), (1, "3"), (2, "4")).toDF("key", "value")
+ val window = Window.partitionBy($"key").orderBy($"value")
+
+@@ -987,7 +990,7 @@ class DataFrameWindowFunctionsSuite extends
SharedSparkSession
+ }
+ }
+
+- test("SPARK-21258: complex object in combination with spilling") {
++ test("SPARK-21258: complex object in combination with spilling",
IgnoreComet("Comet does not support spilling")) {
+ // Make sure we trigger the spilling path.
+ withSQLConf(SQLConf.WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "1",
+ SQLConf.WINDOW_EXEC_BUFFER_SPILL_THRESHOLD.key -> "17") {
+@@ -1336,10 +1339,12 @@ class DataFrameWindowFunctionsSuite extends
SharedSparkSession
+ }
+
+ def isShuffleExecByRequirement(
+- plan: ShuffleExchangeExec,
++ plan: ShuffleExchangeLike,
+ desiredClusterColumns: Seq[String]): Boolean = plan match {
+ case ShuffleExchangeExec(op: HashPartitioning, _, ENSURE_REQUIREMENTS,
_) =>
+ partitionExpressionsColumns(op.expressions) === desiredClusterColumns
++ case CometShuffleExchangeExec(op: HashPartitioning, _, _,
ENSURE_REQUIREMENTS, _, _) =>
++ partitionExpressionsColumns(op.expressions) === desiredClusterColumns
+ case _ => false
+ }
+
+@@ -1362,7 +1367,12 @@ class DataFrameWindowFunctionsSuite extends
SharedSparkSession
+ val shuffleByRequirement = windowed.queryExecution.executedPlan.exists {
+ case w: WindowExec =>
+ w.child.exists {
+- case s: ShuffleExchangeExec => isShuffleExecByRequirement(s,
Seq("key1", "key2"))
++ case s: ShuffleExchangeLike => isShuffleExecByRequirement(s,
Seq("key1", "key2"))
++ case _ => false
++ }
++ case w: CometWindowExec =>
++ w.child.exists {
++ case s: ShuffleExchangeLike => isShuffleExecByRequirement(s,
Seq("key1", "key2"))
+ case _ => false
+ }
+ case _ => false
+@@ -1814,7 +1824,8 @@ class DataFrameWindowFunctionsSuite extends
SharedSparkSession
+ }
+ }
+
+- test("SPARK-49386: Window spill with more than the inMemoryThreshold and
spillSizeThreshold") {
++ test("SPARK-49386: Window spill with more than the inMemoryThreshold and
spillSizeThreshold",
++ IgnoreComet("Comet does not support spilling")) {
+ val df = Seq((1, "1"), (2, "2"), (1, "3"), (2, "4")).toDF("key", "value")
+ val window = Window.partitionBy($"key").orderBy($"value")
+
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
+index 879569045b6..f3ff89067d2 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
+@@ -46,7 +46,7 @@ import
org.apache.spark.sql.catalyst.trees.DataFrameQueryContext
+ import org.apache.spark.sql.catalyst.util.sideBySide
+ import org.apache.spark.sql.execution.{LogicalRDD, RDDScanExec, SQLExecution}
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+-import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
ShuffleExchangeExec}
++import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec,
ShuffleExchangeExec, ShuffleExchangeLike}
+ import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
+ import org.apache.spark.sql.expressions.UserDefinedFunction
+ import org.apache.spark.sql.functions._
+@@ -2521,7 +2521,7 @@ class DatasetSuite extends SharedSparkSession
+
+ // Assert that no extra shuffle introduced by cogroup.
+ val exchanges = collect(df3.queryExecution.executedPlan) {
+- case h: ShuffleExchangeExec => h
++ case h: ShuffleExchangeLike => h
+ }
+ assert(exchanges.size == 2)
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
+index c68f64d52f2..32489f2cb15 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/DynamicPartitionPruningSuite.scala
+@@ -22,6 +22,7 @@ import org.scalatest.GivenWhenThen
+ import org.apache.spark.sql.catalyst.expressions.{DynamicPruningExpression,
Expression}
+ import org.apache.spark.sql.catalyst.expressions.CodegenObjectFactoryMode._
+ import org.apache.spark.sql.catalyst.plans.ExistenceJoin
++import org.apache.spark.sql.comet.{CometNativeScanExec, CometScanExec,
CometSubqueryBroadcastExec}
+ import org.apache.spark.sql.connector.catalog.{InMemoryTableCatalog,
InMemoryTableWithV2FilterCatalog}
+ import org.apache.spark.sql.execution._
+ import org.apache.spark.sql.execution.adaptive._
+@@ -192,6 +193,7 @@ abstract class DynamicPartitionPruningSuiteBase
+ }
+ val subqueryBroadcast = dpExprs.collect {
+ case InSubqueryExec(_, b: SubqueryBroadcastExec, _, _, _, _) => b
++ case InSubqueryExec(_, b: CometSubqueryBroadcastExec, _, _, _, _) => b
+ }
+
+ val hasFilter = if (withSubquery) "Should" else "Shouldn't"
+@@ -246,6 +248,8 @@ abstract class DynamicPartitionPruningSuiteBase
+ val buf =
collectDynamicPruningExpressions(df.queryExecution.executedPlan).collect {
+ case InSubqueryExec(_, b: SubqueryBroadcastExec, _, _, _, _) =>
+ b.indices.map(idx => b.buildKeys(idx))
++ case InSubqueryExec(_, b: CometSubqueryBroadcastExec, _, _, _, _) =>
++ b.indices.map(idx => b.buildKeys(idx))
+ }
+ assert(buf.distinct.size == n)
+ }
+@@ -261,6 +265,12 @@ abstract class DynamicPartitionPruningSuiteBase
+ case s: BatchScanExec => s.runtimeFilters.collect {
+ case d: DynamicPruningExpression => d.child
+ }
++ case s: CometScanExec => s.partitionFilters.collect {
++ case d: DynamicPruningExpression => d.child
++ }
++ case s: CometNativeScanExec => s.partitionFilters.collect {
++ case d: DynamicPruningExpression => d.child
++ }
+ case _ => Nil
+ }
+ }
+@@ -1203,10 +1213,16 @@ abstract class DynamicPartitionPruningSuiteBase
+
+ val plan = df.queryExecution.executedPlan
+ val countSubqueryBroadcasts =
+- collectWithSubqueries(plan)({ case _: SubqueryBroadcastExec => 1
}).sum
++ collectWithSubqueries(plan)({
++ case _: SubqueryBroadcastExec => 1
++ case _: CometSubqueryBroadcastExec => 1
++ }).sum
+
+ val countReusedSubqueryBroadcasts =
+- collectWithSubqueries(plan)({ case ReusedSubqueryExec(_:
SubqueryBroadcastExec) => 1}).sum
++ collectWithSubqueries(plan)({
++ case ReusedSubqueryExec(_: SubqueryBroadcastExec) => 1
++ case ReusedSubqueryExec(_: CometSubqueryBroadcastExec) => 1
++ }).sum
+
+ assert(countSubqueryBroadcasts == 1)
+ assert(countReusedSubqueryBroadcasts == 1)
+@@ -1578,6 +1594,7 @@ abstract class DynamicPartitionPruningSuiteBase
+
+ val subqueryBroadcastExecs =
collectWithSubqueries(df.queryExecution.executedPlan) {
+ case s: SubqueryBroadcastExec => s
++ case s: CometSubqueryBroadcastExec => s
+ }
+ assert(subqueryBroadcastExecs.size === 1)
+ subqueryBroadcastExecs.foreach { subqueryBroadcastExec =>
+@@ -1730,6 +1747,10 @@ abstract class DynamicPartitionPruningV1Suite extends
DynamicPartitionPruningDat
+ case s: BatchScanExec =>
+ // we use f1 col for v2 tables due to schema pruning
+ s.output.exists(_.exists(_.argString(maxFields =
100).contains("f1")))
++ case s: CometScanExec =>
++ s.output.exists(_.exists(_.argString(maxFields =
100).contains("fid")))
++ case s: CometNativeScanExec =>
++ s.output.exists(_.exists(_.argString(maxFields =
100).contains("fid")))
+ case _ => false
+ }
+ assert(scanOption.isDefined)
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
+index 618f0ab675e..be186f99b66 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/ExplainSuite.scala
+@@ -268,7 +268,8 @@ class ExplainSuite extends ExplainSuiteHelper with
DisableAdaptiveExecutionSuite
+ }
+ }
+
+- test("SPARK-33853: explain codegen - check presence of subquery") {
++ test("SPARK-33853: explain codegen - check presence of subquery",
++ IgnoreComet("Comet plan has a different WholeStageCodegen subtree
count")) {
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "true") {
+ withTempView("df") {
+ val df1 = spark.range(1, 100)
+@@ -471,7 +472,8 @@ class ExplainSuite extends ExplainSuiteHelper with
DisableAdaptiveExecutionSuite
+ }
+ }
+
+- test("Explain formatted output for scan operator for datasource V2") {
++ test("Explain formatted output for scan operator for datasource V2",
++ IgnoreComet("Comet explain output is different")) {
+ withTempDir { dir =>
+ Seq("parquet", "orc", "csv", "json").foreach { fmt =>
+ val basePath = dir.getCanonicalPath + "/" + fmt
+@@ -549,7 +551,9 @@ class ExplainSuite extends ExplainSuiteHelper with
DisableAdaptiveExecutionSuite
+ }
+ }
+
+-class ExplainSuiteAE extends ExplainSuiteHelper with
EnableAdaptiveExecutionSuite {
++// Ignored when Comet is enabled. Comet changes expected query plans.
++class ExplainSuiteAE extends ExplainSuiteHelper with
EnableAdaptiveExecutionSuite
++ with IgnoreCometSuite {
+ import testImplicits._
+
+ override protected def sparkConf =
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
+index 1fc45e9703f..b54410841af 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
+@@ -33,6 +33,7 @@ import
org.apache.spark.sql.catalyst.expressions.{AttributeReference, GreaterTha
+ import
org.apache.spark.sql.catalyst.expressions.IntegralLiteralTestUtils.{negativeInt,
positiveInt}
+ import org.apache.spark.sql.catalyst.plans.logical.Filter
+ import org.apache.spark.sql.catalyst.types.DataTypeUtils
++import org.apache.spark.sql.comet.{CometBatchScanExec, CometNativeScanExec,
CometScanExec, CometSortMergeJoinExec}
+ import org.apache.spark.sql.execution.{FileSourceScanLike, SimpleMode}
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+ import org.apache.spark.sql.execution.datasources.FilePartition
+@@ -654,18 +655,25 @@ class FileBasedDataSourceSuite extends SharedSparkSession
+ checkAnswer(sql(s"select A from $tableName"), data.select("A"))
+
+ // RuntimeException is triggered at executor side, which is then
wrapped as
+- // SparkException at driver side
++ // SparkException at driver side. Comet native readers throw
++ // SparkRuntimeException directly without the SparkException
wrapper.
++ def getDuplicateFieldError(query: String): SparkRuntimeException
= {
++ try {
++ sql(query).collect()
++ fail("Expected an
exception").asInstanceOf[SparkRuntimeException]
++ } catch {
++ case e: SparkException =>
++ e.getCause.asInstanceOf[SparkRuntimeException]
++ case e: SparkRuntimeException => e
++ }
++ }
+ checkError(
+- exception = intercept[SparkException] {
+- sql(s"select b from $tableName").collect()
+- }.getCause.asInstanceOf[SparkRuntimeException],
++ exception = getDuplicateFieldError(s"select b from $tableName"),
+ condition = "_LEGACY_ERROR_TEMP_2093",
+ parameters = Map("requiredFieldName" -> "b", "matchedOrcFields"
-> "[b, B]")
+ )
+ checkError(
+- exception = intercept[SparkException] {
+- sql(s"select B from $tableName").collect()
+- }.getCause.asInstanceOf[SparkRuntimeException],
++ exception = getDuplicateFieldError(s"select B from $tableName"),
+ condition = "_LEGACY_ERROR_TEMP_2093",
+ parameters = Map("requiredFieldName" -> "b", "matchedOrcFields"
-> "[b, B]")
+ )
+@@ -976,6 +984,7 @@ class FileBasedDataSourceSuite extends SharedSparkSession
+ assert(bJoinExec.isEmpty)
+ val smJoinExec = collect(joinedDF.queryExecution.executedPlan) {
+ case smJoin: SortMergeJoinExec => smJoin
++ case smJoin: CometSortMergeJoinExec => smJoin
+ }
+ assert(smJoinExec.nonEmpty)
+ }
+@@ -1036,6 +1045,7 @@ class FileBasedDataSourceSuite extends SharedSparkSession
+
+ val fileScan = df.queryExecution.executedPlan collectFirst {
+ case BatchScanExec(_, f: FileScan, _, _, _, _) => f
++ case CometBatchScanExec(BatchScanExec(_, f: FileScan, _, _, _,
_), _, _) => f
+ }
+ assert(fileScan.nonEmpty)
+ assert(fileScan.get.partitionFilters.nonEmpty)
+@@ -1077,6 +1087,7 @@ class FileBasedDataSourceSuite extends SharedSparkSession
+
+ val fileScan = df.queryExecution.executedPlan collectFirst {
+ case BatchScanExec(_, f: FileScan, _, _, _, _) => f
++ case CometBatchScanExec(BatchScanExec(_, f: FileScan, _, _, _,
_), _, _) => f
+ }
+ assert(fileScan.nonEmpty)
+ assert(fileScan.get.partitionFilters.isEmpty)
+@@ -1261,6 +1272,9 @@ class FileBasedDataSourceSuite extends SharedSparkSession
+ val filters = df.queryExecution.executedPlan.collect {
+ case f: FileSourceScanLike => f.dataFilters
+ case b: BatchScanExec => b.scan.asInstanceOf[FileScan].dataFilters
++ case b: CometScanExec => b.dataFilters
++ case b: CometNativeScanExec => b.dataFilters
++ case b: CometBatchScanExec =>
b.scan.asInstanceOf[FileScan].dataFilters
+ }.flatten
+ assert(filters.contains(GreaterThan(scan.logicalPlan.output.head,
Literal(5L))))
+ }
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/IgnoreComet.scala
b/sql/core/src/test/scala/org/apache/spark/sql/IgnoreComet.scala
+new file mode 100644
+index 00000000000..4b31bea33de
+--- /dev/null
++++ b/sql/core/src/test/scala/org/apache/spark/sql/IgnoreComet.scala
+@@ -0,0 +1,42 @@
++/*
++ * Licensed to the Apache Software Foundation (ASF) under one or more
++ * contributor license agreements. See the NOTICE file distributed with
++ * this work for additional information regarding copyright ownership.
++ * The ASF licenses this file to You under the Apache License, Version 2.0
++ * (the "License"); you may not use this file except in compliance with
++ * the License. You may obtain a copy of the License at
++ *
++ * http://www.apache.org/licenses/LICENSE-2.0
++ *
++ * Unless required by applicable law or agreed to in writing, software
++ * distributed under the License is distributed on an "AS IS" BASIS,
++ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
++ * See the License for the specific language governing permissions and
++ * limitations under the License.
++ */
++
++package org.apache.spark.sql
++
++import org.scalactic.source.Position
++import org.scalatest.Tag
++
++import org.apache.spark.sql.test.SQLTestUtils
++
++/**
++ * Tests with this tag will be ignored when Comet is enabled (e.g., via
`ENABLE_COMET`).
++ */
++case class IgnoreComet(reason: String) extends Tag("DisableComet")
++
++/**
++ * Helper trait that disables Comet for all tests regardless of default
config values.
++ */
++trait IgnoreCometSuite extends SQLTestUtils {
++ override protected def test(testName: String, testTags: Tag*)(testFun: =>
Any)
++ (implicit pos: Position): Unit = {
++ if (isCometEnabled) {
++ ignore(testName + " (disabled when Comet is on)", testTags: _*)(testFun)
++ } else {
++ super.test(testName, testTags: _*)(testFun)
++ }
++ }
++}
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
+index df9f1d1ed2d..ad3dd43b652 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinHintSuite.scala
+@@ -22,6 +22,7 @@ import org.apache.logging.log4j.Level
+ import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight,
BuildSide, EliminateResolvedHint}
+ import org.apache.spark.sql.catalyst.plans.logical._
+ import org.apache.spark.sql.catalyst.rules.RuleExecutor
++import org.apache.spark.sql.comet.{CometHashJoinExec, CometSortMergeJoinExec}
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+ import org.apache.spark.sql.execution.joins._
+ import org.apache.spark.sql.internal.SQLConf
+@@ -361,6 +362,7 @@ class JoinHintSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper {
+ val executedPlan = df.queryExecution.executedPlan
+ val shuffleHashJoins = collect(executedPlan) {
+ case s: ShuffledHashJoinExec => s
++ case c: CometHashJoinExec =>
c.originalPlan.asInstanceOf[ShuffledHashJoinExec]
+ }
+ assert(shuffleHashJoins.size == 1)
+ assert(shuffleHashJoins.head.buildSide == buildSide)
+@@ -370,6 +372,7 @@ class JoinHintSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper {
+ val executedPlan = df.queryExecution.executedPlan
+ val shuffleMergeJoins = collect(executedPlan) {
+ case s: SortMergeJoinExec => s
++ case c: CometSortMergeJoinExec => c
+ }
+ assert(shuffleMergeJoins.size == 1)
+ }
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
+index 7f695e90df8..57fea6f8324 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala
+@@ -29,7 +29,8 @@ import
org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
+ import org.apache.spark.sql.catalyst.expressions.{Ascending, GenericRow,
SortOrder}
+ import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight,
JoinSelectionHelper}
+ import org.apache.spark.sql.catalyst.plans.logical.{Filter, HintInfo, Join,
JoinHint, NO_BROADCAST_AND_REPLICATION}
+-import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec,
ProjectExec, SortExec, SparkPlan, WholeStageCodegenExec}
++import org.apache.spark.sql.comet._
++import org.apache.spark.sql.execution.{BinaryExecNode, ColumnarToRowExec,
FilterExec, InputAdapter, ProjectExec, SortExec, SparkPlan,
WholeStageCodegenExec}
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+ import org.apache.spark.sql.execution.exchange.{ShuffleExchangeExec,
ShuffleExchangeLike}
+ import org.apache.spark.sql.execution.joins._
+@@ -808,7 +809,8 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ }
+ }
+
+- test("test SortMergeJoin (with spill)") {
++ test("test SortMergeJoin (with spill)",
++ IgnoreComet("TODO: Comet SMJ doesn't support spill yet")) {
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "1",
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "0",
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_SPILL_THRESHOLD.key -> "1") {
+@@ -816,7 +818,8 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ }
+ }
+
+- test("SPARK-49386: test SortMergeJoin (with spill by size threshold)") {
++ test("SPARK-49386: test SortMergeJoin (with spill by size threshold)",
++ IgnoreComet("TODO: Comet SMJ doesn't support spill yet")) {
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "1",
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_IN_MEMORY_THRESHOLD.key -> "0",
+ SQLConf.SORT_MERGE_JOIN_EXEC_BUFFER_SPILL_THRESHOLD.key ->
Int.MaxValue.toString,
+@@ -946,10 +949,12 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ val physical = df.queryExecution.sparkPlan
+ val physicalJoins = physical.collect {
+ case j: SortMergeJoinExec => j
++ case j: CometSortMergeJoinExec =>
j.originalPlan.asInstanceOf[SortMergeJoinExec]
+ }
+ val executed = df.queryExecution.executedPlan
+ val executedJoins = collect(executed) {
+ case j: SortMergeJoinExec => j
++ case j: CometSortMergeJoinExec =>
j.originalPlan.asInstanceOf[SortMergeJoinExec]
+ }
+ // This only applies to the above tested queries, in which a child
SortMergeJoin always
+ // contains the SortOrder required by its parent SortMergeJoin. Thus,
SortExec should never
+@@ -1195,9 +1200,11 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ val plan = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2", joinType)
+ .groupBy($"k1").count()
+ .queryExecution.executedPlan
+- assert(collect(plan) { case _: ShuffledHashJoinExec => true }.size ===
1)
++ assert(collect(plan) {
++ case _: ShuffledHashJoinExec | _: CometHashJoinExec => true }.size
=== 1)
+ // No extra shuffle before aggregate
+- assert(collect(plan) { case _: ShuffleExchangeExec => true }.size === 2)
++ assert(collect(plan) {
++ case _: ShuffleExchangeLike => true }.size === 2)
+ })
+ }
+
+@@ -1214,10 +1221,11 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ .join(df4.hint("SHUFFLE_MERGE"), $"k1" === $"k4", joinType)
+ .queryExecution
+ .executedPlan
+- assert(collect(plan) { case _: SortMergeJoinExec => true }.size === 2)
++ assert(collect(plan) {
++ case _: SortMergeJoinExec | _: CometSortMergeJoinExec => true }.size
=== 2)
+ assert(collect(plan) { case _: BroadcastHashJoinExec => true }.size ===
1)
+ // No extra sort before last sort merge join
+- assert(collect(plan) { case _: SortExec => true }.size === 3)
++ assert(collect(plan) { case _: SortExec | _: CometSortExec => true
}.size === 3)
+ })
+
+ // Test shuffled hash join
+@@ -1227,10 +1235,13 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ .join(df4.hint("SHUFFLE_MERGE"), $"k1" === $"k4", joinType)
+ .queryExecution
+ .executedPlan
+- assert(collect(plan) { case _: SortMergeJoinExec => true }.size === 2)
+- assert(collect(plan) { case _: ShuffledHashJoinExec => true }.size ===
1)
++ assert(collect(plan) {
++ case _: SortMergeJoinExec | _: CometSortMergeJoinExec => true }.size
=== 2)
++ assert(collect(plan) {
++ case _: ShuffledHashJoinExec | _: CometHashJoinExec => true }.size
=== 1)
+ // No extra sort before last sort merge join
+- assert(collect(plan) { case _: SortExec => true }.size === 3)
++ assert(collect(plan) {
++ case _: SortExec | _: CometSortExec => true }.size === 3)
+ })
+ }
+
+@@ -1349,12 +1360,12 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ inputDFs.foreach { case (df1, df2, joinExprs) =>
+ val smjDF = df1.join(df2.hint("SHUFFLE_MERGE"), joinExprs, "full")
+ assert(collect(smjDF.queryExecution.executedPlan) {
+- case _: SortMergeJoinExec => true }.size === 1)
++ case _: SortMergeJoinExec | _: CometSortMergeJoinExec => true }.size
=== 1)
+ val smjResult = smjDF.collect()
+
+ val shjDF = df1.join(df2.hint("SHUFFLE_HASH"), joinExprs, "full")
+ assert(collect(shjDF.queryExecution.executedPlan) {
+- case _: ShuffledHashJoinExec => true }.size === 1)
++ case _: ShuffledHashJoinExec | _: CometHashJoinExec => true }.size
=== 1)
+ // Same result between shuffled hash join and sort merge join
+ checkAnswer(shjDF, smjResult)
+ }
+@@ -1413,12 +1424,14 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ val smjDF = df1.hint("SHUFFLE_MERGE").join(df2, joinExprs,
"leftouter")
+ assert(collect(smjDF.queryExecution.executedPlan) {
+ case _: SortMergeJoinExec => true
++ case _: CometSortMergeJoinExec => true
+ }.size === 1)
+ val smjResult = smjDF.collect()
+
+ val shjDF = df1.hint("SHUFFLE_HASH").join(df2, joinExprs,
"leftouter")
+ assert(collect(shjDF.queryExecution.executedPlan) {
+ case _: ShuffledHashJoinExec => true
++ case _: CometHashJoinExec => true
+ }.size === 1)
+ // Same result between shuffled hash join and sort merge join
+ checkAnswer(shjDF, smjResult)
+@@ -1429,12 +1442,14 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ val smjDF = df2.join(df1.hint("SHUFFLE_MERGE"), joinExprs,
"rightouter")
+ assert(collect(smjDF.queryExecution.executedPlan) {
+ case _: SortMergeJoinExec => true
++ case _: CometSortMergeJoinExec => true
+ }.size === 1)
+ val smjResult = smjDF.collect()
+
+ val shjDF = df2.join(df1.hint("SHUFFLE_HASH"), joinExprs,
"rightouter")
+ assert(collect(shjDF.queryExecution.executedPlan) {
+ case _: ShuffledHashJoinExec => true
++ case _: CometHashJoinExec => true
+ }.size === 1)
+ // Same result between shuffled hash join and sort merge join
+ checkAnswer(shjDF, smjResult)
+@@ -1478,13 +1493,20 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ assert(shjCodegenDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : ShuffledHashJoinExec) => true
+ case WholeStageCodegenExec(ProjectExec(_, _ :
ShuffledHashJoinExec)) => true
++ case WholeStageCodegenExec(ColumnarToRowExec(InputAdapter(_:
CometHashJoinExec))) =>
++ true
++ case WholeStageCodegenExec(ColumnarToRowExec(
++ InputAdapter(CometProjectExec(_, _, _, _, _: CometHashJoinExec,
_)))) => true
++ case _: CometHashJoinExec => true
+ }.size === 1)
+ checkAnswer(shjCodegenDF, Seq.empty)
+
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
+ val shjNonCodegenDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" ===
$"k2", joinType)
+ assert(shjNonCodegenDF.queryExecution.executedPlan.collect {
+- case _: ShuffledHashJoinExec => true }.size === 1)
++ case _: ShuffledHashJoinExec => true
++ case _: CometHashJoinExec => true
++ }.size === 1)
+ checkAnswer(shjNonCodegenDF, Seq.empty)
+ }
+ }
+@@ -1516,7 +1538,10 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ "/*+ BROADCAST(t2) */ t1.k as k"
+ }
+ val plan = sql(getAggQuery(selectExpr,
joinType)).queryExecution.executedPlan
+- assert(collect(plan) { case _: BroadcastNestedLoopJoinExec => true
}.size === 1)
++ assert(collect(plan) {
++ case _: BroadcastNestedLoopJoinExec | _:
CometBroadcastNestedLoopJoinExec =>
++ true
++ }.size === 1)
+ // No extra shuffle before aggregation
+ assert(collect(plan) { case _: ShuffleExchangeExec => true }.size
=== 0)
+ }
+@@ -1532,7 +1557,8 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ val plan = sql(getAggQuery(selectExpr,
joinType)).queryExecution.executedPlan
+ assert(collect(plan) { case _: BroadcastNestedLoopJoinExec => true
}.size === 1)
+ // Have shuffle before aggregation
+- assert(collect(plan) { case _: ShuffleExchangeExec => true }.size
=== 1)
++ assert(collect(plan) {
++ case _: ShuffleExchangeLike => true }.size === 1)
+ }
+
+ def getJoinQuery(selectExpr: String, joinType: String): String = {
+@@ -1560,10 +1586,16 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ "/*+ BROADCAST(right_t) */ k1 as k0"
+ }
+ val plan = sql(getJoinQuery(selectExpr,
joinType)).queryExecution.executedPlan
+- assert(collect(plan) { case _: BroadcastNestedLoopJoinExec => true
}.size === 1)
+- assert(collect(plan) { case _: SortMergeJoinExec => true }.size ===
3)
++ assert(collect(plan) {
++ case _: BroadcastNestedLoopJoinExec | _:
CometBroadcastNestedLoopJoinExec =>
++ true
++ }.size === 1)
++ assert(collect(plan) {
++ case _: SortMergeJoinExec => true
++ case _: CometSortMergeJoinExec => true
++ }.size === 3)
+ // No extra sort on left side before last sort merge join
+- assert(collect(plan) { case _: SortExec => true }.size === 5)
++ assert(collect(plan) { case _: SortExec | _: CometSortExec => true
}.size === 5)
+ }
+
+ // Test output ordering is not preserved
+@@ -1572,9 +1604,12 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ val selectExpr = "/*+ BROADCAST(left_t) */ k1 as k0"
+ val plan = sql(getJoinQuery(selectExpr,
joinType)).queryExecution.executedPlan
+ assert(collect(plan) { case _: BroadcastNestedLoopJoinExec => true
}.size === 1)
+- assert(collect(plan) { case _: SortMergeJoinExec => true }.size ===
3)
++ assert(collect(plan) {
++ case _: SortMergeJoinExec => true
++ case _: CometSortMergeJoinExec => true
++ }.size === 3)
+ // Have sort on left side before last sort merge join
+- assert(collect(plan) { case _: SortExec => true }.size === 6)
++ assert(collect(plan) { case _: SortExec | _: CometSortExec => true
}.size === 6)
+ }
+
+ // Test singe partition
+@@ -1584,7 +1619,8 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ |FROM range(0, 10, 1, 1) t1 FULL OUTER JOIN range(0, 10, 1, 1) t2
+ |""".stripMargin)
+ val plan = fullJoinDF.queryExecution.executedPlan
+- assert(collect(plan) { case _: ShuffleExchangeExec => true}.size == 1)
++ assert(collect(plan) {
++ case _: ShuffleExchangeLike => true}.size == 1)
+ checkAnswer(fullJoinDF, Row(100))
+ }
+ }
+@@ -1657,6 +1693,9 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ Seq(semiJoinDF, antiJoinDF).foreach { df =>
+ assert(collect(df.queryExecution.executedPlan) {
+ case j: ShuffledHashJoinExec if j.ignoreDuplicatedKey ==
ignoreDuplicatedKey => true
++ case j: CometHashJoinExec
++ if
j.originalPlan.asInstanceOf[ShuffledHashJoinExec].ignoreDuplicatedKey ==
++ ignoreDuplicatedKey => true
+ }.size == 1)
+ }
+ }
+@@ -1701,14 +1740,20 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+
+ test("SPARK-43113: Full outer join with duplicate stream-side references in
condition (SMJ)") {
+ def check(plan: SparkPlan): Unit = {
+- assert(collect(plan) { case _: SortMergeJoinExec => true }.size === 1)
++ assert(collect(plan) {
++ case _: SortMergeJoinExec => true
++ case _: CometSortMergeJoinExec => true
++ }.size === 1)
+ }
+ dupStreamSideColTest("MERGE", check)
+ }
+
+ test("SPARK-43113: Full outer join with duplicate stream-side references in
condition (SHJ)") {
+ def check(plan: SparkPlan): Unit = {
+- assert(collect(plan) { case _: ShuffledHashJoinExec => true }.size ===
1)
++ assert(collect(plan) {
++ case _: ShuffledHashJoinExec => true
++ case _: CometHashJoinExec => true
++ }.size === 1)
+ }
+ dupStreamSideColTest("SHUFFLE_HASH", check)
+ }
+@@ -1847,7 +1892,8 @@ class ThreadLeakInSortMergeJoinSuite
+ sparkConf.set(SHUFFLE_SPILL_NUM_ELEMENTS_FORCE_SPILL_THRESHOLD, 20))
+ }
+
+- test("SPARK-47146: thread leak when doing SortMergeJoin (with spill)") {
++ test("SPARK-47146: thread leak when doing SortMergeJoin (with spill)",
++ IgnoreComet("Comet SMJ doesn't spill yet")) {
+
+ withSQLConf(
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "1") {
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/PlanStabilitySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/PlanStabilitySuite.scala
+index 6cd49948630..b6fcb716a2a 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/PlanStabilitySuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/PlanStabilitySuite.scala
+@@ -69,7 +69,7 @@ import org.apache.spark.util.Utils
+ * }}}
+ */
+ // scalastyle:on line.size.limit
+-trait PlanStabilitySuite extends DisableAdaptiveExecutionSuite {
++trait PlanStabilitySuite extends DisableAdaptiveExecutionSuite with
IgnoreCometSuite {
+
+ protected val baseResourcePath = {
+ // use the same way as `SQLQueryTestSuite` to get the resource path
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
b/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
+index 291aa7cab72..7783c37683e 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/QueryTest.scala
+@@ -42,6 +42,7 @@ import org.apache.spark.sql.catalyst.plans._
+ import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation,
LogicalPlan}
+ import org.apache.spark.sql.catalyst.util._
+ import org.apache.spark.sql.classic.ClassicConversions._
++import org.apache.spark.sql.comet.{CometFilterExec, CometProjectExec}
+ import org.apache.spark.sql.execution.{FilterExec, QueryExecution, SparkPlan,
SQLExecution}
+ import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecution
+ import org.apache.spark.sql.execution.columnar.InMemoryRelation
+@@ -295,6 +296,11 @@ trait QueryTestBase
+ self.spark.asInstanceOf[classic.SparkSession].converter
+ }
+
++ /**
++ * Whether Comet extension is enabled
++ */
++ protected def isCometEnabled: Boolean = classic.SparkSession.isCometEnabled
++
+ protected override def withSQLConf[T](pairs: (String, String)*)(f: => T): T
= {
+ SparkSession.setActiveSession(spark)
+ super.withSQLConf(pairs: _*)(f)
+@@ -514,6 +520,8 @@ trait QueryTestBase
+ val schema = df.schema
+ val withoutFilters = df.queryExecution.executedPlan.transform {
+ case FilterExec(_, child) => child
++ case CometFilterExec(_, _, _, _, child, _) => child
++ case CometProjectExec(_, _, _, _, CometFilterExec(_, _, _, _, child,
_), _) => child
+ }
+
+ spark.asInstanceOf[classic.SparkSession]
+@@ -757,6 +765,11 @@ trait QueryTest extends SparkFunSuite with QueryTestBase
with PlanTest {
+
+ override protected def test(testName: String, testTags: Tag*)(testFun: =>
Any)
+ (implicit pos: Position): Unit = {
++ // Check Comet skip tags first, before DisableAdaptiveExecution handling
++ if (isCometEnabled && testTags.exists(_.isInstanceOf[IgnoreComet])) {
++ ignore(testName + " (disabled when Comet is on)", testTags: _*)(testFun)
++ return
++ }
+ if (testTags.exists(_.isInstanceOf[DisableAdaptiveExecution])) {
+ super.test(testName, testTags: _*) {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+index da6f6aca2ad..c02b7c99490 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
+@@ -1529,7 +1529,8 @@ class SQLQuerySuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ checkAnswer(sql("select -0.001"), Row(BigDecimal("-0.001")))
+ }
+
+- test("external sorting updates peak execution memory") {
++ test("external sorting updates peak execution memory",
++ IgnoreComet("TODO: native CometSort does not update peak execution
memory")) {
+ AccumulatorSuite.verifyPeakExecutionMemorySet(sparkContext, "external
sort") {
+ sql("SELECT * FROM testData2 ORDER BY a ASC, b ASC").collect()
+ }
+@@ -1990,8 +1991,15 @@ class SQLQuerySuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ countAcc.add(1)
+ x
+ })
++ // Comet's `CometProject` and `CometHashAggregate` do not implement
Spark's cross-sibling
++ // subexpression elimination over `ScalaUDF`, so each reference invokes
the UDF body
++ // separately. The other call sites in this test pass against Comet
because the source
++ // (`testData2`, a `LocalRelation`) is not Comet-scannable and the
project runs on Spark's
++ // path; the `agg` case routes through `CometHashAggregate` once an
Exchange enters the plan.
++ // Tracking issue:
https://github.com/apache/datafusion-comet/issues/4516
+ verifyCallCount(
+- df.agg(sum(testUdf($"b") + testUdf($"b") + testUdf($"b"))), Row(3.0),
1)
++ df.agg(sum(testUdf($"b") + testUdf($"b") + testUdf($"b"))), Row(3.0),
++ if (isCometEnabled) 3 else 1)
+
+ verifyCallCount(
+ df.selectExpr("testUdf(a + 1) + testUdf(1 + a)", "testUdf(a + 1)"),
Row(4, 2), 1)
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
+index 395cb67f441..22296439d2e 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLQueryTestSuite.scala
+@@ -188,7 +188,20 @@ class SQLQueryTestSuite extends SharedSparkSession with
SQLHelper
+ if (TestUtils.testCommandAvailable("/bin/bash")) Nil else
Set("transform.sql")
+ /** List of test cases to ignore, in lower cases. */
+ protected def ignoreList: Set[String] = Set(
+- "ignored.sql" // Do NOT remove this one. It is here to test the ignore
functionality.
++ "ignored.sql", // Do NOT remove this one. It is here to test the ignore
functionality.
++ // Comet: ORDER BY column has ties; row order is non-deterministic when
++ // running with high parallelism. Tracked for restoration once Comet
++ // produces stable ordering for these queries.
++ "replacing-missing-expression-with-alias.sql",
++ "in-set-operations.sql",
++ // Comet: theta_sketch_estimate(theta_sketch_agg(...)) over collated
++ // strings reads sketch binary as UTF-8 and fails. Skip until Comet
++ // adds proper handling or falls back for theta-sketch on collation.
++ "thetasketch.sql",
++ // Comet: same collation problem as thetasketch.sql, for
++ // tuple_sketch_estimate_double(tuple_sketch_agg_double(...)) over a
collated
++ // string column (Spark 4.2 added these functions).
++ "tuplesketch.sql"
+ ) ++ otherIgnoreList
+ /** List of test cases that require TPCDS table schemas to be loaded. */
+ private def requireTPCDSCases: Seq[String] = Seq("pipe-operators.sql")
+@@ -829,9 +842,24 @@ class SQLQueryTestSuite extends SharedSparkSession with
SQLHelper
+ s"Schema did not match for query #$i\n${expected.sql}: $output") {
+ output.schema
+ }
+- assertResult(expected.output, s"Result did not match" +
++ // Comet may surface errors as `CometNativeException` instead of the
matching Spark
++ // exception class when a `ScalaUDF` dispatched into the native plan
evaluates a
++ // divide-by-zero (DataFusion wraps the typed error so the JNI bridge
cannot downcast it).
++ // Same category, different surface. Collapse both sides to a
placeholder when this happens
++ // so the literal compare passes.
++ // Tracking issue:
https://github.com/apache/datafusion-comet/issues/4517
++ val (expectedOut, actualOut) = if (isCometEnabled &&
++
expected.output.startsWith("org.apache.spark.SparkArithmeticException") &&
++ expected.output.contains("\"DIVIDE_BY_ZERO\"") &&
++ output.output.startsWith("org.apache.comet.CometNativeException") &&
++ output.output.contains("DivideByZero")) {
++ ("[DIVIDE_BY_ZERO]", "[DIVIDE_BY_ZERO]")
++ } else {
++ (expected.output, output.output)
++ }
++ assertResult(expectedOut, s"Result did not match" +
+ s" for query #$i\n${expected.sql}") {
+- output.output
++ actualOut
+ }
+ }
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionJobTaggingAndCancellationSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionJobTaggingAndCancellationSuite.scala
+index d7b2511eac2..d5f5b940b94 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionJobTaggingAndCancellationSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/SparkSessionJobTaggingAndCancellationSuite.scala
+@@ -82,6 +82,10 @@ class SparkSessionJobTaggingAndCancellationSuite
+ }
+
+ test("Tags set from session are prefixed with session UUID") {
++ // This test relies on job scheduling order which is timing-dependent and
becomes unreliable
++ // when Comet is enabled due to changes in async execution behaviour.
++ assume(!classic.SparkSession.isCometEnabled,
++ "Skipped when Comet is enabled: test results are timing-dependent")
+ sc = new SparkContext("local[2]", "test")
+ val session =
classic.SparkSession.builder().sparkContext(sc).getOrCreate()
+ import session.implicits._
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/StringFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/StringFunctionsSuite.scala
+index 63589472854..f8c07a9b037 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/StringFunctionsSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/StringFunctionsSuite.scala
+@@ -17,6 +17,8 @@
+
+ package org.apache.spark.sql
+
++import org.apache.comet.CometConf
++
+ import org.apache.spark.{SPARK_DOC_ROOT, SparkIllegalArgumentException,
SparkRuntimeException}
+ import org.apache.spark.sql.catalyst.expressions.Cast._
+ import org.apache.spark.sql.catalyst.expressions.IsNotNull
+@@ -179,29 +181,31 @@ class StringFunctionsSuite extends SharedSparkSession {
+ }
+
+ test("string regex_replace / regex_extract") {
+- val df = Seq(
+- ("100-200", "(\\d+)-(\\d+)", "300"),
+- ("100-200", "(\\d+)-(\\d+)", "400"),
+- ("100-200", "(\\d+)", "400")).toDF("a", "b", "c")
++ withSQLConf(CometConf.getExprAllowIncompatConfigKey("regexp") -> "true") {
++ val df = Seq(
++ ("100-200", "(\\d+)-(\\d+)", "300"),
++ ("100-200", "(\\d+)-(\\d+)", "400"),
++ ("100-200", "(\\d+)", "400")).toDF("a", "b", "c")
+
+- checkAnswer(
+- df.select(
+- regexp_replace($"a", "(\\d+)", "num"),
+- regexp_replace($"a", $"b", $"c"),
+- regexp_extract($"a", "(\\d+)-(\\d+)", 1)),
+- Row("num-num", "300", "100") :: Row("num-num", "400", "100") ::
+- Row("num-num", "400-400", "100") :: Nil)
+-
+- // for testing the mutable state of the expression in code gen.
+- // This is a hack way to enable the codegen, thus the codegen is enable
by default,
+- // it will still use the interpretProjection if projection followed by a
LocalRelation,
+- // hence we add a filter operator.
+- // See the optimizer rule `ConvertToLocalRelation`
+- checkAnswer(
+- df.filter("isnotnull(a)").selectExpr(
+- "regexp_replace(a, b, c)",
+- "regexp_extract(a, b, 1)"),
+- Row("300", "100") :: Row("400", "100") :: Row("400-400", "100") :: Nil)
++ checkAnswer(
++ df.select(
++ regexp_replace($"a", "(\\d+)", "num"),
++ regexp_replace($"a", $"b", $"c"),
++ regexp_extract($"a", "(\\d+)-(\\d+)", 1)),
++ Row("num-num", "300", "100") :: Row("num-num", "400", "100") ::
++ Row("num-num", "400-400", "100") :: Nil)
++
++ // for testing the mutable state of the expression in code gen.
++ // This is a hack way to enable the codegen, thus the codegen is enable
by default,
++ // it will still use the interpretProjection if projection followed by
a LocalRelation,
++ // hence we add a filter operator.
++ // See the optimizer rule `ConvertToLocalRelation`
++ checkAnswer(
++ df.filter("isnotnull(a)").selectExpr(
++ "regexp_replace(a, b, c)",
++ "regexp_extract(a, b, 1)"),
++ Row("300", "100") :: Row("400", "100") :: Row("400-400", "100") ::
Nil)
++ }
+ }
+
+ test("non-matching optional group") {
+@@ -1425,7 +1429,12 @@ class StringFunctionsSuite extends SharedSparkSession {
+ s"'$$3 $$1') FROM $tableName"
+ val df = sql(query)
+ val plan = df.queryExecution.executedPlan
+- assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode ==
"CODEGEN_ONLY"))
++ // Comet routes regexp_replace through the codegen dispatcher, so
the executed plan is a
++ // Comet operator rather than WholeStageCodegenExec. The exception
assertions below still
++ // hold; only this Spark-internal plan-shape check is skipped under
Comet.
++ if (!isCometEnabled) {
++ assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode
== "CODEGEN_ONLY"))
++ }
+ val exception = intercept[SparkRuntimeException] {
+ df.collect()
+ }
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
+index cd3e389d765..354c5b97e8b 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
+@@ -23,10 +23,11 @@ import org.apache.spark.SparkRuntimeException
+ import org.apache.spark.sql.catalyst.expressions.{EqualTo, NamedExpression,
OuterReference, SubqueryExpression}
+ import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
+ import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join,
LogicalPlan, Project, Sort, Union}
++import org.apache.spark.sql.comet.{CometColumnarToRowExec,
CometNativeColumnarToRowExec, CometNativeScanExec, CometScanExec}
+ import org.apache.spark.sql.execution._
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecution}
+ import org.apache.spark.sql.execution.datasources.FileScanRDD
+-import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
++import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
+ import org.apache.spark.sql.execution.joins.{BaseJoinExec,
BroadcastHashJoinExec, BroadcastNestedLoopJoinExec}
+ import org.apache.spark.sql.internal.SQLConf
+ import org.apache.spark.sql.test.SharedSparkSession
+@@ -1528,6 +1529,30 @@ class SubquerySuite extends SharedSparkSession
+ fs.inputRDDs().forall(
+ _.asInstanceOf[FileScanRDD].filePartitions.forall(
+ _.files.forall(_.urlEncodedPath.contains("p=0"))))
++ case WholeStageCodegenExec(ColumnarToRowExec(InputAdapter(
++ fs @ CometScanExec(_, _, _, partitionFilters, _, _, _, _, _, _)))) =>
++ partitionFilters.exists(ExecSubqueryExpression.hasSubquery) &&
++ fs.inputRDDs().forall(
++ _.asInstanceOf[FileScanRDD].filePartitions.forall(
++ _.files.forall(_.urlEncodedPath.contains("p=0"))))
++ case CometNativeColumnarToRowExec(
++ fs: CometNativeScanExec) =>
++ fs.partitionFilters.exists(ExecSubqueryExpression.hasSubquery) &&
++ fs.inputRDDs().forall(
++ _.asInstanceOf[FileScanRDD].filePartitions.forall(
++ _.files.forall(_.urlEncodedPath.contains("p=0"))))
++ case WholeStageCodegenExec(CometColumnarToRowExec(InputAdapter(
++ fs @ CometScanExec(_, _, _, partitionFilters, _, _, _, _, _, _)))) =>
++ partitionFilters.exists(ExecSubqueryExpression.hasSubquery) &&
++ fs.inputRDDs().forall(
++ _.asInstanceOf[FileScanRDD].filePartitions.forall(
++ _.files.forall(_.urlEncodedPath.contains("p=0"))))
++ case WholeStageCodegenExec(CometColumnarToRowExec(InputAdapter(
++ fs: CometNativeScanExec))) =>
++ fs.partitionFilters.exists(ExecSubqueryExpression.hasSubquery) &&
++ fs.inputRDDs().forall(
++ _.asInstanceOf[FileScanRDD].filePartitions.forall(
++ _.files.forall(_.urlEncodedPath.contains("p=0"))))
+ case _ => false
+ })
+ }
+@@ -2093,7 +2118,7 @@ class SubquerySuite extends SharedSparkSession
+
+ df.collect()
+ val exchanges = collect(df.queryExecution.executedPlan) {
+- case s: ShuffleExchangeExec => s
++ case s: ShuffleExchangeLike => s
+ }
+ assert(exchanges.size === 1)
+ }
+@@ -2464,18 +2489,29 @@ class SubquerySuite extends SharedSparkSession
+ def checkFileSourceScan(query: String, answer: Seq[Row]): Unit = {
+ val df = sql(query)
+ checkAnswer(df, answer)
+- val fileSourceScanExec = collect(df.queryExecution.executedPlan) {
+- case f: FileSourceScanExec => f
++ val dataSourceScanExec = collect(df.queryExecution.executedPlan) {
++ case f: FileSourceScanLike => f
++ case c: CometScanExec => c
++ case n: CometNativeScanExec => n
+ }
+ sparkContext.listenerBus.waitUntilEmpty()
+- assert(fileSourceScanExec.size === 1)
+- val scalarSubquery =
fileSourceScanExec.head.dataFilters.flatMap(_.collect {
+- case s: ScalarSubquery => s
+- })
++ assert(dataSourceScanExec.size === 1)
++ val scalarSubquery = dataSourceScanExec.head match {
++ case f: FileSourceScanLike =>
++ f.dataFilters.flatMap(_.collect { case s: ScalarSubquery => s })
++ case c: CometScanExec =>
++ c.dataFilters.flatMap(_.collect { case s: ScalarSubquery => s })
++ case n: CometNativeScanExec =>
++ n.dataFilters.flatMap(_.collect { case s: ScalarSubquery => s })
++ }
+ assert(scalarSubquery.length === 1)
+ assert(scalarSubquery.head.plan.isInstanceOf[ReusedSubqueryExec])
+- assert(fileSourceScanExec.head.metrics("numFiles").value === 1)
+- assert(fileSourceScanExec.head.metrics("numOutputRows").value ===
answer.size)
++ assert(dataSourceScanExec.head.metrics("numFiles").value === 1)
++ val numOutputRows = dataSourceScanExec.head match {
++ case n: CometNativeScanExec => n.metrics("output_rows").value
++ case other => other.metrics("numOutputRows").value
++ }
++ assert(numOutputRows === answer.size)
+ }
+
+ withTable("t1", "t2") {
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+index 2d26356890d..2c5994f5fbc 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantEndToEndSuite.scala
+@@ -21,6 +21,7 @@ import org.apache.spark.sql.QueryTest.sameRows
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{Cast, Literal}
+ import org.apache.spark.sql.catalyst.expressions.variant.{ToVariantObject,
VariantExpressionEvalUtils}
++import org.apache.spark.sql.comet.CometNativeColumnarToRowExec
+ import org.apache.spark.sql.execution.WholeStageCodegenExec
+ import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+ import org.apache.spark.sql.functions._
+@@ -391,7 +392,8 @@ class VariantEndToEndSuite extends SharedSparkSession {
+ s"cast(to_variant_object(s) as ${schema(2).dataType.sql})")
+ checkAnswer(df, input)
+ val plan = df.queryExecution.executedPlan
+- assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode ==
"CODEGEN_ONLY"))
++ assert(plan.isInstanceOf[WholeStageCodegenExec] == (codegenMode ==
"CODEGEN_ONLY")
++ || plan.isInstanceOf[CometNativeColumnarToRowExec])
+ }
+ }
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
+index 517d56ca4f0..b4f1aa671a0 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
+@@ -33,7 +33,9 @@ import org.apache.spark.sql.types._
+ import org.apache.spark.types.variant._
+ import org.apache.spark.unsafe.types.{UTF8String, VariantVal}
+
+-class VariantShreddingSuite extends SharedSparkSession with ParquetTest {
++class VariantShreddingSuite extends SharedSparkSession with ParquetTest
++ // TODO enable tests once
https://github.com/apache/datafusion-comet/issues/2209 is fixed
++ with IgnoreCometSuite {
+ def parseJson(s: String): VariantVal = {
+ val v = VariantBuilder.parseJson(s, false)
+ new VariantVal(v.getValue, v.getMetadata)
+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 37684c7fce3..f3574dec867 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
+@@ -26,6 +26,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.{CatalogV2Util, Identifier,
InMemoryTable}
+ import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.CatalogHelper
+@@ -61,7 +63,9 @@ class CollationSuite extends DatasourceV2SQLBase with
AdaptiveSparkPlanHelper {
+ assert(
+ collectFirst(queryPlan) {
+ case _: SortMergeJoinExec => assert(isSortMergeForced)
++ case _: CometSortMergeJoinExec => assert(isSortMergeForced)
+ case _: HashJoin => assert(!isSortMergeForced)
++ case _: CometHashJoinExec | _: CometBroadcastHashJoinExec =>
assert(!isSortMergeForced)
+ }.nonEmpty
+ )
+ }
+@@ -469,6 +473,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)
+ }
+ }
+@@ -1607,10 +1612,14 @@ class CollationSuite extends DatasourceV2SQLBase with
AdaptiveSparkPlanHelper {
+ if
(!CollationFactory.fetchCollation(t.collation).supportsBinaryEquality) {
+ assert(collectFirst(queryPlan) {
+ case b: HashJoin => b.leftKeys.head
++ case ch: CometHashJoinExec => ch.leftKeys.head
++ case cbh: CometBroadcastHashJoinExec => cbh.leftKeys.head
+ }.head.isInstanceOf[CollationKey])
+ } else {
+ assert(!collectFirst(queryPlan) {
+ case b: HashJoin => b.leftKeys.head
++ case ch: CometHashJoinExec => ch.leftKeys.head
++ case cbh: CometBroadcastHashJoinExec => cbh.leftKeys.head
+ }.head.isInstanceOf[CollationKey])
+ }
+ }
+@@ -1666,11 +1675,13 @@ class CollationSuite extends DatasourceV2SQLBase with
AdaptiveSparkPlanHelper {
+ if
(!CollationFactory.fetchCollation(t.collation).supportsBinaryEquality) {
+ assert(collectFirst(queryPlan) {
+ case b: BroadcastHashJoinExec => b.leftKeys.head
++ case b: CometBroadcastHashJoinExec => b.leftKeys.head
+
}.head.asInstanceOf[ArrayTransform].function.asInstanceOf[LambdaFunction].
+ function.isInstanceOf[CollationKey])
+ } else {
+ assert(!collectFirst(queryPlan) {
+ case b: BroadcastHashJoinExec => b.leftKeys.head
++ case b: CometBroadcastHashJoinExec => b.leftKeys.head
+ }.head.isInstanceOf[ArrayTransform])
+ }
+ }
+@@ -1736,6 +1747,7 @@ class CollationSuite extends DatasourceV2SQLBase with
AdaptiveSparkPlanHelper {
+ } else {
+ assert(!collectFirst(queryPlan) {
+ case b: BroadcastHashJoinExec => b.leftKeys.head
++ case b: CometBroadcastHashJoinExec => b.leftKeys.head
+ }.head.isInstanceOf[ArrayTransform])
+ }
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+index 5ae23bc3338..5c2c3fff284 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala
+@@ -29,6 +29,7 @@ import org.apache.spark.SparkUnsupportedOperationException
+ import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{AttributeReference,
GreaterThan => CatalystGreaterThan, Literal => CatalystLiteral, ScalarSubquery}
++import org.apache.spark.sql.comet.CometSortExec
+ import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Project}
+ import org.apache.spark.sql.connector.catalog.{PartitionInternalRow,
SupportsRead, Table, TableCapability, TableProvider}
+ import org.apache.spark.sql.connector.catalog.TableCapability._
+@@ -41,7 +42,7 @@ import org.apache.spark.sql.execution.SortExec
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+ import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec,
DataSourceV2Relation, DataSourceV2ScanRelation, V2ScanPartitioningAndOrdering}
+ import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
+-import org.apache.spark.sql.execution.exchange.{Exchange, ShuffleExchangeExec}
++import org.apache.spark.sql.execution.exchange.{Exchange,
ShuffleExchangeExec, ShuffleExchangeLike}
+ import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
+ import org.apache.spark.sql.expressions.Window
+ import org.apache.spark.sql.functions._
+@@ -283,13 +284,13 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+ val groupByColJ = df.groupBy($"j").agg(sum($"i"))
+ checkAnswer(groupByColJ, Seq(Row(2, 8), Row(4, 2), Row(6, 5)))
+ assert(collectFirst(groupByColJ.queryExecution.executedPlan) {
+- case e: ShuffleExchangeExec => e
++ case e: ShuffleExchangeLike => e
+ }.isDefined)
+
+ val groupByIPlusJ = df.groupBy($"i" + $"j").agg(count("*"))
+ checkAnswer(groupByIPlusJ, Seq(Row(5, 2), Row(6, 2), Row(8, 1),
Row(9, 1)))
+ assert(collectFirst(groupByIPlusJ.queryExecution.executedPlan) {
+- case e: ShuffleExchangeExec => e
++ case e: ShuffleExchangeLike => e
+ }.isDefined)
+ }
+ }
+@@ -349,10 +350,11 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+
+ val (shuffleExpected, sortExpected) = groupByExpects
+ assert(collectFirst(groupBy.queryExecution.executedPlan) {
+- case e: ShuffleExchangeExec => e
++ case e: ShuffleExchangeLike => e
+ }.isDefined === shuffleExpected)
+ assert(collectFirst(groupBy.queryExecution.executedPlan) {
+ case e: SortExec => e
++ case c: CometSortExec => c
+ }.isDefined === sortExpected)
+ }
+
+@@ -367,10 +369,11 @@ class DataSourceV2Suite extends SharedSparkSession with
AdaptiveSparkPlanHelper
+
+ val (shuffleExpected, sortExpected) = windowFuncExpects
+
assert(collectFirst(windowPartByColIOrderByColJ.queryExecution.executedPlan) {
+- case e: ShuffleExchangeExec => e
++ case e: ShuffleExchangeLike => e
+ }.isDefined === shuffleExpected)
+
assert(collectFirst(windowPartByColIOrderByColJ.queryExecution.executedPlan) {
+ case e: SortExec => e
++ case c: CometSortExec => c
+ }.isDefined === sortExpected)
+ }
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
+index 0fbfaedfdca..7d8383d27be 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
+@@ -20,6 +20,7 @@ import scala.collection.mutable.ArrayBuffer
+
+ import org.apache.spark.{SparkConf, SparkException}
+ import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
++import org.apache.spark.sql.comet.{CometNativeScanExec, CometScanExec}
+ import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite,
Table, TableCapability}
+ import org.apache.spark.sql.connector.read.ScanBuilder
+ import org.apache.spark.sql.connector.write.{LogicalWriteInfo, WriteBuilder}
+@@ -190,7 +191,11 @@ class FileDataSourceV2FallBackSuite extends
SharedSparkSession {
+ val df = spark.read.format(format).load(path.getCanonicalPath)
+ checkAnswer(df, inputData.toDF())
+ assert(
+-
df.queryExecution.executedPlan.exists(_.isInstanceOf[FileSourceScanExec]))
++ df.queryExecution.executedPlan.exists {
++ case _: FileSourceScanExec | _: CometScanExec | _:
CometNativeScanExec => true
++ case _ => false
++ }
++ )
+ }
+ } finally {
+ spark.listenerManager.unregister(listener)
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+index 38de6b043bc..939e4a8c4c3 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+@@ -25,6 +25,8 @@ import org.apache.spark.sql.{DataFrame, ExplainSuiteHelper,
Row}
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{Ascending,
AttributeReference, Literal, TransformExpression}
+ import org.apache.spark.sql.catalyst.plans.physical
++import org.apache.spark.sql.comet.CometSortMergeJoinExec
++import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
+ import org.apache.spark.sql.connector.catalog.{Column, Identifier,
InMemoryTableCatalog}
+ import org.apache.spark.sql.connector.catalog.functions._
+ import org.apache.spark.sql.connector.distributions.Distributions
+@@ -325,6 +327,7 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
+ protected def collectAllShuffles(plan: SparkPlan): Seq[ShuffleExchangeLike]
= {
+ collect(plan) {
+ case s: ShuffleExchangeExec => s
++ case c: CometShuffleExchangeExec => c
+ }
+ }
+
+@@ -338,9 +341,10 @@ class KeyGroupedPartitioningSuite extends
DistributionAndOrderingSuiteBase with
+ // here we skip collecting shuffle operators that are not associated with
SMJ
+ collect(plan) {
+ case s: SortMergeJoinExec => s
++ case c: CometSortMergeJoinExec => c.originalPlan
+ }.flatMap(smj =>
+ collect(smj) {
+- case s: ShuffleExchangeExec => s
++ case s: ShuffleExchangeLike => s
+ })
+ }.toSet.toSeq
+
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala
+index ec1b34b8c21..ff4f2a70041 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/WriteDistributionAndOrderingSuite.scala
+@@ -21,7 +21,7 @@ package org.apache.spark.sql.connector
+ import java.sql.Date
+ import java.util.Collections
+
+-import org.apache.spark.sql.{catalyst, AnalysisException, DataFrame, Row}
++import org.apache.spark.sql.{catalyst, AnalysisException, DataFrame,
IgnoreCometSuite, Row}
+ import org.apache.spark.sql.catalyst.expressions.{ApplyFunctionExpression,
Cast, Literal, TransformExpression}
+ import org.apache.spark.sql.catalyst.expressions.objects.Invoke
+ import org.apache.spark.sql.catalyst.plans.physical
+@@ -46,7 +46,8 @@ import org.apache.spark.sql.util.QueryExecutionListener
+ import org.apache.spark.tags.SlowSQLTest
+
+ @SlowSQLTest
+-class WriteDistributionAndOrderingSuite extends
DistributionAndOrderingSuiteBase {
++class WriteDistributionAndOrderingSuite extends
DistributionAndOrderingSuiteBase
++ with IgnoreCometSuite {
+ import testImplicits._
+
+ before {
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
+index ce549da03b4..47adf7cf94d 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
+@@ -31,7 +31,7 @@ import org.mockito.Mockito.{mock, spy, when}
+ import org.scalatest.time.SpanSugar._
+
+ import org.apache.spark._
+-import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, Encoder,
KryoData, Row, SaveMode}
++import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, Encoder,
IgnoreComet, KryoData, Row, SaveMode}
+ import org.apache.spark.sql.catalyst.FunctionIdentifier
+ import org.apache.spark.sql.catalyst.analysis.{NamedParameter,
UnresolvedGenerator}
+ import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder}
+@@ -266,7 +266,8 @@ class QueryExecutionErrorsSuite
+ }
+
+ test("INCONSISTENT_BEHAVIOR_CROSS_VERSION: " +
+- "compatibility with Spark 2.4/3.2 in reading/writing dates") {
++ "compatibility with Spark 2.4/3.2 in reading/writing dates",
++ IgnoreComet("Comet doesn't completely support datetime rebase mode yet"))
{
+
+ // Fail to read ancient datetime values.
+ withSQLConf(SQLConf.PARQUET_REBASE_MODE_IN_READ.key ->
EXCEPTION.toString) {
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/DataSourceScanExecRedactionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/DataSourceScanExecRedactionSuite.scala
+index 60bf3a21e96..7693107f107 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/DataSourceScanExecRedactionSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/DataSourceScanExecRedactionSuite.scala
+@@ -23,7 +23,7 @@ import scala.util.Random
+ import org.apache.hadoop.fs.Path
+
+ import org.apache.spark.SparkConf
+-import org.apache.spark.sql.DataFrame
++import org.apache.spark.sql.{DataFrame, IgnoreComet}
+ import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+ import org.apache.spark.sql.execution.datasources.v2.orc.OrcScan
+ import org.apache.spark.sql.internal.SQLConf
+@@ -195,7 +195,7 @@ class DataSourceV2ScanExecRedactionSuite extends
DataSourceScanRedactionTest {
+ }
+ }
+
+- test("FileScan description") {
++ test("FileScan description", IgnoreComet("Comet doesn't use BatchScan")) {
+ Seq("json", "orc", "parquet").foreach { format =>
+ withTempPath { path =>
+ val dir = path.getCanonicalPath
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/InsertSortForLimitAndOffsetSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/InsertSortForLimitAndOffsetSuite.scala
+index faf0469ec20..dfed25d87fe 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/InsertSortForLimitAndOffsetSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/InsertSortForLimitAndOffsetSuite.scala
+@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution
+
+ import org.apache.spark.sql.Dataset
+ import org.apache.spark.sql.IntegratedUDFTestUtils._
++import org.apache.spark.sql.comet.{CometCollectLimitExec,
CometGlobalLimitExec, CometProjectExec, CometSortExec}
+ import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+ import org.apache.spark.sql.functions.rand
+ import org.apache.spark.sql.internal.SQLConf
+@@ -38,7 +39,7 @@ class InsertSortForLimitAndOffsetSuite extends
SharedSparkSession
+
+ private def assertHasCollectLimitExec(plan: SparkPlan): Unit = {
+ assert(find(plan) {
+- case _: CollectLimitExec => true
++ case _: CollectLimitExec | _: CometCollectLimitExec => true
+ case _ => false
+ }.isDefined)
+ }
+@@ -46,6 +47,7 @@ class InsertSortForLimitAndOffsetSuite extends
SharedSparkSession
+ private def assertHasGlobalLimitExec(plan: SparkPlan): Unit = {
+ assert(find(plan) {
+ case _: GlobalLimitExec => true
++ case _: CometGlobalLimitExec => true
+ case _ => false
+ }.isDefined)
+ }
+@@ -54,6 +56,11 @@ class InsertSortForLimitAndOffsetSuite extends
SharedSparkSession
+ find(plan) {
+ case GlobalLimitExec(_, s: SortExec, _) => !s.global
+ case GlobalLimitExec(_, ProjectExec(_, s: SortExec), _) => !s.global
++ case CometGlobalLimitExec(_, _, _, _, s: CometSortExec, _) =>
++ !s.originalPlan.asInstanceOf[SortExec].global
++ case CometGlobalLimitExec(_, _, _, _,
++ CometProjectExec(_, _, _, _, s: CometSortExec, _), _) =>
++ !s.originalPlan.asInstanceOf[SortExec].global
+ case _ => false
+ }.isDefined
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/LogicalPlanTagInSparkPlanSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/LogicalPlanTagInSparkPlanSuite.scala
+index 743ec41dbe7..9f30d6c8e04 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/LogicalPlanTagInSparkPlanSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/LogicalPlanTagInSparkPlanSuite.scala
+@@ -53,6 +53,10 @@ class LogicalPlanTagInSparkPlanSuite extends
TPCDSQuerySuite with DisableAdaptiv
+ case ColumnarToRowExec(i: InputAdapter) => isScanPlanTree(i.child)
+ case p: ProjectExec => isScanPlanTree(p.child)
+ case f: FilterExec => isScanPlanTree(f.child)
++ // Comet produces scan plan tree like:
++ // ColumnarToRow
++ // +- ReusedExchange
++ case _: ReusedExchangeExec => false
+ case _: LeafExecNode => true
+ case _ => false
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
+index 7c434ebee5f..ac18158100d 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala
+@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution
+
+ import org.apache.spark.SparkUnsupportedOperationException
+ import org.apache.spark.rdd.RDD
+-import org.apache.spark.sql.{execution, DataFrame, Row}
++import org.apache.spark.sql.{execution, DataFrame, IgnoreCometSuite, Row}
+ import org.apache.spark.sql.AnalysisException
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions._
+@@ -38,7 +38,9 @@ import org.apache.spark.sql.internal.SQLConf
+ import org.apache.spark.sql.test.SharedSparkSession
+ import org.apache.spark.sql.types._
+
+-class PlannerSuite extends SharedSparkSession with AdaptiveSparkPlanHelper {
++// Ignore this suite when Comet is enabled. This suite tests the Spark
planner and Comet planner
++// comes out with too many difference. Simply ignoring this suite for now.
++class PlannerSuite extends SharedSparkSession with AdaptiveSparkPlanHelper
with IgnoreCometSuite {
+ import testImplicits._
+
+ setupTestData()
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
+index f7afdb5e6e5..e5a2eeb86a0 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala
+@@ -22,7 +22,7 @@ import scala.io.Source
+ import scala.util.Try
+
+ import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent,
SparkListenerJobStart}
+-import org.apache.spark.sql.{AnalysisException, ExtendedExplainGenerator,
FastOperator, SaveMode}
++import org.apache.spark.sql.{AnalysisException, ExtendedExplainGenerator,
FastOperator, IgnoreComet, SaveMode}
+ import org.apache.spark.sql.catalyst.{QueryPlanningTracker,
QueryPlanningTrackerCallback, TableIdentifier}
+ import org.apache.spark.sql.catalyst.analysis.{CurrentNamespace,
UnresolvedFunction, UnresolvedRelation}
+ import org.apache.spark.sql.catalyst.expressions.{Alias, UnsafeRow}
+@@ -461,7 +461,7 @@ class QueryExecutionSuite extends SharedSparkSession {
+ }
+ }
+
+- test("SPARK-47289: extended explain info") {
++ test("SPARK-47289: extended explain info", IgnoreComet("Comet plan extended
info is different")) {
+ val concat = new PlanStringConcat()
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala
+index 6d259ed19d0..e271f7a2914 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantProjectsSuite.scala
+@@ -17,7 +17,10 @@
+
+ package org.apache.spark.sql.execution
+
++import org.apache.comet.CometConf
++
+ import org.apache.spark.sql.{DataFrame, Row}
++import org.apache.spark.sql.comet.CometProjectExec
+ import org.apache.spark.sql.connector.SimpleWritableDataSource
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+ import org.apache.spark.sql.internal.SQLConf
+@@ -33,7 +36,10 @@ abstract class RemoveRedundantProjectsSuiteBase
+ private def assertProjectExecCount(df: DataFrame, expected: Int): Unit = {
+ withClue(df.queryExecution) {
+ val plan = df.queryExecution.executedPlan
+- val actual = collectWithSubqueries(plan) { case p: ProjectExec => p
}.size
++ val actual = collectWithSubqueries(plan) {
++ case p: ProjectExec => p
++ case p: CometProjectExec => p
++ }.size
+ assert(actual == expected)
+ }
+ }
+@@ -133,12 +139,25 @@ abstract class RemoveRedundantProjectsSuiteBase
+ val df = data.selectExpr("a", "b", "key", "explode(array(key, a, b)) as
d").filter("d > 0")
+ df.collect()
+ val plan = df.queryExecution.executedPlan
+- val numProjects = collectWithSubqueries(plan) { case p: ProjectExec =>
p }.length
++ val numProjects = collectWithSubqueries(plan) {
++ case p: ProjectExec => p
++ case p: CometProjectExec => p
++ }.length
+
+ // Create a new plan that reverse the GenerateExec output and add a new
ProjectExec between
+ // GenerateExec and its child. This is to test if the ProjectExec is
removed, the output of
+ // the query will be incorrect.
+- val newPlan = stripAQEPlan(plan) transform {
++
++ // Comet-specific change to get original Spark plan before applying
++ // a transformation to add a new ProjectExec
++ var sparkPlan: SparkPlan = null
++ withSQLConf(CometConf.COMET_EXEC_ENABLED.key -> "false") {
++ val df = data.selectExpr("a", "b", "key", "explode(array(key, a, b))
as d").filter("d > 0")
++ df.collect()
++ sparkPlan = df.queryExecution.executedPlan
++ }
++
++ val newPlan = stripAQEPlan(sparkPlan) transform {
+ case g @ GenerateExec(_, requiredChildOutput, _, _, child) =>
+ g.copy(requiredChildOutput = requiredChildOutput.reverse,
+ child = ProjectExec(requiredChildOutput.reverse, child))
+@@ -150,6 +169,7 @@ abstract class RemoveRedundantProjectsSuiteBase
+ // The manually added ProjectExec node shouldn't be removed.
+ assert(collectWithSubqueries(newExecutedPlan) {
+ case p: ProjectExec => p
++ case p: CometProjectExec => p
+ }.size == numProjects + 1)
+
+ // Check the original plan's output and the new plan's output are the
same.
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
+index 3cba30079cd..08513544cc0 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantSortsSuite.scala
+@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution
+
+ import org.apache.spark.sql.DataFrame
+ import org.apache.spark.sql.catalyst.plans.physical.{RangePartitioning,
UnknownPartitioning}
++import org.apache.spark.sql.comet.CometSortExec
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+ import org.apache.spark.sql.execution.joins.ShuffledJoin
+ import org.apache.spark.sql.internal.SQLConf
+@@ -32,7 +33,7 @@ abstract class RemoveRedundantSortsSuiteBase
+
+ private def checkNumSorts(df: DataFrame, count: Int): Unit = {
+ val plan = df.queryExecution.executedPlan
+- assert(collectWithSubqueries(plan) { case s: SortExec => s }.length ==
count)
++ assert(collectWithSubqueries(plan) { case _: SortExec | _: CometSortExec
=> 1 }.length == count)
+ }
+
+ private def checkSorts(query: String, enabledCount: Int, disabledCount:
Int): Unit = {
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+index 03e2f4942ed..b0a80fcf683 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/RemoveRedundantWindowGroupLimitsSuite.scala
+@@ -18,6 +18,7 @@
+ package org.apache.spark.sql.execution
+
+ import org.apache.spark.sql.DataFrame
++import org.apache.spark.sql.comet.CometWindowGroupLimitExec
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+ import org.apache.spark.sql.execution.window.WindowGroupLimitExec
+ import org.apache.spark.sql.functions.lit
+@@ -29,7 +30,10 @@ abstract class RemoveRedundantWindowGroupLimitsSuiteBase
+
+ private def checkNumWindowGroupLimits(df: DataFrame, count: Int): Unit = {
+ val plan = df.queryExecution.executedPlan
+- assert(collectWithSubqueries(plan) { case exec: WindowGroupLimitExec =>
exec }.length == count)
++ assert(collectWithSubqueries(plan) {
++ case exec: WindowGroupLimitExec => exec
++ case exec: CometWindowGroupLimitExec => exec
++ }.length == count)
+ }
+
+ private def checkWindowGroupLimits(query: String, count: Int): Unit = {
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala
+index 41edb53441f..eed2cbf858c 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ReplaceHashWithSortAggSuite.scala
+@@ -18,6 +18,7 @@
+ package org.apache.spark.sql.execution
+
+ import org.apache.spark.sql.DataFrame
++import org.apache.spark.sql.comet.CometHashAggregateExec
+ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper,
DisableAdaptiveExecutionSuite, EnableAdaptiveExecutionSuite}
+ import org.apache.spark.sql.execution.aggregate.{HashAggregateExec,
ObjectHashAggregateExec, SortAggregateExec}
+ import org.apache.spark.sql.internal.SQLConf
+@@ -30,7 +31,7 @@ abstract class ReplaceHashWithSortAggSuiteBase
+ private def checkNumAggs(df: DataFrame, hashAggCount: Int, sortAggCount:
Int): Unit = {
+ val plan = df.queryExecution.executedPlan
+ assert(collectWithSubqueries(plan) {
+- case s @ (_: HashAggregateExec | _: ObjectHashAggregateExec) => s
++ case s @ (_: HashAggregateExec | _: ObjectHashAggregateExec | _:
CometHashAggregateExec ) => s
+ }.length == hashAggCount)
+ assert(collectWithSubqueries(plan) { case s: SortAggregateExec => s
}.length == sortAggCount)
+ }
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala
+index e8a39eb8c53..a621c7a7923 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala
+@@ -18,7 +18,7 @@
+ package org.apache.spark.sql.execution
+
+ import org.apache.spark.TestUtils.assertSpilled
+-import org.apache.spark.sql.{AnalysisException, Row}
++import org.apache.spark.sql.{AnalysisException, IgnoreComet, Row}
+ import
org.apache.spark.sql.internal.SQLConf.{WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD,
WINDOW_EXEC_BUFFER_SPILL_THRESHOLD}
+ import org.apache.spark.sql.test.SharedSparkSession
+
+@@ -470,7 +470,7 @@ class SQLWindowFunctionSuite extends SharedSparkSession {
+ Row(1, 3, null) :: Row(2, null, 4) :: Nil)
+ }
+
+- test("test with low buffer spill threshold") {
++ test("test with low buffer spill threshold", IgnoreComet("Comet does not
support spilling")) {
+ val nums = sparkContext.parallelize(1 to 10).map(x => (x, x %
2)).toDF("x", "y")
+ nums.createOrReplaceTempView("nums")
+
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
+index b167dd13dcb..b2b6b88ce66 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/SparkPlanSuite.scala
+@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.expressions.{
+ import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
+ import org.apache.spark.sql.catalyst.plans.logical.Deduplicate
+ import org.apache.spark.sql.catalyst.trees.LeafLike
++import org.apache.spark.sql.comet.{CometColumnarToRowExec,
CometNativeColumnarToRowExec}
+ import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
+ import org.apache.spark.sql.internal.SQLConf
+ import org.apache.spark.sql.test.SharedSparkSession
+@@ -166,7 +167,11 @@ class SparkPlanSuite extends SharedSparkSession {
+ spark.range(1).write.parquet(path.getAbsolutePath)
+ val df = spark.read.parquet(path.getAbsolutePath)
+ val columnarToRowExec =
+- df.queryExecution.executedPlan.collectFirst { case p:
ColumnarToRowExec => p }.get
++ df.queryExecution.executedPlan.collectFirst {
++ case p: ColumnarToRowExec => p
++ case p: CometColumnarToRowExec => p
++ case p: CometNativeColumnarToRowExec => p
++ }.get
+ try {
+ spark.range(1).foreach { _ =>
+ columnarToRowExec.canonicalized
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/UnionCodegenSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/UnionCodegenSuite.scala
+index 835c55fe4c4..57cce5f6b03 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/UnionCodegenSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/UnionCodegenSuite.scala
+@@ -20,7 +20,7 @@ package org.apache.spark.sql.execution
+ import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
+
+ import org.apache.spark.SparkConf
+-import org.apache.spark.sql.{DataFrame, QueryTest, Row}
++import org.apache.spark.sql.{DataFrame, IgnoreComet, QueryTest, Row}
+ import org.apache.spark.sql.functions._
+ import org.apache.spark.sql.internal.SQLConf
+ import org.apache.spark.sql.test.SharedSparkSession
+@@ -605,7 +605,10 @@ class UnionCodegenSuite extends QueryTest with
SharedSparkSession {
+ }
+ }
+
+- test("SPARK-56482: partitioning-aware union falls back to non-codegen") {
++ test("SPARK-56482: partitioning-aware union falls back to non-codegen",
++ // Comet replaces Spark's UnionExec with CometUnionExec, so the executed
plan does not
++ // expose a UnionExec for this WSCG-internal assertion to inspect.
++ IgnoreComet("https://github.com/apache/datafusion-comet/issues/4965";)) {
+ // After repartition, both children expose a `HashPartitioning` on the
same key,
+ // so `UnionExec.outputPartitioning` is non-Unknown and the codegen path
is denied.
+ // AQE is disabled here so the executedPlan exposes the UnionExec directly
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
+index d70bd715879..074a9fa29d9 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/WholeStageCodegenSuite.scala
+@@ -22,9 +22,10 @@ import java.time.Duration
+
+ import org.apache.spark.SparkException
+ import org.apache.spark.rdd.MapPartitionsWithEvaluatorRDD
+-import org.apache.spark.sql.{Dataset, Row, SaveMode}
++import org.apache.spark.sql.{Dataset, IgnoreCometSuite, Row, SaveMode}
+ import org.apache.spark.sql.catalyst.expressions.{And, Cast,
CodegenObjectFactoryMode, Expression, IsNotNull}
+ import org.apache.spark.sql.catalyst.expressions.codegen.{ByteCodeStats,
CodeAndComment, CodeGenerator}
++import org.apache.spark.sql.comet.{CometColumnarToRowExec, CometHashJoinExec,
CometSortExec, CometSortMergeJoinExec}
+ import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecutionSuite
+ import org.apache.spark.sql.execution.aggregate.{HashAggregateExec,
SortAggregateExec}
+ import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
+@@ -37,7 +38,7 @@ import org.apache.spark.sql.types.{DayTimeIntervalType,
DecimalType, IntegerType
+
+ // Disable AQE because the WholeStageCodegenExec is added when running
QueryStageExec
+ class WholeStageCodegenSuite extends SharedSparkSession
+- with DisableAdaptiveExecutionSuite {
++ with DisableAdaptiveExecutionSuite with IgnoreCometSuite {
+
+ import testImplicits._
+
+@@ -176,6 +177,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val oneJoinDF = df1.join(df2.hint("SHUFFLE_HASH"), $"k1" === $"k2")
+ assert(oneJoinDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : ShuffledHashJoinExec) => true
++ case _: CometHashJoinExec => true
+ }.size === 1)
+ checkAnswer(oneJoinDF, Seq(Row(0, 0), Row(1, 1), Row(2, 2), Row(3, 3),
Row(4, 4)))
+
+@@ -184,6 +186,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ .join(df3.hint("SHUFFLE_HASH"), $"k1" === $"k3")
+ assert(twoJoinsDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : ShuffledHashJoinExec) => true
++ case _: CometHashJoinExec => true
+ }.size === 2)
+ checkAnswer(twoJoinsDF,
+ Seq(Row(0, 0, 0), Row(1, 1, 1), Row(2, 2, 2), Row(3, 3, 3), Row(4, 4,
4)))
+@@ -210,6 +213,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(joinUniqueDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_ : SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(joinUniqueDF, Seq(Row(0, 0), Row(1, 1), Row(2, 2), Row(3,
3), Row(4, 4),
+ Row(null, 5), Row(null, 6), Row(null, 7), Row(null, 8), Row(null, 9)))
+@@ -220,6 +225,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(joinNonUniqueDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_ : SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(joinNonUniqueDF, Seq(Row(0, 0), Row(0, 3), Row(0, 6),
Row(0, 9), Row(1, 1),
+ Row(1, 4), Row(1, 7), Row(2, 2), Row(2, 5), Row(2, 8), Row(3, null),
Row(4, null)))
+@@ -230,6 +237,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(joinWithNonEquiDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_ : SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(joinWithNonEquiDF, Seq(Row(0, 0), Row(0, 6), Row(0, 9),
Row(1, 1),
+ Row(1, 7), Row(2, 2), Row(2, 8), Row(3, null), Row(4, null),
Row(null, 3), Row(null, 4),
+@@ -241,6 +250,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(twoJoinsDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_ : SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 2)
+ checkAnswer(twoJoinsDF,
+ Seq(Row(0, 0, 0), Row(1, 1, null), Row(2, 2, 2), Row(3, 3, null),
Row(4, 4, null),
+@@ -262,6 +273,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(rightJoinUniqueDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(rightJoinUniqueDf, Seq(Row(1, 1), Row(2, 2), Row(3, 3),
Row(4, 4),
+ Row(null, 5), Row(null, 6), Row(null, 7), Row(null, 8), Row(null,
9),
+@@ -273,6 +286,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(leftJoinUniqueDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(leftJoinUniqueDf, Seq(Row(0, null), Row(1, 1), Row(2, 2),
Row(3, 3), Row(4, 4)))
+ assert(leftJoinUniqueDf.count() === 5)
+@@ -282,6 +297,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(rightJoinNonUniqueDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(rightJoinNonUniqueDf, Seq(Row(0, 3), Row(0, 6), Row(0,
9), Row(1, 1),
+ Row(1, 4), Row(1, 7), Row(1, 10), Row(2, 2), Row(2, 5), Row(2, 8)))
+@@ -291,6 +308,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(leftJoinNonUniqueDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(leftJoinNonUniqueDf, Seq(Row(0, 3), Row(0, 6), Row(0, 9),
Row(1, 1),
+ Row(1, 4), Row(1, 7), Row(1, 10), Row(2, 2), Row(2, 5), Row(2, 8),
Row(3, null),
+@@ -302,6 +321,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(rightJoinWithNonEquiDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(rightJoinWithNonEquiDf, Seq(Row(0, 6), Row(0, 9), Row(1,
1), Row(1, 7),
+ Row(1, 10), Row(2, 2), Row(2, 8), Row(null, 3), Row(null, 4),
Row(null, 5)))
+@@ -312,6 +333,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(leftJoinWithNonEquiDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 1)
+ checkAnswer(leftJoinWithNonEquiDf, Seq(Row(0, 6), Row(0, 9), Row(1,
1), Row(1, 7),
+ Row(1, 10), Row(2, 2), Row(2, 8), Row(3, null), Row(4, null)))
+@@ -322,6 +345,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(twoRightJoinsDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 2)
+ checkAnswer(twoRightJoinsDf, Seq(Row(2, 2, 2), Row(3, 3, 3), Row(4,
4, 4)))
+
+@@ -331,6 +356,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(twoLeftJoinsDf.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_: ShuffledHashJoinExec) if hint ==
"SHUFFLE_HASH" => true
+ case WholeStageCodegenExec(_: SortMergeJoinExec) if hint ==
"SHUFFLE_MERGE" => true
++ case _: CometHashJoinExec if hint == "SHUFFLE_HASH" => true
++ case _: CometSortMergeJoinExec if hint == "SHUFFLE_MERGE" => true
+ }.size === 2)
+ checkAnswer(twoLeftJoinsDf,
+ Seq(Row(0, null, null), Row(1, 1, null), Row(2, 2, 2), Row(3, 3,
3), Row(4, 4, 4)))
+@@ -347,6 +374,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val oneLeftOuterJoinDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" ===
$"k2", "left_outer")
+ assert(oneLeftOuterJoinDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : SortMergeJoinExec) => true
++ case _: CometSortMergeJoinExec => true
+ }.size === 1)
+ checkAnswer(oneLeftOuterJoinDF, Seq(Row(0, 0), Row(1, 1), Row(2, 2),
Row(3, 3), Row(4, null),
+ Row(5, null), Row(6, null), Row(7, null), Row(8, null), Row(9, null)))
+@@ -355,6 +383,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val oneRightOuterJoinDF = df2.join(df3.hint("SHUFFLE_MERGE"), $"k2" ===
$"k3", "right_outer")
+ assert(oneRightOuterJoinDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : SortMergeJoinExec) => true
++ case _: CometSortMergeJoinExec => true
+ }.size === 1)
+ checkAnswer(oneRightOuterJoinDF, Seq(Row(0, 0), Row(1, 1), Row(2, 2),
Row(3, 3), Row(null, 4),
+ Row(null, 5)))
+@@ -364,6 +393,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ .join(df1.hint("SHUFFLE_MERGE"), $"k3" === $"k1", "right_outer")
+ assert(twoJoinsDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(_ : SortMergeJoinExec) => true
++ case _: CometSortMergeJoinExec => true
+ }.size === 2)
+ checkAnswer(twoJoinsDF,
+ Seq(Row(0, 0, 0), Row(1, 1, 1), Row(2, 2, 2), Row(3, 3, 3), Row(4,
null, 4), Row(5, null, 5),
+@@ -379,6 +409,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val oneJoinDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2",
"left_semi")
+ assert(oneJoinDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(ProjectExec(_, _ : SortMergeJoinExec)) =>
true
++ case _: CometSortMergeJoinExec => true
+ }.size === 1)
+ checkAnswer(oneJoinDF, Seq(Row(0), Row(1), Row(2), Row(3)))
+
+@@ -386,8 +417,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val twoJoinsDF = df3.join(df2.hint("SHUFFLE_MERGE"), $"k3" === $"k2",
"left_semi")
+ .join(df1.hint("SHUFFLE_MERGE"), $"k3" === $"k1", "left_semi")
+ assert(twoJoinsDF.queryExecution.executedPlan.collect {
+- case WholeStageCodegenExec(ProjectExec(_, _ : SortMergeJoinExec)) |
+- WholeStageCodegenExec(_ : SortMergeJoinExec) => true
++ case _: SortMergeJoinExec => true
++ case _: CometSortMergeJoinExec => true
+ }.size === 2)
+ checkAnswer(twoJoinsDF, Seq(Row(0), Row(1), Row(2), Row(3)))
+ }
+@@ -401,6 +432,7 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val oneJoinDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2",
"left_anti")
+ assert(oneJoinDF.queryExecution.executedPlan.collect {
+ case WholeStageCodegenExec(ProjectExec(_, _ : SortMergeJoinExec)) =>
true
++ case _: CometSortMergeJoinExec => true
+ }.size === 1)
+ checkAnswer(oneJoinDF, Seq(Row(4), Row(5), Row(6), Row(7), Row(8),
Row(9)))
+
+@@ -408,8 +440,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val twoJoinsDF = df1.join(df2.hint("SHUFFLE_MERGE"), $"k1" === $"k2",
"left_anti")
+ .join(df3.hint("SHUFFLE_MERGE"), $"k1" === $"k3", "left_anti")
+ assert(twoJoinsDF.queryExecution.executedPlan.collect {
+- case WholeStageCodegenExec(ProjectExec(_, _ : SortMergeJoinExec)) |
+- WholeStageCodegenExec(_ : SortMergeJoinExec) => true
++ case _: SortMergeJoinExec => true
++ case _: CometSortMergeJoinExec => true
+ }.size === 2)
+ checkAnswer(twoJoinsDF, Seq(Row(6), Row(7), Row(8), Row(9)))
+ }
+@@ -542,7 +574,10 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ val plan = df.queryExecution.executedPlan
+ assert(plan.exists(p =>
+ p.isInstanceOf[WholeStageCodegenExec] &&
+- p.asInstanceOf[WholeStageCodegenExec].child.isInstanceOf[SortExec]))
++ p.asInstanceOf[WholeStageCodegenExec].collect {
++ case _: SortExec => true
++ case _: CometSortExec => true
++ }.nonEmpty))
+ assert(df.collect() === Array(Row(1), Row(2), Row(3)))
+ }
+
+@@ -722,7 +757,9 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ .write.mode(SaveMode.Overwrite).parquet(path)
+
+ withSQLConf(SQLConf.WHOLESTAGE_MAX_NUM_FIELDS.key -> "255",
+- SQLConf.WHOLESTAGE_SPLIT_CONSUME_FUNC_BY_OPERATOR.key -> "true") {
++ SQLConf.WHOLESTAGE_SPLIT_CONSUME_FUNC_BY_OPERATOR.key -> "true",
++ // Disable Comet native execution because this checks wholestage
codegen.
++ "spark.comet.exec.enabled" -> "false") {
+ val projection = Seq.tabulate(columnNum)(i => s"c$i + c$i as
newC$i")
+ val df = spark.read.parquet(path).selectExpr(projection: _*)
+
+@@ -842,6 +879,8 @@ class WholeStageCodegenSuite extends SharedSparkSession
+ assert(distinctWithId.queryExecution.executedPlan.exists {
+ case WholeStageCodegenExec(ProjectExec(_,
+ BroadcastHashJoinExec(_, _, _, _, _, _: HashAggregateExec, _, _,
_))) => true
++ case WholeStageCodegenExec(ProjectExec(_,
++ BroadcastHashJoinExec(_, _, _, _, _, _: CometColumnarToRowExec, _,
_, _))) => true
+ case _ => false
+ })
+ checkAnswer(distinctWithId, Seq(Row(1, 0), Row(1, 0)))
+diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
+index d6d19d21e65..fb103e5dad1 100644
+---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
++++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
+@@ -27,13 +27,15 @@ import org.apache.spark.SparkException
+ import org.apache.spark.rdd.RDD
+ import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent,
SparkListenerJobStart}
+ import org.apache.spark.shuffle.sort.SortShuffleManager
+-import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
++import org.apache.spark.sql.{DataFrame, Dataset, IgnoreComet, Row,
SparkSession}
+ import org.apache.spark.sql.catalyst.InternalRow
+ import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, EqualTo, IsNull, Or}
+ import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
+ import org.apache.spark.sql.catalyst.plans.{Inner, LeftAnti}
+ import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Join,
JoinHint, LocalRelation, LogicalPlan}
+ import org.apache.spark.sql.classic.Strategy
++import org.apache.spark.sql.comet._
++import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
+ import org.apache.spark.sql.execution._
+ import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+ import org.apache.spark.sql.execution.columnar.{InMemoryTableScanExec,
InMemoryTableScanLike}
+@@ -124,6 +126,7 @@ class AdaptiveQueryExecSuite
+ private def findTopLevelBroadcastHashJoin(plan: SparkPlan):
Seq[BroadcastHashJoinExec] = {
+ collect(plan) {
+ case j: BroadcastHashJoinExec => j
++ case j: CometBroadcastHashJoinExec =>
j.originalPlan.asInstanceOf[BroadcastHashJoinExec]
+ }
+ }
+
+@@ -136,36 +139,46 @@ class AdaptiveQueryExecSuite
+ private def findTopLevelSortMergeJoin(plan: SparkPlan):
Seq[SortMergeJoinExec] = {
+ collect(plan) {
+ case j: SortMergeJoinExec => j
++ case j: CometSortMergeJoinExec =>
++ assert(j.originalPlan.isInstanceOf[SortMergeJoinExec])
++ j.originalPlan.asInstanceOf[SortMergeJoinExec]
+ }
+ }
+
+ private def findTopLevelShuffledHashJoin(plan: SparkPlan):
Seq[ShuffledHashJoinExec] = {
+ collect(plan) {
+ case j: ShuffledHashJoinExec => j
++ case j: CometHashJoinExec =>
j.originalPlan.asInstanceOf[ShuffledHashJoinExec]
+ }
+ }
+
+ private def findTopLevelBaseJoin(plan: SparkPlan): Seq[BaseJoinExec] = {
+ collect(plan) {
+ case j: BaseJoinExec => j
++ case c: CometHashJoinExec => c.originalPlan.asInstanceOf[BaseJoinExec]
++ case c: CometSortMergeJoinExec =>
c.originalPlan.asInstanceOf[BaseJoinExec]
++ case c: CometBroadcastHashJoinExec =>
c.originalPlan.asInstanceOf[BaseJoinExec]
+ }
+ }
+
+ private def findTopLevelSort(plan: SparkPlan): Seq[SortExec] = {
+ collect(plan) {
+ case s: SortExec => s
++ case s: CometSortExec => s.originalPlan.asInstanceOf[SortExec]
+ }
+ }
+
+ private def findTopLevelAggregate(plan: SparkPlan): Seq[BaseAggregateExec]
= {
+ collect(plan) {
+ case agg: BaseAggregateExec => agg
++ case agg: CometHashAggregateExec =>
agg.originalPlan.asInstanceOf[BaseAggregateExec]
+ }
+ }
+
+ private def findTopLevelLimit(plan: SparkPlan): Seq[CollectLimitExec] = {
+ collect(plan) {
+ case l: CollectLimitExec => l
++ case l: CometCollectLimitExec =>
l.originalPlan.asInstanceOf[CollectLimitExec]
+ }
+ }
+
+@@ -209,6 +222,7 @@ class AdaptiveQueryExecSuite
+ val parts = rdd.partitions
+ assert(parts.forall(rdd.preferredLocations(_).nonEmpty))
+ }
++
+ assert(numShuffles === (numLocalReads.length +
numShufflesWithoutLocalRead))
+ }
+
+@@ -217,7 +231,7 @@ class AdaptiveQueryExecSuite
+ val plan = df.queryExecution.executedPlan
+ assert(plan.isInstanceOf[AdaptiveSparkPlanExec])
+ val shuffle =
plan.asInstanceOf[AdaptiveSparkPlanExec].executedPlan.collect {
+- case s: ShuffleExchangeExec => s
++ case s: ShuffleExchangeLike => s
+ }
+ assert(shuffle.size == 1)
+ assert(shuffle(0).outputPartitioning.numPartitions == numPartition)
+@@ -233,7 +247,8 @@ class AdaptiveQueryExecSuite
+ assert(smj.size == 1)
+ val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
+ assert(bhj.size == 1)
+- checkNumLocalShuffleReads(adaptivePlan)
++ // Comet shuffle changes shuffle metrics
++ // checkNumLocalShuffleReads(adaptivePlan)
+ }
+ }
+
+@@ -260,7 +275,8 @@ class AdaptiveQueryExecSuite
+ }
+ }
+
+- test("Reuse the parallelism of coalesced shuffle in local shuffle read") {
++ test("Reuse the parallelism of coalesced shuffle in local shuffle read",
++ IgnoreComet("Comet shuffle changes shuffle partition size")) {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
+@@ -292,7 +308,8 @@ class AdaptiveQueryExecSuite
+ }
+ }
+
+- test("Reuse the default parallelism in local shuffle read") {
++ test("Reuse the default parallelism in local shuffle read",
++ IgnoreComet("Comet shuffle changes shuffle partition size")) {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
+@@ -306,7 +323,8 @@ class AdaptiveQueryExecSuite
+ val localReads = collect(adaptivePlan) {
+ case read: AQEShuffleReadExec if read.isLocalRead => read
+ }
+- assert(localReads.length == 2)
++ // Comet shuffle changes shuffle metrics
++ assert(localReads.length == 1)
Review Comment:
[P2] Restore the non-Comet local-read count
`IgnoreComet` skips this test only when Comet is enabled. With
`ENABLE_COMET=false`, the body executes, asserts that `localReads.length == 1`,
and then immediately reads `localReads(1)`, so it cannot succeed for any
collection length. Upstream Spark v4.2.0 expects two reads, and the preceding
similarly tagged test correctly retains that count. Please keep the tag and
restore `assert(localReads.length == 2)`. I verified the patch and
`IgnoreComet` routing statically; the full baseline Spark suite was not run,
but this count/index contradiction is deterministic.
--
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]
