Switch rounding calls to using new API (#13497)

This changes the plugin to call the new spark-rapids-jni API for
rounding rather than the old cudf API. The rounding of floating point
functionality will be removed from cudf so we've moved the float
rounding (and the API) into spark-rapids-jni.

This is part of resolving [sr-jni issue
3585](NVIDIA/cudf-spark-jni#3585)

[This PR](rapidsai/cudf#20110) removes the old
java code from cudf, and will be merged in cudf 25.12.
[This PR](NVIDIA/cudf-spark-jni#3770) moves
the floating rounding into spark-rapids-jni and changes the API. This
will be merged in 25.10 and is a dependency of this PR.

Note that I have tested this code applying both of the above PRs and all
of the spark-rapids integration tests pass.

### Checklists

- [ ] This PR has added documentation for new or modified features or
behaviors.
- [ ] This PR has added new tests or modified existing tests to cover
new code paths.
(Please explain in the PR description how the new code paths are tested,
such as names of the new/existing tests that cover them.)
- [ ] Performance testing has been performed and its results are added
in the PR description. Or, an issue has been filed with a link in the PR
description.

---------

Signed-off-by: Paul Mattione <pmattione@nvidia.com>

Author: pmattione-nvidia

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

@@ -22,10 +22,9 @@ import java.util.Optional
 import scala.collection.mutable.ArrayBuffer
 
 import ai.rapids.cudf.{BinaryOp, CaptureGroups, ColumnVector, ColumnView, DType, RegexProgram, Scalar}
-import ai.rapids.cudf
 import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
 import com.nvidia.spark.rapids.RapidsPluginImplicits._
-import com.nvidia.spark.rapids.jni.{CastException, CastStrings, DecimalUtils, GpuTimeZoneDB}
+import com.nvidia.spark.rapids.jni.{Arithmetic, CastException, CastStrings, DecimalUtils, GpuTimeZoneDB, RoundMode}
 import com.nvidia.spark.rapids.shims.{AnsiUtil, CastTimeToIntShim, GpuCastShims, GpuIntervalUtils, GpuTypeShims, NullIntolerantShim, SparkShimImpl}
 import org.apache.commons.text.StringEscapeUtils
 
@@ -1568,7 +1567,7 @@ object GpuCast {
       val rounded = if (!isScaleUpcast) {
         // We have to round the data to the desired scale. Spark uses HALF_UP rounding in
         // this case so we need to also.
-        input.round(to.scale, cudf.RoundMode.HALF_UP)
+        Arithmetic.round(input, to.scale, RoundMode.HALF_UP)
       } else {
         input.copyToColumnVector()
       }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2023, NVIDIA CORPORATION.
+ * Copyright (c) 2020-2025, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -15,9 +15,9 @@
  */
 package com.nvidia.spark.rapids
 
-import ai.rapids.cudf
-import ai.rapids.cudf.{ColumnVector, DecimalUtils, DType, Scalar}
+import ai.rapids.cudf.{ColumnVector, DType, Scalar}
 import com.nvidia.spark.rapids.Arm.withResource
+import com.nvidia.spark.rapids.jni.{Arithmetic, RoundMode}
 
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.types.{DataType, DecimalType, LongType}
@@ -48,7 +48,7 @@ case class GpuCheckOverflow(child: Expression,
     val rounded = if (resultDType.equals(base.getType)) {
       base.incRefCount()
     } else {
-      withResource(base.round(dataType.scale, cudf.RoundMode.HALF_UP)) { rounded =>
+      withResource(Arithmetic.round(base, dataType.scale, RoundMode.HALF_UP)) { rounded =>
         if (resultDType.getTypeId != base.getType.getTypeId) {
           rounded.castTo(resultDType)
         } else {
@@ -101,12 +101,12 @@ case class GpuMakeDecimal(
   override def toString: String = s"MakeDecimal($child,$precision,$sparkScale)"
 
   private lazy val (minValue, maxValue) = {
-    val bounds = DecimalUtils.bounds(dataType.precision, dataType.scale)
+    val bounds = ai.rapids.cudf.DecimalUtils.bounds(dataType.precision, dataType.scale)
     (bounds.getKey.unscaledValue().longValue(), bounds.getValue.unscaledValue().longValue())
   }
 
   override protected def doColumnar(input: GpuColumnVector): ColumnVector = {
-    val outputType = DecimalUtils.createDecimalType(precision, sparkScale)
+    val outputType = ai.rapids.cudf.DecimalUtils.createDecimalType(precision, sparkScale)
     val base = input.getBase
     val outOfBounds = withResource(Scalar.fromLong(maxValue)) { maxScalar =>
       withResource(base.greaterThan(maxScalar)) { over =>

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala

@@ -27,7 +27,7 @@ import com.nvidia.spark.rapids.Arm._
 import com.nvidia.spark.rapids.ExprMeta
 import com.nvidia.spark.rapids.GpuOverrides.{extractStringLit, getTimeParserPolicy}
 import com.nvidia.spark.rapids.RapidsPluginImplicits._
-import com.nvidia.spark.rapids.jni.{DateTimeUtils, GpuTimeZoneDB}
+import com.nvidia.spark.rapids.jni.{Arithmetic, DateTimeUtils, GpuTimeZoneDB}
 import com.nvidia.spark.rapids.shims.{NullIntolerantShim, ShimBinaryExpression, ShimExpression}
 
 import org.apache.spark.sql.catalyst.expressions.{BinaryExpression, ExpectsInputTypes, Expression, FromUnixTime, FromUTCTimestamp, ImplicitCastInputTypes, MonthsBetween, TimeZoneAwareExpression, ToUTCTimestamp, TruncDate, TruncTimestamp}
@@ -1418,7 +1418,7 @@ case class GpuMonthsBetween(ts1: Expression,
               }
               val roundedPartialMonth = if (needsRoundOff.get) {
                 withResource(partialMonth) { _ =>
-                  partialMonth.round(8)
+                  Arithmetic.round(partialMonth, 8)
                 }
               } else {
                 partialMonth

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/mathExpressions.scala

@@ -22,7 +22,7 @@ import ai.rapids.cudf._
 import ai.rapids.cudf.ast.BinaryOperator
 import com.nvidia.spark.rapids._
 import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
-import com.nvidia.spark.rapids.jni.CastStrings
+import com.nvidia.spark.rapids.jni.{Arithmetic, CastStrings, RoundMode}
 
 import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes}
 import org.apache.spark.sql.rapids.shims.RapidsErrorUtils
@@ -608,7 +608,7 @@ abstract class GpuRoundBase(child: Expression, scale: Expression, outputType: Da
       case DecimalType.Fixed(_, s) =>
         // Only needs to perform round when required scale < input scale
         val rounded = if (scaleVal < s) {
-          lhsValue.round(scaleVal, roundMode)
+          Arithmetic.round(lhsValue, scaleVal, roundMode)
         } else {
           lhsValue.incRefCount()
         }
@@ -672,7 +672,7 @@ abstract class GpuRoundBase(child: Expression, scale: Expression, outputType: Da
         }
       }
     } else {
-      lhs.round(scale, roundMode)
+      Arithmetic.round(lhs, scale, roundMode)
     }
   }
 
@@ -769,7 +769,7 @@ abstract class GpuRoundBase(child: Expression, scale: Expression, outputType: Da
       // just returns the original values
       lhs.incRefCount()
     } else {
-      lhs.round(scale, roundMode)
+      Arithmetic.round(lhs, scale, roundMode)
     }
   }
 

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/stringFunctions.scala

@@ -24,10 +24,11 @@ import scala.annotation.tailrec
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 
-import ai.rapids.cudf.{ast, BinaryOp, BinaryOperable, CaptureGroups, ColumnVector, ColumnView, DType, PadSide, RegexFlag, RegexProgram, RoundMode, Scalar, Table}
+import ai.rapids.cudf.{ast, BinaryOp, BinaryOperable, CaptureGroups, ColumnVector, ColumnView, DType, PadSide, RegexFlag, RegexProgram, Scalar, Table}
 import com.nvidia.spark.rapids._
 import com.nvidia.spark.rapids.Arm._
 import com.nvidia.spark.rapids.RapidsPluginImplicits._
+import com.nvidia.spark.rapids.jni.{Arithmetic, RoundMode}
 import com.nvidia.spark.rapids.jni.CastStrings
 import com.nvidia.spark.rapids.jni.GpuSubstringIndexUtils
 import com.nvidia.spark.rapids.jni.NumberConverter
@@ -2283,7 +2284,8 @@ case class GpuFormatNumber(x: Expression, d: Expression)
     val appendZeroNum = (d - scale).max(0).min(d)
     val (intPart, decTemp) = if (roundingScale <= 0) {
       withResource(ArrayBuffer.empty[ColumnVector]) { resourceArray =>
-        val intPart = withResource(cv.round(roundingScale, RoundMode.HALF_EVEN)) { rounded =>
+        val intPart = withResource(Arithmetic.round(cv, roundingScale, 
+          RoundMode.HALF_EVEN)) { rounded =>
           rounded.castTo(DType.STRING)
         }
         resourceArray += intPart
@@ -2307,7 +2309,7 @@ case class GpuFormatNumber(x: Expression, d: Expression)
         (intPartZeroHandled.incRefCount(), decPart.incRefCount())
       }
     } else {
-      withResource(cv.round(roundingScale, RoundMode.HALF_EVEN)) { rounded =>
+      withResource(Arithmetic.round(cv, roundingScale, RoundMode.HALF_EVEN)) { rounded =>
         withResource(rounded.castTo(DType.STRING)) { roundedStr =>
           withResource(roundedStr.stringSplit(".", 2)) { intAndDec =>
             (intAndDec.getColumn(0).incRefCount(), intAndDec.getColumn(1).incRefCount())

sql-plugin/src/main/spark330/scala/org/apache/spark/sql/rapids/shims/intervalExpressions.scala

@@ -45,9 +45,10 @@ package org.apache.spark.sql.rapids.shims
 
 import java.math.BigInteger
 
-import ai.rapids.cudf.{BinaryOperable, ColumnVector, ColumnView, DType, RoundMode, Scalar}
+import ai.rapids.cudf.{BinaryOperable, ColumnVector, ColumnView, DType, Scalar}
 import com.nvidia.spark.rapids.{BoolUtils, GpuBinaryExpression, GpuColumnVector, GpuScalar}
 import com.nvidia.spark.rapids.Arm.withResource
+import com.nvidia.spark.rapids.jni.{Arithmetic, RoundMode}
 import com.nvidia.spark.rapids.shims.NullIntolerantShim
 
 import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes}
@@ -160,7 +161,7 @@ object IntervalUtils {
     // check Inf, -Inf, NaN
     checkDoubleInfNan(doubleCv)
 
-    withResource(doubleCv.round(RoundMode.HALF_UP)) { roundedDouble =>
+    withResource(Arithmetic.round(doubleCv, RoundMode.HALF_UP)) { roundedDouble =>
       // throws exception if the result exceeds int limits
       withResource(roundedDouble.castTo(DType.INT64)) { long =>
         castLongToIntWithOverflowCheck(long)
@@ -192,7 +193,7 @@ object IntervalUtils {
     val MIN_LONG_AS_DOUBLE: Double = -9.223372036854776E18
     val MAX_LONG_AS_DOUBLE_PLUS_ONE: Double = 9.223372036854776E18
 
-    withResource(doubleCv.round(RoundMode.HALF_UP)) { z =>
+    withResource(Arithmetic.round(doubleCv, RoundMode.HALF_UP)) { z =>
       withResource(Scalar.fromDouble(MAX_LONG_AS_DOUBLE_PLUS_ONE)) { max =>
         withResource(z.greaterOrEqualTo(max)) { invalid =>
           if (BoolUtils.isAnyValidTrue(invalid)) {
@@ -323,7 +324,7 @@ object IntervalUtils {
       leftDecimal.div(q, dT)
     }
     withResource(t) { t =>
-      t.round(RoundMode.HALF_UP)
+      Arithmetic.round(t, RoundMode.HALF_UP)
     }
   }
 }