Skip the test for spark versions 3.1.1, 3.1.2 and 3.2.0 only [databricks]

integration_tests/src/main/python/cast_test.py

@@ -16,7 +16,7 @@
 
 from asserts import assert_gpu_and_cpu_are_equal_collect, assert_gpu_and_cpu_are_equal_sql, assert_gpu_and_cpu_error, assert_gpu_fallback_collect, assert_py4j_exception
 from data_gen import *
-from spark_session import is_before_spark_311, is_before_spark_320, is_before_spark_330, with_gpu_session
+from spark_session import is_before_spark_320, with_gpu_session, is_neg_dec_scale_bug_version
 from marks import allow_non_gpu, approximate_float
 from pyspark.sql.types import *
 
@@ -301,7 +301,7 @@ def test_cast_struct_with_unsupported_element_to_string_fallback(data_gen, legac
          "spark.sql.legacy.allowNegativeScaleOfDecimal": 'true'}
     )
 
-@pytest.mark.skipif(not is_before_spark_311() and is_before_spark_330(), reason="RAPIDS doesn't support casting string to decimal for negative scale decimal in this version of Spark because of SPARK-37451")
+@pytest.mark.skipif(is_neg_dec_scale_bug_version(), reason="RAPIDS doesn't support casting string to decimal for negative scale decimal in this version of Spark because of SPARK-37451")
 def test_cast_string_to_negative_scale_decimal():
     assert_gpu_and_cpu_are_equal_collect(
         lambda spark: unary_op_df(spark, StringGen("[0-9]{9}")).select(

integration_tests/src/main/python/spark_session.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2020-2021, NVIDIA CORPORATION.
+# Copyright (c) 2020-2022, NVIDIA CORPORATION.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -111,6 +111,10 @@ def is_before_spark_320():
 def is_before_spark_330():
     return spark_version() < "3.3.0"
 
+# The bug SPARK-37451 only affects the following versions
+def is_neg_dec_scale_bug_version():
+    return ("3.1.1" <= spark_version() < "3.1.3") or ("3.2.0" <= spark_version() < "3.2.1")
+
 def is_databricks91_or_later():
     spark = get_spark_i_know_what_i_am_doing()
     return spark.conf.get("spark.databricks.clusterUsageTags.sparkVersion", "") >= "9.1"

shims/spark313/src/main/scala/com/nvidia/spark/rapids/shims/spark313/Spark313Shims.scala

@@ -45,4 +45,5 @@ class Spark313Shims extends Spark31XShims with Spark30Xuntil33XShims {
 
   override def hasCastFloatTimestampUpcast: Boolean = true
 
+  override def isCastingStringToNegDecimalScaleSupported: Boolean = true
 }

shims/spark320/src/main/scala/com/nvidia/spark/rapids/shims/spark320/Spark320Shims.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021, NVIDIA CORPORATION.
+ * Copyright (c) 2021-2022, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,4 +37,6 @@ class Spark320Shims extends Spark320until322Shims with Spark30Xuntil33XShims {
       metadataColumns: Seq[AttributeReference]): RDD[InternalRow] = {
     new FileScanRDD(sparkSession, readFunction, filePartitions)
   }
+
+  override def isCastingStringToNegDecimalScaleSupported: Boolean = false
 }

shims/spark322/src/main/scala/com/nvidia/spark/rapids/shims/spark322/Spark322Shims.scala

@@ -37,6 +37,4 @@ class Spark322Shims extends Spark322PlusShims with Spark30Xuntil33XShims {
       metadataColumns: Seq[AttributeReference]): RDD[InternalRow] = {
     new FileScanRDD(sparkSession, readFunction, filePartitions)
   }
-
-  override def isCastingStringToNegDecimalScaleSupported: Boolean = false
 }

sql-plugin/src/main/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/Spark30XShims.scala

@@ -370,8 +370,6 @@ abstract class Spark30XShims extends Spark301until320Shims with Logging {
     adaptivePlan.initialPlan
   }
 
-  override def isCastingStringToNegDecimalScaleSupported: Boolean = true
-
   // this is to help with an optimization in Spark 3.1, so we disable it by default in Spark 3.0.x
   override def isEmptyRelation(relation: Any): Boolean = false
   override def tryTransformIfEmptyRelation(mode: BroadcastMode): Option[Any] = None

sql-plugin/src/main/31xdb/scala/com/nvidia/spark/rapids/shims/v2/Spark31XdbShimsBase.scala

@@ -129,5 +129,5 @@ trait Spark31XdbShimsBase extends SparkShims {
 
   override def shouldFallbackOnAnsiTimestamp(): Boolean = false
 
-  override def isCastingStringToNegDecimalScaleSupported: Boolean = true
+  override def isCastingStringToNegDecimalScaleSupported: Boolean = false
 }

sql-plugin/src/main/320+/scala/com/nvidia/spark/rapids/shims/v2/Spark320PlusShims.scala

@@ -1035,8 +1035,6 @@ trait Spark320PlusShims extends SparkShims with RebaseShims with Logging {
     adaptivePlan.initialPlan
   }
 
-  override def isCastingStringToNegDecimalScaleSupported: Boolean = false
-
   override def columnarAdaptivePlan(a: AdaptiveSparkPlanExec,
       goal: CoalesceSizeGoal): SparkPlan = {
     a.copy(supportsColumnar = true)

sql-plugin/src/main/scala/com/nvidia/spark/rapids/SparkShims.scala

@@ -97,7 +97,7 @@ trait SparkShims {
   def int96ParquetRebaseWrite(conf: SQLConf): String
   def int96ParquetRebaseReadKey: String
   def int96ParquetRebaseWriteKey: String
-  def isCastingStringToNegDecimalScaleSupported: Boolean
+  def isCastingStringToNegDecimalScaleSupported: Boolean = true
 
   def getParquetFilters(
     schema: MessageType,