[SPARK-30011][SQL] Inline 2.12 "AsIfIntegral" classes, not present in 2.13

### What changes were proposed in this pull request?

Classes like DoubleAsIfIntegral are not found in Scala 2.13, but used in the current build. This change 'inlines' the 2.12 implementation and makes it work with both 2.12 and 2.13.

### Why are the changes needed?

To cross-compile with 2.13.

### Does this PR introduce any user-facing change?

It should not as it copies in 2.12's existing behavior.

### How was this patch tested?

Existing tests.

Closes apache#26769 from srowen/SPARK-30011.

Authored-by: Sean Owen <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: srowen

sql/catalyst/src/main/scala/org/apache/spark/sql/types/Decimal.scala

@@ -20,8 +20,9 @@ package org.apache.spark.sql.types
 import java.lang.{Long => JLong}
 import java.math.{BigInteger, MathContext, RoundingMode}
 
+import scala.util.Try
+
 import org.apache.spark.annotation.Unstable
-import org.apache.spark.sql.AnalysisException
 
 /**
  * A mutable implementation of BigDecimal that can hold a Long if values are small enough.
@@ -622,6 +623,9 @@ object Decimal {
     override def toLong(x: Decimal): Long = x.toLong
     override def fromInt(x: Int): Decimal = new Decimal().set(x)
     override def compare(x: Decimal, y: Decimal): Int = x.compare(y)
+    // Added from Scala 2.13; don't override to work in 2.12
+    // TODO revisit once Scala 2.12 support is dropped
+    def parseString(str: String): Option[Decimal] = Try(Decimal(str)).toOption
   }
 
   /** A [[scala.math.Fractional]] evidence parameter for Decimals. */

sql/catalyst/src/main/scala/org/apache/spark/sql/types/DoubleType.scala

@@ -18,11 +18,10 @@
 package org.apache.spark.sql.types
 
 import scala.math.{Fractional, Numeric}
-import scala.math.Numeric.DoubleAsIfIntegral
 import scala.reflect.runtime.universe.typeTag
+import scala.util.Try
 
 import org.apache.spark.annotation.Stable
-import org.apache.spark.util.Utils
 
 /**
  * The data type representing `Double` values. Please use the singleton `DataTypes.DoubleType`.
@@ -40,7 +39,7 @@ class DoubleType private() extends FractionalType {
   private[sql] val fractional = implicitly[Fractional[Double]]
   private[sql] val ordering =
     (x: Double, y: Double) => java.lang.Double.compare(x, y)
-  private[sql] val asIntegral = DoubleAsIfIntegral
+  private[sql] val asIntegral = DoubleType.DoubleAsIfIntegral
 
   override private[sql] def exactNumeric = DoubleExactNumeric
 
@@ -56,4 +55,34 @@ class DoubleType private() extends FractionalType {
  * @since 1.3.0
  */
 @Stable
-case object DoubleType extends DoubleType
+case object DoubleType extends DoubleType {
+
+  // Traits below copied from Scala 2.12; not present in 2.13
+  // TODO: SPARK-30011 revisit once Scala 2.12 support is dropped
+  trait DoubleIsConflicted extends Numeric[Double] {
+    def plus(x: Double, y: Double): Double = x + y
+    def minus(x: Double, y: Double): Double = x - y
+    def times(x: Double, y: Double): Double = x * y
+    def negate(x: Double): Double = -x
+    def fromInt(x: Int): Double = x.toDouble
+    def toInt(x: Double): Int = x.toInt
+    def toLong(x: Double): Long = x.toLong
+    def toFloat(x: Double): Float = x.toFloat
+    def toDouble(x: Double): Double = x
+    // logic in Numeric base trait mishandles abs(-0.0)
+    override def abs(x: Double): Double = math.abs(x)
+    // Added from Scala 2.13; don't override to work in 2.12
+    def parseString(str: String): Option[Double] =
+      Try(java.lang.Double.parseDouble(str)).toOption
+
+  }
+
+  trait DoubleAsIfIntegral extends DoubleIsConflicted with Integral[Double] {
+    def quot(x: Double, y: Double): Double = (BigDecimal(x) quot BigDecimal(y)).doubleValue
+    def rem(x: Double, y: Double): Double = (BigDecimal(x) remainder BigDecimal(y)).doubleValue
+  }
+
+  object DoubleAsIfIntegral extends DoubleAsIfIntegral {
+    override def compare(x: Double, y: Double): Int = java.lang.Double.compare(x, y)
+  }
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/types/FloatType.scala

@@ -18,11 +18,10 @@
 package org.apache.spark.sql.types
 
 import scala.math.{Fractional, Numeric}
-import scala.math.Numeric.FloatAsIfIntegral
 import scala.reflect.runtime.universe.typeTag
+import scala.util.Try
 
 import org.apache.spark.annotation.Stable
-import org.apache.spark.util.Utils
 
 /**
  * The data type representing `Float` values. Please use the singleton `DataTypes.FloatType`.
@@ -40,7 +39,7 @@ class FloatType private() extends FractionalType {
   private[sql] val fractional = implicitly[Fractional[Float]]
   private[sql] val ordering =
     (x: Float, y: Float) => java.lang.Float.compare(x, y)
-  private[sql] val asIntegral = FloatAsIfIntegral
+  private[sql] val asIntegral = FloatType.FloatAsIfIntegral
 
   override private[sql] def exactNumeric = FloatExactNumeric
 
@@ -57,4 +56,33 @@ class FloatType private() extends FractionalType {
  * @since 1.3.0
  */
 @Stable
-case object FloatType extends FloatType
+case object FloatType extends FloatType {
+
+  // Traits below copied from Scala 2.12; not present in 2.13
+  // TODO: SPARK-30011 revisit once Scala 2.12 support is dropped
+  trait FloatIsConflicted extends Numeric[Float] {
+    def plus(x: Float, y: Float): Float = x + y
+    def minus(x: Float, y: Float): Float = x - y
+    def times(x: Float, y: Float): Float = x * y
+    def negate(x: Float): Float = -x
+    def fromInt(x: Int): Float = x.toFloat
+    def toInt(x: Float): Int = x.toInt
+    def toLong(x: Float): Long = x.toLong
+    def toFloat(x: Float): Float = x
+    def toDouble(x: Float): Double = x.toDouble
+    // logic in Numeric base trait mishandles abs(-0.0f)
+    override def abs(x: Float): Float = math.abs(x)
+    // Added from Scala 2.13; don't override to work in 2.12
+    def parseString(str: String): Option[Float] =
+      Try(java.lang.Float.parseFloat(str)).toOption
+  }
+
+  trait FloatAsIfIntegral extends FloatIsConflicted with Integral[Float] {
+    def quot(x: Float, y: Float): Float = (BigDecimal(x) quot BigDecimal(y)).floatValue
+    def rem(x: Float, y: Float): Float = (BigDecimal(x) remainder BigDecimal(y)).floatValue
+  }
+
+  object FloatAsIfIntegral extends FloatAsIfIntegral {
+    override def compare(x: Float, y: Float): Int = java.lang.Float.compare(x, y)
+  }
+}