[SPARK-12205][SQL] Pivot fails Analysis when aggregate is UnresolvedFunction

Delays application of ResolvePivot until all aggregates are resolved to prevent problems with UnresolvedFunction and adds unit test

Author: Andrew Ray <ray.andrew@gmail.com>

Closes #10202 from aray/sql-pivot-unresolved-function.

(cherry picked from commit 4bcb894)
Signed-off-by: Yin Huai <yhuai@databricks.com>

Author: aray

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -259,7 +259,7 @@ class Analyzer(
 
   object ResolvePivot extends Rule[LogicalPlan] {
     def apply(plan: LogicalPlan): LogicalPlan = plan transform {
-      case p: Pivot if !p.childrenResolved => p
+      case p: Pivot if !p.childrenResolved | !p.aggregates.forall(_.resolved) => p
       case Pivot(groupByExprs, pivotColumn, pivotValues, aggregates, child) =>
         val singleAgg = aggregates.size == 1
         val pivotAggregates: Seq[NamedExpression] = pivotValues.flatMap { value =>

sql/core/src/test/scala/org/apache/spark/sql/DataFramePivotSuite.scala

@@ -85,4 +85,12 @@ class DataFramePivotSuite extends QueryTest with SharedSQLContext{
     sqlContext.conf.setConf(SQLConf.DATAFRAME_PIVOT_MAX_VALUES,
       SQLConf.DATAFRAME_PIVOT_MAX_VALUES.defaultValue.get)
   }
+
+  test("pivot with UnresolvedFunction") {
+    checkAnswer(
+      courseSales.groupBy("year").pivot("course", Seq("dotNET", "Java"))
+        .agg("earnings" -> "sum"),
+      Row(2012, 15000.0, 20000.0) :: Row(2013, 48000.0, 30000.0) :: Nil
+    )
+  }
 }