Delays application of ResolvePivot until all aggregates are resolved …

…to prevent problems with UnresolvedFunction and unit test
apache · Dec 8, 2015 · 8a3f555 · 8a3f555
1 parent da2012a
commit 8a3f555
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 1 deletion.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/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 =>

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFramePivotSuite.scala b/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
+    )
+  }
 }