structStarExpansion

Author: gatorsmile

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

@@ -362,12 +362,26 @@ class Analyzer(
       exprs.flatMap {
         case s: Star => s.expand(child, resolver)
         case e =>
-          e.transformDown {
+          e.transformUp {
+            // ResolveFunctions can handle the case when the number of variables is not valid
             case f1: UnresolvedFunction if containsStar(f1.children) =>
               f1.copy(children = f1.children.flatMap {
                 case s: Star => s.expand(child, resolver)
                 case o => o :: Nil
               })
+            case c: CreateStruct if containsStar(c.children) =>
+              c.copy(children = c.children.flatMap {
+                case s: Star => s.expand(child, resolver)
+                case o => o :: Nil
+              })
+            case c: CreateStructUnsafe if containsStar(c.children) =>
+              c.copy(children = c.children.flatMap {
+                case s: Star => s.expand(child, resolver)
+                case o => o :: Nil
+              })
+            // count(*) has been replaced by count(1)
+            case f2: ExpectsInputTypes if containsStar(f2.children) =>
+              failAnalysis(s"Invalid usage of '*' in function '${f2.prettyName}'")
           } :: Nil
       }
     }

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

@@ -528,6 +528,14 @@ class DatasetSuite extends QueryTest with SharedSQLContext {
     assert(e.getMessage.contains("cannot resolve 'c' given input columns: [a, b]"), e.getMessage)
   }
 
+  test("verify star in functions fail with a good error") {
+    val ds = Seq(("a", 1, "c"), ("b", 2, "d")).map(a => (a._1, a._3))
+    val e = intercept[AnalysisException] {
+      ds.toDF().groupBy($"_1").agg(sum($"*") as "sumOccurances")
+    }
+    assert(e.getMessage.contains("Invalid usage of '*' in function 'sum'"), e.getMessage)
+  }
+
   test("runtime nullability check") {
     val schema = StructType(Seq(
       StructField("f", StructType(Seq(

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

@@ -1834,6 +1834,8 @@ class SQLQuerySuite extends QueryTest with SharedSQLContext {
       """.stripMargin).select($"r.*"),
       Row(3, 2) :: Nil)
 
+    assert(structDf.groupBy($"a").agg(min(struct($"record.*"))).first() == Row(3, Row(3, 1)))
+
     // With GROUP BY
     checkAnswer(sql(
       """