andygrove commented on code in PR #4950:
URL: https://github.com/apache/datafusion-comet/pull/4950#discussion_r3890016711
##########
dev/diffs/4.2.0.diff:
##########
@@ -0,0 +1,4324 @@
+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"
Review Comment:
Fixed in e0c796364, and you are right about `ignoreList` -
`createScalaTestCase` calls `ignore(testCase.name)` unconditionally at line
205, before anything Comet-specific, so both sketch files were dropped from the
`ENABLE_COMET=false` baseline too.
I went digging before changing it, and the recorded rationale turned out to
be wrong. The sketch aggregates are not the problem: they run on Spark's
`ObjectHashAggregate` and are never converted. What fails is the native scan
reading the `t_string_collation` fixture, which contains invalid UTF-8
(`CAST(X'C1' AS STRING)`, `CAST(X'80' AS STRING)`):
```
Parquet error: Arrow: Parquet argument error: Parquet error: encountered non
UTF-8 data
```
That is #4121, and it is not about collation at all - the uncollated query
at `tuplesketch.sql:825` fails identically, and
`theta_sketch_estimate(theta_sketch_agg(key1))` fails the same way. So "make
those expressions fall back" is unfortunately not reachable here: nothing can
tell at plan time that a Parquet file holds non-UTF-8 bytes.
Rather than gating on `isCometEnabled`, I used the mechanism this repo
already applies to `hll.sql`, which fails on the same fixture for the same
reason: a `--SET spark.comet.enabled = false` header in the input file, linked
to #4121. That is strictly better than the gate you suggested as the fallback -
all 273 `SELECT`s in `tuplesketch.sql` and 91 in `thetasketch.sql` now run and
are verified against the golden files in *both* the baseline and the Comet job,
instead of the file being skipped in either. No golden file changes, since `--`
lines are stripped before the query text is compared.
Verified all four representative queries (uncollated, `UTF8_LCASE`,
`UNICODE_CI_RTRIM`, and the theta-sketch one) fail with
`spark.comet.enabled=true` and return Spark's answers with it false, and
re-checked `git apply --check` against a fresh `v4.2.0` checkout. The Spark 4.2
SQL job on this PR is what actually exercises the two files end to end.
Two related things I left alone to keep this scoped, happy to file
follow-ups: `thetasketch.sql` is listed the same way in `dev/diffs/4.1.3.diff`
on main, and the two remaining `ignoreList` entries
(`replacing-missing-expression-with-alias.sql`, `in-set-operations.sql`, both
pre-existing) suppress the baseline in exactly the same way.
--
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]
