[GitHub] [spark] m44444 commented on a change in pull request #24144: [SPARK-24935][SQL] fix Hive UDAF with two aggregation buffers

[GitBox <gi...@apache.org>]

m44444 commented on a change in pull request #24144: [SPARK-24935][SQL] fix Hive UDAF with two aggregation buffers
URL: https://github.com/apache/spark/pull/24144#discussion_r277474112
 
 

 ##########
 File path: sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala
 ##########
 @@ -401,29 +391,47 @@ private[hive] case class HiveUDAFFunction(
     s"$name($distinct${children.map(_.sql).mkString(", ")})"
   }
 
-  override def createAggregationBuffer(): AggregationBuffer =
-    partial1HiveEvaluator.evaluator.getNewAggregationBuffer
+  // The hive UDAF may create different buffers to handle different inputs: original data or
+  // aggregate buffer. However, the Spark UDAF framework does not expose this information when
+  // creating the buffer. Here we return null, and create the buffer in `update` and `merge`
+  // on demand, so that we can know what input we are dealing with.
+  override def createAggregationBuffer(): AggregationBuffer = null
 
   @transient
   private lazy val inputProjection = UnsafeProjection.create(children)
 
   override def update(buffer: AggregationBuffer, input: InternalRow): AggregationBuffer = {
+    // The input is original data, we create buffer with the partial1 evaluator.
+    val actualBuffer = if (buffer == null) {
+      partial1HiveEvaluator.evaluator.getNewAggregationBuffer
+    } else {
+      buffer
+    }
+
     partial1HiveEvaluator.evaluator.iterate(
-      buffer, wrap(inputProjection(input), inputWrappers, cached, inputDataTypes))
-    buffer
+      actualBuffer, wrap(inputProjection(input), inputWrappers, cached, inputDataTypes))
+    actualBuffer
   }
 
   override def merge(buffer: AggregationBuffer, input: AggregationBuffer): AggregationBuffer = {
+    // The input is aggregate buffer, we create buffer with the partial2 evaluator.
+    val actualBuffer = if (buffer == null) {
 
 Review comment:
   Hi, I came through the DataSketches hll issue, and find it is still problematic in spark 2.4.1, which was released with this change. Briefly the test case @pgandhi999 posted here https://github.com/apache/spark/pull/23778 is not passed from Spark 2.4.1. However another testcase passed when I created a DataFrame only based on an single-object array, which means the DataFrame is actually not distributed into multiple threads. I believe what @pgandhi999 said here came true, and more scary, which is unexpected in Spark per @cloud-fan. I post the error log bellow:
   Caused by: java.lang.ClassCastException: com.yahoo.sketches.hive.hll.SketchState cannot be cast to com.yahoo.sketches.hive.hll.UnionState
     at com.yahoo.sketches.hive.hll.SketchEvaluator.merge(SketchEvaluator.java:56)
     at com.yahoo.sketches.hive.hll.DataToSketchUDAF$DataToSketchEvaluator.merge(DataToSketchUDAF.java:100)
     at org.apache.spark.sql.hive.HiveUDAFFunction.merge(hiveUDFs.scala:430)
     at org.apache.spark.sql.hive.HiveUDAFFunction.merge(hiveUDFs.scala:307)
     at org.apache.spark.sql.catalyst.expressions.aggregate.TypedImperativeAggregate.merge(interfaces.scala:539)
     at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$1$$anonfun$applyOrElse$2.apply(AggregationIterator.scala:174)
     at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$1$$anonfun$applyOrElse$2.apply(AggregationIterator.scala:174)
     at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateProcessRow$1.apply(AggregationIterator.scala:188)
     at org.apache.spark.sql.execution.aggregate.AggregationIterator$$anonfun$generateProcessRow$1.apply(AggregationIterator.scala:182)
     at org.apache.spark.sql.execution.aggregate.SortBasedAggregator$$anon$1.findNextSortedGroup(ObjectAggregationIterator.scala:275)
     at org.apache.spark.sql.execution.aggregate.SortBasedAggregator$$anon$1.hasNext(ObjectAggregationIterator.scala:247)
     at org.apache.spark.sql.execution.aggregate.ObjectAggregationIterator.hasNext(ObjectAggregationIterator.scala:81)
     at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
     at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:187)
     at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
     at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:55)
     at org.apache.spark.scheduler.Task.run(Task.scala:121)
     at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:403)
     at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:409)
     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
     at java.lang.Thread.run(Thread.java:745) 

----------------------------------------------------------------
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.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org