Add UserDefinedType to Cast.

Author: viirya

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala

@@ -81,6 +81,9 @@ object Cast {
                 toField.nullable)
         }
 
+    case (udt1: UserDefinedType[_], udt2: UserDefinedType[_]) if udt1.userClass == udt2.userClass =>
+      true
+
     case _ => false
   }
 
@@ -473,6 +476,9 @@ case class Cast(child: Expression, dataType: DataType) extends UnaryExpression {
       castArrayCode(from.asInstanceOf[ArrayType].elementType, array.elementType, ctx)
     case map: MapType => castMapCode(from.asInstanceOf[MapType], map, ctx)
     case struct: StructType => castStructCode(from.asInstanceOf[StructType], struct, ctx)
+    case udt: UserDefinedType[_]
+      if udt.userClass == from.asInstanceOf[UserDefinedType[_]].userClass =>
+      (c, evPrim, evNull) => s"$evPrim = $c;"
   }
 
   // Since we need to cast child expressions recursively inside ComplexTypes, such as Map's

sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala

@@ -17,11 +17,15 @@
 
 package org.apache.spark.sql
 
+import java.util.concurrent.ConcurrentMap
+
 import org.apache.spark.sql.catalyst.util.{GenericArrayData, ArrayData}
 
 import scala.beans.{BeanInfo, BeanProperty}
 import scala.reflect.runtime.universe.TypeTag
 
+import com.google.common.collect.MapMaker
+
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.{CatalystTypeConverters, ScalaReflection}
 import org.apache.spark.sql.catalyst.encoders._
@@ -94,6 +98,9 @@ class UserDefinedTypeSuite extends QueryTest with SharedSQLContext with ParquetT
     assert(featuresArrays.contains(new MyDenseVector(Array(0.2, 2.0))))
   }
 
+  private val outers: ConcurrentMap[String, AnyRef] = new MapMaker().weakValues().makeMap()
+  outers.put(getClass.getName, this)
+
   test("user type with ScalaReflection") {
     val points = Seq(
       MyLabeledPoint(1.0, new MyDenseVector(Array(0.1, 1.0))),
@@ -109,6 +116,10 @@ class UserDefinedTypeSuite extends QueryTest with SharedSQLContext with ParquetT
     points.zip(decodedPoints).foreach { case (p, p2) =>
       assert(p.label == p2(0) && p.features == p2(1))
     }
+
+    val boundEncoder = pointEncoder.resolve(attributeSeq, outers).bind(attributeSeq)
+    val point = MyLabeledPoint(1.0, new MyDenseVector(Array(0.1, 1.0)))
+    assert(boundEncoder.fromRow(boundEncoder.toRow(point)) === point)
   }
 
   test("UDTs and UDFs") {