fix: enable cast tests for Spark 4.0

Author: manuzhang

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

@@ -33,7 +33,6 @@ import org.apache.spark.sql.functions.col
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{ArrayType, BooleanType, ByteType, DataType, DataTypes, DecimalType, IntegerType, LongType, ShortType, StringType, StructField, StructType}
 
-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
@@ -575,8 +574,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   // CAST from StringType
 
   test("cast StringType to BooleanType") {
-    // TODO fix for Spark 4.0.0
-    assume(!isSpark40Plus)
     val testValues =
       (Seq("TRUE", "True", "true", "FALSE", "False", "false", "1", "0", "", null) ++
         gen.generateStrings(dataSize, "truefalseTRUEFALSEyesno10" + whitespaceChars, 8)).toDF("a")
@@ -617,35 +614,27 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   )
 
   test("cast StringType to ByteType") {
-    // TODO fix for Spark 4.0.0
-    assume(!isSpark40Plus)
     // test with hand-picked values
     castTest(castStringToIntegralInputs.toDF("a"), DataTypes.ByteType)
     // fuzz test
     castTest(gen.generateStrings(dataSize, numericPattern, 4).toDF("a"), DataTypes.ByteType)
   }
 
   test("cast StringType to ShortType") {
-    // TODO fix for Spark 4.0.0
-    assume(!isSpark40Plus)
     // test with hand-picked values
     castTest(castStringToIntegralInputs.toDF("a"), DataTypes.ShortType)
     // fuzz test
     castTest(gen.generateStrings(dataSize, numericPattern, 5).toDF("a"), DataTypes.ShortType)
   }
 
   test("cast StringType to IntegerType") {
-    // TODO fix for Spark 4.0.0
-    assume(!isSpark40Plus)
     // test with hand-picked values
     castTest(castStringToIntegralInputs.toDF("a"), DataTypes.IntegerType)
     // fuzz test
     castTest(gen.generateStrings(dataSize, numericPattern, 8).toDF("a"), DataTypes.IntegerType)
   }
 
   test("cast StringType to LongType") {
-    // TODO fix for Spark 4.0.0
-    assume(!isSpark40Plus)
     // test with hand-picked values
     castTest(castStringToIntegralInputs.toDF("a"), DataTypes.LongType)
     // fuzz test
@@ -707,8 +696,6 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
   }
 
   test("cast StringType to DateType") {
-    // TODO fix for Spark 4.0.0
-    assume(!isSpark40Plus)
     val validDates = Seq(
       "262142-01-01",
       "262142-01-01 ",
@@ -1294,11 +1281,16 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
                 assert(cometMessage.contains("too large to store"))
               } else {
                 if (CometSparkSessionExtensions.isSpark40Plus) {
+                  // this is removed in Spark 4.0
+                  val cometOnly =
+                    "If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error."
                   // for Spark 4 we expect to sparkException carries the message
+                  val sparkOnly = "SQLSTATE"
                   assert(
-                    sparkException.getMessage
-                      .replace(".WITH_SUGGESTION] ", "]")
-                      .startsWith(cometMessage))
+                    sparkMessage.contains(sparkOnly) &&
+                      sparkMessage
+                        .replace(".WITH_SUGGESTION] ", "]")
+                        .startsWith(cometMessage.replaceAll(cometOnly, "")))
                 } else {
                   // for Spark 3.4 we expect to reproduce the error message exactly
                   assert(cometMessage == sparkMessage)