[SPARK-8677] [SQL] Fix non-terminating decimal expansion for decimal divide operation

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

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

Closes apache#7056 from viirya/fix_decimal3 and squashes the following commits:

34d7419 [Liang-Chi Hsieh] Fix Non-terminating decimal expansion for decimal divide operation.

Author: viirya

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

@@ -265,8 +265,15 @@ final class Decimal extends Ordered[Decimal] with Serializable {
 
   def * (that: Decimal): Decimal = Decimal(toBigDecimal * that.toBigDecimal)
 
-  def / (that: Decimal): Decimal =
-    if (that.isZero) null else Decimal(toBigDecimal / that.toBigDecimal)
+  def / (that: Decimal): Decimal = {
+    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))
+    }
+  }
 
   def % (that: Decimal): Decimal =
     if (that.isZero) null else Decimal(toBigDecimal % that.toBigDecimal)

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

@@ -167,4 +167,9 @@ class DecimalSuite extends SparkFunSuite with PrivateMethodTester {
     val decimal = (Decimal(Long.MaxValue, 38, 0) * Decimal(Long.MaxValue, 38, 0)).toJavaBigDecimal
     assert(decimal.unscaledValue.toString === "85070591730234615847396907784232501249")
   }
+
+  test("fix non-terminating decimal expansion problem") {
+    val decimal = Decimal(1.0, 10, 3) / Decimal(3.0, 10, 3)
+    assert(decimal.toString === "0.333")
+  }
 }