address comment

Author: cfmcgrady

spark/src/test/scala/org/apache/comet/CometCastSuite.scala

@@ -21,6 +21,7 @@ package org.apache.comet
 
 import java.io.File
 
+import scala.collection.mutable.ListBuffer
 import scala.util.Random
 import scala.util.matching.Regex
 
@@ -34,6 +35,7 @@ import org.apache.spark.sql.types.{ArrayType, BinaryType, BooleanType, ByteType,
 
 import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus
 import org.apache.comet.expressions.{CometCast, CometEvalMode}
+import org.apache.comet.rules.CometScanTypeChecker
 import org.apache.comet.serde.Compatible
 
 class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
@@ -1047,11 +1049,13 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast ArrayType to StringType") {
-    val scanImpl = sys.env
-      .getOrElse(
-        "COMET_PARQUET_SCAN_IMPL",
-        conf.getConfString(CometConf.COMET_NATIVE_SCAN_IMPL.key, "native_comet"))
-    val isNativeCometScan = scanImpl == "native_comet"
+    val hasIncompatibleType = (dt: DataType) =>
+      if (CometConf.COMET_NATIVE_SCAN_IMPL.get() == "auto") {
+        true
+      } else {
+        !CometScanTypeChecker(CometConf.COMET_NATIVE_SCAN_IMPL.get())
+          .isTypeSupported(dt, "a", ListBuffer.empty)
+      }
     Seq(
       BooleanType,
       StringType,
@@ -1065,7 +1069,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
       DecimalType(38, 18),
       BinaryType).foreach { dt =>
       val input = generateArrays(100, dt)
-      castTest(input, StringType, hasIncompatibleType = isNativeCometScan)
+      castTest(input, StringType, hasIncompatibleType = hasIncompatibleType(input.schema))
     }
   }
 

spark/src/test/spark-3.5/org/apache/spark/sql/CometToPrettyStringSuite.scala

@@ -34,8 +34,6 @@ import org.apache.spark.sql.types.DataTypes
 class CometToPrettyStringSuite extends CometFuzzTestBase {
 
   test("ToPrettyString") {
-    val cometScanTypeChecker = CometScanTypeChecker(conf.getConfString(CometConf.COMET_NATIVE_SCAN_IMPL.key))
-    val scanImpl = conf.getConfString(CometConf.COMET_NATIVE_SCAN_IMPL.key)
     val df = spark.read.parquet(filename)
     df.createOrReplaceTempView("t1")
     val table = spark.sessionState.catalog.lookupRelation(TableIdentifier("t1"))
@@ -47,7 +45,7 @@ class CometToPrettyStringSuite extends CometFuzzTestBase {
       val analyzed = spark.sessionState.analyzer.execute(plan)
       val result: DataFrame = Dataset.ofRows(spark, analyzed)
       CometCast.isSupported(field.dataType, DataTypes.StringType, Some(spark.sessionState.conf.sessionLocalTimeZone), CometEvalMode.TRY) match {
-        case _: Compatible if cometScanTypeChecker.isTypeSupported(field.dataType, scanImpl, ListBuffer.empty) =>
+        case _: Compatible if CometScanTypeChecker(CometConf.COMET_NATIVE_SCAN_IMPL.get()).isTypeSupported(field.dataType, field.name, ListBuffer.empty) =>
           checkSparkAnswerAndOperator(result)
         case _ => checkSparkAnswer(result)
       }

spark/src/test/spark-4.0/org/apache/spark/sql/CometToPrettyStringSuite.scala

@@ -37,8 +37,6 @@ import org.apache.spark.sql.types.DataTypes
 class CometToPrettyStringSuite extends CometFuzzTestBase {
 
   test("ToPrettyString") {
-    val cometScanTypeChecker = CometScanTypeChecker(conf.getConfString(CometConf.COMET_NATIVE_SCAN_IMPL.key))
-    val scanImpl = conf.getConfString(CometConf.COMET_NATIVE_SCAN_IMPL.key)
     val style = List(
       BinaryOutputStyle.UTF8,
       BinaryOutputStyle.BASIC,
@@ -59,7 +57,7 @@ class CometToPrettyStringSuite extends CometFuzzTestBase {
           val analyzed = spark.sessionState.analyzer.execute(plan)
           val result: DataFrame = Dataset.ofRows(spark, analyzed)
           CometCast.isSupported(field.dataType, DataTypes.StringType, Some(spark.sessionState.conf.sessionLocalTimeZone), CometEvalMode.TRY) match {
-            case _: Compatible if cometScanTypeChecker.isTypeSupported(field.dataType, scanImpl, ListBuffer.empty) =>
+            case _: Compatible if CometScanTypeChecker(CometConf.COMET_NATIVE_SCAN_IMPL.get()).isTypeSupported(field.dataType, field.name, ListBuffer.empty) =>
               checkSparkAnswerAndOperator(result)
             case _ => checkSparkAnswer(result)
           }