chore: collapse IgnoreCometNativeDataFusion/Scan into IgnoreComet

With native_datafusion as the only Parquet scan implementation, the
per-impl tag variants are equivalent to the plain IgnoreComet tag.

- Replace all IgnoreCometNativeDataFusion and IgnoreCometNativeScan
  call sites with IgnoreComet.
- Drop the corresponding case-class definitions from IgnoreComet.scala.
- Remove the now-redundant secondary check in SQLTestUtils; the
  existing IgnoreComet check handles all usages.

Author: andygrove

dev/diffs/3.4.3.diff

@@ -608,10 +608,10 @@ index 2796b1cf154..53dcfde932e 100644
          }
 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..c528360742a
+index 00000000000..4b31bea33de
 --- /dev/null
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/IgnoreComet.scala
-@@ -0,0 +1,44 @@
+@@ -0,0 +1,42 @@
 +/*
 + * Licensed to the Apache Software Foundation (ASF) under one or more
 + * contributor license agreements.  See the NOTICE file distributed with
@@ -640,8 +640,6 @@ index 00000000000..c528360742a
 + * Tests with this tag will be ignored when Comet is enabled (e.g., via `ENABLE_COMET`).
 + */
 +case class IgnoreComet(reason: String) extends Tag("DisableComet")
