Implicitly convert Double to Float based on provided schema.

Author: viirya

R/pkg/inst/tests/test_sparkSQL.R

@@ -126,6 +126,14 @@ test_that("create DataFrame from RDD", {
   expect_equal(columns(df2), c("name", "age", "height"))
   expect_equal(dtypes(df2), list(c("name", "string"), c("age", "int"), c("height", "float")))
   expect_equal(collect(where(df2, df2$name == "Bob")), c("Bob", 16, 176.5))
+
+  localDF <- data.frame(name=c("John", "Smith", "Sarah"), age=c(19, 23, 18), height=c(164.10, 181.4, 173.7))
+  df <- createDataFrame(sqlContext, localDF, schema)
+  expect_is(df, "DataFrame")
+  expect_equal(count(df), 3)
+  expect_equal(columns(df), c("name", "age", "height"))
+  expect_equal(dtypes(df), list(c("name", "string"), c("age", "int"), c("height", "float")))
+  expect_equal(collect(where(df, df$name == "John")), c("John", 19, 164.10))
 })
 
 test_that("convert NAs to null type in DataFrames", {

core/src/main/scala/org/apache/spark/api/r/SerDe.scala

@@ -46,9 +46,18 @@ private[spark] object SerDe {
     dis.readByte().toChar
   }
 
-  def readObject(dis: DataInputStream): Object = {
+  def readObject(dis: DataInputStream, typeName: String = ""): Object = {
     val dataType = readObjectType(dis)
-    readTypedObject(dis, dataType)
+    val data = readTypedObject(dis, dataType)
+    doConversion(data, dataType, typeName)
+  }
+
+  def doConversion(data: Object, dataType: Char, typeName: String): Object = {
+    dataType match {
+      case 'd' if typeName == "Float" =>
+        new java.lang.Float(data.asInstanceOf[java.lang.Double])
+      case _ => data
+    }
   }
 
   def readTypedObject(

sql/core/src/main/scala/org/apache/spark/sql/api/r/SQLUtils.scala

@@ -69,20 +69,20 @@ private[r] object SQLUtils {
 
   def createDF(rdd: RDD[Array[Byte]], schema: StructType, sqlContext: SQLContext): DataFrame = {
     val num = schema.fields.size
-    val rowRDD = rdd.map(bytesToRow)
+    val rowRDD = rdd.map(bytesToRow(_, schema))
     sqlContext.createDataFrame(rowRDD, schema)
   }
 
   def dfToRowRDD(df: DataFrame): JavaRDD[Array[Byte]] = {
     df.map(r => rowToRBytes(r))
   }
 
-  private[this] def bytesToRow(bytes: Array[Byte]): Row = {
+  private[this] def bytesToRow(bytes: Array[Byte], schema: StructType): Row = {
     val bis = new ByteArrayInputStream(bytes)
     val dis = new DataInputStream(bis)
     val num = SerDe.readInt(dis)
     Row.fromSeq((0 until num).map { i =>
-      SerDe.readObject(dis)
+      SerDe.readObject(dis, schema.fields(i).dataType.typeName)
     }.toSeq)
   }