[SPARK-9955][SQL] correct error message for aggregate

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

@@ -931,6 +931,7 @@ class Analyzer(
    */
   object PullOutNondeterministic extends Rule[LogicalPlan] {
     override def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
+      case p if !p.resolved => p // Skip unresolved nodes.
       case p: Project => p
       case f: Filter => f
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala

@@ -145,6 +145,11 @@ class AnalysisErrorSuite extends AnalysisTest {
     UnresolvedTestPlan(),
     "unresolved" :: Nil)
 
+  errorTest(
+    "SPARK-9955: correct error message for aggregate",
+    // When parse SQL string, we will wrap aggregate expressions with UnresolvedAlias.
+    testRelation2.where('bad_column > 1).groupBy('a)(UnresolvedAlias(max('b))),
+    "cannot resolve 'bad_column'" :: Nil)
 
   test("SPARK-6452 regression test") {
     // CheckAnalysis should throw AnalysisException when Aggregate contains missing attribute(s)

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

@@ -1618,12 +1618,11 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
   }
 
   test("SPARK-9511: error with table starting with number") {
-    val df = sqlContext.sparkContext.parallelize(1 to 10).map(i => (i, i.toString))
-      .toDF("num", "str")
-    df.registerTempTable("1one")
-
-    checkAnswer(sql("select count(num) from 1one"), Row(10))
-
-    sqlContext.dropTempTable("1one")
+    withTempTable("1one") {
+      sqlContext.sparkContext.parallelize(1 to 10).map(i => (i, i.toString))
+        .toDF("num", "str")
+        .registerTempTable("1one")
+      checkAnswer(sql("select count(num) from 1one"), Row(10))
+    }
   }
 }