[ https://issues.apache.org/jira/browse/SPARK-37476?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
koert kuipers updated SPARK-37476: ---------------------------------- Description: i have a need to have a dataframe aggregation return a nullable case class. there seems to be no way to get this to work. the suggestion to wrap the result in an option doesnt work either. first attempt using nulls: {code:java} case class SumAndProduct(sum: Double, product: Double) val sumAndProductAgg = new Aggregator[Double, SumAndProduct, SumAndProduct] { def zero: SumAndProduct = null def reduce(b: SumAndProduct, a: Double): SumAndProduct = if (b == null) { SumAndProduct(a, a) } else { SumAndProduct(b.sum + a, b.product * a) } def merge(b1: SumAndProduct, b2: SumAndProduct): SumAndProduct = if (b1 == null) { b2 } else if (b2 == null) { b1 } else { SumAndProduct(b1.sum + b2.sum, b1.product * b2.product) } def finish(r: SumAndProduct): SumAndProduct = r def bufferEncoder: Encoder[SumAndProduct] = ExpressionEncoder() def outputEncoder: Encoder[SumAndProduct] = ExpressionEncoder() } val df = Seq.empty[Double] .toDF() .select(udaf(sumAndProductAgg).apply(col("value"))) df.printSchema() df.show() {code} this gives: {code:java} root |-- $anon$3(value): struct (nullable = true) | |-- sum: double (nullable = false) | |-- product: double (nullable = false) 16:44:54.882 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 in stage 1491.0 (TID 1929) java.lang.RuntimeException: Error while encoding: java.lang.NullPointerException: Null value appeared in non-nullable field: top level Product or row object If the schema is inferred from a Scala tuple/case class, or a Java bean, please try to use scala.Option[_] or other nullable types (e.g. java.lang.Integer instead of int/scala.Int). knownnotnull(assertnotnull(input[0, org.apache.spark.sql.SumAndProduct, true])).sum AS sum#20070 knownnotnull(assertnotnull(input[0, org.apache.spark.sql.SumAndProduct, true])).product AS product#20071 at org.apache.spark.sql.errors.QueryExecutionErrors$.expressionEncodingError(QueryExecutionErrors.scala:1125) {code} taking the advice to heart and using option we get to the second attempt using options: {code:java} case class SumAndProduct(sum: Double, product: Double) val sumAndProductAgg = new Aggregator[Double, Option[SumAndProduct], Option[SumAndProduct]] { def zero: Option[SumAndProduct] = None def reduce(b: Option[SumAndProduct], a: Double): Option[SumAndProduct] = b .map{ b => SumAndProduct(b.sum + a, b.product * a) } .orElse{ Option(SumAndProduct(a, a)) } def merge(b1: Option[SumAndProduct], b2: Option[SumAndProduct]): Option[SumAndProduct] = b1.map{ b1 => b2.map{ b2 => SumAndProduct(b1.sum + b2.sum, b1.product * b2.product) }.getOrElse(b1) }.orElse(b2) def finish(r: Option[SumAndProduct]): Option[SumAndProduct] = r def bufferEncoder: Encoder[Option[SumAndProduct]] = ExpressionEncoder() def outputEncoder: Encoder[Option[SumAndProduct]] = ExpressionEncoder() } val df = Seq.empty[Double] .toDF() .select(udaf(sumAndProductAgg).apply(col("value"))) df.printSchema() df.show() {code} this gives: {code:java} root |-- $anon$4(value): struct (nullable = true) | |-- sum: double (nullable = false) | |-- product: double (nullable = false) 16:44:54.998 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 in stage 1493.0 (TID 1930) java.lang.AssertionError: index (1) should < 1 at org.apache.spark.sql.catalyst.expressions.UnsafeRow.assertIndexIsValid(UnsafeRow.java:142) at org.apache.spark.sql.catalyst.expressions.UnsafeRow.isNullAt(UnsafeRow.java:338) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source) at org.apache.spark.sql.execution.aggregate.AggregationIterator.$anonfun$generateResultProjection$5(AggregationIterator.scala:260) at org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.outputForEmptyGroupingKeyWithoutInput(ObjectAggregationIterator.scala:107) {code} was: i have a need to have a dataframe aggregation return a nullable case class. there seems to be no way to get this to work. the suggestion to wrap the result in an option doesnt work either. first attempt using nulls: {code:java} case class SumAndProduct(sum: Double, product: Double) val sumAndProductAgg = new Aggregator[Double, SumAndProduct, SumAndProduct] { def zero: SumAndProduct = null def reduce(b: SumAndProduct, a: Double): SumAndProduct = if (b == null) { SumAndProduct(a, a) } else { SumAndProduct(b.sum + a, b.product * a) } def merge(b1: SumAndProduct, b2: SumAndProduct): SumAndProduct = if (b1 == null) { b2 } else if (b2 == null) { b1 } else { SumAndProduct(b1.sum + b2.sum, b1.product * b2.product) } def finish(r: SumAndProduct): SumAndProduct = r def bufferEncoder: Encoder[SumAndProduct] = ExpressionEncoder() def outputEncoder: Encoder[SumAndProduct] = ExpressionEncoder() } val df = Seq.empty[Double] .toDF() .select(udaf(sumAndProductAgg).apply(col("value"))) df.printSchema() df.show() {code} this gives: {code:java} root |-- $anon$3(value): struct (nullable = true) | |-- sum: double (nullable = false) | |-- product: double (nullable = false) 16:44:54.882 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 in stage 1491.0 (TID 1929) java.lang.RuntimeException: Error while encoding: java.lang.NullPointerException: Null value appeared in non-nullable field: top level Product or row object If the schema is inferred from a Scala tuple/case class, or a Java bean, please try to use scala.Option[_] or other nullable types (e.g. java.lang.Integer instead of int/scala.Int). knownnotnull(assertnotnull(input[0, org.apache.spark.sql.SumAndProduct, true])).sum AS sum#20070 knownnotnull(assertnotnull(input[0, org.apache.spark.sql.SumAndProduct, true])).product AS product#20071 at org.apache.spark.sql.errors.QueryExecutionErrors$.expressionEncodingError(QueryExecutionErrors.scala:1125) {code} taking the advice to heart and using option we get to the second attempt using options: {code:java} val sumAndProductAgg = new Aggregator[Double, Option[SumAndProduct], Option[SumAndProduct]] { def zero: Option[SumAndProduct] = None def reduce(b: Option[SumAndProduct], a: Double): Option[SumAndProduct] = b .map{ b => SumAndProduct(b.sum + a, b.product * a) } .orElse{ Option(SumAndProduct(a, a)) } def merge(b1: Option[SumAndProduct], b2: Option[SumAndProduct]): Option[SumAndProduct] = b1.map{ b1 => b2.map{ b2 => SumAndProduct(b1.sum + b2.sum, b1.product * b2.product) }.getOrElse(b1) }.orElse(b2) def finish(r: Option[SumAndProduct]): Option[SumAndProduct] = r def bufferEncoder: Encoder[Option[SumAndProduct]] = ExpressionEncoder() def outputEncoder: Encoder[Option[SumAndProduct]] = ExpressionEncoder() } val df = Seq.empty[Double] .toDF() .select(udaf(sumAndProductAgg).apply(col("value"))) df.printSchema() df.show() {code} this gives: {code:java} root |-- $anon$4(value): struct (nullable = true) | |-- sum: double (nullable = false) | |-- product: double (nullable = false) 16:44:54.998 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 in stage 1493.0 (TID 1930) java.lang.AssertionError: index (1) should < 1 at org.apache.spark.sql.catalyst.expressions.UnsafeRow.assertIndexIsValid(UnsafeRow.java:142) at org.apache.spark.sql.catalyst.expressions.UnsafeRow.isNullAt(UnsafeRow.java:338) at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source) at org.apache.spark.sql.execution.aggregate.AggregationIterator.$anonfun$generateResultProjection$5(AggregationIterator.scala:260) at org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.outputForEmptyGroupingKeyWithoutInput(ObjectAggregationIterator.scala:107) {code} > udaf doesnt work with nullable (or option of) case class result > ---------------------------------------------------------------- > > Key: SPARK-37476 > URL: https://issues.apache.org/jira/browse/SPARK-37476 > Project: Spark > Issue Type: Bug > Components: SQL > Affects Versions: 3.2.0 > Environment: spark master branch on nov 27 > Reporter: koert kuipers > Priority: Minor > > i have a need to have a dataframe aggregation return a nullable case class. > there seems to be no way to get this to work. the suggestion to wrap the > result in an option doesnt work either. > first attempt using nulls: > {code:java} > case class SumAndProduct(sum: Double, product: Double) > val sumAndProductAgg = new Aggregator[Double, SumAndProduct, SumAndProduct] { > def zero: SumAndProduct = null > def reduce(b: SumAndProduct, a: Double): SumAndProduct = > if (b == null) { > SumAndProduct(a, a) > } else { > SumAndProduct(b.sum + a, b.product * a) > } > def merge(b1: SumAndProduct, b2: SumAndProduct): SumAndProduct = > if (b1 == null) { > b2 > } else if (b2 == null) { > b1 > } else { > SumAndProduct(b1.sum + b2.sum, b1.product * b2.product) > } > def finish(r: SumAndProduct): SumAndProduct = r > def bufferEncoder: Encoder[SumAndProduct] = ExpressionEncoder() > def outputEncoder: Encoder[SumAndProduct] = ExpressionEncoder() > } > val df = Seq.empty[Double] > .toDF() > .select(udaf(sumAndProductAgg).apply(col("value"))) > df.printSchema() > df.show() > {code} > this gives: > {code:java} > root > |-- $anon$3(value): struct (nullable = true) > | |-- sum: double (nullable = false) > | |-- product: double (nullable = false) > 16:44:54.882 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 > in stage 1491.0 (TID 1929) > java.lang.RuntimeException: Error while encoding: > java.lang.NullPointerException: Null value appeared in non-nullable field: > top level Product or row object > If the schema is inferred from a Scala tuple/case class, or a Java bean, > please try to use scala.Option[_] or other nullable types (e.g. > java.lang.Integer instead of int/scala.Int). > knownnotnull(assertnotnull(input[0, org.apache.spark.sql.SumAndProduct, > true])).sum AS sum#20070 > knownnotnull(assertnotnull(input[0, org.apache.spark.sql.SumAndProduct, > true])).product AS product#20071 > at > org.apache.spark.sql.errors.QueryExecutionErrors$.expressionEncodingError(QueryExecutionErrors.scala:1125) > {code} > taking the advice to heart and using option we get to the second attempt > using options: > {code:java} > case class SumAndProduct(sum: Double, product: Double) > val sumAndProductAgg = new Aggregator[Double, Option[SumAndProduct], > Option[SumAndProduct]] { > def zero: Option[SumAndProduct] = None > def reduce(b: Option[SumAndProduct], a: Double): Option[SumAndProduct] = > b > .map{ b => SumAndProduct(b.sum + a, b.product * a) } > .orElse{ Option(SumAndProduct(a, a)) } > def merge(b1: Option[SumAndProduct], b2: Option[SumAndProduct]): > Option[SumAndProduct] = > b1.map{ b1 => > b2.map{ b2 => > SumAndProduct(b1.sum + b2.sum, b1.product * b2.product) > }.getOrElse(b1) > }.orElse(b2) > def finish(r: Option[SumAndProduct]): Option[SumAndProduct] = r > def bufferEncoder: Encoder[Option[SumAndProduct]] = ExpressionEncoder() > def outputEncoder: Encoder[Option[SumAndProduct]] = ExpressionEncoder() > } > val df = Seq.empty[Double] > .toDF() > .select(udaf(sumAndProductAgg).apply(col("value"))) > df.printSchema() > df.show() > {code} > this gives: > {code:java} > root > |-- $anon$4(value): struct (nullable = true) > | |-- sum: double (nullable = false) > | |-- product: double (nullable = false) > 16:44:54.998 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 > in stage 1493.0 (TID 1930) > java.lang.AssertionError: index (1) should < 1 > at > org.apache.spark.sql.catalyst.expressions.UnsafeRow.assertIndexIsValid(UnsafeRow.java:142) > at > org.apache.spark.sql.catalyst.expressions.UnsafeRow.isNullAt(UnsafeRow.java:338) > at > org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown > Source) > at > org.apache.spark.sql.execution.aggregate.AggregationIterator.$anonfun$generateResultProjection$5(AggregationIterator.scala:260) > at > org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.outputForEmptyGroupingKeyWithoutInput(ObjectAggregationIterator.scala:107) > {code} > > > -- This message was sent by Atlassian Jira (v8.20.1#820001) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org