-+case class IgnoreCometNativeDataFusion(reason: String) extends Tag("DisableComet")
-+case class IgnoreCometNativeScan(reason: String) extends Tag("DisableComet")
 +
 +/**
 + * Helper trait that disables Comet for all tests regardless of default config values.
@@ -1968,14 +1966,14 @@ index 07e2849ce6f..3e73645b638 100644
        ParquetOutputFormat.WRITER_VERSION -> ParquetProperties.WriterVersion.PARQUET_2_0.toString
      )
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
-index 104b4e416cd..4adb273170a 100644
+index 104b4e416cd..835aaa18e39 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
 @@ -38,6 +38,7 @@ import org.apache.parquet.schema.MessageType
 
  import org.apache.spark.{SparkConf, SparkException}
  import org.apache.spark.sql._
-+import org.apache.spark.sql.{IgnoreCometNativeDataFusion, IgnoreCometNativeScan}
++import org.apache.spark.sql.IgnoreComet
  import org.apache.spark.sql.catalyst.dsl.expressions._
  import org.apache.spark.sql.catalyst.expressions._
  import org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints
@@ -1998,7 +1996,7 @@ index 104b4e416cd..4adb273170a 100644
 
 -  test("Filters should be pushed down for vectorized Parquet reader at row group level") {
 +  test("Filters should be pushed down for vectorized Parquet reader at row group level",
-+    IgnoreCometNativeScan("Native scans do not support the tested accumulator")) {
++    IgnoreComet("Native scans do not support the tested accumulator")) {
      import testImplicits._
 
      withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
@@ -2034,7 +2032,7 @@ index 104b4e416cd..4adb273170a 100644
 
 -  test("filter pushdown - StringPredicate") {
 +  test("filter pushdown - StringPredicate",
-+    IgnoreCometNativeDataFusion("cannot be pushed down")) {
++    IgnoreComet("cannot be pushed down")) {
      import testImplicits._
      // keep() should take effect on StartsWith/EndsWith/Contains
      Seq(
@@ -2044,7 +2042,7 @@ index 104b4e416cd..4adb273170a 100644
 
 -  test("SPARK-17091: Convert IN predicate to Parquet filter push-down") {
 +  test("SPARK-17091: Convert IN predicate to Parquet filter push-down",
-+    IgnoreCometNativeScan("Comet has different push-down behavior")) {
++    IgnoreComet("Comet has different push-down behavior")) {
      val schema = StructType(Seq(
        StructField("a", IntegerType, nullable = false)
      ))
@@ -2089,7 +2087,7 @@ index 104b4e416cd..4adb273170a 100644
 
 -  test("SPARK-34562: Bloom filter push down") {
 +  test("SPARK-34562: Bloom filter push down",
-+    IgnoreCometNativeScan("Native scans do not support the tested accumulator")) {
++    IgnoreComet("Native scans do not support the tested accumulator")) {
      withTempPath { dir =>
        val path = dir.getCanonicalPath
        spark.range(100).selectExpr("id * 2 AS id")
@@ -2134,14 +2132,14 @@ index 8670d95c65e..b624c3811dd 100644
        checkAnswer(
          // "fruit" column in this file is encoded using DELTA_LENGTH_BYTE_ARRAY.
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala
-index 29cb224c878..dcb8a0e9bef 100644
+index 29cb224c878..62e3ab96004 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetQuerySuite.scala
 @@ -27,6 +27,7 @@ import org.apache.parquet.hadoop.ParquetOutputFormat
 
  import org.apache.spark.{DebugFilesystem, SparkConf, SparkException}
  import org.apache.spark.sql._
-+import org.apache.spark.sql.IgnoreCometNativeDataFusion
++import org.apache.spark.sql.IgnoreComet
  import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
  import org.apache.spark.sql.catalyst.expressions.SpecificInternalRow
  import org.apache.spark.sql.catalyst.util.ArrayData
@@ -2151,7 +2149,7 @@ index 29cb224c878..dcb8a0e9bef 100644
 
 -  test("SPARK-36182: can't read TimestampLTZ as TimestampNTZ") {
 +  test("SPARK-36182: can't read TimestampLTZ as TimestampNTZ",
-+    IgnoreCometNativeDataFusion("https://github.com/apache/datafusion-comet/issues/4219")) {
++    IgnoreComet("https://github.com/apache/datafusion-comet/issues/4219")) {
      val data = (1 to 1000).map { i =>
        val ts = new java.sql.Timestamp(i)
        Row(ts)
@@ -2161,7 +2159,7 @@ index 29cb224c878..dcb8a0e9bef 100644
 
 -  test("SPARK-26677: negated null-safe equality comparison should not filter matched row groups") {
 +  test("SPARK-26677: negated null-safe equality comparison should not filter matched row groups",
-+    IgnoreCometNativeScan("Native scans had the filter pushed into DF operator, cannot strip")) {
++    IgnoreComet("Native scans had the filter pushed into DF operator, cannot strip")) {
      withAllParquetReaders {
        withTempPath { path =>
          // Repeated values for dictionary encoding.
@@ -2171,7 +2169,7 @@ index 29cb224c878..dcb8a0e9bef 100644
 
 -  test("SPARK-34212 Parquet should read decimals correctly") {
 +  test("SPARK-34212 Parquet should read decimals correctly",
-+    IgnoreCometNativeDataFusion("https://github.com/apache/datafusion-comet/issues/4354")) {
++    IgnoreComet("https://github.com/apache/datafusion-comet/issues/4354")) {
      def readParquet(schema: String, path: File): DataFrame = {
        spark.read.schema(schema).parquet(path.toString)
      }
@@ -2201,7 +2199,7 @@ index 29cb224c878..dcb8a0e9bef 100644
 
 -  test("row group skipping doesn't overflow when reading into larger type") {
 +  test("row group skipping doesn't overflow when reading into larger type",
-+    IgnoreCometNativeDataFusion("https://github.com/apache/datafusion-comet/issues/4354")) {
++    IgnoreComet("https://github.com/apache/datafusion-comet/issues/4354")) {
      withTempPath { path =>
        Seq(0).toDF("a").write.parquet(path.toString)
        // The vectorized and non-vectorized readers will produce different exceptions, we don't need
@@ -2290,14 +2288,14 @@ index 5c0b7def039..151184bc98c 100644
      assert(fileSourceScanSchemata.size === expectedSchemaCatalogStrings.size,
        s"Found ${fileSourceScanSchemata.size} file sources in dataframe, " +
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala
-index bf5c51b89bb..7e143a0e0f9 100644
+index bf5c51b89bb..f7402b7d883 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaSuite.scala
 @@ -27,6 +27,7 @@ import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
  import org.apache.parquet.schema.Type._
 
  import org.apache.spark.SparkException
-+import org.apache.spark.sql.{IgnoreComet, IgnoreCometNativeDataFusion}
++import org.apache.spark.sql.IgnoreComet
  import org.apache.spark.sql.catalyst.ScalaReflection
  import org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException
  import org.apache.spark.sql.functions.desc
@@ -2317,7 +2315,7 @@ index bf5c51b89bb..7e143a0e0f9 100644
 
 -  test("schema mismatch failure error message for parquet vectorized reader") {
 +  test("schema mismatch failure error message for parquet vectorized reader",
-+      IgnoreCometNativeDataFusion("https://github.com/apache/datafusion-comet/issues/4316")) {
++      IgnoreComet("https://github.com/apache/datafusion-comet/issues/4316")) {
      withTempPath { dir =>
        val e = testSchemaMismatch(dir.getCanonicalPath, vectorizedReaderEnabled = true)
        assert(e.getCause.isInstanceOf[SparkException])
@@ -2863,7 +2861,7 @@ index abe606ad9c1..2d930b64cca 100644
      val tblTargetName = "tbl_target"
      val tblSourceQualified = s"default.$tblSourceName"
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala
-index dd55fcfe42c..d4f94b1d608 100644
+index dd55fcfe42c..d9a3f2df535 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala
 @@ -41,6 +41,7 @@ import org.apache.spark.sql.catalyst.plans.PlanTest
@@ -2874,25 +2872,19 @@ index dd55fcfe42c..d4f94b1d608 100644
  import org.apache.spark.sql.execution.FilterExec
  import org.apache.spark.sql.execution.adaptive.DisableAdaptiveExecution
  import org.apache.spark.sql.execution.datasources.DataSourceUtils
-@@ -119,6 +120,17 @@ private[sql] trait SQLTestUtils extends SparkFunSuite with SQLTestUtilsBase with
+@@ -119,6 +120,11 @@ private[sql] trait SQLTestUtils extends SparkFunSuite with SQLTestUtilsBase with
 
    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 (isCometEnabled &&
-+      testTags.exists(t =>
-+        t.isInstanceOf[IgnoreCometNativeDataFusion] || t.isInstanceOf[IgnoreCometNativeScan])) {
-+      ignore(testName + " (disabled for NATIVE_DATAFUSION)", testTags: _*)(testFun)
-+      return
 +    }
      if (testTags.exists(_.isInstanceOf[DisableAdaptiveExecution])) {
        super.test(testName, testTags: _*) {
          withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
-@@ -242,6 +254,11 @@ private[sql] trait SQLTestUtilsBase
+@@ -242,6 +248,11 @@ private[sql] trait SQLTestUtilsBase
      protected override def _sqlContext: SQLContext = self.spark.sqlContext
    }
 
@@ -2904,7 +2896,7 @@ index dd55fcfe42c..d4f94b1d608 100644
    protected override def withSQLConf(pairs: (String, String)*)(f: => Unit): Unit = {
      SparkSession.setActiveSession(spark)
      super.withSQLConf(pairs: _*)(f)
-@@ -434,6 +451,8 @@ private[sql] trait SQLTestUtilsBase
+@@ -434,6 +445,8 @@ private[sql] trait SQLTestUtilsBase
      val schema = df.schema
      val withoutFilters = df.queryExecution.executedPlan.transform {
        case FilterExec(_, child) => child