[SPARK-8800] [SQL] Fix inaccurate precision/scale of Decimal division operation

JIRA: https://issues.apache.org/jira/browse/SPARK-8800

Previously, we turn to Java BigDecimal's divide with specified ROUNDING_MODE to avoid non-terminating decimal expansion problem. However, as JihongMA reported, for the division operation on some specific values, we get inaccurate results.

Author: Liang-Chi Hsieh <[email protected]>

Closes apache#7212 from viirya/fix_decimal4 and squashes the following commits:

4205a0a [Liang-Chi Hsieh] Fix inaccuracy precision/scale of Decimal division operation.

Author: viirya

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

@@ -145,6 +145,14 @@ final class Decimal extends Ordered[Decimal] with Serializable {
     }
   }
 
+  def toLimitedBigDecimal: BigDecimal = {
+    if (decimalVal.ne(null)) {
+      decimalVal
+    } else {
+      BigDecimal(longVal, _scale)
+    }
+  }
+
   def toJavaBigDecimal: java.math.BigDecimal = toBigDecimal.underlying()
 
   def toUnscaledLong: Long = {
@@ -269,9 +277,9 @@ final class Decimal extends Ordered[Decimal] with Serializable {
     if (that.isZero) {
       null
     } else {
-      // To avoid non-terminating decimal expansion problem, we turn to Java BigDecimal's divide
-      // with specified ROUNDING_MODE.
-      Decimal(toJavaBigDecimal.divide(that.toJavaBigDecimal, ROUNDING_MODE.id))
+      // To avoid non-terminating decimal expansion problem, we get scala's BigDecimal with limited
+      // precision and scala.
+      Decimal(toLimitedBigDecimal / that.toLimitedBigDecimal)
     }
   }
 

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

@@ -170,6 +170,14 @@ class DecimalSuite extends SparkFunSuite with PrivateMethodTester {
 
   test("fix non-terminating decimal expansion problem") {
     val decimal = Decimal(1.0, 10, 3) / Decimal(3.0, 10, 3)
-    assert(decimal.toString === "0.333")
+    // The difference between decimal should not be more than 0.001.
+    assert(decimal.toDouble - 0.333 < 0.001)
+  }
+
+  test("fix loss of precision/scale when doing division operation") {
+    val a = Decimal(2) / Decimal(3)
+    assert(a.toDouble < 1.0 && a.toDouble > 0.6)
+    val b = Decimal(1) / Decimal(8)
+    assert(b.toDouble === 0.125)
   }
 }