You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ahshahid (via GitHub)" <gi...@apache.org> on 2024/09/19 04:02:23 UTC
[PR] [WIP][SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
ahshahid opened a new pull request, #48158:
URL: https://github.com/apache/spark/pull/48158
### What changes were proposed in this pull request?
If a bean is parameterized with generic type having bounds and that generic type happens to be a getter/setter of the bean, then depending upon the nature of the bounds, if it is java Serializable, KryoSerializable or a UDT Type, then appropriate encoder is created. If the bound is of 1st two types, then the data is represented as a BinaryType.
More details will follow
### Why are the changes needed?
To fix the regression in spark 3.3 onwards, where the bean having a generic type as return value, throws EncoderNotFoundException.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Added bug tests. More tests will be added
### Was this patch authored or co-authored using generative AI tooling?
No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771700787
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
+ Encoders.kryo(c).asInstanceOf[AgnosticEncoder[_]]
+ } else if (classOf[java.io.Serializable].isAssignableFrom(c)) {
+ Encoders.javaSerialization(c).asInstanceOf[AgnosticEncoder[_]]
+ } else {
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ } else {
+ // add type variables from inheritance hierarchy of the class
+ val parentClassesTypeMap =
+ JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap
+ val classTV = parentClassesTypeMap ++ typeVariables
+ // Note that the fields are ordered by name.
+ val fields = properties.map { property =>
+ val readMethod = property.getReadMethod
+ val methodReturnType = readMethod.getGenericReturnType
+ val encoder = encoderFor(methodReturnType, seenTypeSet + c, classTV)
+ // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
+ val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
+ EncoderField(
+ property.getName,
+ encoder,
+ encoder.nullable && !hasNonNull,
+ Metadata.empty,
+ Option(readMethod.getName),
+ Option(property.getWriteMethod).map(_.getName))
+ }
+ // implies it cannot be assumed a BeanClass.
+ // Check if its super class or interface could be represented by an Encoder
+
+ JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
}
- JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
- case _ =>
- throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
+ case t =>
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
}
+ private def getAllSuperClasses(typee: Type): Array[Class[_]] = Option(typee)
Review Comment:
This sort of begs for a queue and a loop. I am reasonable sure that is more readable...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771988876
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -66,37 +70,46 @@ object JavaTypeInference {
private def encoderFor(
t: Type,
seenTypeSet: Set[Class[_]],
- typeVariables: Map[TypeVariable[_], Type] = Map.empty): AgnosticEncoder[_] = t match {
-
- case c: Class[_] if c == java.lang.Boolean.TYPE => PrimitiveBooleanEncoder
- case c: Class[_] if c == java.lang.Byte.TYPE => PrimitiveByteEncoder
- case c: Class[_] if c == java.lang.Short.TYPE => PrimitiveShortEncoder
- case c: Class[_] if c == java.lang.Integer.TYPE => PrimitiveIntEncoder
- case c: Class[_] if c == java.lang.Long.TYPE => PrimitiveLongEncoder
- case c: Class[_] if c == java.lang.Float.TYPE => PrimitiveFloatEncoder
- case c: Class[_] if c == java.lang.Double.TYPE => PrimitiveDoubleEncoder
-
- case c: Class[_] if c == classOf[java.lang.Boolean] => BoxedBooleanEncoder
- case c: Class[_] if c == classOf[java.lang.Byte] => BoxedByteEncoder
- case c: Class[_] if c == classOf[java.lang.Short] => BoxedShortEncoder
- case c: Class[_] if c == classOf[java.lang.Integer] => BoxedIntEncoder
- case c: Class[_] if c == classOf[java.lang.Long] => BoxedLongEncoder
- case c: Class[_] if c == classOf[java.lang.Float] => BoxedFloatEncoder
- case c: Class[_] if c == classOf[java.lang.Double] => BoxedDoubleEncoder
-
- case c: Class[_] if c == classOf[java.lang.String] => StringEncoder
- case c: Class[_] if c == classOf[Array[Byte]] => BinaryEncoder
- case c: Class[_] if c == classOf[java.math.BigDecimal] => DEFAULT_JAVA_DECIMAL_ENCODER
- case c: Class[_] if c == classOf[java.math.BigInteger] => JavaBigIntEncoder
- case c: Class[_] if c == classOf[java.time.LocalDate] => STRICT_LOCAL_DATE_ENCODER
- case c: Class[_] if c == classOf[java.sql.Date] => STRICT_DATE_ENCODER
- case c: Class[_] if c == classOf[java.time.Instant] => STRICT_INSTANT_ENCODER
- case c: Class[_] if c == classOf[java.sql.Timestamp] => STRICT_TIMESTAMP_ENCODER
- case c: Class[_] if c == classOf[java.time.LocalDateTime] => LocalDateTimeEncoder
- case c: Class[_] if c == classOf[java.time.Duration] => DayTimeIntervalEncoder
- case c: Class[_] if c == classOf[java.time.Period] => YearMonthIntervalEncoder
-
- case c: Class[_] if c.isEnum => JavaEnumEncoder(ClassTag(c))
+ typeVariables: Map[TypeVariable[_], Type] = Map.empty,
+ forGenericBound: Boolean = false): AgnosticEncoder[_] = t match {
+
+ case c: Class[_] if !forGenericBound && c == java.lang.Boolean.TYPE => PrimitiveBooleanEncoder
Review Comment:
I have sort of done it now..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771987465
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
+ Encoders.kryo(c).asInstanceOf[AgnosticEncoder[_]]
+ } else if (classOf[java.io.Serializable].isAssignableFrom(c)) {
+ Encoders.javaSerialization(c).asInstanceOf[AgnosticEncoder[_]]
+ } else {
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ } else {
+ // add type variables from inheritance hierarchy of the class
+ val parentClassesTypeMap =
+ JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap
+ val classTV = parentClassesTypeMap ++ typeVariables
+ // Note that the fields are ordered by name.
+ val fields = properties.map { property =>
+ val readMethod = property.getReadMethod
+ val methodReturnType = readMethod.getGenericReturnType
+ val encoder = encoderFor(methodReturnType, seenTypeSet + c, classTV)
+ // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
+ val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
+ EncoderField(
+ property.getName,
+ encoder,
+ encoder.nullable && !hasNonNull,
+ Metadata.empty,
+ Option(readMethod.getName),
+ Option(property.getWriteMethod).map(_.getName))
+ }
+ // implies it cannot be assumed a BeanClass.
+ // Check if its super class or interface could be represented by an Encoder
+
+ JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
}
- JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
- case _ =>
- throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
+ case t =>
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
}
+ private def getAllSuperClasses(typee: Type): Array[Class[_]] = Option(typee)
+ .map({
+ case clazz: Class[_] =>
+ val currentInterfaces = clazz.getInterfaces
+ currentInterfaces ++ (clazz.getSuperclass +: currentInterfaces).flatMap(clz =>
+ getAllSuperClasses(clz)) :+ clazz
+
+ case tv: TypeVariable[_] =>
+ tv.getBounds.foldLeft(Array.empty[Class[_]]) { case (res, bnd) =>
+ res ++ getAllSuperClasses(bnd)
+ }
+ })
+ .getOrElse(Array.empty)
+
def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = {
val beanInfo = Introspector.getBeanInfo(beanClass)
beanInfo.getPropertyDescriptors
.filterNot(_.getName == "class")
.filterNot(_.getName == "declaringClass")
.filter(_.getReadMethod != null)
}
+
+ object UseSerializationEncoder {
+ def unapply(th: Throwable): Option[Class[_] => AgnosticEncoder[_]] = th match {
Review Comment:
I had thought about using the Serializable encoders as part of the match/case statement, but it appears tricky as the idea is that the Serializable encoders are to be used as last resort. And the way current logic works ( if I am not wrong), it collects all the interfaces/classes via the CommonUtils method, and it is possible , I think, that Serializable interface may show early as the introspection data is converted to a map in the Bean class
val parentClassesTypeMap =
JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap
I am hesitant at this point to further complicate the code, unless you all think that its worthwhile to do it in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771980203
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
+ Encoders.kryo(c).asInstanceOf[AgnosticEncoder[_]]
+ } else if (classOf[java.io.Serializable].isAssignableFrom(c)) {
+ Encoders.javaSerialization(c).asInstanceOf[AgnosticEncoder[_]]
+ } else {
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ } else {
+ // add type variables from inheritance hierarchy of the class
+ val parentClassesTypeMap =
+ JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap
+ val classTV = parentClassesTypeMap ++ typeVariables
+ // Note that the fields are ordered by name.
+ val fields = properties.map { property =>
+ val readMethod = property.getReadMethod
+ val methodReturnType = readMethod.getGenericReturnType
+ val encoder = encoderFor(methodReturnType, seenTypeSet + c, classTV)
+ // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
+ val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
+ EncoderField(
+ property.getName,
+ encoder,
+ encoder.nullable && !hasNonNull,
+ Metadata.empty,
+ Option(readMethod.getName),
+ Option(property.getWriteMethod).map(_.getName))
+ }
+ // implies it cannot be assumed a BeanClass.
+ // Check if its super class or interface could be represented by an Encoder
+
+ JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
}
- JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
- case _ =>
- throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
+ case t =>
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
}
+ private def getAllSuperClasses(typee: Type): Array[Class[_]] = Option(typee)
Review Comment:
@hvanhovell Done. Kindly check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771680158
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -66,37 +70,46 @@ object JavaTypeInference {
private def encoderFor(
t: Type,
seenTypeSet: Set[Class[_]],
- typeVariables: Map[TypeVariable[_], Type] = Map.empty): AgnosticEncoder[_] = t match {
-
- case c: Class[_] if c == java.lang.Boolean.TYPE => PrimitiveBooleanEncoder
- case c: Class[_] if c == java.lang.Byte.TYPE => PrimitiveByteEncoder
- case c: Class[_] if c == java.lang.Short.TYPE => PrimitiveShortEncoder
- case c: Class[_] if c == java.lang.Integer.TYPE => PrimitiveIntEncoder
- case c: Class[_] if c == java.lang.Long.TYPE => PrimitiveLongEncoder
- case c: Class[_] if c == java.lang.Float.TYPE => PrimitiveFloatEncoder
- case c: Class[_] if c == java.lang.Double.TYPE => PrimitiveDoubleEncoder
-
- case c: Class[_] if c == classOf[java.lang.Boolean] => BoxedBooleanEncoder
- case c: Class[_] if c == classOf[java.lang.Byte] => BoxedByteEncoder
- case c: Class[_] if c == classOf[java.lang.Short] => BoxedShortEncoder
- case c: Class[_] if c == classOf[java.lang.Integer] => BoxedIntEncoder
- case c: Class[_] if c == classOf[java.lang.Long] => BoxedLongEncoder
- case c: Class[_] if c == classOf[java.lang.Float] => BoxedFloatEncoder
- case c: Class[_] if c == classOf[java.lang.Double] => BoxedDoubleEncoder
-
- case c: Class[_] if c == classOf[java.lang.String] => StringEncoder
- case c: Class[_] if c == classOf[Array[Byte]] => BinaryEncoder
- case c: Class[_] if c == classOf[java.math.BigDecimal] => DEFAULT_JAVA_DECIMAL_ENCODER
- case c: Class[_] if c == classOf[java.math.BigInteger] => JavaBigIntEncoder
- case c: Class[_] if c == classOf[java.time.LocalDate] => STRICT_LOCAL_DATE_ENCODER
- case c: Class[_] if c == classOf[java.sql.Date] => STRICT_DATE_ENCODER
- case c: Class[_] if c == classOf[java.time.Instant] => STRICT_INSTANT_ENCODER
- case c: Class[_] if c == classOf[java.sql.Timestamp] => STRICT_TIMESTAMP_ENCODER
- case c: Class[_] if c == classOf[java.time.LocalDateTime] => LocalDateTimeEncoder
- case c: Class[_] if c == classOf[java.time.Duration] => DayTimeIntervalEncoder
- case c: Class[_] if c == classOf[java.time.Period] => YearMonthIntervalEncoder
-
- case c: Class[_] if c.isEnum => JavaEnumEncoder(ClassTag(c))
+ typeVariables: Map[TypeVariable[_], Type] = Map.empty,
+ forGenericBound: Boolean = false): AgnosticEncoder[_] = t match {
+
+ case c: Class[_] if !forGenericBound && c == java.lang.Boolean.TYPE => PrimitiveBooleanEncoder
Review Comment:
For readability purposes I would create a branch at the beginning where you handle `case tv: TypeVariable[_] if forGenericBound =>` This way the rest of the code is less impacted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771703915
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
+ Encoders.kryo(c).asInstanceOf[AgnosticEncoder[_]]
+ } else if (classOf[java.io.Serializable].isAssignableFrom(c)) {
+ Encoders.javaSerialization(c).asInstanceOf[AgnosticEncoder[_]]
+ } else {
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ } else {
+ // add type variables from inheritance hierarchy of the class
+ val parentClassesTypeMap =
+ JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap
+ val classTV = parentClassesTypeMap ++ typeVariables
+ // Note that the fields are ordered by name.
+ val fields = properties.map { property =>
+ val readMethod = property.getReadMethod
+ val methodReturnType = readMethod.getGenericReturnType
+ val encoder = encoderFor(methodReturnType, seenTypeSet + c, classTV)
+ // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
+ val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
+ EncoderField(
+ property.getName,
+ encoder,
+ encoder.nullable && !hasNonNull,
+ Metadata.empty,
+ Option(readMethod.getName),
+ Option(property.getWriteMethod).map(_.getName))
+ }
+ // implies it cannot be assumed a BeanClass.
+ // Check if its super class or interface could be represented by an Encoder
+
+ JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
}
- JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
- case _ =>
- throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
+ case t =>
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
}
+ private def getAllSuperClasses(typee: Type): Array[Class[_]] = Option(typee)
+ .map({
+ case clazz: Class[_] =>
+ val currentInterfaces = clazz.getInterfaces
+ currentInterfaces ++ (clazz.getSuperclass +: currentInterfaces).flatMap(clz =>
+ getAllSuperClasses(clz)) :+ clazz
+
+ case tv: TypeVariable[_] =>
+ tv.getBounds.foldLeft(Array.empty[Class[_]]) { case (res, bnd) =>
+ res ++ getAllSuperClasses(bnd)
+ }
+ })
+ .getOrElse(Array.empty)
+
def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = {
val beanInfo = Introspector.getBeanInfo(beanClass)
beanInfo.getPropertyDescriptors
.filterNot(_.getName == "class")
.filterNot(_.getName == "declaringClass")
.filter(_.getReadMethod != null)
}
+
+ object UseSerializationEncoder {
+ def unapply(th: Throwable): Option[Class[_] => AgnosticEncoder[_]] = th match {
Review Comment:
Can we use the actual superclasses here instead of going through the error message? I'd also prefer if you unify this with the other java serialization code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771692208
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -66,37 +70,46 @@ object JavaTypeInference {
private def encoderFor(
t: Type,
seenTypeSet: Set[Class[_]],
- typeVariables: Map[TypeVariable[_], Type] = Map.empty): AgnosticEncoder[_] = t match {
-
- case c: Class[_] if c == java.lang.Boolean.TYPE => PrimitiveBooleanEncoder
- case c: Class[_] if c == java.lang.Byte.TYPE => PrimitiveByteEncoder
- case c: Class[_] if c == java.lang.Short.TYPE => PrimitiveShortEncoder
- case c: Class[_] if c == java.lang.Integer.TYPE => PrimitiveIntEncoder
- case c: Class[_] if c == java.lang.Long.TYPE => PrimitiveLongEncoder
- case c: Class[_] if c == java.lang.Float.TYPE => PrimitiveFloatEncoder
- case c: Class[_] if c == java.lang.Double.TYPE => PrimitiveDoubleEncoder
-
- case c: Class[_] if c == classOf[java.lang.Boolean] => BoxedBooleanEncoder
- case c: Class[_] if c == classOf[java.lang.Byte] => BoxedByteEncoder
- case c: Class[_] if c == classOf[java.lang.Short] => BoxedShortEncoder
- case c: Class[_] if c == classOf[java.lang.Integer] => BoxedIntEncoder
- case c: Class[_] if c == classOf[java.lang.Long] => BoxedLongEncoder
- case c: Class[_] if c == classOf[java.lang.Float] => BoxedFloatEncoder
- case c: Class[_] if c == classOf[java.lang.Double] => BoxedDoubleEncoder
-
- case c: Class[_] if c == classOf[java.lang.String] => StringEncoder
- case c: Class[_] if c == classOf[Array[Byte]] => BinaryEncoder
- case c: Class[_] if c == classOf[java.math.BigDecimal] => DEFAULT_JAVA_DECIMAL_ENCODER
- case c: Class[_] if c == classOf[java.math.BigInteger] => JavaBigIntEncoder
- case c: Class[_] if c == classOf[java.time.LocalDate] => STRICT_LOCAL_DATE_ENCODER
- case c: Class[_] if c == classOf[java.sql.Date] => STRICT_DATE_ENCODER
- case c: Class[_] if c == classOf[java.time.Instant] => STRICT_INSTANT_ENCODER
- case c: Class[_] if c == classOf[java.sql.Timestamp] => STRICT_TIMESTAMP_ENCODER
- case c: Class[_] if c == classOf[java.time.LocalDateTime] => LocalDateTimeEncoder
- case c: Class[_] if c == classOf[java.time.Duration] => DayTimeIntervalEncoder
- case c: Class[_] if c == classOf[java.time.Period] => YearMonthIntervalEncoder
-
- case c: Class[_] if c.isEnum => JavaEnumEncoder(ClassTag(c))
+ typeVariables: Map[TypeVariable[_], Type] = Map.empty,
+ forGenericBound: Boolean = false): AgnosticEncoder[_] = t match {
+
+ case c: Class[_] if !forGenericBound && c == java.lang.Boolean.TYPE => PrimitiveBooleanEncoder
Review Comment:
Thanks. I will do that. was thinking how to do that.. . This neat idea did not struck to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid closed pull request #48158: [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders
URL: https://github.com/apache/spark/pull/48158
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771704443
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
+ Encoders.kryo(c).asInstanceOf[AgnosticEncoder[_]]
+ } else if (classOf[java.io.Serializable].isAssignableFrom(c)) {
+ Encoders.javaSerialization(c).asInstanceOf[AgnosticEncoder[_]]
+ } else {
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ } else {
+ // add type variables from inheritance hierarchy of the class
+ val parentClassesTypeMap =
+ JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap
+ val classTV = parentClassesTypeMap ++ typeVariables
+ // Note that the fields are ordered by name.
+ val fields = properties.map { property =>
+ val readMethod = property.getReadMethod
+ val methodReturnType = readMethod.getGenericReturnType
+ val encoder = encoderFor(methodReturnType, seenTypeSet + c, classTV)
+ // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
+ val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
+ EncoderField(
+ property.getName,
+ encoder,
+ encoder.nullable && !hasNonNull,
+ Metadata.empty,
+ Option(readMethod.getName),
+ Option(property.getWriteMethod).map(_.getName))
+ }
+ // implies it cannot be assumed a BeanClass.
+ // Check if its super class or interface could be represented by an Encoder
+
+ JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
}
- JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
- case _ =>
- throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
+ case t =>
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
}
+ private def getAllSuperClasses(typee: Type): Array[Class[_]] = Option(typee)
Review Comment:
will flatten the recursion
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771693608
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
Review Comment:
sure. I was not aware of that.. will write test for it and check.
will be checking in a code with some refactoring..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771708744
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
+ Encoders.kryo(c).asInstanceOf[AgnosticEncoder[_]]
+ } else if (classOf[java.io.Serializable].isAssignableFrom(c)) {
+ Encoders.javaSerialization(c).asInstanceOf[AgnosticEncoder[_]]
+ } else {
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ } else {
+ // add type variables from inheritance hierarchy of the class
+ val parentClassesTypeMap =
+ JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap
+ val classTV = parentClassesTypeMap ++ typeVariables
+ // Note that the fields are ordered by name.
+ val fields = properties.map { property =>
+ val readMethod = property.getReadMethod
+ val methodReturnType = readMethod.getGenericReturnType
+ val encoder = encoderFor(methodReturnType, seenTypeSet + c, classTV)
+ // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
+ val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
+ EncoderField(
+ property.getName,
+ encoder,
+ encoder.nullable && !hasNonNull,
+ Metadata.empty,
+ Option(readMethod.getName),
+ Option(property.getWriteMethod).map(_.getName))
+ }
+ // implies it cannot be assumed a BeanClass.
+ // Check if its super class or interface could be represented by an Encoder
+
+ JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
}
- JavaBeanEncoder(ClassTag(c), fields.toImmutableArraySeq)
- case _ =>
- throw ExecutionErrors.cannotFindEncoderForTypeError(t.toString)
+ case t =>
+ throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
}
+ private def getAllSuperClasses(typee: Type): Array[Class[_]] = Option(typee)
+ .map({
+ case clazz: Class[_] =>
+ val currentInterfaces = clazz.getInterfaces
+ currentInterfaces ++ (clazz.getSuperclass +: currentInterfaces).flatMap(clz =>
+ getAllSuperClasses(clz)) :+ clazz
+
+ case tv: TypeVariable[_] =>
+ tv.getBounds.foldLeft(Array.empty[Class[_]]) { case (res, bnd) =>
+ res ++ getAllSuperClasses(bnd)
+ }
+ })
+ .getOrElse(Array.empty)
+
def getJavaBeanReadableProperties(beanClass: Class[_]): Array[PropertyDescriptor] = {
val beanInfo = Introspector.getBeanInfo(beanClass)
beanInfo.getPropertyDescriptors
.filterNot(_.getName == "class")
.filterNot(_.getName == "declaringClass")
.filter(_.getReadMethod != null)
}
+
+ object UseSerializationEncoder {
+ def unapply(th: Throwable): Option[Class[_] => AgnosticEncoder[_]] = th match {
Review Comment:
Unifying it will other java serialziation code, is something which I intend to do in my refactoring..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1771684833
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
Review Comment:
This will be an issue for Connect. While the API supports Kryo, Connect can't support Kryo in its current form. Either we have detect whether we are in connect mode, or we have to just fall back to java serialization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-46679][SQL]: Handling of generic parameter with bounds while creating encoders [spark]
Posted by "ahshahid (via GitHub)" <gi...@apache.org>.
ahshahid commented on code in PR #48158:
URL: https://github.com/apache/spark/pull/48158#discussion_r1772337719
##########
sql/api/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala:
##########
@@ -117,65 +130,144 @@ object JavaTypeInference {
.asInstanceOf[UserDefinedType[Any]]
UDTEncoder(udt, udt.getClass)
- case c: Class[_] if c.isArray =>
+ case c: Class[_] if !forGenericBound && c.isArray =>
val elementEncoder = encoderFor(c.getComponentType, seenTypeSet, typeVariables)
ArrayEncoder(elementEncoder, elementEncoder.nullable)
- case c: Class[_] if classOf[JList[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JList[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JSet[_]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JSet[_]].isAssignableFrom(c) =>
val element = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
IterableEncoder(ClassTag(c), element, element.nullable, lenientSerialization = false)
- case c: Class[_] if classOf[JMap[_, _]].isAssignableFrom(c) =>
+ case c: Class[_] if !forGenericBound && classOf[JMap[_, _]].isAssignableFrom(c) =>
val keyEncoder = encoderFor(c.getTypeParameters.array(0), seenTypeSet, typeVariables)
val valueEncoder = encoderFor(c.getTypeParameters.array(1), seenTypeSet, typeVariables)
MapEncoder(ClassTag(c), keyEncoder, valueEncoder, valueEncoder.nullable)
case tv: TypeVariable[_] =>
- encoderFor(typeVariables(tv), seenTypeSet, typeVariables)
+ typeVariables.get(tv) match {
+ case Some(knownType) => encoderFor(knownType, seenTypeSet, typeVariables)
+
+ case None =>
+ // Concrete type unknown, check if the bounds can be used to identify a
+ // UDTEncoder and if exception is thrown check if the type extends Serializable
+ val concreteBound = tv.getBounds.collectFirst { case cls: Class[_] => cls }
+ tv.getBounds
+ .flatMap(getAllSuperClasses)
+ .foldLeft[(Option[AgnosticEncoder[_]], Option[AgnosticEncoder[_]])](None -> None) {
+ case (r @ (Some(_), _), _) => r
- case pt: ParameterizedType =>
+ case (r @ (None, serEncoder), bound) =>
+ Try(encoderFor(bound, seenTypeSet, typeVariables, forGenericBound = true)) match {
+ case Success(value) => (Option(value), serEncoder)
+
+ case Failure(UseSerializationEncoder(encoderProvider)) if serEncoder.isEmpty =>
+ None -> Option(encoderProvider(concreteBound.getOrElse(bound)))
+
+ case Failure(_) => r
+ }
+ } match {
+ case (Some(encd), _) => encd
+
+ case (None, Some(encd)) => encd
+
+ case _ => throw ExecutionErrors.cannotFindEncoderForTypeError(t.getTypeName)
+ }
+ }
+
+ case pt: ParameterizedType if !forGenericBound =>
encoderFor(pt.getRawType, seenTypeSet, JavaTypeUtils.getTypeArguments(pt).asScala.toMap)
- case c: Class[_] =>
+ case c: Class[_] if !forGenericBound =>
if (seenTypeSet.contains(c)) {
throw ExecutionErrors.cannotHaveCircularReferencesInBeanClassError(c)
}
- // TODO: we should only collect properties that have getter and setter. However, some tests
- // pass in scala case class as java bean class which doesn't have getter and setter.
+ // TODO: we should only collect properties that have getter and setter. However, some
+ // tests pass in scala case class as java bean class which doesn't have getter and
+ // setter.
val properties = getJavaBeanReadableProperties(c)
- // add type variables from inheritance hierarchy of the class
- val classTV = JavaTypeUtils.getTypeArguments(c, classOf[Object]).asScala.toMap ++
- typeVariables
- // Note that the fields are ordered by name.
- val fields = properties.map { property =>
- val readMethod = property.getReadMethod
- val encoder = encoderFor(readMethod.getGenericReturnType, seenTypeSet + c, classTV)
- // The existence of `javax.annotation.Nonnull`, means this field is not nullable.
- val hasNonNull = readMethod.isAnnotationPresent(classOf[Nonnull])
- EncoderField(
- property.getName,
- encoder,
- encoder.nullable && !hasNonNull,
- Metadata.empty,
- Option(readMethod.getName),
- Option(property.getWriteMethod).map(_.getName))
+ // if the properties is empty and this is not a top level enclosing class, then we
+ // should not consider class as bean, as otherwise it will be treated as empty schema
+ // and loose the data on deser.
+ if (properties.isEmpty && seenTypeSet.nonEmpty) {
+ if (classOf[KryoSerializable].isAssignableFrom(c)) {
Review Comment:
@hvanhovell Added check for client connect while creating encoders, to not use kryo based encoder. Using threadlocal to detect the same, instead of api change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org