convert datetype for runtime

java.util.{List,Set} => Seq
java.util.Map => Map

but it can not convert Seq into java.util.Set, so set() and tuple()
and array() can not been handled gracefully (back with the original
type).

We can not access items in ArrayType by position, but this is not defined
for set().

Do we still want to support set()/tuple()/array() ?

Author: davies

python/pyspark/sql.py

@@ -49,11 +49,12 @@ def __init__(self, sparkContext, sqlContext=None):
 
         >>> from datetime import datetime
         >>> allTypes = sc.parallelize([{"int": 1, "string": "string", "double": 1.0, "long": 1L,
-        ... "boolean": True, "time": datetime(2010, 1, 1, 1, 1, 1)}])
+        ... "boolean": True, "time": datetime(2010, 1, 1, 1, 1, 1), "dict": {"a": 1},
+        ... "list": [1, 2, 3]}])
         >>> srdd = sqlCtx.inferSchema(allTypes).map(lambda x: (x.int, x.string, x.double, x.long,
-        ... x.boolean, x.time))
+        ... x.boolean, x.time, x.dict["a"], x.list))
         >>> srdd.collect()[0]
-        (1, u'string', 1.0, 1, True, datetime.datetime(2010, 1, 1, 1, 1, 1))
+        (1, u'string', 1.0, 1, True, datetime.datetime(2010, 1, 1, 1, 1, 1), 1, [1, 2, 3])
         """
         self._sc = sparkContext
         self._jsc = self._sc._jsc
@@ -89,13 +90,13 @@ def inferSchema(self, rdd):
 
         >>> from array import array
         >>> srdd = sqlCtx.inferSchema(nestedRdd1)
-        >>> srdd.collect() == [{"f1" : array('i', [1, 2]), "f2" : {"row1" : 1.0}},
-        ...                    {"f1" : array('i', [2, 3]), "f2" : {"row2" : 2.0}}]
+        >>> srdd.collect() == [{"f1" : [1, 2], "f2" : {"row1" : 1.0}},
+        ...                    {"f1" : [2, 3], "f2" : {"row2" : 2.0}}]
         True
 
         >>> srdd = sqlCtx.inferSchema(nestedRdd2)
-        >>> srdd.collect() == [{"f1" : [[1, 2], [2, 3]], "f2" : set([1, 2]), "f3" : (1, 2)},
-        ...                    {"f1" : [[2, 3], [3, 4]], "f2" : set([2, 3]), "f3" : (2, 3)}]
+        >>> srdd.collect() == [{"f1" : [[1, 2], [2, 3]], "f2" : [1, 2], "f3" : [1, 2]},
+        ...                    {"f1" : [[2, 3], [3, 4]], "f2" : [2, 3], "f3" : [2, 3]}]
         True
         """
         if (rdd.__class__ is SchemaRDD):
@@ -510,8 +511,8 @@ def _test():
         {"f1": array('i', [1, 2]), "f2": {"row1": 1.0}},
         {"f1": array('i', [2, 3]), "f2": {"row2": 2.0}}])
     globs['nestedRdd2'] = sc.parallelize([
-        {"f1": [[1, 2], [2, 3]], "f2": set([1, 2]), "f3": (1, 2)},
-        {"f1": [[2, 3], [3, 4]], "f2": set([2, 3]), "f3": (2, 3)}])
+        {"f1": [[1, 2], [2, 3]], "f2": set([1, 2]), "f3": [1, 2]},
+        {"f1": [[2, 3], [3, 4]], "f2": set([2, 3]), "f3": [2, 3]}])
     (failure_count, test_count) = doctest.testmod(globs=globs, optionflags=doctest.ELLIPSIS)
     globs['sc'].stop()
     if failure_count:

sql/core/src/main/scala/org/apache/spark/sql/SQLContext.scala

@@ -369,25 +369,24 @@ class SQLContext(@transient val sparkContext: SparkContext)
     }.toSeq
 
     def needTransform(obj: Any): Boolean = obj match {
-      case c: java.util.List[_] => c.exists(needTransform)
-      case c: java.util.Set[_] => c.exists(needTransform)
-      case c: java.util.Map[_, _] => c.exists {
-        case (key, value) => needTransform(key) || needTransform(value)
-      }
-      case c if c.getClass.isArray =>
-        c.asInstanceOf[Array[_]].exists(needTransform)
+      case c: java.util.List[_] => true
+      case c: java.util.Set[_] => true
+      case c: java.util.Map[_, _] => true
+      case c if c.getClass.isArray => true
       case c: java.util.Calendar => true
       case c => false
     }
 
+    // convert JList, JSet into Seq, convert JMap into Map
+    // convert Calendar into Timestamp
     def transform(obj: Any): Any = obj match {
-      case c: java.util.List[_] => c.map(transform)
-      case c: java.util.Set[_] => c.map(transform)
+      case c: java.util.List[_] => c.map(transform).toSeq
+      case c: java.util.Set[_] => c.map(transform).toSet.toSeq
       case c: java.util.Map[_, _] => c.map {
-        case (key, value) => (transform(key), transform(value))
-      }
+        case (key, value) => (key, transform(value))
+      }.toMap
       case c if c.getClass.isArray =>
-        c.asInstanceOf[Array[_]].map(transform)
+        c.asInstanceOf[Array[_]].map(transform).toSeq
       case c: java.util.Calendar =>
         new java.sql.Timestamp(c.getTime().getTime())
       case c => c

sql/core/src/main/scala/org/apache/spark/sql/SchemaRDD.scala

@@ -32,7 +32,7 @@ import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.plans.{Inner, JoinType}
-import org.apache.spark.sql.catalyst.types.{ArrayType, BooleanType, StructType}
+import org.apache.spark.sql.catalyst.types.{ArrayType, BooleanType, StructType, MapType}
 import org.apache.spark.sql.execution.{ExistingRdd, SparkLogicalPlan}
 import org.apache.spark.api.java.JavaRDD
 
@@ -388,27 +388,30 @@ class SchemaRDD(
                 case seq: Seq[Any] =>
                   seq.map(element => rowToMap(element.asInstanceOf[Row], struct)).asJava
                 case list: JList[_] =>
-                  list.map(element => rowToMap(element.asInstanceOf[Row], struct))
+                  list.map(element => rowToMap(element.asInstanceOf[Row], struct)).asJava
                 case set: JSet[_] =>
-                  set.map(element => rowToMap(element.asInstanceOf[Row], struct))
+                  set.map(element => rowToMap(element.asInstanceOf[Row], struct)).asJava
                 case arr if arr != null && arr.getClass.isArray =>
                   arr.asInstanceOf[Array[Any]].map {
                     element => rowToMap(element.asInstanceOf[Row], struct)
                   }
-                case t: java.sql.Timestamp =>
-                  val c = java.util.Calendar.getInstance()
-                  c.setTimeInMillis(t.getTime())
-                  c
                 case other => other
               }
               map.put(attrName, arrayValues)
+            case m @ MapType(_, struct: StructType) =>
+              val nm = obj.asInstanceOf[Map[_,_]].map {
+                case (k, v) => (k, rowToMap(v.asInstanceOf[Row], struct))
+              }.asJava
+              map.put(attrName, nm)
             case array: ArrayType => {
               val arrayValues = obj match {
                 case seq: Seq[Any] => seq.asJava
                 case other => other
               }
               map.put(attrName, arrayValues)
             }
+            case m: MapType => map.put(attrName, obj.asInstanceOf[Map[_,_]].asJava)
+            // Pyrolite can handle Timestamp
             case other => map.put(attrName, obj)
           }
       }