RowSerializer: Safe guard against double types with Decimal schema tags (#799)

Author: pfcoperez

Diff for: common/src/main/scala/com/stratio/crossdata/common/serializers/RowSerializer.scala

@@ -95,6 +95,8 @@ case class RowSerializer(providedSchema: StructType) extends Serializer[Row] {
       case (DoubleType, v: Double) => JDouble(v)
       case (LongType, v: Long) => JInt(v)
       case (_: DecimalType, v: Decimal) => JDecimal(v.toBigDecimal)
+      case (_: DecimalType, v: Double) => JDecimal(BigDecimal(v))
+      case (_: DecimalType, v: Float) => JDecimal(BigDecimal(v))
       case (ByteType, v: Byte) => JInt(v.toInt)
       case (BinaryType, v: Array[Byte]) => JString(new String(v))
       case (BooleanType, v: Boolean) => JBool(v)

Diff for: common/src/test/scala/com/stratio/crossdata/common/serializers/RowSerializerSpec.scala

@@ -20,13 +20,16 @@ import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.json4s.Extraction
+import org.json4s.jackson.JsonMethods.{compact, parse, render}
 import org.junit.runner.RunWith
+import org.scalatest.Inside
 import org.scalatest.junit.JUnitRunner
 
 import scala.collection.mutable.WrappedArray
 
 @RunWith(classOf[JUnitRunner])
-class RowSerializerSpec extends XDSerializationTest[Row] with CrossdataCommonSerializer {
+class RowSerializerSpec extends XDSerializationTest[Row] with CrossdataCommonSerializer with Inside {
 
   lazy val schema = StructType(List(
     StructField("int",IntegerType,true),
@@ -112,4 +115,20 @@ class RowSerializerSpec extends XDSerializationTest[Row] with CrossdataCommonSer
     TestCase("marshall & unmarshall a row with schema", rowWithSchema)
   )
 
+  it should " be able to recover Double values when their schema type is misleading" in {
+
+    val schema = StructType(List(StructField("decimaldouble", DecimalType(10,1),true)))
+    val row = Row.fromSeq(Array(32.1))
+
+    val formats = json4sJacksonFormats + new RowSerializer(schema)
+
+    val serialized = compact(render(Extraction.decompose(row)(formats)))
+    val extracted = parse(serialized, false).extract[Row](formats, implicitly[Manifest[Row]])
+
+    inside(extracted) {
+      case r: Row => r.get(0) shouldBe Decimal(32.1)
+    }
+
+  }
+
 }