[SPARK-16474] Global Aggregation doesn't seem to work at all

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Not A Problem
• Affects Version/s: 1.6.2, 2.0.0
• Component/s: SQL
• Assignee: Unassigned
• Reporter: Amit Sela
• Created: 10/Jul/16 14:10
• Updated: 24/Jun/19 21:33
• Resolved: 10/Jul/16 16:27

Description

Executing a global aggregation (not grouped by key) fails.
Take the following code for example:

    val session = SparkSession.builder()
                              .appName("TestGlobalAggregator")
                              .master("local[*]")
                              .getOrCreate()
    import session.implicits._

    val ds1 = List(1, 2, 3).toDS
    val ds2 = ds1.agg(
          new Aggregator[Int, Int, Int]{

      def zero: Int = 0

      def reduce(b: Int, a: Int): Int = b + a

      def merge(b1: Int, b2: Int): Int = b1 + b2

      def finish(reduction: Int): Int = reduction

      def bufferEncoder: Encoder[Int] = implicitly[Encoder[Int]]

      def outputEncoder: Encoder[Int] = implicitly[Encoder[Int]]
    }.toColumn)

    ds2.printSchema
    ds2.show

I would expect the result to be 6, but instead I get the following exception:

java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to java.lang.Integer 
at scala.runtime.BoxesRunTime.unboxToInt(BoxesRunTime.java:101)
.........

Trying the same code on DataFrames in 1.6.2 results in:

Exception in thread "main" org.apache.spark.sql.AnalysisException: unresolved operator 'Aggregate [(anon$1(),mode=Complete,isDistinct=false) AS anon$1()#8]; 
at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.failAnalysis(CheckAnalysis.scala:38)
..........