diff --git a/dev/diffs/4.0.0-preview1.diff b/dev/diffs/4.0.0-preview1.diff
index 335aa01d1e..1a6810a7ab 100644
--- a/dev/diffs/4.0.0-preview1.diff
+++ b/dev/diffs/4.0.0-preview1.diff
@@ -1,5 +1,5 @@
 diff --git a/pom.xml b/pom.xml
-index a4b1b2c3c9f..db50bdb0d3b 100644
+index a4b1b2c3c9f..6a532749978 100644
 --- a/pom.xml
 +++ b/pom.xml
 @@ -147,6 +147,8 @@
@@ -38,7 +38,7 @@ index a4b1b2c3c9f..db50bdb0d3b 100644
        <dependency>
          <groupId>org.apache.datasketches</groupId>
 diff --git a/sql/core/pom.xml b/sql/core/pom.xml
-index 19f6303be36..31e1d27700f 100644
+index 19f6303be36..6c0e77882e6 100644
 --- a/sql/core/pom.xml
 +++ b/sql/core/pom.xml
 @@ -77,6 +77,10 @@
@@ -146,133 +146,6 @@ index 698ca009b4f..57d774a3617 100644
  
  -- Test tables
  CREATE table  explain_temp1 (key int, val int) USING PARQUET;
-diff --git a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part1.sql b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part1.sql
-index 1152d77da0c..f77493f690b 100644
---- a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part1.sql
-+++ b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part1.sql
-@@ -7,6 +7,9 @@
- 
- -- avoid bit-exact output here because operations may not be bit-exact.
- -- SET extra_float_digits = 0;
-+-- Disable Comet exec due to floating point precision difference
-+--SET spark.comet.exec.enabled = false
-+
- 
- -- Test aggregate operator with codegen on and off.
- --CONFIG_DIM1 spark.sql.codegen.wholeStage=true
-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..44cd244d3b0 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
-@@ -5,6 +5,9 @@
- -- AGGREGATES [Part 3]
- -- https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/aggregates.sql#L352-L605
- 
-+-- Disable Comet exec due to floating point precision difference
-+--SET spark.comet.exec.enabled = false
-+
- -- Test aggregate operator with codegen on and off.
- --CONFIG_DIM1 spark.sql.codegen.wholeStage=true
- --CONFIG_DIM1 spark.sql.codegen.wholeStage=false,spark.sql.codegen.factoryMode=CODEGEN_ONLY
-diff --git a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/float8.sql b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/float8.sql
-index 932cdb95fcf..bbafaadb960 100644
---- a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/float8.sql
-+++ b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/float8.sql
-@@ -5,6 +5,9 @@
- -- FLOAT8
- -- https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/float8.sql
- 
-+-- TODO: https://github.com/apache/datafusion-comet/issues/551
-+--SET spark.comet.enabled = false
-+
- CREATE TABLE FLOAT8_TBL(f1 double) USING parquet;
- 
- -- PostgreSQL implicitly casts string literals to data with floating point types, but
-diff --git a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/groupingsets.sql b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/groupingsets.sql
-index f06b0276b00..1f39a1b3da8 100644
---- a/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/groupingsets.sql
-+++ b/sql/core/src/test/resources/sql-tests/inputs/postgreSQL/groupingsets.sql
-@@ -5,6 +5,9 @@
- 
- -- test data sources
- 
-+-- TODO: https://github.com/apache/datafusion-comet/issues/551
-+--SET spark.comet.enabled = false
-+
- create temp view gstest1(a,b,v)
-   as values (1,1,10),(1,1,11),(1,2,12),(1,2,13),(1,3,14),
-             (2,3,15),
-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..2b73732c33f 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
-@@ -1,6 +1,10 @@
- --
- -- Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
- --
-+
-+-- Disable Comet exec due to floating point precision difference
-+--SET spark.comet.exec.enabled = false
-+
- --
- -- INT8
- -- Test int8 64-bit integers.
-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..423d3b3d76d 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
-@@ -1,6 +1,10 @@
- --
- -- Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
- --
-+
-+-- Disable Comet exec due to floating point precision difference
-+--SET spark.comet.exec.enabled = false
-+
- --
- -- SELECT_HAVING
- -- https://github.com/postgres/postgres/blob/REL_12_BETA2/src/test/regress/sql/select_having.sql
-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..316788b2989 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,4 +1,8 @@
- -- This test suite checks the WITH SCHEMA COMPENSATION clause
-+
-+-- TODO: https://github.com/apache/datafusion-comet/issues/551
-+--SET spark.comet.enabled = false
-+
- -- Disable ANSI mode to ensure we are forcing it explicitly in the CASTS
- SET spark.sql.ansi.enabled = false;
- 
 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 d023fb82185..0f4f03bda6c 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
