simplify Decimal

Author: Davies Liu

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -562,16 +562,16 @@ case class Cast(child: Expression, dataType: DataType)
               java.math.BigDecimal.valueOf(${timestampToDoubleCode(c)}));
             ${changePrecision("tmpDecimal", target, evPrim, evNull)}
           """
-      case DecimalType() =>
+      case dt: DecimalType =>
         (c, evPrim, evNull) =>
           s"""
             Decimal tmpDecimal = $c.clone();
             ${changePrecision("tmpDecimal", target, evPrim, evNull)}
           """
-      case LongType =>
+      case ByteType | ShortType | IntegerType | LongType =>
         (c, evPrim, evNull) =>
           s"""
-            Decimal tmpDecimal = Decimal.apply($c);
+            Decimal tmpDecimal = Decimal.apply((long) $c);
             ${changePrecision("tmpDecimal", target, evPrim, evNull)}
           """
       case x: NumericType =>

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

@@ -20,6 +20,7 @@ package org.apache.spark.sql.types
 import java.math.{MathContext, RoundingMode, BigDecimal => JavaBigDecimal}
 
 import org.apache.spark.annotation.DeveloperApi
+import org.apache.spark.unsafe.PlatformDependent
 
 /**
  * A mutable implementation of BigDecimal that can hold a Long if values are small enough.
@@ -32,74 +33,38 @@ import org.apache.spark.annotation.DeveloperApi
 final class Decimal extends Ordered[Decimal] with Serializable {
   import org.apache.spark.sql.types.Decimal._
 
-  private var decimalVal: JavaBigDecimal = null
-  private var longVal: Long = 0L
+  private var decimalVal: JavaBigDecimal = BIG_DEC_ZERO
   private var _precision: Int = 1
-  private var _scale: Int = 0
 
   def precision: Int = _precision
-  def scale: Int = _scale
+  def scale: Int = decimalVal.scale()
 
   /**
    * Set this Decimal to the given Long. Will have precision 20 and scale 0.
    */
   def set(longVal: Long): Decimal = {
-    if (longVal <= -POW_10(MAX_LONG_DIGITS) || longVal >= POW_10(MAX_LONG_DIGITS)) {
-      // We can't represent this compactly as a long without risking overflow
-      this.decimalVal = new JavaBigDecimal(longVal)
-      this.longVal = 0L
-    } else {
-      this.decimalVal = null
-      this.longVal = longVal
-    }
-    this._precision = 20
-    this._scale = 0
+    decimalVal = JavaBigDecimal.valueOf(longVal)
+    _precision = 20
     this
   }
 
   /**
    * Set this Decimal to the given Int. Will have precision 10 and scale 0.
    */
   def set(intVal: Int): Decimal = {
-    this.decimalVal = null
-    this.longVal = intVal
-    this._precision = 10
-    this._scale = 0
+    decimalVal = JavaBigDecimal.valueOf(intVal)
+    _precision = 10
     this
   }
 
   /**
    * Set this Decimal to the given unscaled Long, with a given precision and scale.
+   *
+   * Note: this is used in serialization, caller will make sure that it will not overflow
    */
   def set(unscaled: Long, precision: Int, scale: Int): Decimal = {
-    if (setOrNull(unscaled, precision, scale) == null) {
-      throw new IllegalArgumentException("Unscaled value too large for precision")
-    }
-    this
-  }
-
-  /**
-   * Set this Decimal to the given unscaled Long, with a given precision and scale,
-   * and return it, or return null if it cannot be set due to overflow.
-   */
-  def setOrNull(unscaled: Long, precision: Int, scale: Int): Decimal = {
-    if (unscaled <= -POW_10(MAX_LONG_DIGITS) || unscaled >= POW_10(MAX_LONG_DIGITS)) {
-      // We can't represent this compactly as a long without risking overflow
-      if (precision < 19) {
-        return null  // Requested precision is too low to represent this value
-      }
-      this.decimalVal = new JavaBigDecimal(unscaled)
-      this.longVal = 0L
-    } else {
-      val p = POW_10(math.min(precision, MAX_LONG_DIGITS))
-      if (unscaled <= -p || unscaled >= p) {
-        return null  // Requested precision is too low to represent this value
-      }
-      this.decimalVal = null
-      this.longVal = unscaled
-    }
-    this._precision = precision
-    this._scale = scale
+    decimalVal = JavaBigDecimal.valueOf(unscaled, scale)
+    _precision = precision
     this
   }
 
@@ -121,11 +86,9 @@ final class Decimal extends Ordered[Decimal] with Serializable {
    * Set this Decimal to the given java.math.BigDecimal value, with a given precision and scale.
    */
   private[sql] def set(decimal: JavaBigDecimal, precision: Int, scale: Int): Decimal = {
-    this.decimalVal = decimal.setScale(scale, ROUNDING_MODE)
+    decimalVal = decimal.setScale(scale, ROUNDING_MODE)
     require(decimalVal.precision <= precision, "Overflowed precision")
-    this.longVal = 0L
-    this._precision = precision
-    this._scale = scale
+    _precision = precision
     this
   }
 
@@ -134,9 +97,7 @@ final class Decimal extends Ordered[Decimal] with Serializable {
    */
   private[sql] def set(decimal: JavaBigDecimal): Decimal = {
     this.decimalVal = decimal
-    this.longVal = 0L
     this._precision = decimal.precision
-    this._scale = decimal.scale
     this
   }
 
@@ -145,52 +106,36 @@ final class Decimal extends Ordered[Decimal] with Serializable {
    */
   def set(decimal: Decimal): Decimal = {
     this.decimalVal = decimal.decimalVal
-    this.longVal = decimal.longVal
     this._precision = decimal._precision
-    this._scale = decimal._scale
     this
   }
 
   def toBigDecimal: BigDecimal = BigDecimal(toJavaBigDecimal)
 
-  private[sql] def toJavaBigDecimal: JavaBigDecimal = {
-    if (decimalVal.ne(null)) {
-      decimalVal
-    } else {
-      JavaBigDecimal.valueOf(longVal, _scale)
-    }
-  }
+  private[sql] def toJavaBigDecimal: JavaBigDecimal = decimalVal
 
   def toUnscaledLong: Long = {
-    if (decimalVal.ne(null)) {
-      decimalVal.unscaledValue().longValue()
+    val unscaled = PlatformDependent.UNSAFE.getLong(decimalVal,
+      PlatformDependent.BIG_DECIMAL_INTCOMPACT_OFFSET)
+    if (unscaled != Long.MinValue) {
+      unscaled
     } else {
-      longVal
+      decimalVal.unscaledValue().longValue()
     }
   }
 
-  override def toString: String = toJavaBigDecimal.toString()
+  override def toString: String = decimalVal.toString()
 
   @DeveloperApi
   def toDebugString: String = {
-    if (decimalVal.ne(null)) {
-      s"Decimal(expanded,$decimalVal,$precision,$scale})"
-    } else {
-      s"Decimal(compact,$longVal,$precision,$scale})"
-    }
+    s"Decimal($decimalVal,${_precision})"
   }
 
   def toDouble: Double = toJavaBigDecimal.doubleValue()
 
   def toFloat: Float = toJavaBigDecimal.floatValue()
 
-  def toLong: Long = {
-    if (decimalVal.eq(null)) {
-      longVal / POW_10(_scale)
-    } else {
-      decimalVal.longValue()
-    }
-  }
+  def toLong: Long = decimalVal.longValue()
 
   def toInt: Int = toLong.toInt
 
@@ -205,65 +150,23 @@ final class Decimal extends Ordered[Decimal] with Serializable {
    */
   def changePrecision(precision: Int, scale: Int): Boolean = {
     // fast path for UnsafeProjection
-    if (precision == this.precision && scale == this.scale) {
+    if (precision == _precision && scale == decimalVal.scale()) {
       return true
     }
-    // First, update our longVal if we can, or transfer over to using a BigDecimal
-    if (decimalVal.eq(null)) {
-      if (scale < _scale) {
-        // Easier case: we just need to divide our scale down
-        val diff = _scale - scale
-        val droppedDigits = longVal % POW_10(diff)
-        longVal /= POW_10(diff)
-        if (math.abs(droppedDigits) * 2 >= POW_10(diff)) {
-          longVal += (if (longVal < 0) -1L else 1L)
-        }
-      } else if (scale > _scale) {
-        // We might be able to multiply longVal by a power of 10 and not overflow, but if not,
-        // switch to using a BigDecimal
-        val diff = scale - _scale
-        val p = POW_10(math.max(MAX_LONG_DIGITS - diff, 0))
-        if (diff <= MAX_LONG_DIGITS && longVal > -p && longVal < p) {
-          // Multiplying longVal by POW_10(diff) will still keep it below MAX_LONG_DIGITS
-          longVal *= POW_10(diff)
-        } else {
-          // Give up on using Longs; switch to BigDecimal, which we'll modify below
-          decimalVal = JavaBigDecimal.valueOf(longVal, _scale)
-        }
-      }
-      // In both cases, we will check whether our precision is okay below
-    }
 
-    if (decimalVal.ne(null)) {
-      // We get here if either we started with a BigDecimal, or we switched to one because we would
-      // have overflowed our Long; in either case we must rescale decimalVal to the new scale.
-      val newVal = decimalVal.setScale(scale, ROUNDING_MODE)
-      if (newVal.precision > precision) {
-        return false
-      }
-      decimalVal = newVal
-    } else {
-      // We're still using Longs, but we should check whether we match the new precision
-      val p = POW_10(math.min(precision, MAX_LONG_DIGITS))
-      if (longVal <= -p || longVal >= p) {
-        // Note that we shouldn't have been able to fix this by switching to BigDecimal
-        return false
-      }
+    val newVal = decimalVal.setScale(scale, ROUNDING_MODE)
+    if (newVal.precision > precision) {
+      return false
     }
-
+    decimalVal = newVal
     _precision = precision
-    _scale = scale
     true
   }
 
   override def clone(): Decimal = new Decimal().set(this)
 
   override def compare(other: Decimal): Int = {
-    if (decimalVal.eq(null) && other.decimalVal.eq(null) && _scale == other._scale) {
-      if (longVal < other.longVal) -1 else if (longVal == other.longVal) 0 else 1
-    } else {
-      toJavaBigDecimal.compareTo(other.toJavaBigDecimal)
-    }
+    toJavaBigDecimal.compareTo(other.toJavaBigDecimal)
   }
 
   override def equals(other: Any): Boolean = other match {
@@ -276,24 +179,15 @@ final class Decimal extends Ordered[Decimal] with Serializable {
   override def hashCode(): Int = toBigDecimal.hashCode()
 
   def isZero: Boolean = {
-    if (decimalVal.ne(null)) decimalVal.compareTo(BIG_DEC_ZERO) == 0
-    else longVal == 0
+    decimalVal.compareTo(BIG_DEC_ZERO) == 0
   }
 
   def + (that: Decimal): Decimal = {
-    if (decimalVal.eq(null) && that.decimalVal.eq(null) && scale == that.scale) {
-      Decimal(longVal + that.longVal, Math.max(precision, that.precision), scale)
-    } else {
-      Decimal(toJavaBigDecimal.add(that.toJavaBigDecimal, MATH_CONTEXT), precision, scale)
-    }
+    Decimal(toJavaBigDecimal.add(that.toJavaBigDecimal, MATH_CONTEXT), precision, scale)
   }
 
   def - (that: Decimal): Decimal = {
-    if (decimalVal.eq(null) && that.decimalVal.eq(null) && scale == that.scale) {
-      Decimal(longVal - that.longVal, Math.max(precision, that.precision), scale)
-    } else {
-      Decimal(toJavaBigDecimal.subtract(that.toJavaBigDecimal, MATH_CONTEXT), precision, scale)
-    }
+    Decimal(toJavaBigDecimal.subtract(that.toJavaBigDecimal, MATH_CONTEXT), precision, scale)
   }
 
   // HiveTypeCoercion will take care of the precision, scale of result
@@ -313,11 +207,7 @@ final class Decimal extends Ordered[Decimal] with Serializable {
   def remainder(that: Decimal): Decimal = this % that
 
   def unary_- : Decimal = {
-    if (decimalVal.ne(null)) {
-      Decimal(decimalVal.negate(), precision, scale)
-    } else {
-      Decimal(-longVal, precision, scale)
-    }
+    Decimal(decimalVal.negate(), precision, scale)
   }
 
   def abs: Decimal = if (this.compare(Decimal.ZERO) < 0) this.unary_- else this

sql/catalyst/src/test/scala/org/apache/spark/sql/types/decimal/DecimalSuite.scala

@@ -46,11 +46,8 @@ class DecimalSuite extends SparkFunSuite with PrivateMethodTester {
     checkDecimal(Decimal(1e17.toLong, 18, 0), 1e17.toLong.toString, 18, 0)
     checkDecimal(Decimal(Long.MaxValue), Long.MaxValue.toString, 20, 0)
     checkDecimal(Decimal(Long.MinValue), Long.MinValue.toString, 20, 0)
-    intercept[IllegalArgumentException](Decimal(170L, 2, 1))
-    intercept[IllegalArgumentException](Decimal(170L, 2, 0))
     intercept[IllegalArgumentException](Decimal(BigDecimal("10.030"), 2, 1))
     intercept[IllegalArgumentException](Decimal(BigDecimal("-9.95"), 2, 1))
-    intercept[IllegalArgumentException](Decimal(1e17.toLong, 17, 0))
   }
 
   test("creating decimals with negative scale") {
@@ -88,36 +85,19 @@ class DecimalSuite extends SparkFunSuite with PrivateMethodTester {
     checkValues(Decimal(Double.MinValue), Double.MinValue, 0L)
   }
 
-  // Accessor for the BigDecimal value of a Decimal, which will be null if it's using Longs
-  private val decimalVal = PrivateMethod[BigDecimal]('decimalVal)
-
-  /** Check whether a decimal is represented compactly (passing whether we expect it to be) */
-  private def checkCompact(d: Decimal, expected: Boolean): Unit = {
-    val isCompact = d.invokePrivate(decimalVal()).eq(null)
-    assert(isCompact == expected, s"$d ${if (expected) "was not" else "was"} compact")
-  }
-
-  test("small decimals represented as unscaled long") {
-    checkCompact(new Decimal(), true)
-    checkCompact(Decimal(BigDecimal(10.03)), false)
-    checkCompact(Decimal(BigDecimal(1e20)), false)
-    checkCompact(Decimal(17L), true)
-    checkCompact(Decimal(17), true)
-    checkCompact(Decimal(17L, 2, 1), true)
-    checkCompact(Decimal(170L, 4, 2), true)
-    checkCompact(Decimal(17L, 24, 1), true)
-    checkCompact(Decimal(1e16.toLong), true)
-    checkCompact(Decimal(1e17.toLong), true)
-    checkCompact(Decimal(1e18.toLong - 1), true)
-    checkCompact(Decimal(- 1e18.toLong + 1), true)
-    checkCompact(Decimal(1e18.toLong - 1, 30, 10), true)
-    checkCompact(Decimal(- 1e18.toLong + 1, 30, 10), true)
-    checkCompact(Decimal(1e18.toLong), false)
-    checkCompact(Decimal(-1e18.toLong), false)
-    checkCompact(Decimal(1e18.toLong, 30, 10), false)
-    checkCompact(Decimal(-1e18.toLong, 30, 10), false)
-    checkCompact(Decimal(Long.MaxValue), false)
-    checkCompact(Decimal(Long.MinValue), false)
+  test("change precision and scale") {
+    assert(true === Decimal(5).changePrecision(1, 0))
+    assert(false === Decimal(15).changePrecision(1, 0))
+    assert(true === Decimal(5).changePrecision(2, 1))
+    assert(false === Decimal(5).changePrecision(2, 2))
+    assert(true === Decimal(0).changePrecision(1, 0))
+    assert(true === Decimal(BigDecimal("10.5")).changePrecision(3, 0))
+    assert(true === Decimal(BigDecimal("10.5")).changePrecision(3, 1))
+    assert(false === Decimal(BigDecimal("10.5")).changePrecision(3, 2))
+    assert(true === Decimal(BigDecimal("10.5")).changePrecision(4, 0))
+    assert(true === Decimal(BigDecimal("10.5")).changePrecision(4, 1))
+    assert(true === Decimal(BigDecimal("10.5")).changePrecision(4, 2))
+    assert(false === Decimal(BigDecimal("10.5")).changePrecision(4, 3))
   }
 
   test("hash code") {
@@ -132,10 +112,6 @@ class DecimalSuite extends SparkFunSuite with PrivateMethodTester {
   }
 
   test("equals") {
-    // The decimals on the left are stored compactly, while the ones on the right aren't
-    checkCompact(Decimal(123), true)
-    checkCompact(Decimal(BigDecimal(123)), false)
-    checkCompact(Decimal("123"), false)
     assert(Decimal(123) === Decimal(BigDecimal(123)))
     assert(Decimal(123) === Decimal(BigDecimal("123.00")))
     assert(Decimal(-123) === Decimal(BigDecimal(-123)))
@@ -187,7 +163,7 @@ class DecimalSuite extends SparkFunSuite with PrivateMethodTester {
     assert(b.toDouble === 0.125)
   }
 
-  test("set/setOrNull") {
+  test("set") {
     assert(new Decimal().set(10L, 10, 0).toUnscaledLong === 10L)
     assert(new Decimal().set(100L, 10, 0).toUnscaledLong === 100L)
     assert(Decimal(Long.MaxValue, 100, 0).toUnscaledLong === Long.MaxValue)