[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user asfgit closed the pull request at: https://github.com/apache/spark/pull/14136 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user lins05 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89686604
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,262 @@
+/*
+ * 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 java.io.{ByteArrayInputStream, ByteArrayOutputStream,
DataInputStream, DataOutputStream}
+import java.util
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import
org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure,
TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[OpenHashMap[Number, Long]] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ @transient
+ private lazy val returnPercentileArray =
percentageExpression.dataType.isInstanceOf[ArrayType]
+
+ @transient
+ private lazy val percentages = percentageExpression.eval() match {
+case p: Double => Seq(p)
+case a: ArrayData => a.toDoubleArray().toSeq
+ }
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override lazy val dataType: DataType = percentageExpression.dataType
match {
+case _: ArrayType => ArrayType(DoubleType, false)
+case _ => DoubleType
+ }
+
+ override def inputTypes: Seq[AbstractDataType] =
percentageExpression.dataType match {
+case _: ArrayType => Seq(NumericType, ArrayType(DoubleType, false))
+case _ => Seq(NumericType, DoubleType)
+ }
+
+ // Check the i
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89672530
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89672499
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,326 @@
+/*
+ * 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 java.io.{ByteArrayInputStream, ByteArrayOutputStream,
DataInputStream, DataOutputStream}
+import java.util
+
+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.analysis.TypeCheckResult.{TypeCheckFailure,
TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import
org.apache.spark.sql.catalyst.expressions.aggregate.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val returnPercentileArray =
percentageExpression.dataType.isInstanceOf[ArrayType]
+
+ @transient
+ private lazy val percentages = evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override lazy val dataType: DataType = percentageExpression.dataType
match {
+case _: ArrayType => ArrayType(DoubleType, false)
+case _ => DoubleType
+ }
+
+ override def inputTypes: Seq[AbstractDataType] =
percentageExpression.dataType match {
+case _: ArrayType => Seq(NumericType, ArrayType(DoubleType, false))
+case _ => Seq(NumericType, DoubleType)
+ }
+
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89672494
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,326 @@
+/*
+ * 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 java.io.{ByteArrayInputStream, ByteArrayOutputStream,
DataInputStream, DataOutputStream}
+import java.util
+
+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.analysis.TypeCheckResult.{TypeCheckFailure,
TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import
org.apache.spark.sql.catalyst.expressions.aggregate.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val returnPercentileArray =
percentageExpression.dataType.isInstanceOf[ArrayType]
+
+ @transient
+ private lazy val percentages = evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override lazy val dataType: DataType = percentageExpression.dataType
match {
+case _: ArrayType => ArrayType(DoubleType, false)
+case _ => DoubleType
+ }
+
+ override def inputTypes: Seq[AbstractDataType] =
percentageExpression.dataType match {
+case _: ArrayType => Seq(NumericType, ArrayType(DoubleType, false))
+case _ => Seq(NumericType, DoubleType)
+ }
+
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89672476
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,326 @@
+/*
+ * 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 java.io.{ByteArrayInputStream, ByteArrayOutputStream,
DataInputStream, DataOutputStream}
+import java.util
+
+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.analysis.TypeCheckResult.{TypeCheckFailure,
TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import
org.apache.spark.sql.catalyst.expressions.aggregate.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val returnPercentileArray =
percentageExpression.dataType.isInstanceOf[ArrayType]
--- End diff --
Mark it `@transient`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89672415
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,326 @@
+/*
+ * 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 java.io.{ByteArrayInputStream, ByteArrayOutputStream,
DataInputStream, DataOutputStream}
+import java.util
+
+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.analysis.TypeCheckResult.{TypeCheckFailure,
TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions._
+import
org.apache.spark.sql.catalyst.expressions.aggregate.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val returnPercentileArray =
percentageExpression.dataType.isInstanceOf[ArrayType]
+
+ @transient
+ private lazy val percentages = evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override lazy val dataType: DataType = percentageExpression.dataType
match {
+case _: ArrayType => ArrayType(DoubleType, false)
+case _ => DoubleType
+ }
+
+ override def inputTypes: Seq[AbstractDataType] =
percentageExpression.dataType match {
+case _: ArrayType => Seq(NumericType, ArrayType(DoubleType, false))
+case _ => Seq(NumericType, DoubleType)
+ }
+
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89669049
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.da
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89652990
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89647985
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89646058
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89654874
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89645958
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89654611
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89647893
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89652140
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89645855
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,201 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Number, Long]
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
--- End diff --
BTW - you can make the analyzer add casts for you:
```scala
override def inputTypes: Seq[AbstractDataType] =
percentageExpression.dataType match {
case _: ArrayType => Seq(NumericType, ArrayType(DoubleType, false))
case _ => Seq(NumericType, DoubleType)
}
```
Then you are alway sure you get a double or a double array for the
`percentageExpression`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89644591
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
--- End diff --
This can be problematic with serialization. Just put the percentages in a
`@transient lazy val` and inline the use of `returnPercentileArray`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89644918
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
--- End diff --
```scala
override lazy val dataType: DataType = percentageExpression.dataType match {
case _: ArrayType => ArrayType(DoubleType, false)
case _ => DoubleType
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-m
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89644764
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
--- End diff --
I think we should return the type of the input. We can always interpolate
the value and cast that to the input type. Is this is different from what Hive
does?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89645138
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
--- End diff --
Call `super.checkInputDataTypes()`, that will validate the inputTypes().
Also check the `percentageExpression`, that must foldable and the percentage(s)
must be in the range [0, 1].
---
If your project is set up for it, you can reply to this email a
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89644268
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
--- End diff --
Why not return do
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89644178
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,201 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Number, Long]
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
--- End diff --
Supporting `NumericType` does not really make sense for the percentage
value. Use `FractionalType` instead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89616284
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,292 @@
+/*
+ * 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 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.Percentile.Countings
+import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.Platform.BYTE_ARRAY_OFFSET
+import org.apache.spark.util.collection.OpenHashMap
+
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends
TypedImperativeAggregate[Countings] {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): Percentile =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
Percentile =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def createAggregationBuffer(): Countings = {
+// Initialize new Countings instance here.
+Countings()
+ }
+
+ private def evalPercentages(expr: Expression): (Boolean, Seq[Number]) = {
+val (isArrayType, values) = (expr.dat
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89465541
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,201 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Number, Long]
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
--- End diff --
We can't specify the type of the array(`FractionalType`), do we want to
support `DecimalType` in the array?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
--
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r89339672
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,201 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Number, Long]
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
--- End diff --
I'm not sure we should support AtomicTypes because not all of them could be
converted to double, and we have to use double values to do interpolation
operations.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r88502344
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,201 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Number, Long]
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] = Nil
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] = Nil
+
+ override def initialize(buffer: Inter
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r88500845
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,201 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function returns the exact percentile(s) of
numeric column `expr` at
+ * the given percentage(s) with value range in [0.0, 1.0].
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ *
+ * @param child child expression that produce numeric column value with
`child.eval(inputRow)`
+ * @param percentageExpression Expression that represents a single
percentage value or an array of
+ * percentage values. Each percentage value
must be in the range
+ * [0.0, 1.0].
+ */
+@ExpressionDescription(
+ usage =
+"""
+ _FUNC_(col, percentage) - Returns the exact percentile value of
numeric column `col` at the
+ given percentage. The value of percentage must be between 0.0 and
1.0.
+
+ _FUNC_(col, array(percentage1 [, percentage2]...)) - Returns the
exact percentile value array
+ of numeric column `col` at the given percentage(s). Each value of
the percentage array must
+ be between 0.0 and 1.0.
+""")
+case class Percentile(
+ child: Expression,
+ percentageExpression: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, percentageExpression: Expression) = {
+this(child, percentageExpression, 0, 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Number, Long]
+
+ // Mark as lazy so that percentageExpression is not evaluated during
tree transformation.
+ private lazy val (returnPercentileArray: Boolean, percentages:
Seq[Number]) =
+evalPercentages(percentageExpression)
+
+ override def children: Seq[Expression] = child :: percentageExpression
:: Nil
+
+ // Returns null for empty inputs
+ override def nullable: Boolean = true
+
+ override def dataType: DataType =
+if (returnPercentileArray) ArrayType(DoubleType) else DoubleType
+
+ override def inputTypes: Seq[AbstractDataType] =
+Seq(NumericType, TypeCollection(NumericType, ArrayType))
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForNumericExpr(child.dataType, "function percentile")
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] = Nil
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] = Nil
+
+ override def initialize(buffer: Inter
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r84839715
--- Diff:
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileSuite.scala
---
@@ -0,0 +1,192 @@
+/*
+ * 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 org.apache.spark.SparkFunSuite
+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.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.types._
+
+class PercentileSuite extends SparkFunSuite {
+
+ test("high level interface, update, merge, eval...") {
+val count = 1
+val data = (1 to count)
+val percentages = Array(0, 0.25, 0.5, 0.75, 1)
+val expectedPercentiles = Array(1, 2500.75, 5000.5, 7500.25, 1)
+val childExpression = Cast(BoundReference(0, IntegerType, nullable =
false), DoubleType)
+val percentageExpression =
CreateArray(percentages.toSeq.map(Literal(_)))
+val agg = new Percentile(childExpression, percentageExpression)
+
+assert(agg.nullable)
+val group = (0 until data.length)
+// Don't use group buffer for now.
+val groupBuffer = InternalRow.empty
+group.foreach { index =>
+ val input = InternalRow(data(index))
+ agg.update(groupBuffer, input)
+}
+
+// Don't support partial aggregations for now.
--- End diff --
Perhaps we could pass the HashMap `counts` as `inputBuffer` in `merge`
function, then we can generate local `counts` map first, after that we merge
these maps and compute the `eval()` function.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r84838527
--- Diff:
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PercentileSuite.scala
---
@@ -0,0 +1,192 @@
+/*
+ * 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 org.apache.spark.SparkFunSuite
+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.dsl.expressions._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.types._
+
+class PercentileSuite extends SparkFunSuite {
+
+ test("high level interface, update, merge, eval...") {
+val count = 1
+val data = (1 to count)
+val percentages = Array(0, 0.25, 0.5, 0.75, 1)
+val expectedPercentiles = Array(1, 2500.75, 5000.5, 7500.25, 1)
+val childExpression = Cast(BoundReference(0, IntegerType, nullable =
false), DoubleType)
+val percentageExpression =
CreateArray(percentages.toSeq.map(Literal(_)))
+val agg = new Percentile(childExpression, percentageExpression)
+
+assert(agg.nullable)
+val group = (0 until data.length)
+// Don't use group buffer for now.
+val groupBuffer = InternalRow.empty
+group.foreach { index =>
+ val input = InternalRow(data(index))
+ agg.update(groupBuffer, input)
+}
+
+// Don't support partial aggregations for now.
--- End diff --
for now? is percentile ever possible to do with partial agg?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r84591480
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/functions.scala ---
@@ -613,6 +613,46 @@ object functions {
def min(columnName: String): Column = min(Column(columnName))
/**
+ * Aggregate function: returns the exact percentile(s) of the expression
in a group at pc with
--- End diff --
let's not add the scala api for this for now. i'm not sure if we want to
encourage users to use it -- it is super expensive.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r84591459
--- Diff:
sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
---
@@ -136,7 +136,7 @@ class HiveUDFSuite extends QueryTest with
TestHiveSingleton with SQLTestUtils {
test("SPARK-2693 udaf aggregates test") {
checkAnswer(sql("SELECT percentile(key, 1) FROM src LIMIT 1"),
- sql("SELECT max(key) FROM src").collect().toSeq)
+ sql("SELECT array(max(key)) FROM src").collect().toSeq)
--- End diff --
what is this change about?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r84591451
--- Diff:
sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
---
@@ -851,6 +851,42 @@ abstract class AggregationQuerySuite extends QueryTest
with SQLTestUtils with Te
checkAnswer(df3.groupBy().agg(covar_pop("a", "b")), Row(0.0))
}
+ test("percentile") {
--- End diff --
i don't think you need the test case here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r84591458
--- Diff:
sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
---
@@ -851,6 +851,42 @@ abstract class AggregationQuerySuite extends QueryTest
with SQLTestUtils with Te
checkAnswer(df3.groupBy().agg(covar_pop("a", "b")), Row(0.0))
}
+ test("percentile") {
--- End diff --
basically i think this entire suite is mostly historic and pretty much
obsolete now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user vectorijk commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70834638
--- Diff:
sql/core/src/test/scala/org/apache/spark/sql/DataFrameAggregateSuite.scala ---
@@ -475,4 +475,20 @@ class DataFrameAggregateSuite extends QueryTest with
SharedSQLContext {
spark.sql("select avg(a) over () from values 1.0, 2.0, 3.0 T(a)"),
Row(2.0) :: Row(2.0) :: Row(2.0) :: Nil)
}
+
+ test("percentile functions") {
+val df = Seq(1, 3, 3, 6, 5, 4, 17, 38, 29, 400).toDF("a")
+checkAnswer(
+ df.select(percentile($"a", 0.5d), percentile($"a", Seq(0d, 0.75d,
1d))),
+ Seq(Row(Seq(5.5), Seq(1.0, 26.0, 400.0)))
+)
+ }
+
+ test("percentile functions with zero input rows.") {
+val df = Seq(1, 3, 3, 6, 5, 4, 17, 38, 29, 400).toDF("a").where($"a" <
0)
+checkAnswer(
+ df.select(percentile($"a", 0.5d)),
+ Seq(Row(Seq.empty))
+)
+ }
--- End diff --
I think we should also give a test to see what if pass an empty array of
percentiles. It may throw an exception or error.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70823811
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,172 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Expression) = {
+this(child = child, pc = pc, mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ private val percentiles: Seq[Double] = pc match {
+case Literal(ar: GenericArrayData, _: ArrayType) =>
+ ar.asInstanceOf[GenericArrayData].array.map{ d =>
d.asInstanceOf[Double]}
+case _ => Seq.empty
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Double, Long]()
+
+ override def children: Seq[Expression] = child :: pc :: Nil
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType,
NumericType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForOrderingExpr(child.dataType, "function percentile")
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
percentiles.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+var i = 0
+while (i < percentiles.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+ i += 1
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+val key = v 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
+ case _ => sys.error("Percentile is restricted to Numeric types
only.")
--- End diff --
I don't this this is
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70823475
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,172 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Expression) = {
+this(child = child, pc = pc, mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ private val percentiles: Seq[Double] = pc match {
+case Literal(ar: GenericArrayData, _: ArrayType) =>
+ ar.asInstanceOf[GenericArrayData].array.map{ d =>
d.asInstanceOf[Double]}
+case _ => Seq.empty
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Double, Long]()
+
+ override def children: Seq[Expression] = child :: pc :: Nil
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType,
NumericType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForOrderingExpr(child.dataType, "function percentile")
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
percentiles.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+var i = 0
+while (i < percentiles.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+ i += 1
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+val key = v match {
--- End diff --
Why force everything into a double? We can also store the value. And
potentially convert it to a double when interpolation is needed.
---
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, pleas
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70823276
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,172 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Expression) = {
+this(child = child, pc = pc, mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ private val percentiles: Seq[Double] = pc match {
+case Literal(ar: GenericArrayData, _: ArrayType) =>
+ ar.asInstanceOf[GenericArrayData].array.map{ d =>
d.asInstanceOf[Double]}
+case _ => Seq.empty
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ private var counts = new OpenHashMap[Double, Long]()
+
+ override def children: Seq[Expression] = child :: pc :: Nil
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(NumericType,
NumericType)
+
+ override def checkInputDataTypes(): TypeCheckResult =
+TypeUtils.checkForOrderingExpr(child.dataType, "function percentile")
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
percentiles.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+var i = 0
+while (i < percentiles.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+ i += 1
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+val key = v match {
--- End diff --
is it faster to a pattern match for every value, or to wrap this in a
function?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70822992
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
--- End diff --
the coins `openhashmap` will contain values of other groups if you do not
initialize it here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70822715
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,172 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Expression) = {
+this(child = child, pc = pc, mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ private val percentiles: Seq[Double] = pc match {
+case Literal(ar: GenericArrayData, _: ArrayType) =>
+ ar.asInstanceOf[GenericArrayData].array.map{ d =>
d.asInstanceOf[Double]}
--- End diff --
We should also check if the array is not empty.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70822633
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,172 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Expression) = {
+this(child = child, pc = pc, mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ private val percentiles: Seq[Double] = pc match {
+case Literal(ar: GenericArrayData, _: ArrayType) =>
+ ar.asInstanceOf[GenericArrayData].array.map{ d =>
d.asInstanceOf[Double]}
+case _ => Seq.empty
--- End diff --
I think we should throw an error here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70822426
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,172 @@
+/*
+ * 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 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.util._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ *
+ * The operator is bound to the slower sort based aggregation path because
the number of elements
+ * and their partial order cannot be determined in advance. Therefore we
have to store all the
+ * elements in memory, and that too many elements can cause GC paused and
eventually OutOfMemory
+ * Errors.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0) extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Expression) = {
+this(child = child, pc = pc, mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ private val percentiles: Seq[Double] = pc match {
+case Literal(ar: GenericArrayData, _: ArrayType) =>
+ ar.asInstanceOf[GenericArrayData].array.map{ d =>
d.asInstanceOf[Double]}
--- End diff --
Shouldn't we check here if a percentile is valid? Waiting until `eval` is
really late in the game.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user vectorijk commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70294379
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
--- End diff --
@hvanhovell 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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70286689
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
--- End diff --
OpenHashMap is typically faster and has less overhead.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user vectorijk commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70285450
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
--- End diff --
@jiangxb1987 I am just curious about why we use `OpenHashMap` here instead
of using `mutable.Map` to correspond with code
[here](https://github.com/apache/hive/blob/master/ql/src/java/org/apache/hadoop/hive/ql/udf/UDAFPercentile.java#L58)
in hive. Is there any specific reason?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70254020
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/functions.scala ---
@@ -536,6 +536,25 @@ object functions {
def min(columnName: String): Column = min(Column(columnName))
/**
+ * Aggregate function: returns the exact percentile(s) of the expression
in a group at pc with
+ * range in [0, 1].
+ *
+ * @group agg_funcs
+ * @since 2.1.0
+ */
+ def percentile(e: Column, pc: Seq[Double]): Column =
+withAggregateFunction {Percentile(e.expr, pc)}
--- End diff --
Would this work:
```scala
def percentile(e: Column, pc: Seq[Double]): Column = withAggregateFunction {
Percentile(e.expr, pc)
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70245302
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/functions.scala ---
@@ -536,6 +536,25 @@ object functions {
def min(columnName: String): Column = min(Column(columnName))
/**
+ * Aggregate function: returns the exact percentile(s) of the expression
in a group at pc with
+ * range in [0, 1].
+ *
+ * @group agg_funcs
+ * @since 2.1.0
+ */
+ def percentile(e: Column, pc: Seq[Double]): Column =
+withAggregateFunction {Percentile(e.expr, pc)}
--- End diff --
Combining that two lines will lead to a line length over 100 characters
which will trigger a scalastyle check failure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70245059
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+v match {
+ case o: Int => counts.changeValue(o.toLong, 1L, _ + 1L)
+ case o: Long => counts.changeValue(o, 1L, _ + 1L)
+ case _ => return false
+}
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit =
{
+sys.error("Percentile cannot be used in partial aggregations.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+if (counts.size == 0) {
+ return new GenericArrayData(Seq.empty)
+}
+
+// Sort all items and generate a sequence, then accumulate the counts
+val sortedCounts = counts.toSeq.sortBy(_._1)
+val aggreCounts = sortedCounts.scanLeft(0L, 0L) { (k1: (Long, Long),
k2: (Long, Long)) =>
+ (k2._1, k1._2 + k2._2)
+}.drop(1)
+val maxPosition = aggreCounts.last._2 - 1
+
+new GenericArrayData(pc.map { percentile =>
+ if (percentile < 0.0 || percentile > 1.0) {
+sys.error("Percentile value must be within the range of 0 to 1.")
+ }
+ getPercentile(aggreCounts, maxPosition * percentile)
+})
+ }
+
+ /**
+ * Get the
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70244013
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+v match {
+ case o: Int => counts.changeValue(o.toLong, 1L, _ + 1L)
+ case o: Long => counts.changeValue(o, 1L, _ + 1L)
+ case _ => return false
+}
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit =
{
+sys.error("Percentile cannot be used in partial aggregations.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+if (counts.size == 0) {
+ return new GenericArrayData(Seq.empty)
+}
+
+// Sort all items and generate a sequence, then accumulate the counts
+val sortedCounts = counts.toSeq.sortBy(_._1)
+val aggreCounts = sortedCounts.scanLeft(0L, 0L) { (k1: (Long, Long),
k2: (Long, Long)) =>
+ (k2._1, k1._2 + k2._2)
+}.drop(1)
+val maxPosition = aggreCounts.last._2 - 1
+
+new GenericArrayData(pc.map { percentile =>
+ if (percentile < 0.0 || percentile > 1.0) {
--- End diff --
Yep, you are right.
---
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
enable
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70243723
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
--- End diff --
I'll implement it, 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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70243630
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+v match {
+ case o: Int => counts.changeValue(o.toLong, 1L, _ + 1L)
--- End diff --
In HIVE percentile function supports only BigInt Column, so I thought we
should be consistent with that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70243463
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
--- End diff --
Sure, I'll move it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70243309
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
--- End diff --
Sorry...seems I left out something.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70243206
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/functions.scala ---
@@ -536,6 +536,25 @@ object functions {
def min(columnName: String): Column = min(Column(columnName))
/**
+ * Aggregate function: returns the exact percentile(s) of the expression
in a group at pc with
+ * range in [0, 1].
+ *
+ * @group agg_funcs
+ * @since 2.1.0
+ */
+ def percentile(e: Column, pc: Seq[Double]): Column =
+withAggregateFunction {Percentile(e.expr, pc)}
--- End diff --
Style: match the other code, e.g.: `def min(e: Column): Column =
withAggregateFunction { Min(e.expr) }`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70243163
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
--- End diff --
Yep, I'll update that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70242786
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+v match {
+ case o: Int => counts.changeValue(o.toLong, 1L, _ + 1L)
+ case o: Long => counts.changeValue(o, 1L, _ + 1L)
+ case _ => return false
+}
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit =
{
+sys.error("Percentile cannot be used in partial aggregations.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+if (counts.size == 0) {
+ return new GenericArrayData(Seq.empty)
+}
+
+// Sort all items and generate a sequence, then accumulate the counts
+val sortedCounts = counts.toSeq.sortBy(_._1)
+val aggreCounts = sortedCounts.scanLeft(0L, 0L) { (k1: (Long, Long),
k2: (Long, Long)) =>
+ (k2._1, k1._2 + k2._2)
+}.drop(1)
+val maxPosition = aggreCounts.last._2 - 1
+
+new GenericArrayData(pc.map { percentile =>
+ if (percentile < 0.0 || percentile > 1.0) {
+sys.error("Percentile value must be within the range of 0 to 1.")
+ }
+ getPercentile(aggreCounts, maxPosition * percentile)
+})
+ }
+
+ /**
+ * Get the p
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70242570
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+v match {
+ case o: Int => counts.changeValue(o.toLong, 1L, _ + 1L)
+ case o: Long => counts.changeValue(o, 1L, _ + 1L)
+ case _ => return false
+}
+ }
+
+ override def merge(buffer: MutableRow, inputBuffer: InternalRow): Unit =
{
+sys.error("Percentile cannot be used in partial aggregations.")
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+if (counts.size == 0) {
+ return new GenericArrayData(Seq.empty)
+}
+
+// Sort all items and generate a sequence, then accumulate the counts
+val sortedCounts = counts.toSeq.sortBy(_._1)
+val aggreCounts = sortedCounts.scanLeft(0L, 0L) { (k1: (Long, Long),
k2: (Long, Long)) =>
+ (k2._1, k1._2 + k2._2)
+}.drop(1)
+val maxPosition = aggreCounts.last._2 - 1
+
+new GenericArrayData(pc.map { percentile =>
+ if (percentile < 0.0 || percentile > 1.0) {
--- End diff --
We can check this at a much earlier stage.
---
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 ha
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70242502
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
--- End diff --
Please implement `checkInputDataTypes` in order to make sure the data type
supports sorting.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70242099
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
+ buffer.setNullAt(mutableAggBufferOffset + i)
+}
+ }
+
+ override def update(buffer: MutableRow, input: InternalRow): Unit = {
+val v = child.eval(input)
+
+v match {
+ case o: Int => counts.changeValue(o.toLong, 1L, _ + 1L)
--- End diff --
We only support Ints and Longs? Why do we care about the type here? The
only thing that matter is that we can sort the keys later on.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70242134
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
--- End diff --
nvm
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70241891
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
--- End diff --
Initialize counts here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70241660
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
--- End diff --
The doc says you return a Double or an Array of doubles. Which one is it?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/14136#discussion_r70241490
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala
---
@@ -0,0 +1,148 @@
+/*
+ * 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 org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.util.GenericArrayData
+import org.apache.spark.sql.types._
+import org.apache.spark.util.collection.OpenHashMap
+
+/**
+ * The Percentile aggregate function computes the exact percentile(s) of
expr at pc with range in
+ * [0, 1].
+ * The parameter pc can be a DoubleType or DoubleType array.
+ */
+@ExpressionDescription(
+ usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc
(range: [0,1]). pc can be
+ a double or double array.""")
+case class Percentile(
+ child: Expression,
+ pc: Seq[Double],
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate {
+
+ def this(child: Expression, pc: Double) = {
+this(child = child, pc = Seq(pc), mutableAggBufferOffset = 0,
inputAggBufferOffset = 0)
+ }
+
+ override def prettyName: String = "percentile"
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset:
Int): ImperativeAggregate =
+copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ var counts = new OpenHashMap[Long, Long]()
+
+ override def children: Seq[Expression] = Seq(child)
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = ArrayType(DoubleType)
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(AnyDataType)
+
+ override def supportsPartial: Boolean = false
+
+ override def aggBufferSchema: StructType =
StructType.fromAttributes(aggBufferAttributes)
+
+ override val aggBufferAttributes: Seq[AttributeReference] =
pc.map(percentile =>
+AttributeReference(percentile.toString, DoubleType)())
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: MutableRow): Unit = {
+for (i <- 0 until pc.size) {
--- End diff --
Please use a while loop here; for is not that efficient.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
Github user hvanhovell commented on a diff in the pull request: https://github.com/apache/spark/pull/14136#discussion_r70241201 --- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala --- @@ -0,0 +1,148 @@ +/* + * 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 org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.util.GenericArrayData +import org.apache.spark.sql.types._ +import org.apache.spark.util.collection.OpenHashMap + +/** + * The Percentile aggregate function computes the exact percentile(s) of expr at pc with range in + * [0, 1]. + * The parameter pc can be a DoubleType or DoubleType array. + */ +@ExpressionDescription( + usage = """_FUNC_(epxr, pc) - Returns the percentile(s) of expr at pc (range: [0,1]). pc can be + a double or double array.""") +case class Percentile( + child: Expression, --- End diff -- Nit: Style --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #14136: [SPARK-16282][SQL] Implement percentile SQL funct...
GitHub user jiangxb1987 opened a pull request: https://github.com/apache/spark/pull/14136 [SPARK-16282][SQL] Implement percentile SQL function. ## What changes were proposed in this pull request? Implement percentile SQL function. It computes the exact percentile(s) of expr at pc with range in [0, 1]. ## How was this patch tested? Added new testcases in DataFrameAggregateSuite. You can merge this pull request into a Git repository by running: $ git pull https://github.com/jiangxb1987/spark percentile Alternatively you can review and apply these changes as the patch at: https://github.com/apache/spark/pull/14136.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 #14136 commit 1ae3df7463f36ac15c4cb3138bec3bade7eae600 Author: èæå Date: 2016-07-11T11:11:26Z [SPARK-16282][SQL] Implement percentile SQL function. --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
