[SPARK-18122][SQL]Fallback to Kryo for unsupported encoder for class's subfiled

root · root · commit bb11c9349b1a · 2016-11-17T17:16:16.000+08:00
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala
@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.catalyst
 
+import scala.reflect.ClassTag
+
 import org.apache.spark.sql.catalyst.analysis.{GetColumnByOrdinal, UnresolvedAttribute, UnresolvedExtractValue}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.objects._
@@ -403,6 +405,9 @@ object ScalaReflection extends ScalaReflection {
         } else {
           newInstance
         }
+
+      case _ =>
+        DecodeUsingSerializer(getPath, ClassTag(getClassFromType(tpe)), true)
     }
   }
 
@@ -583,8 +588,7 @@ object ScalaReflection extends ScalaReflection {
         expressions.If(IsNull(inputObject), nullOutput, nonNullOutput)
 
       case other =>
-        throw new UnsupportedOperationException(
-          s"No Encoder found for $tpe\n" + walkedTypePath.mkString("\n"))
+        EncodeUsingSerializer(inputObject, true)
     }
 
   }
@@ -701,7 +705,7 @@ object ScalaReflection extends ScalaReflection {
             StructField(fieldName, dataType, nullable)
           }), nullable = true)
       case other =>
-        throw new UnsupportedOperationException(s"Schema for type $other is not supported")
+        Schema(BinaryType, nullable = false)
     }
   }
 
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoderSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoderSuite.scala
@@ -50,6 +50,11 @@ case class NestedArray(a: Array[Array[Int]]) {
   }
 }
 
+case class KryoUnsupportedEncoderForSubFiled(
+    a: String,
+    b: Seq[Int],
+    c: Option[Set[Int]])
+
 case class BoxedData(
     intField: java.lang.Integer,
     longField: java.lang.Long,
@@ -179,6 +184,10 @@ class ExpressionEncoderSuite extends PlanTest with AnalysisTest {
   encodeDecodeTest(new KryoSerializable(15), "kryo object")(
     encoderFor(Encoders.kryo[KryoSerializable]))
 
+  // use kryo to ser/deser the type which has a unsupported Encoder
+  encodeDecodeTest(Seq(KryoUnsupportedEncoderForSubFiled("a", Seq(1), Some(Set(2))),
+    KryoUnsupportedEncoderForSubFiled("b", Seq(3), None)), "type with unsupported encoder,use kryo")
+
   // Java encoders
   encodeDecodeTest("hello", "java string")(encoderFor(Encoders.javaSerialization[String]))
   encodeDecodeTest(new JavaSerializable(15), "java object")(

Original file line number	Diff line number	Diff line change
`@@ -17,6 +17,8 @@`
`17`	`17`
`18`	`18`	`package org.apache.spark.sql.catalyst`
`19`	`19`
	`20`	`+import scala.reflect.ClassTag`
	`21`	`+`
`20`	`22`	`import org.apache.spark.sql.catalyst.analysis.{GetColumnByOrdinal, UnresolvedAttribute, UnresolvedExtractValue}`
`21`	`23`	`import org.apache.spark.sql.catalyst.expressions._`
`22`	`24`	`import org.apache.spark.sql.catalyst.expressions.objects._`
`@@ -403,6 +405,9 @@ object ScalaReflection extends ScalaReflection {`
`403`	`405`	`} else {`
`404`	`406`	`newInstance`
`405`	`407`	`}`
	`408`	`+`
	`409`	`+ case _ =>`
	`410`	`+ DecodeUsingSerializer(getPath, ClassTag(getClassFromType(tpe)), true)`
`406`	`411`	`}`
`407`	`412`	`}`
`408`	`413`
`@@ -583,8 +588,7 @@ object ScalaReflection extends ScalaReflection {`
`583`	`588`	`expressions.If(IsNull(inputObject), nullOutput, nonNullOutput)`
`584`	`589`
`585`	`590`	`case other =>`
`586`		`- throw new UnsupportedOperationException(`
`587`		`- s"No Encoder found for $tpe\n" + walkedTypePath.mkString("\n"))`
	`591`	`+ EncodeUsingSerializer(inputObject, true)`
`588`	`592`	`}`
`589`	`593`
`590`	`594`	`}`
`@@ -701,7 +705,7 @@ object ScalaReflection extends ScalaReflection {`
`701`	`705`	`StructField(fieldName, dataType, nullable)`
`702`	`706`	`}), nullable = true)`
`703`	`707`	`case other =>`
`704`		`- throw new UnsupportedOperationException(s"Schema for type $other is not supported")`
	`708`	`+ Schema(BinaryType, nullable = false)`
`705`	`709`	`}`
`706`	`710`	`}`
`707`	`711`