Add GPU version of ToPrettyString [databricks]

integration_tests/src/main/python/cast_test.py

@@ -16,8 +16,8 @@
 
 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_spark_exception
 from data_gen import *
-from spark_session import is_before_spark_320, is_before_spark_330, is_spark_340_or_later, is_spark_350_or_later, \
-    is_databricks113_or_later, with_gpu_session
+from spark_session import is_before_spark_320, is_before_spark_330, is_spark_340_or_later, \
+    is_databricks113_or_later
 from marks import allow_non_gpu, approximate_float
 from pyspark.sql.types import *
 from spark_init_internal import spark_version
@@ -297,7 +297,6 @@ def _assert_cast_to_string_equal (data_gen, conf):
 
 @pytest.mark.parametrize('data_gen', all_array_gens_for_cast_to_string, ids=idfn)
 @pytest.mark.parametrize('legacy', ['true', 'false'])
-@pytest.mark.xfail(condition=is_spark_350_or_later(), reason='https://github.com/NVIDIA/spark-rapids/issues/9065')
 def test_cast_array_to_string(data_gen, legacy):
     _assert_cast_to_string_equal(
         data_gen, 
@@ -317,7 +316,6 @@ def test_cast_array_with_unmatched_element_to_string(data_gen, legacy):
 
 @pytest.mark.parametrize('data_gen', basic_map_gens_for_cast_to_string, ids=idfn)
 @pytest.mark.parametrize('legacy', ['true', 'false'])
-@pytest.mark.xfail(condition=is_spark_350_or_later(), reason='https://github.com/NVIDIA/spark-rapids/issues/9065')
 def test_cast_map_to_string(data_gen, legacy):
     _assert_cast_to_string_equal(
         data_gen, 
@@ -337,7 +335,6 @@ def test_cast_map_with_unmatched_element_to_string(data_gen, legacy):
 
 @pytest.mark.parametrize('data_gen', [StructGen([[str(i), gen] for i, gen in enumerate(basic_array_struct_gens_for_cast_to_string)] + [["map", MapGen(ByteGen(nullable=False), null_gen)]])], ids=idfn)
 @pytest.mark.parametrize('legacy', ['true', 'false'])
-@pytest.mark.xfail(condition=is_spark_350_or_later(), reason='https://github.com/NVIDIA/spark-rapids/issues/9065')
 def test_cast_struct_to_string(data_gen, legacy):
     _assert_cast_to_string_equal(
         data_gen, 

sql-plugin/src/main/java/com/nvidia/spark/rapids/iceberg/spark/source/GpuIcebergReader.java

@@ -23,7 +23,7 @@
 import java.util.NoSuchElementException;
 
 import ai.rapids.cudf.Scalar;
-import com.nvidia.spark.rapids.GpuCast;
+import com.nvidia.spark.rapids.CastOperation;
 import com.nvidia.spark.rapids.GpuColumnVector;
 import com.nvidia.spark.rapids.GpuScalar;
 import com.nvidia.spark.rapids.iceberg.data.GpuDeleteFilter;
@@ -157,7 +157,7 @@ static ColumnarBatch addUpcastsIfNeeded(ColumnarBatch batch, Schema expectedSche
         DataType expectedSparkType = SparkSchemaUtil.convert(expectedColumnTypes.get(i).type());
         GpuColumnVector oldColumn = columns[i];
         columns[i] = GpuColumnVector.from(
-            GpuCast.doCast(oldColumn.getBase(), oldColumn.dataType(), expectedSparkType,
+            CastOperation.apply(oldColumn.getBase(), oldColumn.dataType(), expectedSparkType,
             false, false, false), expectedSparkType);
       }
       ColumnarBatch newBatch = new ColumnarBatch(columns, batch.numRows());

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

@@ -84,4 +84,25 @@ object FloatUtils {
       }
     }
   }
+
+  private[rapids] def castFloatingTypeToString(input: ColumnView): ColumnVector = {
+    withResource(input.castTo(DType.STRING)) { cudfCast =>
+
+      // replace "e+" with "E"
+      val replaceExponent = withResource(Scalar.fromString("e+")) { cudfExponent =>
+        withResource(Scalar.fromString("E")) { sparkExponent =>
+          cudfCast.stringReplace(cudfExponent, sparkExponent)
+        }
+      }
+
+      // replace "Inf" with "Infinity"
+      withResource(replaceExponent) { replaceExponent =>
+        withResource(Scalar.fromString("Inf")) { cudfInf =>
+          withResource(Scalar.fromString("Infinity")) { sparkInfinity =>
+            replaceExponent.stringReplace(cudfInf, sparkInfinity)
+          }
+        }
+      }
+    }
+  }
 }

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

@@ -19,7 +19,6 @@ package com.nvidia.spark.rapids
 import ai.rapids.cudf
 import ai.rapids.cudf.{DType, GroupByAggregation, ReductionAggregation}
 import com.nvidia.spark.rapids.Arm.withResource
-import com.nvidia.spark.rapids.GpuCast.doCast
 import com.nvidia.spark.rapids.shims.ShimExpression
 
 import org.apache.spark.sql.catalyst.InternalRow
@@ -144,7 +143,7 @@ case class ApproxPercentileFromTDigestExpr(
           // array and return that (after converting from Double to finalDataType)
           withResource(cv.getBase.approxPercentile(Array(p))) { percentiles =>
             withResource(percentiles.extractListElement(0)) { childView =>
-              withResource(doCast(childView, DataTypes.DoubleType, finalDataType,
+              withResource(CastOperation(childView, DataTypes.DoubleType, finalDataType,
                   ansiMode = false, legacyCastToString = false,
                   stringToDateAnsiModeEnabled = false)) { childCv =>
                 GpuColumnVector.from(childCv.copyToColumnVector(), dataType)
@@ -159,7 +158,7 @@ case class ApproxPercentileFromTDigestExpr(
               GpuColumnVector.from(percentiles.incRefCount(), dataType)
             } else {
               withResource(percentiles.getChildColumnView(0)) { childView =>
-                withResource(doCast(childView, DataTypes.DoubleType, finalDataType,
+                withResource(CastOperation(childView, DataTypes.DoubleType, finalDataType,
                     ansiMode = false, legacyCastToString = false,
                     stringToDateAnsiModeEnabled = false)) { childCv =>
                   withResource(percentiles.replaceListChild(childCv)) { x =>