[SPARK-24996][SQL] Use DSL in DeclarativeAggregate

## What changes were proposed in this pull request?

The PR refactors the aggregate expressions which were not using DSL in order to simplify them.

## How was this patch tested?

NA

Author: Marco Gaido <[email protected]>

Closes apache#21970 from mgaido91/SPARK-24996.

Author: mgaido91

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala

@@ -167,6 +167,8 @@ package object dsl {
     def upper(e: Expression): Expression = Upper(e)
     def lower(e: Expression): Expression = Lower(e)
     def coalesce(args: Expression*): Expression = Coalesce(args)
+    def greatest(args: Expression*): Expression = Greatest(args)
+    def least(args: Expression*): Expression = Least(args)
     def sqrt(e: Expression): Expression = Sqrt(e)
     def abs(e: Expression): Expression = Abs(e)
     def star(names: String*): Expression = names match {

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

@@ -68,7 +68,7 @@ abstract class AverageLike(child: Expression) extends DeclarativeAggregate {
     Add(
       sum,
       coalesce(child.cast(sumDataType), Literal(0).cast(sumDataType))),
-    /* count = */ If(IsNull(child), count, count + 1L)
+    /* count = */ If(child.isNull, count, count + 1L)
   )
 
   override lazy val updateExpressions = updateExpressionsDef

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

@@ -75,7 +75,7 @@ abstract class CentralMomentAgg(child: Expression)
     val n2 = n.right
     val newN = n1 + n2
     val delta = avg.right - avg.left
-    val deltaN = If(newN === Literal(0.0), Literal(0.0), delta / newN)
+    val deltaN = If(newN === 0.0, 0.0, delta / newN)
     val newAvg = avg.left + deltaN * n2
 
     // higher order moments computed according to:
@@ -102,7 +102,7 @@ abstract class CentralMomentAgg(child: Expression)
   }
 
   protected def updateExpressionsDef: Seq[Expression] = {
-    val newN = n + Literal(1.0)
+    val newN = n + 1.0
     val delta = child - avg
     val deltaN = delta / newN
     val newAvg = avg + deltaN
@@ -123,11 +123,11 @@ abstract class CentralMomentAgg(child: Expression)
     }
 
     trimHigherOrder(Seq(
-      If(IsNull(child), n, newN),
-      If(IsNull(child), avg, newAvg),
-      If(IsNull(child), m2, newM2),
-      If(IsNull(child), m3, newM3),
-      If(IsNull(child), m4, newM4)
+      If(child.isNull, n, newN),
+      If(child.isNull, avg, newAvg),
+      If(child.isNull, m2, newM2),
+      If(child.isNull, m3, newM3),
+      If(child.isNull, m4, newM4)
     ))
   }
 }
@@ -142,8 +142,7 @@ case class StddevPop(child: Expression) extends CentralMomentAgg(child) {
   override protected def momentOrder = 2
 
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      Sqrt(m2 / n))
+    If(n === 0.0, Literal.create(null, DoubleType), sqrt(m2 / n))
   }
 
   override def prettyName: String = "stddev_pop"
@@ -159,9 +158,8 @@ case class StddevSamp(child: Expression) extends CentralMomentAgg(child) {
   override protected def momentOrder = 2
 
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      If(n === Literal(1.0), Literal(Double.NaN),
-        Sqrt(m2 / (n - Literal(1.0)))))
+    If(n === 0.0, Literal.create(null, DoubleType),
+      If(n === 1.0, Double.NaN, sqrt(m2 / (n - 1.0))))
   }
 
   override def prettyName: String = "stddev_samp"
@@ -175,8 +173,7 @@ case class VariancePop(child: Expression) extends CentralMomentAgg(child) {
   override protected def momentOrder = 2
 
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      m2 / n)
+    If(n === 0.0, Literal.create(null, DoubleType), m2 / n)
   }
 
   override def prettyName: String = "var_pop"
@@ -190,9 +187,8 @@ case class VarianceSamp(child: Expression) extends CentralMomentAgg(child) {
   override protected def momentOrder = 2
 
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      If(n === Literal(1.0), Literal(Double.NaN),
-        m2 / (n - Literal(1.0))))
+    If(n === 0.0, Literal.create(null, DoubleType),
+      If(n === 1.0, Double.NaN, m2 / (n - 1.0)))
   }
 
   override def prettyName: String = "var_samp"
@@ -207,9 +203,8 @@ case class Skewness(child: Expression) extends CentralMomentAgg(child) {
   override protected def momentOrder = 3
 
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      If(m2 === Literal(0.0), Literal(Double.NaN),
-        Sqrt(n) * m3 / Sqrt(m2 * m2 * m2)))
+    If(n === 0.0, Literal.create(null, DoubleType),
+      If(m2 === 0.0, Double.NaN, sqrt(n) * m3 / sqrt(m2 * m2 * m2)))
   }
 }
 