@@ -348,7 +221,7 @@ index f6fd6b501d7..11870c85d82 100644
            spark.range(100).write.saveAsTable(s"$dbName.$table2Name")
  
 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 760ee802608..ebd4a34b08d 100644
+index 760ee802608..b77133ffd37 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,
@@ -407,16 +280,6 @@ index 760ee802608..ebd4a34b08d 100644
      }
      assert(exchanges.size == 2)
    }
-@@ -2299,7 +2302,8 @@ class DataFrameSuite extends QueryTest
-     assert(df2.isLocal)
-   }
- 
--  test("SPARK-35886: PromotePrecision should be subexpr replaced") {
-+  test("SPARK-35886: PromotePrecision should be subexpr replaced",
-+    IgnoreComet("TODO: fix Comet for this test")) {
-     withTable("tbl") {
-       sql(
-         """
 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 16a493b5290..3f0b70e2d59 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala
@@ -1054,7 +917,7 @@ index 34c6c49bc49..f5dea07a213 100644
    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/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
-index 56c364e2084..fc3abd7cdc4 100644
+index 56c364e2084..a00a50e020a 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
 @@ -1510,7 +1510,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
@@ -1067,36 +930,6 @@ index 56c364e2084..fc3abd7cdc4 100644
      AccumulatorSuite.verifyPeakExecutionMemorySet(sparkContext, "external sort") {
        sql("SELECT * FROM testData2 ORDER BY a ASC, b ASC").collect()
      }
-@@ -4454,7 +4455,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
-   }
- 
-   test("SPARK-39166: Query context of binary arithmetic should be serialized to executors" +
--    " when WSCG is off") {
-+    " when WSCG is off",
-+    IgnoreComet("TODO: https://github.com/apache/datafusion-comet/issues/551")) {
-     withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
-       SQLConf.ANSI_ENABLED.key -> "true") {
-       withTable("t") {
-@@ -4475,7 +4477,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
-   }
- 
-   test("SPARK-39175: Query context of Cast should be serialized to executors" +
--    " when WSCG is off") {
-+    " when WSCG is off",
-+    IgnoreComet("TODO: https://github.com/apache/datafusion-comet/issues/551")) {
-     withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
-       SQLConf.ANSI_ENABLED.key -> "true") {
-       withTable("t") {
-@@ -4502,7 +4505,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
-   }
- 
-   test("SPARK-39190,SPARK-39208,SPARK-39210: Query context of decimal overflow error should " +
--    "be serialized to executors when WSCG is off") {
-+    "be serialized to executors when WSCG is off",
-+    IgnoreComet("TODO: https://github.com/apache/datafusion-comet/issues/551")) {
-     withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false",
-       SQLConf.ANSI_ENABLED.key -> "true") {
-       withTable("t") {
 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 68f14f13bbd..174636cefb5 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala
@@ -2145,29 +1978,6 @@ index a7efd0aa75e..baae0967a2a 100644
          }.isEmpty)
          assert(collect(initialExecutedPlan) {
            case i: InMemoryTableScanLike => i
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala
-index 05872d41131..0dd83608bbd 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala
-@@ -21,7 +21,7 @@ import java.io.File
- 
- import org.apache.hadoop.fs.{FileStatus, Path}
- 
--import org.apache.spark.sql.{DataFrame, Dataset, QueryTest, Row}
-+import org.apache.spark.sql.{DataFrame, Dataset, IgnoreComet, QueryTest, Row}
- import org.apache.spark.sql.catalyst.InternalRow
- import org.apache.spark.sql.catalyst.expressions.{Expression, FileSourceConstantMetadataStructField, FileSourceGeneratedMetadataStructField, Literal}
- import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
-@@ -134,7 +134,8 @@ class FileSourceCustomMetadataStructSuite extends QueryTest with SharedSparkSess
-     }
-   }
- 
--  test("[SPARK-43226] extra constant metadata fields with extractors") {
-+  test("[SPARK-43226] extra constant metadata fields with extractors",
-+    IgnoreComet("TODO: https://github.com/apache/datafusion-comet/issues/551")) {
-     withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
-       val format = new TestFileFormat(extraConstantMetadataFields) {
-         val extractPartitionNumber = { pf: PartitionedFile =>
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SchemaPruningSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SchemaPruningSuite.scala
 index 0a0b23d1e60..5685926250f 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SchemaPruningSuite.scala
@@ -2541,35 +2351,6 @@ index 4bd35e0789b..6544d86dbe0 100644
      )
    }
    test(s"parquet widening conversion $fromType -> $toType") {
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
-index c800168b507..991d52a1a75 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/state/StateDataSourceReadSuite.scala
-@@ -22,7 +22,7 @@ import org.scalatest.Assertions
- 
- import org.apache.spark.SparkUnsupportedOperationException
- import org.apache.spark.io.CompressionCodec
--import org.apache.spark.sql.{AnalysisException, DataFrame, Encoders, Row}
-+import org.apache.spark.sql.{AnalysisException, DataFrame, Encoders, IgnoreComet, Row}
- import org.apache.spark.sql.catalyst.expressions.{BoundReference, GenericInternalRow}
- import org.apache.spark.sql.catalyst.plans.physical.HashPartitioning
- import org.apache.spark.sql.execution.datasources.v2.state.utils.SchemaUtil
-@@ -525,11 +525,13 @@ abstract class StateDataSourceReadSuite extends StateDataSourceTestBase with Ass
-     }
-   }
- 
--  test("flatMapGroupsWithState, state ver 1") {
-+  test("flatMapGroupsWithState, state ver 1",
-+    IgnoreComet("Ignored if Comet is enabled due to SPARK-49070.")) {
-     testFlatMapGroupsWithState(1)
-   }
- 
--  test("flatMapGroupsWithState, state ver 2") {
-+  test("flatMapGroupsWithState, state ver 2",
-+    IgnoreComet("Ignored if Comet is enabled due to SPARK-49070.")) {
-     testFlatMapGroupsWithState(2)
-   }
- 
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala
 index b8f3ea3c6f3..bbd44221288 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala
@@ -2900,94 +2681,6 @@ index 04193d5189a..d83d03f8e0d 100644
        }.headOption.getOrElse {
          fail(s"No FileScan in query\n${df.queryExecution}")
        }
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateDistributionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateDistributionSuite.scala
-index b597a244710..b2e8be41065 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateDistributionSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateDistributionSuite.scala
-@@ -21,6 +21,7 @@ import java.io.File
- 
- import org.apache.commons.io.FileUtils
- 
-+import org.apache.spark.sql.IgnoreComet
- import org.apache.spark.sql.catalyst.streaming.InternalOutputModes.Update
- import org.apache.spark.sql.execution.streaming.{FlatMapGroupsWithStateExec, MemoryStream}
- import org.apache.spark.sql.internal.SQLConf
-@@ -91,7 +92,7 @@ class FlatMapGroupsWithStateDistributionSuite extends StreamTest
-   }
- 
-   test("SPARK-38204: flatMapGroupsWithState should require StatefulOpClusteredDistribution " +
--    "from children - without initial state") {
-+    "from children - without initial state", IgnoreComet("TODO: fix Comet for this test")) {
-     // function will return -1 on timeout and returns count of the state otherwise
-     val stateFunc =
-       (key: (String, String), values: Iterator[(String, String, Long)],
-@@ -243,7 +244,8 @@ class FlatMapGroupsWithStateDistributionSuite extends StreamTest
-   }
- 
-   test("SPARK-38204: flatMapGroupsWithState should require ClusteredDistribution " +
--    "from children if the query starts from checkpoint in 3.2.x - without initial state") {
-+    "from children if the query starts from checkpoint in 3.2.x - without initial state",
-+    IgnoreComet("TODO: fix Comet for this test")) {
-     // function will return -1 on timeout and returns count of the state otherwise
-     val stateFunc =
-       (key: (String, String), values: Iterator[(String, String, Long)],
-@@ -335,7 +337,8 @@ class FlatMapGroupsWithStateDistributionSuite extends StreamTest
-   }
- 
-   test("SPARK-38204: flatMapGroupsWithState should require ClusteredDistribution " +
--    "from children if the query starts from checkpoint in prior to 3.2") {
-+    "from children if the query starts from checkpoint in prior to 3.2",
-+    IgnoreComet("TODO: fix Comet for this test")) {
-     // function will return -1 on timeout and returns count of the state otherwise
-     val stateFunc =
-       (key: (String, String), values: Iterator[(String, String, Long)],
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateSuite.scala
-index b35e996106f..326f72235e0 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateSuite.scala
-@@ -25,7 +25,7 @@ import org.scalatest.exceptions.TestFailedException
- 
- import org.apache.spark.SparkException
- import org.apache.spark.api.java.function.FlatMapGroupsWithStateFunction
--import org.apache.spark.sql.{DataFrame, Encoder}
-+import org.apache.spark.sql.{DataFrame, Encoder, IgnoreCometSuite}
- import org.apache.spark.sql.catalyst.InternalRow
- import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, UnsafeProjection, UnsafeRow}
- import org.apache.spark.sql.catalyst.plans.logical.FlatMapGroupsWithState
-@@ -46,8 +46,9 @@ case class RunningCount(count: Long)
- 
- case class Result(key: Long, count: Int)
- 
-+// TODO: fix Comet to enable this suite
- @SlowSQLTest
--class FlatMapGroupsWithStateSuite extends StateStoreMetricsTest {
-+class FlatMapGroupsWithStateSuite extends StateStoreMetricsTest with IgnoreCometSuite {
- 
-   import testImplicits._
- 
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateWithInitialStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateWithInitialStateSuite.scala
-index 2a2a83d35e1..e3b7b290b3e 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateWithInitialStateSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FlatMapGroupsWithStateWithInitialStateSuite.scala
-@@ -18,7 +18,7 @@
- package org.apache.spark.sql.streaming
- 
- import org.apache.spark.SparkException
--import org.apache.spark.sql.{AnalysisException, Dataset, KeyValueGroupedDataset}
-+import org.apache.spark.sql.{AnalysisException, Dataset, IgnoreComet, KeyValueGroupedDataset}
- import org.apache.spark.sql.catalyst.streaming.InternalOutputModes.Update
- import org.apache.spark.sql.execution.streaming.MemoryStream
- import org.apache.spark.sql.execution.streaming.state.FlatMapGroupsWithStateExecHelper
-@@ -253,7 +253,8 @@ class FlatMapGroupsWithStateWithInitialStateSuite extends StateStoreMetricsTest
-     assert(e.message.contains(expectedError))
-   }
- 
--  test("flatMapGroupsWithState - initial state - initial state has flatMapGroupsWithState") {
-+  test("flatMapGroupsWithState - initial state - initial state has flatMapGroupsWithState",
-+    IgnoreComet("TODO: fix Comet for this test")) {
-     val initialStateDS = Seq(("keyInStateAndData", new RunningCount(1))).toDS()
-     val initialState: KeyValueGroupedDataset[String, RunningCount] =
-       initialStateDS.groupByKey(_._1).mapValues(_._2)
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
 index 1fce992126b..6d3ea74e0fc 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala
@@ -3102,153 +2795,6 @@ index e05cb4d3c35..dc65a4fe18e 100644
        })
    }
  
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateSuite.scala
-index dea16e52989..55cdf47c4d5 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateSuite.scala
-@@ -18,7 +18,7 @@
- package org.apache.spark.sql.streaming
- 
- import org.apache.spark.SparkIllegalArgumentException
--import org.apache.spark.sql.Encoders
-+import org.apache.spark.sql.{Encoders, IgnoreCometSuite}
- import org.apache.spark.sql.execution.streaming.MemoryStream
- import org.apache.spark.sql.execution.streaming.state.{AlsoTestWithChangelogCheckpointingEnabled, RocksDBStateStoreProvider}
- import org.apache.spark.sql.internal.SQLConf
-@@ -128,7 +128,7 @@ class ToggleSaveAndEmitProcessor
- }
- 
- class TransformWithListStateSuite extends StreamTest
--  with AlsoTestWithChangelogCheckpointingEnabled {
-+  with AlsoTestWithChangelogCheckpointingEnabled with IgnoreCometSuite {
-   import testImplicits._
- 
-   test("test appending null value in list state throw exception") {
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateTTLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateTTLSuite.scala
-index 299a3346b2e..2213f3c52a6 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateTTLSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithListStateTTLSuite.scala
-@@ -19,7 +19,7 @@ package org.apache.spark.sql.streaming
- 
- import java.time.Duration
- 
--import org.apache.spark.sql.Encoders
-+import org.apache.spark.sql.{Encoders, IgnoreCometSuite}
- import org.apache.spark.sql.execution.streaming.{ListStateImplWithTTL, MemoryStream}
- import org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider
- import org.apache.spark.sql.internal.SQLConf
-@@ -95,7 +95,7 @@ class ListStateTTLProcessor(ttlConfig: TTLConfig)
-  * Test suite for testing list state with TTL.
-  * We use the base TTL suite with a list state processor.
-  */
--class TransformWithListStateTTLSuite extends TransformWithStateTTLTest {
-+class TransformWithListStateTTLSuite extends TransformWithStateTTLTest with IgnoreCometSuite {
- 
-   import testImplicits._
- 
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithMapStateTTLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithMapStateTTLSuite.scala
-index bf46c802fde..623c3003430 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithMapStateTTLSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithMapStateTTLSuite.scala
-@@ -19,7 +19,7 @@ package org.apache.spark.sql.streaming
- 
- import java.time.Duration
- 
--import org.apache.spark.sql.Encoders
-+import org.apache.spark.sql.{Encoders, IgnoreCometSuite}
- import org.apache.spark.sql.execution.streaming.{MapStateImplWithTTL, MemoryStream}
- import org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider
- import org.apache.spark.sql.internal.SQLConf
-@@ -174,7 +174,7 @@ class MapStateTTLProcessor(ttlConfig: TTLConfig)
-   }
- }
- 
--class TransformWithMapStateTTLSuite extends TransformWithStateTTLTest {
-+class TransformWithMapStateTTLSuite extends TransformWithStateTTLTest with IgnoreCometSuite {
- 
-   import testImplicits._
-   override def getProcessor(ttlConfig: TTLConfig):
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala
-index 5388d6f1fb6..8aa11c5b875 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateChainingSuite.scala
-@@ -21,7 +21,7 @@ import java.sql.Timestamp
- import java.time.{Instant, LocalDateTime, ZoneId}
- 
- import org.apache.spark.{SparkRuntimeException, SparkThrowable}
--import org.apache.spark.sql.AnalysisException
-+import org.apache.spark.sql.{AnalysisException, IgnoreCometSuite}
- import org.apache.spark.sql.catalyst.ExtendedAnalysisException
- import org.apache.spark.sql.execution.streaming.{MemoryStream, StreamExecution}
- import org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider
-@@ -106,7 +106,8 @@ case class AggEventRow(
-     window: Window,
-     count: Long)
- 
--class TransformWithStateChainingSuite extends StreamTest {
-+class TransformWithStateChainingSuite extends StreamTest
-+  with IgnoreCometSuite {
-   import testImplicits._
- 
-   test("watermark is propagated correctly for next stateful operator" +
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala
-index 0057af44d3e..51975748309 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithStateSuite.scala
-@@ -22,7 +22,7 @@ import java.util.UUID
- 
- import org.apache.spark.SparkRuntimeException
- import org.apache.spark.internal.Logging
--import org.apache.spark.sql.{Dataset, Encoders}
-+import org.apache.spark.sql.{Dataset, Encoders, IgnoreCometSuite}
- import org.apache.spark.sql.catalyst.util.stringToFile
- import org.apache.spark.sql.execution.streaming._
- import org.apache.spark.sql.execution.streaming.state.{AlsoTestWithChangelogCheckpointingEnabled, RocksDBStateStoreProvider, StatefulProcessorCannotPerformOperationWithInvalidHandleState, StateStoreMultipleColumnFamiliesNotSupportedException}
-@@ -307,9 +307,11 @@ class RunningCountStatefulProcessorWithError extends RunningCountStatefulProcess
- 
- /**
-  * Class that adds tests for transformWithState stateful streaming operator
-+ *
-+ * Ignored if Comet is enabled due to SPARK-49070.
-  */
- class TransformWithStateSuite extends StateStoreMetricsTest
--  with AlsoTestWithChangelogCheckpointingEnabled {
-+  with AlsoTestWithChangelogCheckpointingEnabled with IgnoreCometSuite {
- 
-   import testImplicits._
- 
-@@ -786,7 +788,7 @@ class TransformWithStateSuite extends StateStoreMetricsTest
-   }
- }
- 
--class TransformWithStateValidationSuite extends StateStoreMetricsTest {
-+class TransformWithStateValidationSuite extends StateStoreMetricsTest with IgnoreCometSuite {
-   import testImplicits._
- 
-   test("transformWithState - streaming with hdfsStateStoreProvider should fail") {
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithValueStateTTLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithValueStateTTLSuite.scala
-index 54004b419f7..4e5b35aa0da 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithValueStateTTLSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/TransformWithValueStateTTLSuite.scala
-@@ -20,7 +20,7 @@ package org.apache.spark.sql.streaming
- import java.time.Duration
- 
- import org.apache.spark.internal.Logging
--import org.apache.spark.sql.Encoders
-+import org.apache.spark.sql.{Encoders, IgnoreCometSuite}
- import org.apache.spark.sql.execution.streaming.{MemoryStream, ValueStateImpl, ValueStateImplWithTTL}
- import org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider
- import org.apache.spark.sql.internal.SQLConf
-@@ -160,7 +160,8 @@ case class MultipleValueStatesTTLProcessor(
-   }
- }
- 
--class TransformWithValueStateTTLSuite extends TransformWithStateTTLTest {
-+class TransformWithValueStateTTLSuite extends TransformWithStateTTLTest
-+  with IgnoreCometSuite{
- 
-   import testImplicits._
- 
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
 index af07aceaed1..ed0b5e6d9be 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamTableAPISuite.scala
@@ -3423,20 +2969,6 @@ index 52abd248f3a..7a199931a08 100644
        case h: HiveTableScanExec => h.partitionPruningPred.collect {
          case d: DynamicPruningExpression => d.child
        }
-diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
-index 4b000fff0eb..e867b6c0d9b 100644
---- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
-+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
-@@ -660,7 +660,8 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te
-         Row(3, 4, 4, 3, null) :: Nil)
-   }
- 
--  test("single distinct multiple columns set") {
-+  test("single distinct multiple columns set",
-+    IgnoreComet("TODO: fix Comet for this test")) {
-     checkAnswer(
-       spark.sql(
-         """
 diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala
 index 3f8de93b330..fc06e81f41a 100644
 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/test/TestHive.scala