Github user yhuai commented on a diff in the pull request:
https://github.com/apache/spark/pull/6297#discussion_r34627743
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregates.scala
---
@@ -761,3 +761,216 @@ case class LastFunction(expr: Expression, base:
AggregateExpression) extends Agg
if (result != null) expr.eval(result.asInstanceOf[InternalRow]) else
null
}
}
+
+// Compute standard deviation based on online algorithm specified here:
+// http://en.wikipedia.org/wiki/Algorithms_for_calculating_variance
+case class Stddev(child: Expression) extends PartialAggregate with
trees.UnaryNode[Expression] {
+ override def nullable: Boolean = true
+ override def dataType: DataType = child.dataType match {
+ case DecimalType.Fixed(_, _) | DecimalType.Unlimited =>
+ DecimalType.Unlimited
+ case _ =>
+ DoubleType
+ }
+ override def toString: String = s"STDDEV($child)"
+ override def asPartial: SplitEvaluation = {
+ val partialStd = Alias(ComputePartialStd(Cast(child, dataType)),
"PartialStddev")()
+ SplitEvaluation(CombinePartialStd(partialStd.toAttribute), partialStd
:: Nil)
+ }
+ override def newInstance(): StddevFunction = new StddevFunction(child,
this)
+}
+
+case class ComputePartialStd(child: Expression) extends
AggregateExpression {
+ def this() = this(null)
+
+ override def children: Seq[Expression] = child :: Nil
+ override def nullable: Boolean = false
+ override def dataType: DataType = child.dataType match {
+ case DecimalType.Unlimited => ArrayType(DecimalType.Unlimited)
+ case _ => ArrayType(DoubleType)
+ }
+ override def toString: String = s"computePartialStddev($child)"
+ override def newInstance(): ComputePartialStdFunction =
+ new ComputePartialStdFunction(child, this)
+}
+
+case class CombinePartialStd(child: Expression) extends
AggregateExpression {
+ def this() = this(null)
+
+ override def children: Seq[Expression] = child:: Nil
+ override def nullable: Boolean = false
+ override def dataType: DataType = child.dataType match {
+ case ArrayType(DecimalType.Unlimited, _) => DecimalType.Unlimited
+ case _ => DoubleType
+ }
+ override def toString: String = s"CombinePartialStd($child)"
+ override def newInstance(): CombinePartialStdFunction = {
+ new CombinePartialStdFunction(child, this)
+ }
+}
+
+case class ComputePartialStdFunction (
+ expr: Expression,
+ base: AggregateExpression
+) extends AggregateFunction {
+ def this() = this(null, null) // Required for serialization
+
+ private val computeType = expr.dataType
--- End diff --
Is `computeType` used for intermediate result? Why it just takes the
datatype of the expr?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
