[SPARK-5456] [SQL] fix decimal compare for jdbc rdd

Author: Daoyuan Wang <daoyuan.wang@intel.com>

Closes apache#5803 from adrian-wang/decimalcompare and squashes the following commits:

aef0e96 [Daoyuan Wang] add null handle
ec455b9 [Daoyuan Wang] fix decimal compare for jdbc rdd

sql/core/src/main/scala/org/apache/spark/sql/jdbc/JDBCRDD.scala

@@ -363,7 +363,13 @@ private[sql] class JDBCRDD(
             case BooleanConversion    => mutableRow.setBoolean(i, rs.getBoolean(pos))
             case DateConversion       =>
               mutableRow.update(i, DateUtils.fromJavaDate(rs.getDate(pos)))
-            case DecimalConversion    => mutableRow.update(i, rs.getBigDecimal(pos))
+            case DecimalConversion    =>
+              val decimalVal = rs.getBigDecimal(pos)
+              if (decimalVal == null) {
+                mutableRow.update(i, null)
+              } else {
+                mutableRow.update(i, Decimal(decimalVal))
+              }
             case DoubleConversion     => mutableRow.setDouble(i, rs.getDouble(pos))
             case FloatConversion      => mutableRow.setFloat(i, rs.getFloat(pos))
             case IntegerConversion    => mutableRow.setInt(i, rs.getInt(pos))

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -271,8 +271,11 @@ class JDBCSuite extends FunSuite with BeforeAndAfter {
     assert(rows(0).getDouble(0) === 1.00000000000000022) // Yes, I meant ==.
     assert(rows(0).getDouble(1) === 1.00000011920928955) // Yes, I meant ==.
     assert(rows(0).getAs[BigDecimal](2)
-        .equals(new BigDecimal("123456789012345.54321543215432100000")))
+      .equals(new BigDecimal("123456789012345.54321543215432100000")))
     assert(rows(0).schema.fields(2).dataType === DecimalType(40, 20))
+    val compareDecimal = sql("SELECT C FROM flttypes where C > C - 1").collect()
+    assert(compareDecimal(0).getAs[BigDecimal](0)
+      .equals(new BigDecimal("123456789012345.54321543215432100000")))
   }
 
   test("SQL query as table name") {