fix bug in StructType

Author: Davies Liu

python/pyspark/sql/context.py

@@ -295,6 +295,7 @@ def _createFromRDD(self, rdd, schema, samplingRatio):
             if isinstance(schema, (list, tuple)):
                 for i, name in enumerate(schema):
                     struct.fields[i].name = name
+                    struct.names[i] = name
             schema = struct
 
         elif isinstance(schema, StructType):
@@ -325,6 +326,7 @@ def _createFromLocal(self, data, schema):
             if isinstance(schema, (list, tuple)):
                 for i, name in enumerate(schema):
                     struct.fields[i].name = name
+                    struct.names[i] = name
             schema = struct
 
         elif isinstance(schema, StructType):

python/pyspark/sql/types.py

@@ -455,7 +455,7 @@ def __init__(self, fields=None):
             self.names = [f.name for f in fields]
             assert all(isinstance(f, StructField) for f in fields),\
                 "fields should be a list of StructField"
-        self._needSerializeFields = None
+        self._needSerializeAnyField = any(f.needConversion() for f in self.fields)
 
     def add(self, field, data_type=None, nullable=True, metadata=None):
         """
@@ -498,6 +498,7 @@ def add(self, field, data_type=None, nullable=True, metadata=None):
                 data_type_f = data_type
             self.fields.append(StructField(field, data_type_f, nullable, metadata))
             self.names.append(field)
+        self._needSerializeAnyField = any(f.needConversion() for f in self.fields)
         return self
 
     def simpleString(self):
@@ -523,12 +524,9 @@ def toInternal(self, obj):
         if obj is None:
             return
 
-        if self._needSerializeFields is None:
-            self._needSerializeFields = any(f.needConversion() for f in self.fields)
-
-        if self._needSerializeFields:
+        if self._needSerializeAnyField:
             if isinstance(obj, dict):
-                return tuple(f.toInternal(obj.get(n)) for n, f in zip(names, self.fields))
+                return tuple(f.toInternal(obj.get(n)) for n, f in zip(self.names, self.fields))
             elif isinstance(obj, (tuple, list)):
                 return tuple(f.toInternal(v) for f, v in zip(self.fields, obj))
             else:
@@ -547,7 +545,10 @@ def fromInternal(self, obj):
         if isinstance(obj, Row):
             # it's already converted by pickler
             return obj
-        values = [f.dataType.fromInternal(v) for f, v in zip(self.fields, obj)]
+        if self._needSerializeAnyField:
+            values = [f.fromInternal(v) for f, v in zip(self.fields, obj)]
+        else:
+            values = obj
         return _create_row(self.names, values)
 
 

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

@@ -89,15 +89,16 @@ abstract class UserDefinedType[UserType] extends DataType with Serializable {
  *
  * Note: This can only be accessed via Python UDF, or accessed as serialized object.
  */
-class PythonUserDefinedType(val sqlType: DataType, pyClass: String) extends UserDefinedType[Any] {
-  override def pyUDT = pyClass
-  def serialize(obj: Any): Any = obj
-  def deserialize(datam: Any): Any = datam
-  override private[sql] def jsonValue: JValue = {
-    ("type" -> "udt") ~
-      ("class" -> "") ~
-      ("pyClass" -> pyUDT) ~
-      ("sqlType" -> sqlType.jsonValue)
-  }
-  def userClass: java.lang.Class[Any] = null
+private[sql] class PythonUserDefinedType(val sqlType: DataType, pyClass: String)
+    extends UserDefinedType[Any] {
+
+  /* The Python UDT class */
+  override def pyUDT: String = pyClass
+
+  /* The serialization is handled by UDT class in Python */
+  override def serialize(obj: Any): Any = obj
+  override def deserialize(datam: Any): Any = datam
+
+  /* There is no Java class for Python UDT */
+  override def userClass: java.lang.Class[Any] = null
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/pythonUDFs.scala

@@ -271,7 +271,6 @@ object EvaluatePython {
           pickler.save(row.values(i))
           i += 1
         }
-        row.values.foreach(pickler.save)
         out.write(Opcodes.TUPLE)
         out.write(Opcodes.REDUCE)
       }