[SPARK-30184][SQL] Implement a helper method for aliasing functions

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

@@ -692,10 +692,10 @@ class Analyzer(
                 // Assumption is the aggregate function ignores nulls. This is true for all current
                 // AggregateFunction's with the exception of First and Last in their default mode
                 // (which we handle) and possibly some Hive UDAF's.
-                case First(expr, _) =>
-                  First(ifExpr(expr), Literal(true))
-                case Last(expr, _) =>
-                  Last(ifExpr(expr), Literal(true))
+                case First(funcName, expr, _) =>
+                  First(funcName, ifExpr(expr), Literal(true))
+                case Last(funcName, expr, _) =>
+                  Last(funcName, ifExpr(expr), Literal(true))
                 case a: AggregateFunction =>
                   a.withNewChildren(a.children.map(ifExpr))
               }.transform {

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

@@ -288,11 +288,11 @@ object FunctionRegistry {
     expression[CountIf]("count_if"),
     expression[CovPopulation]("covar_pop"),
     expression[CovSample]("covar_samp"),
-    expression[First]("first"),
-    expression[First]("first_value"),
+    expressionWithAlias[First]("first"),
+    expressionWithAlias[First]("first_value"),
     expression[Kurtosis]("kurtosis"),
-    expression[Last]("last"),
-    expression[Last]("last_value"),
+    expressionWithAlias[Last]("last"),
+    expressionWithAlias[Last]("last_value"),
     expression[Max]("max"),
     expression[MaxBy]("max_by"),
     expression[Average]("mean"),
@@ -302,14 +302,14 @@ object FunctionRegistry {
     expression[Skewness]("skewness"),
     expression[ApproximatePercentile]("percentile_approx"),
     expression[ApproximatePercentile]("approx_percentile"),
-    expression[StddevSamp]("std"),
-    expression[StddevSamp]("stddev"),
+    expressionWithAlias[StddevSamp]("std"),
+    expressionWithAlias[StddevSamp]("stddev"),
     expression[StddevPop]("stddev_pop"),
-    expression[StddevSamp]("stddev_samp"),
+    expressionWithAlias[StddevSamp]("stddev_samp"),
     expression[Sum]("sum"),
-    expression[VarianceSamp]("variance"),
+    expressionWithAlias[VarianceSamp]("variance"),
     expression[VariancePop]("var_pop"),
-    expression[VarianceSamp]("var_samp"),
+    expressionWithAlias[VarianceSamp]("var_samp"),
     expression[CollectList]("collect_list"),
     expression[CollectSet]("collect_set"),
     expression[CountMinSketchAgg]("count_min_sketch"),

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

@@ -431,11 +431,11 @@ object TypeCoercion {
       case Sum(e @ StringType()) => Sum(Cast(e, DoubleType))
       case Average(e @ StringType()) => Average(Cast(e, DoubleType))
       case StddevPop(e @ StringType()) => StddevPop(Cast(e, DoubleType))
-      case StddevSamp(e @ StringType()) => StddevSamp(Cast(e, DoubleType))
+      case StddevSamp(funcName, e @ StringType()) => StddevSamp(funcName, Cast(e, DoubleType))
       case UnaryMinus(e @ StringType()) => UnaryMinus(Cast(e, DoubleType))
       case UnaryPositive(e @ StringType()) => UnaryPositive(Cast(e, DoubleType))
       case VariancePop(e @ StringType()) => VariancePop(Cast(e, DoubleType))
-      case VarianceSamp(e @ StringType()) => VarianceSamp(Cast(e, DoubleType))
+      case VarianceSamp(funcName, e @ StringType()) => VarianceSamp(funcName, Cast(e, DoubleType))
       case Skewness(e @ StringType()) => Skewness(Cast(e, DoubleType))
       case Kurtosis(e @ StringType()) => Kurtosis(Cast(e, DoubleType))
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/CentralMomentAgg.scala

@@ -165,7 +165,7 @@ case class StddevPop(child: Expression) extends CentralMomentAgg(child) {
   """,
   since = "1.6.0")
 // scalastyle:on line.size.limit
-case class StddevSamp(child: Expression) extends CentralMomentAgg(child) {
+case class StddevSamp(funcName: String, child: Expression) extends CentralMomentAgg(child) {
 
   override protected def momentOrder = 2
 
@@ -174,7 +174,7 @@ case class StddevSamp(child: Expression) extends CentralMomentAgg(child) {
       If(n === 1.0, Double.NaN, sqrt(m2 / (n - 1.0))))
   }
 
-  override def prettyName: String = "stddev_samp"
+  override def nodeName: String = funcName
 }
 
 // Compute the population variance of a column
@@ -206,7 +206,7 @@ case class VariancePop(child: Expression) extends CentralMomentAgg(child) {
        1.0
   """,
   since = "1.6.0")
-case class VarianceSamp(child: Expression) extends CentralMomentAgg(child) {
+case class VarianceSamp(funcName: String, child: Expression) extends CentralMomentAgg(child) {
 
   override protected def momentOrder = 2
 
@@ -215,7 +215,7 @@ case class VarianceSamp(child: Expression) extends CentralMomentAgg(child) {
       If(n === 1.0, Double.NaN, m2 / (n - 1.0)))
   }
 
-  override def prettyName: String = "var_samp"
+  override def nodeName: String = funcName
 }
 
 @ExpressionDescription(

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/First.scala

@@ -44,10 +44,10 @@ import org.apache.spark.sql.types._
        5
   """,
   since = "2.0.0")
-case class First(child: Expression, ignoreNullsExpr: Expression)
+case class First(funcName: String, child: Expression, ignoreNullsExpr: Expression)
   extends DeclarativeAggregate with ExpectsInputTypes {
 
-  def this(child: Expression) = this(child, Literal.create(false, BooleanType))
+  def this(child: Expression) = this("first", child, Literal.create(false, BooleanType))
 
   override def children: Seq[Expression] = child :: ignoreNullsExpr :: Nil
 
@@ -113,5 +113,5 @@ case class First(child: Expression, ignoreNullsExpr: Expression)
 
   override lazy val evaluateExpression: AttributeReference = first
 
-  override def toString: String = s"first($child)${if (ignoreNulls) " ignore nulls"}"
+  override def toString: String = s"$funcName($child)${if (ignoreNulls) " ignore nulls"}"
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Last.scala

@@ -44,10 +44,10 @@ import org.apache.spark.sql.types._
        5
   """,
   since = "2.0.0")
-case class Last(child: Expression, ignoreNullsExpr: Expression)
+case class Last(funcName: String, child: Expression, ignoreNullsExpr: Expression)
   extends DeclarativeAggregate with ExpectsInputTypes {
 
-  def this(child: Expression) = this(child, Literal.create(false, BooleanType))
+  def this(child: Expression) = this("last", child, Literal.create(false, BooleanType))
 
   override def children: Seq[Expression] = child :: ignoreNullsExpr :: Nil
 
@@ -111,5 +111,5 @@ case class Last(child: Expression, ignoreNullsExpr: Expression)
 
   override lazy val evaluateExpression: AttributeReference = last
 
-  override def toString: String = s"last($child)${if (ignoreNulls) " ignore nulls"}"
+  override def toString: String = s"$funcName($child)${if (ignoreNulls) " ignore nulls"}"
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregates.scala

@@ -198,7 +198,8 @@ object RewriteDistinctAggregates extends Rule[LogicalPlan] {
 
         // Select the result of the first aggregate in the last aggregate.
         val result = AggregateExpression(
-          aggregate.First(evalWithinGroup(regularGroupId, operator.toAttribute), Literal(true)),
+          aggregate.First("first",
+            evalWithinGroup(regularGroupId, operator.toAttribute), Literal(true)),
           mode = Complete,
           isDistinct = false)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -1510,15 +1510,15 @@ class AstBuilder(conf: SQLConf) extends SqlBaseBaseVisitor[AnyRef] with Logging
    */
   override def visitFirst(ctx: FirstContext): Expression = withOrigin(ctx) {
     val ignoreNullsExpr = ctx.IGNORE != null
-    First(expression(ctx.expression), Literal(ignoreNullsExpr)).toAggregateExpression()
+    First("first", expression(ctx.expression), Literal(ignoreNullsExpr)).toAggregateExpression()
   }
 
   /**
    * Create a [[Last]] expression.
    */
   override def visitLast(ctx: LastContext): Expression = withOrigin(ctx) {
     val ignoreNullsExpr = ctx.IGNORE != null
-    Last(expression(ctx.expression), Literal(ignoreNullsExpr)).toAggregateExpression()
+    Last("last", expression(ctx.expression), Literal(ignoreNullsExpr)).toAggregateExpression()
   }
 
   /**

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/LastTestSuite.scala

@@ -23,8 +23,9 @@ import org.apache.spark.sql.types.IntegerType
 
 class LastTestSuite extends SparkFunSuite {
   val input = AttributeReference("input", IntegerType, nullable = true)()
-  val evaluator = DeclarativeAggregateEvaluator(Last(input, Literal(false)), Seq(input))
-  val evaluatorIgnoreNulls = DeclarativeAggregateEvaluator(Last(input, Literal(true)), Seq(input))
+  val evaluator = DeclarativeAggregateEvaluator(Last("last", input, Literal(false)), Seq(input))
+  val evaluatorIgnoreNulls = DeclarativeAggregateEvaluator(
+    Last("last", input, Literal(true)), Seq(input))
 
   test("empty buffer") {
     assert(evaluator.initialize() === InternalRow(null, false))

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ExpressionParserSuite.scala

@@ -787,19 +787,23 @@ class ExpressionParserSuite extends AnalysisTest {
   }
 
   test("SPARK-19526 Support ignore nulls keywords for first and last") {
-    assertEqual("first(a ignore nulls)", First('a, Literal(true)).toAggregateExpression())
-    assertEqual("first(a)", First('a, Literal(false)).toAggregateExpression())
-    assertEqual("last(a ignore nulls)", Last('a, Literal(true)).toAggregateExpression())
-    assertEqual("last(a)", Last('a, Literal(false)).toAggregateExpression())
+    assertEqual("first(a ignore nulls)", First("first", 'a, Literal(true)).toAggregateExpression())
+    assertEqual("first(a)", First("first", 'a, Literal(false)).toAggregateExpression())
+    assertEqual("last(a ignore nulls)", Last("last", 'a, Literal(true)).toAggregateExpression())
+    assertEqual("last(a)", Last("last", 'a, Literal(false)).toAggregateExpression())
   }
 
   test("Support respect nulls keywords for first_value and last_value") {
-    assertEqual("first_value(a ignore nulls)", First('a, Literal(true)).toAggregateExpression())
-    assertEqual("first_value(a respect nulls)", First('a, Literal(false)).toAggregateExpression())
-    assertEqual("first_value(a)", First('a, Literal(false)).toAggregateExpression())
-    assertEqual("last_value(a ignore nulls)", Last('a, Literal(true)).toAggregateExpression())
-    assertEqual("last_value(a respect nulls)", Last('a, Literal(false)).toAggregateExpression())
-    assertEqual("last_value(a)", Last('a, Literal(false)).toAggregateExpression())
+    assertEqual("first_value(a ignore nulls)",
+      First("first", 'a, Literal(true)).toAggregateExpression())
+    assertEqual("first_value(a respect nulls)",
+      First("first", 'a, Literal(false)).toAggregateExpression())
+    assertEqual("first_value(a)", First("first", 'a, Literal(false)).toAggregateExpression())
+    assertEqual("last_value(a ignore nulls)",
+      Last("last", 'a, Literal(true)).toAggregateExpression())
+    assertEqual("last_value(a respect nulls)",
+      Last("last", 'a, Literal(false)).toAggregateExpression())
+    assertEqual("last_value(a)", Last("last", 'a, Literal(false)).toAggregateExpression())
   }
 
   test("timestamp literals") {

sql/core/src/main/scala/org/apache/spark/sql/execution/stat/StatFunctions.scala

@@ -254,7 +254,8 @@ object StatFunctions extends Logging {
         stats.toLowerCase(Locale.ROOT) match {
           case "count" => (child: Expression) => Count(child).toAggregateExpression()
           case "mean" => (child: Expression) => Average(child).toAggregateExpression()
-          case "stddev" => (child: Expression) => StddevSamp(child).toAggregateExpression()
+          case "stddev" => (child: Expression) =>
+            StddevSamp("stddev_samp", child).toAggregateExpression()
           case "min" => (child: Expression) => Min(child).toAggregateExpression()
           case "max" => (child: Expression) => Max(child).toAggregateExpression()
           case _ => throw new IllegalArgumentException(s"$stats is not a recognised statistic")

sql/core/src/main/scala/org/apache/spark/sql/functions.scala

@@ -431,7 +431,7 @@ object functions {
    * @since 2.0.0
    */
   def first(e: Column, ignoreNulls: Boolean): Column = withAggregateFunction {
-    new First(e.expr, Literal(ignoreNulls))
+    new First("first", e.expr, Literal(ignoreNulls))
   }
 
   /**
@@ -556,7 +556,7 @@ object functions {
    * @since 2.0.0
    */
   def last(e: Column, ignoreNulls: Boolean): Column = withAggregateFunction {
-    new Last(e.expr, Literal(ignoreNulls))
+    new Last("last", e.expr, Literal(ignoreNulls))
   }
 
   /**
@@ -675,7 +675,7 @@ object functions {
    * @group agg_funcs
    * @since 1.6.0
    */
-  def stddev(e: Column): Column = withAggregateFunction { StddevSamp(e.expr) }
+  def stddev(e: Column): Column = withAggregateFunction { StddevSamp("stddev", e.expr) }
 
   /**
    * Aggregate function: alias for `stddev_samp`.
@@ -692,7 +692,7 @@ object functions {
    * @group agg_funcs
    * @since 1.6.0
    */
-  def stddev_samp(e: Column): Column = withAggregateFunction { StddevSamp(e.expr) }
+  def stddev_samp(e: Column): Column = withAggregateFunction { StddevSamp("stddev_samp", e.expr) }
 
   /**
    * Aggregate function: returns the sample standard deviation of
@@ -759,7 +759,7 @@ object functions {
    * @group agg_funcs
    * @since 1.6.0
    */
-  def variance(e: Column): Column = withAggregateFunction { VarianceSamp(e.expr) }
+  def variance(e: Column): Column = withAggregateFunction { VarianceSamp("variance", e.expr) }
 
   /**
    * Aggregate function: alias for `var_samp`.
@@ -775,7 +775,7 @@ object functions {
    * @group agg_funcs
    * @since 1.6.0
    */
-  def var_samp(e: Column): Column = withAggregateFunction { VarianceSamp(e.expr) }
+  def var_samp(e: Column): Column = withAggregateFunction { VarianceSamp("var_samp", e.expr) }
 
   /**
    * Aggregate function: returns the unbiased variance of the values in a group.