@@ -220,9 +215,8 @@ case class Kurtosis(child: Expression) extends CentralMomentAgg(child) {
   override protected def momentOrder = 4
 
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      If(m2 === Literal(0.0), Literal(Double.NaN),
-        n * m4 / (m2 * m2) - Literal(3.0)))
+    If(n === 0.0, Literal.create(null, DoubleType),
+      If(m2 === 0.0, Double.NaN, n * m4 / (m2 * m2) - 3.0))
   }
 
   override def prettyName: String = "kurtosis"

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

@@ -54,9 +54,9 @@ abstract class PearsonCorrelation(x: Expression, y: Expression)
     val n2 = n.right
     val newN = n1 + n2
     val dx = xAvg.right - xAvg.left
-    val dxN = If(newN === Literal(0.0), Literal(0.0), dx / newN)
+    val dxN = If(newN === 0.0, 0.0, dx / newN)
     val dy = yAvg.right - yAvg.left
-    val dyN = If(newN === Literal(0.0), Literal(0.0), dy / newN)
+    val dyN = If(newN === 0.0, 0.0, dy / newN)
     val newXAvg = xAvg.left + dxN * n2
     val newYAvg = yAvg.left + dyN * n2
     val newCk = ck.left + ck.right + dx * dyN * n1 * n2
@@ -67,7 +67,7 @@ abstract class PearsonCorrelation(x: Expression, y: Expression)
   }
 
   protected def updateExpressionsDef: Seq[Expression] = {
-    val newN = n + Literal(1.0)
+    val newN = n + 1.0
     val dx = x - xAvg
     val dxN = dx / newN
     val dy = y - yAvg
@@ -78,7 +78,7 @@ abstract class PearsonCorrelation(x: Expression, y: Expression)
     val newXMk = xMk + dx * (x - newXAvg)
     val newYMk = yMk + dy * (y - newYAvg)
 
-    val isNull = IsNull(x) || IsNull(y)
+    val isNull = x.isNull || y.isNull
     Seq(
       If(isNull, n, newN),
       If(isNull, xAvg, newXAvg),
@@ -99,9 +99,8 @@ case class Corr(x: Expression, y: Expression)
   extends PearsonCorrelation(x, y) {
 
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      If(n === Literal(1.0), Literal(Double.NaN),
-        ck / Sqrt(xMk * yMk)))
+    If(n === 0.0, Literal.create(null, DoubleType),
+      If(n === 1.0, Double.NaN, ck / sqrt(xMk * yMk)))
   }
 
   override def prettyName: String = "corr"

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

@@ -50,9 +50,9 @@ abstract class Covariance(x: Expression, y: Expression)
     val n2 = n.right
     val newN = n1 + n2
     val dx = xAvg.right - xAvg.left
-    val dxN = If(newN === Literal(0.0), Literal(0.0), dx / newN)
+    val dxN = If(newN === 0.0, 0.0, dx / newN)
     val dy = yAvg.right - yAvg.left
-    val dyN = If(newN === Literal(0.0), Literal(0.0), dy / newN)
+    val dyN = If(newN === 0.0, 0.0, dy / newN)
     val newXAvg = xAvg.left + dxN * n2
     val newYAvg = yAvg.left + dyN * n2
     val newCk = ck.left + ck.right + dx * dyN * n1 * n2
@@ -61,15 +61,15 @@ abstract class Covariance(x: Expression, y: Expression)
   }
 
   protected def updateExpressionsDef: Seq[Expression] = {
-    val newN = n + Literal(1.0)
+    val newN = n + 1.0
     val dx = x - xAvg
     val dy = y - yAvg
     val dyN = dy / newN
     val newXAvg = xAvg + dx / newN
     val newYAvg = yAvg + dyN
     val newCk = ck + dx * (y - newYAvg)
 
-    val isNull = IsNull(x) || IsNull(y)
+    val isNull = x.isNull || y.isNull
     Seq(
       If(isNull, n, newN),
       If(isNull, xAvg, newXAvg),
@@ -83,8 +83,7 @@ abstract class Covariance(x: Expression, y: Expression)
   usage = "_FUNC_(expr1, expr2) - Returns the population covariance of a set of number pairs.")
 case class CovPopulation(left: Expression, right: Expression) extends Covariance(left, right) {
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      ck / n)
+    If(n === 0.0, Literal.create(null, DoubleType), ck / n)
   }
   override def prettyName: String = "covar_pop"
 }
