Github user rayortigas commented on a diff in the pull request:
https://github.com/apache/spark/pull/5713#discussion_r40514432
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala
---
@@ -454,4 +458,166 @@ object CatalystTypeConverters {
def convertToScala(catalystValue: Any, dataType: DataType): Any = {
createToScalaConverter(dataType)(catalystValue)
}
+
+ /**
+ * Like createToScalaConverter(DataType), creates a function that
converts a Catalyst object to a
+ * Scala object; however, in this case, the Scala object is an instance
of a subtype of Product
+ * (e.g. a case class).
+ *
+ * If the given Scala type is not compatible with the given structType,
this method ultimately
+ * throws a ClassCastException when the converter is invoked.
+ *
+ * Typical use case would be converting a collection of rows that have
the same schema. You will
+ * call this function once to get a converter, and apply it to every row.
+ */
+ private[sql] def createToProductConverter[T <: Product](
+ structType: StructType)(implicit classTag: ClassTag[T]): InternalRow
=> T = {
+
+ // Use ScalaReflectionLock, to avoid reflection thread safety issues
in 2.10.
+ // https://issues.scala-lang.org/browse/SI-6240
+ // http://docs.scala-lang.org/overviews/reflection/thread-safety.html
+ ScalaReflectionLock.synchronized { createToProductConverter(classTag,
structType) }
+ }
+
+ private[sql] def createToProductConverter[T <: Product](
+ classTag: ClassTag[T], structType: StructType): InternalRow => T = {
+
+ import universe._
+
+ val constructorMirror = {
+ val mirror = runtimeMirror(Utils.getContextOrSparkClassLoader)
+ val classSymbol = mirror.classSymbol(classTag.runtimeClass)
+ val classMirror = mirror.reflectClass(classSymbol)
+ val constructorSymbol = {
+ // Adapted from ScalaReflection to find primary constructor.
+ // https://issues.apache.org/jira/browse/SPARK-4791
+ val symbol = classSymbol.toType.declaration(nme.CONSTRUCTOR)
+ if (symbol.isMethod) {
+ symbol.asMethod
+ } else {
+ val candidateSymbol =
+ symbol.asTerm.alternatives.find { s => s.isMethod &&
s.asMethod.isPrimaryConstructor }
+ if (candidateSymbol.isDefined) {
+ candidateSymbol.get.asMethod
+ } else {
+ throw new IllegalArgumentException(s"No primary constructor
for ${symbol.name}")
+ }
+ }
+ }
+ classMirror.reflectConstructor(constructorSymbol)
+ }
+
+ val params = constructorMirror.symbol.paramss.head.toSeq
+ val paramTypes = params.map { _.asTerm.typeSignature }
+ val fields = structType.fields
+ val dataTypes = fields.map { _.dataType }
+ val converters: Seq[Any => Any] =
+ paramTypes.zip(dataTypes).map { case (pt, dt) =>
createToScalaConverter(pt, dt) }
+
+ (row: InternalRow) => if (row == null) {
+ null.asInstanceOf[T]
+ } else {
+ val convertedArgs =
+ converters.zip(row.toSeq(dataTypes)).map { case (converter, arg)
=> converter(arg) }
+ try {
+ constructorMirror.apply(convertedArgs: _*).asInstanceOf[T]
+ } catch {
+ case e: IllegalArgumentException => // argument type mismatch
+ val message =
+ s"""|Error constructing ${classTag.runtimeClass.getName}:
${e.getMessage};
+ |paramTypes: ${paramTypes}, dataTypes: ${dataTypes},
+ |convertedArgs:
${convertedArgs}""".stripMargin.replace("\n", " ")
--- End diff --
Addressing feedback, this error message provides more details when things
go wrong calling the constructor.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
