You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by lw-lin <gi...@git.apache.org> on 2016/07/21 08:54:25 UTC
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
GitHub user lw-lin opened a pull request:
https://github.com/apache/spark/pull/14298
[SPARK-16283][SQL] Implement `percentile_approx` SQL function
## What changes were proposed in this pull request?
This patch Implements `percentile_approx` SQL function using Spark's implementation of G-K algorithm.
- commit 1: moves the G-K algorithm implementation(`QuantileSummaries` and related tests) from `sql/core` to `sql/catalyst`
- commit 2: implements `percentile_approx` using G-K algorithm
## How was this patch tested?
- Jenkins
- added new tests
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/lw-lin/spark impl_percentile_approx
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/14298.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #14298
----
commit d3a6dc825577a4a5e44e8eb0f8e61ef2053e127d
Author: Liwei Lin <lw...@gmail.com>
Date: 2016-07-21T08:29:00Z
Move G-K all from `sql/core` to `sql/catalyst`
commit 110158062cb1f6a571ad8e0bab9bc5962107b59a
Author: Liwei Lin <lw...@gmail.com>
Date: 2016-07-21T08:38:06Z
Implement percentile_approx
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/62676/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r73995159
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
Sure, this can wait. Thanks for the information!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62676 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62676/consoleFull)** for PR 14298 at commit [`1101580`](https://github.com/apache/spark/commit/110158062cb1f6a571ad8e0bab9bc5962107b59a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class PercentileApprox(`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r73991551
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
to be clear, are you saying that our current algorithm to compute `percentile_approx` can't support partial aggregation fundamentally?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by thunterdb <gi...@git.apache.org>.
Github user thunterdb commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r71740068
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either an single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ private var summary: QuantileSummaries = null
+
+ override def initialize(buffer: MutableRow): Unit = {
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ val relativeError = PercentileApprox.bToRelativeError(B)
+ summary = new QuantileSummaries(QuantileSummaries.defaultCompressThreshold, relativeError)
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+ val value = child.eval(input) match {
+ case o: Byte => o.toDouble
+ case o: Short => o.toDouble
+ case o: Int => o.toDouble
+ case o: Long => o.toDouble
+ case o: Float => o.toDouble
+ case o: Decimal => o.toDouble
+ case o: Double => o
+ }
+
+ summary.insert(value)
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit = {
+ sys.error("PercentileApprox does not support partial aggregation.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+ // summary must be compressed before being queried
+ summary = summary.compress()
+
+ if (resultAsArray) {
+ // return the result as an array of doubles, or return null for empty inputs
+ if (summary.count > 0) new GenericArrayData(percentiles.map { summary.query(_) }) else null
+ }
+ else {
+ // return the result as a double, or return null for empty inputs
+ if (summary.count > 0) summary.query(percentiles.head) else null
+ }
+ }
+
+ private def childrenSQL: String =
+ if (bExpr.isDefined) {
+ s"${child.sql}, ${percentilesExpr.toString}, ${bExpr.get.toString}"
+ }
+ else {
+ s"${child.sql}, ${percentilesExpr.toString}"
+ }
+
+ override def sql: String = s"$prettyName($childrenSQL)"
+
+ override def sql(isDistinct: Boolean): String = {
+ val distinct = if (isDistinct) "DISTINCT " else ""
+ s"$prettyName($distinct$childrenSQL)"
+ }
+}
+
+object PercentileApprox {
+
+ // Using 1000 as a default achieves a relativeError of 0.001
+ private[sql] val B_DEFAULT = 1000
+
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ private[sql] def bToRelativeError(B: Int): Double = Math.max(1.0d / B, 0.001)
+
+ /**
+ * Validates the percentile(s) expression and extract the percentile(s).
+ * Returns the extracted percentile(s) and an indicator of whether it's an array.
+ */
+ private def validatePercentilesLiteral(exp: Expression): (Seq[Double], Boolean) = {
+ def withinRange(v: Double): Boolean = 0.0 <= v && v <= 1.0
+ exp match {
+ case Literal(f: Float, FloatType) if withinRange(f) => (Seq(f.toDouble), false)
+ case Literal(d: Double, DoubleType) if withinRange(d) => (Seq(d), false)
+ case Literal(dec: Decimal, _) if withinRange(dec.toDouble) => (Seq(dec.toDouble), false)
+
+ case CreateArray(children: Seq[Expression]) if (children.length > 0) =>
+ (children.map(_ match {
+ case Literal(f: Float, FloatType) if withinRange(f) => f.toDouble
+ case Literal(d: Double, DoubleType) if withinRange(d) => d
+ case Literal(dec: Decimal, _) if withinRange(dec.toDouble) => dec.toDouble
+ case _ =>
+ throw new AnalysisException(
+ "The second argument should be a double literal or an array of doubles, and should " +
+ "be within range [0.0, 1.0]")
+ }), true)
+
+ case _ =>
+ throw new AnalysisException(
+ "The second argument should be a double literal or an array of doubles, and should " +
+ "be within range [0.0, 1.0]")
+ }
+ }
+
+ /** Validates the B expression and extract its value. */
+ private def validateBLiteral(exp: Expression): Int = exp match {
+ case Literal(i: Int, IntegerType) if i > 0 => i
+
+ case _ =>
+ throw new AnalysisException("The third argument should be a positive integer literal")
+ }
+}
+
+/**
+ * Helper class to compute approximate quantile summary.
+ * This implementation is based on the algorithm proposed in the paper:
--- End diff --
why did you need to move the implementation around?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r75830247
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
@clockfly cool!
now that https://github.com/apache/spark/pull/14754 has been merged, looks like I should update this patch once https://github.com/apache/spark/pull/14753 gets merged.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62670 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62670/consoleFull)** for PR 14298 at commit [`1101580`](https://github.com/apache/spark/commit/110158062cb1f6a571ad8e0bab9bc5962107b59a).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by dongjoon-hyun <gi...@git.apache.org>.
Github user dongjoon-hyun commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r72294685
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either an single double value or an
--- End diff --
minor: `an single` -> `a single`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin closed the pull request at:
https://github.com/apache/spark/pull/14298
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62881 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62881/consoleFull)** for PR 14298 at commit [`da87bfd`](https://github.com/apache/spark/commit/da87bfdb0ab91a265f74df01ec87221651583919).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class StringToMap(text: Expression, pairDelim: Expression, keyValueDelim: Expression)`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r73832267
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
can you explain a bit more about this? AFAIK, hive supports partial aggregate for `percentile_approx`, and it looks to me that your implementation keeps the buffer data(`QuantileSummaries`) in this aggregate function object, instead of letting aggregate operator manage it, that's the main reason why we can't support partial aggregate for `percentile_approx` I think.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62881 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62881/consoleFull)** for PR 14298 at commit [`da87bfd`](https://github.com/apache/spark/commit/da87bfdb0ab91a265f74df01ec87221651583919).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r73993189
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
Fundamentally the `QuantileSummaries` implementation does support RDD-style partial aggregation -- `QuantileSummaries` itself is the agg buffer of a partition's data at mappers, and multiple `QuantileSummaries`s will be merged at reducers (please refer to [StatFunctions.multipleApproxQuantiles](https://github.com/apache/spark/blob/5a3533e779d8e43ce0980203dfd3cbe343cc7d0a/sql/core/src/main/scala/org/apache/spark/sql/execution/stat/StatFunctions.scala#L93)).
Fundamentally it should also support the SparkSQL-style partial aggregation. I'm trying to reuse `QuantileSummaries`; if there's no easy way to reuse `QuantileSummaries`, I'm afraid we'll have to re-write this GK algorithm totally to support our SparkSQL-style partial aggregation
So any way we can just directly put a `QuantileSummaries` instance into SparkSQL's agg buffer? Or do we have to break a `QuantileSummaries` instance up, say into `an int + a double + a double[] + an int[] + an int[]` which SparkSQL's agg buffer can manage?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/62678/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62706 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62706/consoleFull)** for PR 14298 at commit [`da87bfd`](https://github.com/apache/spark/commit/da87bfdb0ab91a265f74df01ec87221651583919).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class StringToMap(text: Expression, pairDelim: Expression, keyValueDelim: Expression)`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r73990892
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
> hive supports partial aggregate for percentile_approx
Hive's implementation computes approximate percentile values from a histogram, thus Hive supports partial aggregation but makes no approximation guarantees.
> ... QuantileSummaries in this aggregate function object, instead of letting aggregate operator manage it, that's the main reason why we can't support partial aggregate
Yes that's quite right. `QuantileSummaries` has been implemented and well tested prior to this patch, so it'd be great if we can directly put a `QuantileSummaries` instance into the aggregation buffer (in order to support partial aggregation). @cloud-fan any pointer on how to do that?
Thanks!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r73101674
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either an single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
--- End diff --
I don't have strong preference here -- let's see what reviewers say.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/62670/
Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r72227222
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either an single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ private var summary: QuantileSummaries = null
+
+ override def initialize(buffer: MutableRow): Unit = {
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ val relativeError = PercentileApprox.bToRelativeError(B)
+ summary = new QuantileSummaries(QuantileSummaries.defaultCompressThreshold, relativeError)
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+ val value = child.eval(input) match {
+ case o: Byte => o.toDouble
+ case o: Short => o.toDouble
+ case o: Int => o.toDouble
+ case o: Long => o.toDouble
+ case o: Float => o.toDouble
+ case o: Decimal => o.toDouble
+ case o: Double => o
+ }
+
+ summary.insert(value)
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit = {
+ sys.error("PercentileApprox does not support partial aggregation.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+ // summary must be compressed before being queried
+ summary = summary.compress()
+
+ if (resultAsArray) {
+ // return the result as an array of doubles, or return null for empty inputs
+ if (summary.count > 0) new GenericArrayData(percentiles.map { summary.query(_) }) else null
+ }
+ else {
+ // return the result as a double, or return null for empty inputs
+ if (summary.count > 0) summary.query(percentiles.head) else null
+ }
+ }
+
+ private def childrenSQL: String =
+ if (bExpr.isDefined) {
+ s"${child.sql}, ${percentilesExpr.toString}, ${bExpr.get.toString}"
+ }
+ else {
+ s"${child.sql}, ${percentilesExpr.toString}"
+ }
+
+ override def sql: String = s"$prettyName($childrenSQL)"
+
+ override def sql(isDistinct: Boolean): String = {
+ val distinct = if (isDistinct) "DISTINCT " else ""
+ s"$prettyName($distinct$childrenSQL)"
+ }
+}
+
+object PercentileApprox {
+
+ // Using 1000 as a default achieves a relativeError of 0.001
+ private[sql] val B_DEFAULT = 1000
+
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ private[sql] def bToRelativeError(B: Int): Double = Math.max(1.0d / B, 0.001)
+
+ /**
+ * Validates the percentile(s) expression and extract the percentile(s).
+ * Returns the extracted percentile(s) and an indicator of whether it's an array.
+ */
+ private def validatePercentilesLiteral(exp: Expression): (Seq[Double], Boolean) = {
+ def withinRange(v: Double): Boolean = 0.0 <= v && v <= 1.0
+ exp match {
--- End diff --
You could also pattern match on the result of the `exp.eval()`. That would be way easier.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62678 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62678/consoleFull)** for PR 14298 at commit [`f21d746`](https://github.com/apache/spark/commit/f21d74613e414821adab2ed61c19cb3297e30ee6).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/62881/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62676 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62676/consoleFull)** for PR 14298 at commit [`1101580`](https://github.com/apache/spark/commit/110158062cb1f6a571ad8e0bab9bc5962107b59a).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the issue:
https://github.com/apache/spark/pull/14298
@cloud-fan could you also help review this? Thanks!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/63105/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Merged build finished. Test FAILed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r71815241
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either an single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ private var summary: QuantileSummaries = null
+
+ override def initialize(buffer: MutableRow): Unit = {
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ val relativeError = PercentileApprox.bToRelativeError(B)
+ summary = new QuantileSummaries(QuantileSummaries.defaultCompressThreshold, relativeError)
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+ val value = child.eval(input) match {
+ case o: Byte => o.toDouble
+ case o: Short => o.toDouble
+ case o: Int => o.toDouble
+ case o: Long => o.toDouble
+ case o: Float => o.toDouble
+ case o: Decimal => o.toDouble
+ case o: Double => o
+ }
+
+ summary.insert(value)
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit = {
+ sys.error("PercentileApprox does not support partial aggregation.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+ // summary must be compressed before being queried
+ summary = summary.compress()
+
+ if (resultAsArray) {
+ // return the result as an array of doubles, or return null for empty inputs
+ if (summary.count > 0) new GenericArrayData(percentiles.map { summary.query(_) }) else null
+ }
+ else {
+ // return the result as a double, or return null for empty inputs
+ if (summary.count > 0) summary.query(percentiles.head) else null
+ }
+ }
+
+ private def childrenSQL: String =
+ if (bExpr.isDefined) {
+ s"${child.sql}, ${percentilesExpr.toString}, ${bExpr.get.toString}"
+ }
+ else {
+ s"${child.sql}, ${percentilesExpr.toString}"
+ }
+
+ override def sql: String = s"$prettyName($childrenSQL)"
+
+ override def sql(isDistinct: Boolean): String = {
+ val distinct = if (isDistinct) "DISTINCT " else ""
+ s"$prettyName($distinct$childrenSQL)"
+ }
+}
+
+object PercentileApprox {
+
+ // Using 1000 as a default achieves a relativeError of 0.001
+ private[sql] val B_DEFAULT = 1000
+
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ private[sql] def bToRelativeError(B: Int): Double = Math.max(1.0d / B, 0.001)
+
+ /**
+ * Validates the percentile(s) expression and extract the percentile(s).
+ * Returns the extracted percentile(s) and an indicator of whether it's an array.
+ */
+ private def validatePercentilesLiteral(exp: Expression): (Seq[Double], Boolean) = {
+ def withinRange(v: Double): Boolean = 0.0 <= v && v <= 1.0
+ exp match {
+ case Literal(f: Float, FloatType) if withinRange(f) => (Seq(f.toDouble), false)
+ case Literal(d: Double, DoubleType) if withinRange(d) => (Seq(d), false)
+ case Literal(dec: Decimal, _) if withinRange(dec.toDouble) => (Seq(dec.toDouble), false)
+
+ case CreateArray(children: Seq[Expression]) if (children.length > 0) =>
+ (children.map(_ match {
+ case Literal(f: Float, FloatType) if withinRange(f) => f.toDouble
+ case Literal(d: Double, DoubleType) if withinRange(d) => d
+ case Literal(dec: Decimal, _) if withinRange(dec.toDouble) => dec.toDouble
+ case _ =>
+ throw new AnalysisException(
+ "The second argument should be a double literal or an array of doubles, and should " +
+ "be within range [0.0, 1.0]")
+ }), true)
+
+ case _ =>
+ throw new AnalysisException(
+ "The second argument should be a double literal or an array of doubles, and should " +
+ "be within range [0.0, 1.0]")
+ }
+ }
+
+ /** Validates the B expression and extract its value. */
+ private def validateBLiteral(exp: Expression): Int = exp match {
+ case Literal(i: Int, IntegerType) if i > 0 => i
+
+ case _ =>
+ throw new AnalysisException("The third argument should be a positive integer literal")
+ }
+}
+
+/**
+ * Helper class to compute approximate quantile summary.
+ * This implementation is based on the algorithm proposed in the paper:
--- End diff --
thanks!
module `sql/core` depends on module `sql/catalyst`, but module `sql/catalyst` does not depend on module `sql/core`; so in order to enable expressions like `PercentileApprox` (which lives in `sql/catalyst`) to use of the G-K algorithm, the algorithm should probably live in `sql/catalyst`. As a reference, algorithms like HyperLogLog also live in `sql/catalyst`(please refer to [HyperLogLogPlusPlus](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HyperLogLogPlusPlus.scala)).
@thunterdb does this make sense?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the issue:
https://github.com/apache/spark/pull/14298
@hvanhovell could you take a look at this? Thanks!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the issue:
https://github.com/apache/spark/pull/14298
Jenkins retest this please
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by thunterdb <gi...@git.apache.org>.
Github user thunterdb commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r72137984
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either an single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] = Seq()
+
+ private var summary: QuantileSummaries = null
+
+ override def initialize(buffer: MutableRow): Unit = {
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ val relativeError = PercentileApprox.bToRelativeError(B)
+ summary = new QuantileSummaries(QuantileSummaries.defaultCompressThreshold, relativeError)
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+ val value = child.eval(input) match {
+ case o: Byte => o.toDouble
+ case o: Short => o.toDouble
+ case o: Int => o.toDouble
+ case o: Long => o.toDouble
+ case o: Float => o.toDouble
+ case o: Decimal => o.toDouble
+ case o: Double => o
+ }
+
+ summary.insert(value)
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit = {
+ sys.error("PercentileApprox does not support partial aggregation.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+ // summary must be compressed before being queried
+ summary = summary.compress()
+
+ if (resultAsArray) {
+ // return the result as an array of doubles, or return null for empty inputs
+ if (summary.count > 0) new GenericArrayData(percentiles.map { summary.query(_) }) else null
+ }
+ else {
+ // return the result as a double, or return null for empty inputs
+ if (summary.count > 0) summary.query(percentiles.head) else null
+ }
+ }
+
+ private def childrenSQL: String =
+ if (bExpr.isDefined) {
+ s"${child.sql}, ${percentilesExpr.toString}, ${bExpr.get.toString}"
+ }
+ else {
+ s"${child.sql}, ${percentilesExpr.toString}"
+ }
+
+ override def sql: String = s"$prettyName($childrenSQL)"
+
+ override def sql(isDistinct: Boolean): String = {
+ val distinct = if (isDistinct) "DISTINCT " else ""
+ s"$prettyName($distinct$childrenSQL)"
+ }
+}
+
+object PercentileApprox {
+
+ // Using 1000 as a default achieves a relativeError of 0.001
+ private[sql] val B_DEFAULT = 1000
+
+ // Our `PercentileApprox` function takes a `B` parameter, but the underlying GK algorithm takes
+ // a `relativeError` parameter, so we need to convert `B` to `relativeError`.
+ // Please refer to SPARK-16283 for details.
+ private[sql] def bToRelativeError(B: Int): Double = Math.max(1.0d / B, 0.001)
+
+ /**
+ * Validates the percentile(s) expression and extract the percentile(s).
+ * Returns the extracted percentile(s) and an indicator of whether it's an array.
+ */
+ private def validatePercentilesLiteral(exp: Expression): (Seq[Double], Boolean) = {
+ def withinRange(v: Double): Boolean = 0.0 <= v && v <= 1.0
+ exp match {
+ case Literal(f: Float, FloatType) if withinRange(f) => (Seq(f.toDouble), false)
+ case Literal(d: Double, DoubleType) if withinRange(d) => (Seq(d), false)
+ case Literal(dec: Decimal, _) if withinRange(dec.toDouble) => (Seq(dec.toDouble), false)
+
+ case CreateArray(children: Seq[Expression]) if (children.length > 0) =>
+ (children.map(_ match {
+ case Literal(f: Float, FloatType) if withinRange(f) => f.toDouble
+ case Literal(d: Double, DoubleType) if withinRange(d) => d
+ case Literal(dec: Decimal, _) if withinRange(dec.toDouble) => dec.toDouble
+ case _ =>
+ throw new AnalysisException(
+ "The second argument should be a double literal or an array of doubles, and should " +
+ "be within range [0.0, 1.0]")
+ }), true)
+
+ case _ =>
+ throw new AnalysisException(
+ "The second argument should be a double literal or an array of doubles, and should " +
+ "be within range [0.0, 1.0]")
+ }
+ }
+
+ /** Validates the B expression and extract its value. */
+ private def validateBLiteral(exp: Expression): Int = exp match {
+ case Literal(i: Int, IntegerType) if i > 0 => i
+
+ case _ =>
+ throw new AnalysisException("The third argument should be a positive integer literal")
+ }
+}
+
+/**
+ * Helper class to compute approximate quantile summary.
+ * This implementation is based on the algorithm proposed in the paper:
--- End diff --
Yes, this makes sense, thanks.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #63105 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/63105/consoleFull)** for PR 14298 at commit [`c0acf16`](https://github.com/apache/spark/commit/c0acf1697ad369302068aeaecad59e812038d14a).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r76548403
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
@liancheng @cloud-fan should I update this or do you guys want to take this over?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on the issue:
https://github.com/apache/spark/pull/14298
@lw-lin
There are other issues to fix besides partial aggregation, like out of memory, argument type (B, and percentages should use catalyst expression instead so that we support constant folding), naming, input type check, sql. I have created another PR, and acknowledged your work in the PR description. Do you want to take a look at https://github.com/apache/spark/pull/14868?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #63105 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/63105/consoleFull)** for PR 14298 at commit [`c0acf16`](https://github.com/apache/spark/commit/c0acf1697ad369302068aeaecad59e812038d14a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/62706/
Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14298
Merged build finished. Test PASSed.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the issue:
https://github.com/apache/spark/pull/14298
@clockfly
Your PR's improvements overall look reasonable to me, so I'm closing this one. Good job!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r73993906
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
I think putting buffer object in agg buffer row is better, but that need to be well designed. Can you hold this PR for a while? We are discussing about it internally.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62678 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62678/consoleFull)** for PR 14298 at commit [`f21d746`](https://github.com/apache/spark/commit/f21d74613e414821adab2ed61c19cb3297e30ee6).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class PercentileApprox(`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the issue:
https://github.com/apache/spark/pull/14298
Jenkins retest this please
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62706 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62706/consoleFull)** for PR 14298 at commit [`da87bfd`](https://github.com/apache/spark/commit/da87bfdb0ab91a265f74df01ec87221651583919).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by clockfly <gi...@git.apache.org>.
Github user clockfly commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r75709632
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,462 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either a single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
+ resultAsArray: Boolean, // whether to return the result as an array
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ private def this(child: Expression, percentilesExpr: Expression, bExpr: Option[Expression]) = {
+ this(
+ child = child,
+ percentilesExpr = percentilesExpr,
+ bExpr = bExpr,
+ // validate and extract percentiles
+ percentiles = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._1,
+ // validate and extract B
+ B = bExpr.map(PercentileApprox.validateBLiteral(_)).getOrElse(PercentileApprox.B_DEFAULT),
+ // validate and mark whether we should return results as array of double or not
+ resultAsArray = PercentileApprox.validatePercentilesLiteral(percentilesExpr)._2)
+ }
+
+ // Constructor for the "_FUNC_(col, p) / _FUNC_(col, array(p1, ...))" form
+ def this(child: Expression, percentilesExpr: Expression) = {
+ this(child, percentilesExpr, None)
+ }
+
+ // Constructor for the "_FUNC_(col, p, B) / _FUNC_(col, array(p1, ...), B)" form
+ def this(child: Expression, percentilesExpr: Expression, bExpr: Expression) = {
+ this(child, percentilesExpr, Some(bExpr))
+ }
+
+ override def prettyName: String = "percentile_approx"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def children: Seq[Expression] =
+ bExpr.map(child :: percentilesExpr :: _ :: Nil).getOrElse(child :: percentilesExpr :: Nil)
+
+ // we would return null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = if (resultAsArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, AnyDataType, IntegralType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+ TypeUtils.checkForNumericExpr(child.dataType, "function percentile_approx")
+
+ // The number of intermediate outputs is highly relative to the actual data-set (an upper bound is
+ // (11/2e)log(2en), where e is the relativeError parameter, n is the number of items in the
+ // dataset) -- thus it's hard to allocate agg buffer in advance without knowing the size of
+ // inputs. Due to this reason, currently we don't support partial mode.
--- End diff --
Some updates, https://github.com/apache/spark/pull/14753
is created to support putting generic object in aggregation buffer.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14298
**[Test build #62670 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/62670/consoleFull)** for PR 14298 at commit [`1101580`](https://github.com/apache/spark/commit/110158062cb1f6a571ad8e0bab9bc5962107b59a).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class PercentileApprox(`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #14298: [SPARK-16283][SQL] Implement `percentile_approx` ...
Posted by dongjoon-hyun <gi...@git.apache.org>.
Github user dongjoon-hyun commented on a diff in the pull request:
https://github.com/apache/spark/pull/14298#discussion_r72295166
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileApprox.scala ---
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions.aggregate
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.aggregate.QuantileSummaries.Stats
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+
+/**
+ * Computes an approximate percentile (quantile) using the G-K algorithm (see below), for very
+ * large numbers of rows where the regular percentile() UDAF might run out of memory.
+ *
+ * The input is a single double value or an array of double values representing the percentiles
+ * requested. The output, corresponding to the input, is either an single double value or an
+ * array of doubles that are the percentile values.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(col, p [, B]) - Returns an approximate pth percentile of a numeric column in the
+ group. The B parameter, which defaults to 1000, controls approximation accuracy at the cost of
+ memory; higher values yield better approximations.
+ _FUNC_(col, array(p1 [, p2]...) [, B]) - Same as above, but accepts and returns an array of
+ percentile values instead of a single one.
+ """)
+case class PercentileApprox(
+ child: Expression,
+ percentilesExpr: Expression,
+ bExpr: Option[Expression],
+ percentiles: Seq[Double], // the extracted percentiles
+ B: Int, // the extracted B
--- End diff --
Hi, @lw-lin .
I know the reason why you define this as a capatal 'B', but I'm just wondering it's consistent with Spark naming rule.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #14298: [SPARK-16283][SQL] Implement `percentile_approx` SQL fun...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the issue:
https://github.com/apache/spark/pull/14298
@hvanhovell comments addressed. Please let me know when there's more to do.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org