@@ -94,9 +93,8 @@ case class CovPopulation(left: Expression, right: Expression) extends Covariance
   usage = "_FUNC_(expr1, expr2) - Returns the sample covariance of a set of number pairs.")
 case class CovSample(left: Expression, right: Expression) extends Covariance(left, right) {
   override val evaluateExpression: Expression = {
-    If(n === Literal(0.0), Literal.create(null, DoubleType),
-      If(n === Literal(1.0), Literal(Double.NaN),
-        ck / (n - Literal(1.0))))
+    If(n === 0.0, Literal.create(null, DoubleType),
+      If(n === 1.0, Double.NaN, ck / (n - 1.0)))
   }
   override def prettyName: String = "covar_samp"
 }

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.expressions.aggregate
 
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.types._
 
@@ -80,8 +81,8 @@ case class First(child: Expression, ignoreNullsExpr: Expression)
   override lazy val updateExpressions: Seq[Expression] = {
     if (ignoreNulls) {
       Seq(
-        /* first = */ If(Or(valueSet, IsNull(child)), first, child),
-        /* valueSet = */ Or(valueSet, IsNotNull(child))
+        /* first = */ If(valueSet || child.isNull, first, child),
+        /* valueSet = */ valueSet || child.isNotNull
       )
     } else {
       Seq(
@@ -97,7 +98,7 @@ case class First(child: Expression, ignoreNullsExpr: Expression)
     // false, we are safe to do so because first.right will be null in this case).
     Seq(
       /* first = */ If(valueSet.left, first.left, first.right),
-      /* valueSet = */ Or(valueSet.left, valueSet.right)
+      /* valueSet = */ valueSet.left || valueSet.right
     )
   }
 

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.expressions.aggregate
 
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.types._
 
@@ -80,8 +81,8 @@ case class Last(child: Expression, ignoreNullsExpr: Expression)
   override lazy val updateExpressions: Seq[Expression] = {
     if (ignoreNulls) {
       Seq(
-        /* last = */ If(IsNull(child), last, child),
-        /* valueSet = */ Or(valueSet, IsNotNull(child))
+        /* last = */ If(child.isNull, last, child),
+        /* valueSet = */ valueSet || child.isNotNull
       )
     } else {
       Seq(
@@ -95,7 +96,7 @@ case class Last(child: Expression, ignoreNullsExpr: Expression)
     // Prefer the right hand expression if it has been set.
     Seq(
       /* last = */ If(valueSet.right, last.right, last.left),
-      /* valueSet = */ Or(valueSet.right, valueSet.left)
+      /* valueSet = */ valueSet.right || valueSet.left
     )
   }
 

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.expressions.aggregate
 
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util.TypeUtils
 import org.apache.spark.sql.types._
@@ -45,12 +46,12 @@ case class Max(child: Expression) extends DeclarativeAggregate {
   )
 
   override lazy val updateExpressions: Seq[Expression] = Seq(
-    /* max = */ Greatest(Seq(max, child))
+    /* max = */ greatest(max, child)
   )
 
   override lazy val mergeExpressions: Seq[Expression] = {
     Seq(
-      /* max = */ Greatest(Seq(max.left, max.right))
+      /* max = */ greatest(max.left, max.right)
     )
   }
 

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.expressions.aggregate
 
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util.TypeUtils
 import org.apache.spark.sql.types._
@@ -45,12 +46,12 @@ case class Min(child: Expression) extends DeclarativeAggregate {
   )
 
   override lazy val updateExpressions: Seq[Expression] = Seq(
-    /* min = */ Least(Seq(min, child))
+    /* min = */ least(min, child)
   )
 
   override lazy val mergeExpressions: Seq[Expression] = {
     Seq(
-      /* min = */ Least(Seq(min.left, min.right))
+      /* min = */ least(min.left, min.right)
     )
   }
 

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.expressions.aggregate
 
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util.TypeUtils
 import org.apache.spark.sql.types._
@@ -61,20 +62,20 @@ case class Sum(child: Expression) extends DeclarativeAggregate with ImplicitCast
     if (child.nullable) {
       Seq(
         /* sum = */
-        Coalesce(Seq(Add(Coalesce(Seq(sum, zero)), Cast(child, sumDataType)), sum))
+        coalesce(coalesce(sum, zero) + child.cast(sumDataType), sum)
       )
     } else {
       Seq(
         /* sum = */
-        Add(Coalesce(Seq(sum, zero)), Cast(child, sumDataType))
+        coalesce(sum, zero) + child.cast(sumDataType)
       )
     }
   }
 
   override lazy val mergeExpressions: Seq[Expression] = {
     Seq(
       /* sum = */
-      Coalesce(Seq(Add(Coalesce(Seq(sum.left, zero)), sum.right), sum.left))
+      coalesce(coalesce(sum.left, zero) + sum.right, sum.left)
     )
   }