You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by wzhfy <gi...@git.apache.org> on 2016/10/26 03:39:40 UTC
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
GitHub user wzhfy opened a pull request:
https://github.com/apache/spark/pull/15637
[SPARK-18000] [SQL] Aggregation function for computing endpoints for histograms
## What changes were proposed in this pull request?
This function for a column returns bins - (distinct value, frequency) pairs
of equi-width histogram when the number of distinct values is less than or equal to the
specified maximum number of bins. Otherwise, for column of string type, it returns an empty
map; for column of numeric type, it returns endpoints of equi-height histogram - approximate
percentiles at percentages 0.0, 1/numBins, 2/numBins, ..., (numBins-1)/numBins, 1.0.
## How was this patch tested?
add test cases
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/wzhfy/spark histogramEndpoints
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/15637.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 #15637
----
commit 1580a7fb0a6a21b603e754c744c8c6cb02fd24c2
Author: wangzhenhua <wa...@huawei.com>
Date: 2016-10-12T01:02:37Z
add agg function to generate string histogram
commit a3281606372f83eca960ea90734e8ee9cb1c3125
Author: wangzhenhua <wa...@huawei.com>
Date: 2016-10-21T08:10:01Z
comments
commit 907cd99b8b26ae3caa224df67cc10bc784f10fb4
Author: Zhenhua Wang <wz...@163.com>
Date: 2016-10-22T12:15:22Z
create HistogramEndpoints to generate endpoints for string and numeric types
commit 35e453cb1079398196ece4f13e8f294ee4e4e916
Author: Zhenhua Wang <wz...@163.com>
Date: 2016-10-23T15:05:09Z
change suite names
commit f6fe25de3f5f1382727cecfdda7b74e40758896b
Author: wangzhenhua <wa...@huawei.com>
Date: 2016-10-26T03:20:05Z
test cases and fix bugs
commit 15eb3721f56ac27bd90933ef7e66f3453eae4a75
Author: wangzhenhua <wa...@huawei.com>
Date: 2016-10-26T03:29:14Z
fix doc
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
I mean count_min_sketch(hash(col1)) returns result of hash(col1) rather than col1, right? but we need the result of col1.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86504634
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
+
+ // Update bins and clear it when its size exceeds numBins.
+ def updateMap(value: T, numBins: Int): Unit = {
+ mergeMap(mutable.HashMap(value -> 1L), numBins)
+ }
+
+ // Merge two maps and clear bins when its size exceeds numBins.
+ override def merge(otherDigest: MapDigest, numBins: Int): Unit = {
+ val otherMap = otherDigest.asInstanceOf[MapDigestBase[T]].bins
+ mergeMap(otherMap, numBins)
+ }
+
+ def mergeMap(otherMap: mutable.HashMap[T, Long], numBins: Int): Unit = {
+ otherMap.foreach { case (key, value) =>
+ if (bins.contains(key)) {
+ bins.update(key, bins(key) + value)
+ } else {
+ if (bins.size >= numBins) {
+ isInvalid = true
+ bins.clear()
+ return
--- End diff --
don't use return within a foreach https://github.com/databricks/scala-style-guide#return
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85934978
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $numBins)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (buffer.isInvalid) {
+ return
+ }
+ val evaluated = child.eval(input)
+ if (evaluated != null) {
+ buffer.update(child.dataType, evaluated, numBins)
+ }
--- End diff --
Yeah, that's better, thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
For equi-height hsitogram, we need extra info like ndv's in each bin. Does count min sketch also have this information? I had a discussion with Herman and Tim about histogram construction before, we want to compute all stats including histograms in a single table scan. They suggested we modify ApproximatePercentile to get equi-height histograms, i.e. endpoints of bins and the ndv's within them. And this pr is only for getting equi-width histograms. You can check [this JIRA](https://issues.apache.org/jira/browse/SPARK-17074) for the discussion.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67769 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67769/consoleFull)** for PR 15637 at commit [`9d65f6f`](https://github.com/apache/spark/commit/9d65f6f174e6790ef3ea40dbd81175ab07c37401).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67806 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67806/consoleFull)** for PR 15637 at commit [`98d502b`](https://github.com/apache/spark/commit/98d502b3c0f3dce2d743de7448f5af5ad4c419ae).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
retest this please
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
It supports arbitrary objects
```
/**
* Increments {@code item}'s count by one.
*/
public abstract void add(Object item);
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86505467
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
+
+ // Update bins and clear it when its size exceeds numBins.
+ def updateMap(value: T, numBins: Int): Unit = {
+ mergeMap(mutable.HashMap(value -> 1L), numBins)
--- End diff --
once you do that you can also just put mergeMap into merge.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86504716
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
--- End diff --
use java.util.HashMap for better performance
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
@srinathshankar Yeah, I think your proposal is better. Thanks for the advice! I'll update this pr asap.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
@rxin About the name "map_aggregate", actually it was suggested by srinath, do you have a better name?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86506104
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
Yea I think it's fine for now to not support those.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
OK, I'll try to use count min sketch. BTW, seems it only supports Integral and String types for 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86030388
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In the use cases of CBO, YES. However, this function becomes a general one. It could be also used/called by external users. Then, it might not make sense for this limit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85812423
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
--- End diff --
Describe the general description of this function at first, and then explains the return values?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86031611
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
--- End diff --
I see. Then, we would better to explicitly explain it. `Returns null if the result set is empty or all values are null`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85811315
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/MapAggregateQuerySuite.scala ---
@@ -0,0 +1,144 @@
+/*
+ * 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
+
+import java.sql.{Date, Timestamp}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.test.SharedSQLContext
+import org.apache.spark.sql.types._
+
+
+class MapAggregateQuerySuite extends QueryTest with SharedSQLContext {
+
+ private val table = "map_aggregate_test"
+ private val col1 = "col1"
+ private val col2 = "col2"
+ private val schema = StructType(Seq(StructField(col1, StringType), StructField(col2, DoubleType)))
+
+ private def query(numBins: Int): DataFrame = {
+ sql(s"SELECT map_aggregate($col1, $numBins), map_aggregate($col2, $numBins) FROM $table")
+ }
+
+ test("null handling") {
+ withTempView(table) {
+ // Null input
+ val nullRdd: RDD[Row] = spark.sparkContext.parallelize(Seq(Row(null, null)))
+ spark.createDataFrame(nullRdd, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(null, null))
+
+ // Empty input
+ val emptyRdd: RDD[Row] = spark.sparkContext.parallelize(Seq.empty)
+ spark.createDataFrame(emptyRdd, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(null, null))
+
+ // Add some non-null data
+ val rdd: RDD[Row] = spark.sparkContext.parallelize(Seq(Row(null, 3.0D), Row("a", null)))
+ spark.createDataFrame(rdd, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(Map(("a", 1)), Map((3.0D, 1))))
+ }
+ }
+
+ test("returns empty result when ndv exceeds numBins") {
+ withTempView(table) {
+ val rdd: RDD[Row] = spark.sparkContext.parallelize(
+ Seq(Row("a", 4.0D), Row("d", 2.0D), Row("c", 4.0D), Row("b", 1.0D), Row("a", 3.0D),
+ Row("a", 2.0D)), 2)
+ spark.createDataFrame(rdd, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 4), Row(
+ Map(("a", 3), ("b", 1), ("c", 1), ("d", 1)),
+ Map((1.0D, 1), (2.0D, 2), (3.0D, 1), (4.0D, 2))))
+ // One partial exceeds numBins during update()
+ checkAnswer(query(numBins = 2), Row(Map.empty, Map.empty))
+ // Exceeding numBins during merge()
+ checkAnswer(query(numBins = 3), Row(Map.empty, Map.empty))
+ }
+ }
+
+ test("multiple columns of different types") {
+ def queryMultiColumns(numBins: Int): DataFrame = {
+ sql(
+ s"""
+ |SELECT
+ | map_aggregate(c1, $numBins),
+ | map_aggregate(c2, $numBins),
+ | map_aggregate(c3, $numBins),
+ | map_aggregate(c4, $numBins),
+ | map_aggregate(c5, $numBins),
+ | map_aggregate(c6, $numBins),
+ | map_aggregate(c7, $numBins),
+ | map_aggregate(c8, $numBins),
+ | map_aggregate(c9, $numBins),
+ | map_aggregate(c10, $numBins)
+ |FROM $table
+ """.stripMargin)
+ }
+
+ val allTypeSchema = StructType(Seq(
+ StructField("c1", ByteType),
+ StructField("c2", ShortType),
+ StructField("c3", IntegerType),
+ StructField("c4", LongType),
+ StructField("c5", FloatType),
+ StructField("c6", DoubleType),
+ StructField("c7", DecimalType(10, 5)),
+ StructField("c8", DateType),
+ StructField("c9", TimestampType),
+ StructField("c10", StringType)))
--- End diff --
Here, it still misses `BinaryType` and `BooleanType`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy closed the pull request at:
https://github.com/apache/spark/pull/15637
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86504505
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
--- End diff --
note - a basic thing: please use `/** ... */` for commenting member variables or methods, rather than `//`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by srinathshankar <gi...@git.apache.org>.
Github user srinathshankar commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85993076
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $numBins)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (buffer.isInvalid) {
+ return
+ }
+ val evaluated = child.eval(input)
+ if (evaluated != null) {
+ buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (buffer.isInvalid) return
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ return
+ }
+ buffer.merge(other, numBins)
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer match {
+ case stringDigest: StringMapDigest => StringMapDigest.serialize(stringDigest)
+ case numericDigest: NumericMapDigest => NumericMapDigest.serialize(numericDigest)
+ }
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest.deserialize(bytes)
+ case _ => NumericMapDigest.deserialize(bytes)
+ }
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean = false
+
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+
+ // Update baseMap and clear it when its size exceeds numBins.
+ def updateMap[T](baseMap: mutable.HashMap[T, Long], value: T, numBins: Int): Unit = {
+ mergeMap(baseMap, mutable.HashMap(value -> 1L), numBins)
+ }
+
+ // Merge two maps and clear baseMap when its size exceeds numBins.
+ def mergeMap[T](
+ baseMap: mutable.HashMap[T, Long],
+ otherMap: mutable.HashMap[T, Long],
+ numBins: Int): Unit = {
+ otherMap.foreach { case (key, value) =>
+ if (baseMap.contains(key)) {
+ baseMap.update(key, baseMap(key) + value)
+ } else {
+ if (baseMap.size >= numBins) {
+ isInvalid = true
+ baseMap.clear()
+ return
+ } else {
+ baseMap.put(key, value)
+ }
+ }
+ }
+ }
+}
+
+// Digest class for column of string type.
+case class StringMapDigest(
+ bins: mutable.HashMap[UTF8String, Long] = mutable.HashMap.empty) extends MapDigest {
+ def this(bins: mutable.HashMap[UTF8String, Long], isInvalid: Boolean) = {
+ this(bins)
+ this.isInvalid = isInvalid
+ }
+
+ override def update(dataType: DataType, value: Any, numBins: Int): Unit = {
+ updateMap(bins, value.asInstanceOf[UTF8String], numBins)
+ }
+
+ override def merge(otherDigest: MapDigest, numBins: Int): Unit = {
+ mergeMap(baseMap = bins, otherMap = otherDigest.asInstanceOf[StringMapDigest].bins, numBins)
+ }
+
+ override def clear(): Unit = bins.clear()
+}
+
+object StringMapDigest {
+
+ private final def length(obj: StringMapDigest): Int = {
+ // isInvalid, size of bins
+ var len: Int = Ints.BYTES + Ints.BYTES
+ obj.bins.foreach { case (key, value) =>
+ // length of key, key, value
+ len += Ints.BYTES + key.getBytes.length + Longs.BYTES
+ }
+ len
+ }
+
+ final def serialize(obj: StringMapDigest): Array[Byte] = {
+ val buffer = ByteBuffer.wrap(new Array(length(obj)))
+ buffer.putInt(if (obj.isInvalid) 0 else 1)
+ buffer.putInt(obj.bins.size)
+ obj.bins.foreach { case (key, value) =>
+ val bytes = key.getBytes
+ buffer.putInt(bytes.length)
+ buffer.put(bytes)
+ buffer.putLong(value)
+ }
+ buffer.array()
+ }
+
+ final def deserialize(bytes: Array[Byte]): StringMapDigest = {
+ val buffer = ByteBuffer.wrap(bytes)
+ val isInvalid = if (buffer.getInt == 0) true else false
+ val size = buffer.getInt
+ val bins = new mutable.HashMap[UTF8String, Long]
+ var i = 0
+ while (i < size) {
+ val keyLength = buffer.getInt
+ var j = 0
+ val keyBytes = new Array[Byte](keyLength)
+ while (j < keyLength) {
+ keyBytes(j) = buffer.get()
+ j += 1
+ }
+ val value = buffer.getLong
+ bins.put(UTF8String.fromBytes(keyBytes), value)
+ i += 1
+ }
+ new StringMapDigest(bins, isInvalid)
+ }
+}
+
+// Digest class for column of numeric type.
+case class NumericMapDigest(
+ bins: mutable.HashMap[Double, Long] = mutable.HashMap.empty) extends MapDigest {
+ def this(bins: mutable.HashMap[Double, Long], isInvalid: Boolean) = {
+ this(bins)
+ this.isInvalid = isInvalid
+ }
+
+ override def update(dataType: DataType, value: Any, numBins: Int): Unit = {
+ // Use Double to represent endpoints (in histograms) for simplicity
+ val doubleValue = dataType match {
--- End diff --
Couldn't you leave this type promotion to the engine ?
That is, for numeric types instead of calling
`mapaggregate(numericcol, bins)`
call
`mapaggregate(cast(numericcol as double), bins)`
That way you only have to deal with doubles in your aggregate.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/15637
BTW, please follow the PR https://github.com/apache/spark/issues/15677 to rewrite your function description. Add an example for the new function if possible.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
Please correct me if i'm wrong about count_min_sketch, i don't have laptop with me now and will look at it when i get back.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85930167
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
--- End diff --
Now if the input is empty, it also returns null, so that empty result is returned only when ndv exceeds numBins
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67555 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67555/consoleFull)** for PR 15637 at commit [`32478d1`](https://github.com/apache/spark/commit/32478d160356aec3cc07579a657b3e8fbd20e2bd).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
@rxin By "histograms" for numeric columns, do you mean equi-height histogram? The main purpose of this pr is to construct equi-width histogram without prior knowledge of ndv, so that we can compute it with other stats in the same scan. And equi-width histogram is important for both string and numeric(including timestamp and date) types, because it provides more accurate estimation than equi-height histogram.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86411252
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In today's software development environment, speed is an important factor. One of the agile development methodology principles is to avoid over design. The over design can slow down the development and divert an engineer's attention on the unimportant cases. Since we develop MapAggregate.scala for CBO, let's focus on CBO's need. There is no need to develop NDV and histogram for binary data type. For Boolean data type, we already have number of trues and number of falses defined in ColumnStat. I think wxhfy's code is doing fine 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85645925
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
+ maximum number of bins. Otherwise, it returns an empty map.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $numBins)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (buffer.invalid) {
+ return
+ }
+ val evaluated = child.eval(input)
+ if (evaluated != null) {
+ buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (buffer.invalid) return
+ if (other.invalid) {
+ buffer.invalid = true
+ buffer.clear()
+ return
+ }
+ buffer.merge(other, numBins)
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.invalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer match {
+ case stringDigest: StringMapDigest => StringMapDigest.serialize(stringDigest)
+ case numericDigest: NumericMapDigest => NumericMapDigest.serialize(numericDigest)
+ }
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest.deserialize(bytes)
+ case _ => NumericMapDigest.deserialize(bytes)
+ }
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var invalid: Boolean = false
--- End diff --
`invalid` is confusing. How about `isInvalid`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
BTW why is this function called map_aggregate?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86415255
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
I posted same comments multiple times due to a browser glitch. I just cleaned it up.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67769 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67769/consoleFull)** for PR 15637 at commit [`9d65f6f`](https://github.com/apache/spark/commit/9d65f6f174e6790ef3ea40dbd81175ab07c37401).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86504123
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
--- End diff --
"must be a literal or constant foldable"
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85933282
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
+ maximum number of bins. Otherwise, it returns an empty map.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
--- End diff --
Oh, I see, thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67786 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67786/consoleFull)** for PR 15637 at commit [`82de8df`](https://github.com/apache/spark/commit/82de8dfd7f3db3660ce329023f7371df2bda5685).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
In that way, we can only get the hashed value instead of real value of the column, right? So I think we still need to implement a hash code for fractional types.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86409090
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In today's software development environment, speed is an important factor. One of the agile development methodology principles is to avoid over design. The over design can slow down the development and divert an engineer's attention on the unimportant cases. Since we develop MapAggregate.scala for CBO, let's focus on CBO's need. There is no need to develop NDV and histogram for binary data type. For Boolean data type, we already have number of trues and number of falses defined in ColumnStat. I think we are doing fine 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by srinathshankar <gi...@git.apache.org>.
Github user srinathshankar commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85619246
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramEndpoints.scala ---
@@ -0,0 +1,465 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, QuantileSummaries}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries._
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The HistogramEndpoints function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, for column of string type, it returns an empty
+ * map; for column of numeric type, it returns endpoints of equi-height histogram - approximate
+ * percentiles at percentages 0.0, 1/numBins, 2/numBins, ..., (numBins-1)/numBins, 1.0.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ * @param accuracyExpression Accuracy used in computing approximate percentiles.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins [, accuracy]) - Returns bins - (distinct value, frequency) pairs
+ of equi-width histogram when the number of distinct values is less than or equal to the
+ specified maximum number of bins. Otherwise, for column of string type, it returns an empty
+ map; for column of numeric type, it returns endpoints of equi-height histogram - approximate
+ percentiles at percentages 0.0, 1/numBins, 2/numBins, ..., (numBins-1)/numBins, 1.0. The
+ `accuracy` parameter (default: 10000) is a positive integer literal which controls percentiles
+ approximation accuracy at the cost of memory. Higher value of `accuracy` yields better
+ accuracy, `1.0/accuracy` is the relative error of the approximation.
+ """)
+case class HistogramEndpoints(
+ child: Expression,
+ numBinsExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[EndpointsDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression, accuracyExpression: Expression) = {
+ this(child, numBinsExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ private lazy val percentages: Array[Double] = {
+ val array = new Array[Double](numBins + 1)
+ for (i <- 0 to numBins) {
+ array(i) = i / numBins.toDouble
+ }
+ array
+ }
+
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType, IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins or accuracy provided must be a constant literal")
+ } else if (numBins < 2) {
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $numBins)")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: EndpointsDigest, input: InternalRow): Unit = {
+ if (buffer.invalid) {
+ return
+ }
+ val evaluated = child.eval(input)
+ if (evaluated != null) {
+ buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: EndpointsDigest, other: EndpointsDigest): Unit = {
+ if (buffer.invalid) return
+ if (other.invalid) {
+ buffer.invalid = true
+ buffer.clear()
+ return
+ }
+ buffer.merge(other, numBins)
+ }
+
+ override def eval(buffer: EndpointsDigest): Any = {
+ if (buffer.invalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ buffer match {
+ case stringDigest: StringEndpointsDigest =>
+ // sort the result to make it more readable
+ val sorted = TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ case numericDigest: NumericEndpointsDigest =>
+ if (!numericDigest.mapInvalid) {
+ val sorted = TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ } else {
+ val percentiles = numericDigest.percentileDigest.getPercentiles(percentages)
+ // we only need percentiles, this is for constructing MapData
+ val padding = new Array[Long](percentiles.length)
+ ArrayBasedMapData(percentiles, padding)
+ }
+ }
+ }
+ }
+
+ override def serialize(buffer: EndpointsDigest): Array[Byte] = {
+ buffer match {
+ case stringDigest: StringEndpointsDigest => StringEndpointsDigest.serialize(stringDigest)
+ case numericDigest: NumericEndpointsDigest => NumericEndpointsDigest.serialize(numericDigest)
+ }
+ }
+
+ override def deserialize(bytes: Array[Byte]): EndpointsDigest = {
+ child.dataType match {
+ case StringType => StringEndpointsDigest.deserialize(bytes)
+ case _ => NumericEndpointsDigest.deserialize(bytes)
+ }
+ }
+
+ override def createAggregationBuffer(): EndpointsDigest = {
+ child.dataType match {
+ case StringType =>
+ StringEndpointsDigest()
+ case _ =>
+ NumericEndpointsDigest(new PercentileDigest(1.0D / accuracy))
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): HistogramEndpoints = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): HistogramEndpoints = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression, accuracyExpression)
+
+ override def prettyName: String = "histogram_endpoints"
+}
+
+trait EndpointsDigest {
+
+ // Mark this EndpointsDigest as invalid when:
+ // 1. for string type - the size of the hashmap (ndv of the column) exceeds numBins;
+ // 2. for numeric type - Some Decimal value out of the range of Double occurs.
+ var invalid: Boolean = false
+
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: EndpointsDigest, numBins: Int): Unit
+ def clear(): Unit
+
+ // Updates baseMap, and returns success or not.
+ def updateMap[T](baseMap: mutable.HashMap[T, Long], value: T, numBins: Int): Boolean = {
+ if (baseMap.contains(value)) {
+ baseMap.update(value, baseMap(value) + 1)
+ } else {
+ if (baseMap.size >= numBins) {
+ return false
+ } else {
+ baseMap.put(value, 1)
+ }
+ }
+ true
+ }
+
+ // Merges two Maps into baseMap, and returns success or not.
+ def mergeMaps[T](
+ baseMap: mutable.HashMap[T, Long],
+ otherMap: mutable.HashMap[T, Long],
+ numBins: Int): Boolean = {
+ otherMap.foreach { case (key, value) =>
+ if (baseMap.contains(key)) {
+ baseMap.update(key, baseMap(key) + value)
+ } else {
+ if (baseMap.size >= numBins) {
+ return false
+ } else {
+ baseMap.put(key, value)
+ }
+ }
+ }
+ true
+ }
+}
+
+/**
+ * Digest class for column of string type.
+ * @param bins A HashMap to maintain frequency of each distinct value.
+ */
+case class StringEndpointsDigest(
+ bins: mutable.HashMap[UTF8String, Long] = mutable.HashMap.empty[UTF8String, Long])
+ extends EndpointsDigest {
+
+ def this(bins: mutable.HashMap[UTF8String, Long], invalid: Boolean) = {
+ this(bins)
+ this.invalid = invalid
+ }
+
+ override def update(dataType: DataType, value: Any, numBins: Int): Unit = {
+ val success = updateMap(baseMap = bins, value.asInstanceOf[UTF8String], numBins)
+ if (!success) {
+ // clear the map and mark this digest as invalid
+ bins.clear()
+ invalid = true
+ }
+ }
+
+ override def merge(otherDigest: EndpointsDigest, numBins: Int): Unit = {
+ val other = otherDigest.asInstanceOf[StringEndpointsDigest]
+ val success = mergeMaps(baseMap = bins, otherMap = other.bins, numBins)
+ if (!success) {
+ // clear the map and mark this digest as invalid
+ bins.clear()
+ invalid = true
+ }
+ }
+
+ override def clear(): Unit = {
+ bins.clear()
+ }
+}
+
+object StringEndpointsDigest {
+
+ private final def length(obj: StringEndpointsDigest): Int = {
+ // invalid, size of bins
+ var len: Int = Ints.BYTES + Ints.BYTES
+ obj.bins.foreach { case (key, value) =>
+ // length of key, key, value
+ len += Ints.BYTES + key.getBytes.length + Longs.BYTES
+ }
+ len
+ }
+
+ final def serialize(obj: StringEndpointsDigest): Array[Byte] = {
+ val buffer = ByteBuffer.wrap(new Array(length(obj)))
+ buffer.putInt(if (obj.invalid) 0 else 1)
+ buffer.putInt(obj.bins.size)
+ obj.bins.foreach { case (key, value) =>
+ val bytes = key.getBytes
+ buffer.putInt(bytes.length)
+ buffer.put(bytes)
+ buffer.putLong(value)
+ }
+ buffer.array()
+ }
+
+ final def deserialize(bytes: Array[Byte]): StringEndpointsDigest = {
+ val buffer = ByteBuffer.wrap(bytes)
+ val invalid = if (buffer.getInt == 0) true else false
+ val size = buffer.getInt
+ val bins = new mutable.HashMap[UTF8String, Long]
+ var i = 0
+ while (i < size) {
+ val keyLength = buffer.getInt
+ var j = 0
+ val keyBytes = new Array[Byte](keyLength)
+ while (j < keyLength) {
+ keyBytes(j) = buffer.get()
+ j += 1
+ }
+ val value = buffer.getLong
+ bins.put(UTF8String.fromBytes(keyBytes), value)
+ i += 1
+ }
+ new StringEndpointsDigest(bins, invalid)
+ }
+}
+
+/**
+ * Digest class for column of numeric type.
+ * @param bins A HashMap to maintain frequency of each distinct value.
+ * @param percentileDigest A helper class to compute approximate percentiles.
+ */
+case class NumericEndpointsDigest(
+ percentileDigest: PercentileDigest,
+ bins: mutable.HashMap[Double, Long] = mutable.HashMap.empty[Double, Long])
+ extends EndpointsDigest {
+
+ var mapInvalid = false
+
+ def this(
+ percentileDigest: PercentileDigest,
+ bins: mutable.HashMap[Double, Long],
+ invalid: Boolean,
+ mapInvalid: Boolean) = {
+ this(percentileDigest, bins)
+ this.invalid = invalid
+ this.mapInvalid = mapInvalid
+ }
+
+ override def update(dataType: DataType, value: Any, numBins: Int): Unit = {
+ if (dataType.isInstanceOf[DecimalType]) {
+ val decimal = value.asInstanceOf[Decimal]
+ val double = decimal.toDouble
+ if (double == Double.PositiveInfinity || double == Double.NegativeInfinity) {
+ // This value has too great a magnitude to represent as a Double.
+ invalid = true
+ clear()
+ return
+ }
+ }
+
+ // We use Double to represent endpoints (in histograms) for simplicity.
+ // Loss of precision is acceptable because we are computing approximate answers anyway.
+ val doubleValue = dataType match {
+ case n: NumericType =>
+ n.numeric.toDouble(value.asInstanceOf[n.InternalType])
+ case d: DateType =>
+ value.asInstanceOf[Int].toDouble
+ case t: TimestampType =>
+ value.asInstanceOf[Long].toDouble
+ }
+ // update percentileDigest
+ percentileDigest.add(doubleValue)
--- End diff --
I take it you're updating the percentile in all cases because you don;t want to merge maps with percentiledigests ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86409288
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In today's software development environment, speed is an important factor. One of the agile development methodology principles is to avoid over design. The over design can slow down the development and divert an engineer's attention on the unimportant cases. Since we develop MapAggregate.scala for CBO, let's focus on CBO's need. There is no need to develop NDV and histogram for binary data type. For Boolean data type, we already have number of trues and number of falses defined in ColumnStat. I think we are doing fine 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/67786/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
Yes, but in its impl, it actually only supports above types.
```
@Override
public void add(Object item, long count) {
if (item instanceof String) {
addString((String) item, count);
} else {
addLong(Utils.integralToLong(item), count);
}
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86408951
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In today's software development environment, speed is an important factor. One of the agile development methodology principles is to avoid over design. The over design can slow down the development and divert an engineer's attention on the unimportant cases. Since we develop MapAggregate.scala for CBO, let's focus on CBO's need. There is no need to develop NDV and histogram for binary data type. For Boolean data type, we already have number of trues and number of falses defined in ColumnStat. I think we are doing fine 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
Yes, in our design, equi-width histogram is a seq of single valued bins, which is used for columns with low cardinality, so that we can get accurate estimation. When cardinality is high, equi-height histogram is better than equi-width histogram (with some fixed width), especially in data skew scenario, where the skewed value would be hidden in a equi-width bin, while in equi-height histogram, it would occupies many contiguous bins.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86505042
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
+
+ // Update bins and clear it when its size exceeds numBins.
+ def updateMap(value: T, numBins: Int): Unit = {
+ mergeMap(mutable.HashMap(value -> 1L), numBins)
+ }
+
+ // Merge two maps and clear bins when its size exceeds numBins.
+ override def merge(otherDigest: MapDigest, numBins: Int): Unit = {
+ val otherMap = otherDigest.asInstanceOf[MapDigestBase[T]].bins
+ mergeMap(otherMap, numBins)
+ }
+
+ def mergeMap(otherMap: mutable.HashMap[T, Long], numBins: Int): Unit = {
+ otherMap.foreach { case (key, value) =>
+ if (bins.contains(key)) {
+ bins.update(key, bins(key) + value)
+ } else {
+ if (bins.size >= numBins) {
+ isInvalid = true
+ bins.clear()
+ return
+ } else {
+ bins.put(key, value)
+ }
+ }
+ }
+ }
+
+ override def clear(): Unit = bins.clear()
+
+ override def serialize(): Array[Byte] = {
--- End diff --
maybe it's fine, but if you are writing our own binary protocol, I'd document 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86647571
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
--- End diff --
Frequency usually means the rate something occurs. In statistics, frequency means the number of items occurring in a given category. Here frequency actually is equivalent to count. We use frequency and count interchangeably.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85657540
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/MapAggregateQuerySuite.scala ---
@@ -0,0 +1,124 @@
+/*
+ * 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
+
+import java.sql.{Date, Timestamp}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.test.SharedSQLContext
+import org.apache.spark.sql.types._
+
+
+class MapAggregateQuerySuite extends QueryTest with SharedSQLContext {
+ import testImplicits._
+
+ private val table = "map_aggregate_test"
+ private val col1 = "col1"
+ private val col2 = "col2"
+ private def query(numBins: Int): DataFrame = {
+ sql(s"SELECT map_aggregate($col1, $numBins), map_aggregate($col2, $numBins) FROM $table")
+ }
+
+ test("null handling") {
+ withTempView(table) {
+ val schema = StructType(Seq(StructField(col1, StringType), StructField(col2, DoubleType)))
+ // Empty input row
+ val rdd1 = spark.sparkContext.parallelize(Seq(Row(null, null)))
+ spark.createDataFrame(rdd1, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(Map.empty, Map.empty))
+
+ // Add some non-empty row
+ val rdd2 = spark.sparkContext.parallelize(Seq(Row(null, 3.0D), Row("a", null)))
+ spark.createDataFrame(rdd2, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(Map(("a", 1)), Map((3.0D, 1))))
+ }
+ }
+
+ test("returns empty result when ndv exceeds numBins") {
+ withTempView(table) {
+ spark.sparkContext.makeRDD(
+ Seq(("a", 4), ("d", 2), ("c", 4), ("b", 1), ("a", 3), ("a", 2)), 2).toDF(col1, col2)
+ .createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 4), Row(
+ Map(("a", 3), ("b", 1), ("c", 1), ("d", 1)),
+ Map((1.0D, 1), (2.0D, 2), (3.0D, 1), (4.0D, 2))))
+ // One partial exceeds numBins during update()
+ checkAnswer(query(numBins = 2), Row(Map.empty, Map.empty))
+ // Exceeding numBins during merge()
+ checkAnswer(query(numBins = 3), Row(Map.empty, Map.empty))
+ }
+ }
+
+ test("multiple columns of different types") {
+ def queryMultiColumns(numBins: Int): DataFrame = {
+ sql(
+ s"""
+ |SELECT
+ | map_aggregate(c1, $numBins),
+ | map_aggregate(c2, $numBins),
+ | map_aggregate(c3, $numBins),
+ | map_aggregate(c4, $numBins),
+ | map_aggregate(c5, $numBins),
+ | map_aggregate(c6, $numBins),
+ | map_aggregate(c7, $numBins),
+ | map_aggregate(c8, $numBins),
+ | map_aggregate(c9, $numBins),
+ | map_aggregate(c10, $numBins)
+ |FROM $table
+ """.stripMargin)
+ }
+
+ val ints = Seq(5, 3, 1)
+ val doubles = Seq(1.0D, 3.0D, 5.0D)
+ val dates = Seq("1970-01-01", "1970-02-02", "1970-03-03")
+ val timestamps = Seq("1970-01-01 00:00:00", "1970-01-01 00:00:05", "1970-01-01 00:00:10")
+ val strings = Seq("a", "bb", "ccc")
+
+ val data = ints.indices.map { i =>
+ (ints(i).toByte,
+ ints(i).toShort,
+ ints(i),
+ ints(i).toLong,
+ doubles(i).toFloat,
+ doubles(i),
+ Decimal(doubles(i)),
+ Date.valueOf(dates(i)),
+ Timestamp.valueOf(timestamps(i)),
+ strings(i))
+ }
+ withTempView(table) {
+ data.toDF("c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9", "c10")
--- End diff --
ok
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/68114/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85646444
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
+ maximum number of bins. Otherwise, it returns an empty map.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
--- End diff --
How about this change?
```Scala
private lazy val numBins: Int = numBinsExpression.eval() match {
case o: Int => o
case x => throw new AnalysisException(
s"The maximum number of bins provided must be a foldable integer expression: $x")
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86504529
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
--- End diff --
add some classdoc explaining what this is.
also why is this called "Map"?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85814522
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $numBins)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (buffer.isInvalid) {
+ return
+ }
+ val evaluated = child.eval(input)
+ if (evaluated != null) {
+ buffer.update(child.dataType, evaluated, numBins)
+ }
--- End diff --
A general comment about the impl. Here, I think we should avoid `return` if possible. For example, we can re-write it like
```Scala
if (!buffer.isInvalid) {
val evaluated = child.eval(input)
if (evaluated != null) {
buffer.update(child.dataType, evaluated, numBins)
}
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
Why not use count min sketch then? You would get more signal (with some error) for a much larger range of values.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/67551/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
It supports any data types, uses less space and can provide probabilistic frequencies for a large number of distinct values, and it's already implemented in Spark. We just need to add a wrapper for it to turn it into an aggregate 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
We already have those don't we? sparks own hash expresssion.
On Friday, November 4, 2016, Zhenhua Wang <no...@github.com> wrote:
> In that way, we can only get the hashed value instead of real value of the
> column, right? So I think we still need to implement a hash code for
> fractional types.
>
> \u2014
> You are receiving this because you were mentioned.
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/spark/pull/15637#issuecomment-258588577>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/AATvPNMa9Azpaf5ZR-L88sFJAVnF_zqnks5q6_tngaJpZM4Kgufb>
> .
>
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67555 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67555/consoleFull)** for PR 15637 at commit [`32478d1`](https://github.com/apache/spark/commit/32478d160356aec3cc07579a657b3e8fbd20e2bd).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67758 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67758/consoleFull)** for PR 15637 at commit [`9d65f6f`](https://github.com/apache/spark/commit/9d65f6f174e6790ef3ea40dbd81175ab07c37401).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
Weird - github didn't email me notifications.
Why do you care about all (value, freq) pairs? All you want should just be some summary statistics that can tell you given a value, what the estimated freq is, isn't 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85645776
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/MapAggregateQuerySuite.scala ---
@@ -0,0 +1,124 @@
+/*
+ * 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
+
+import java.sql.{Date, Timestamp}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.test.SharedSQLContext
+import org.apache.spark.sql.types._
+
+
+class MapAggregateQuerySuite extends QueryTest with SharedSQLContext {
+ import testImplicits._
+
+ private val table = "map_aggregate_test"
+ private val col1 = "col1"
+ private val col2 = "col2"
+ private def query(numBins: Int): DataFrame = {
+ sql(s"SELECT map_aggregate($col1, $numBins), map_aggregate($col2, $numBins) FROM $table")
+ }
+
+ test("null handling") {
+ withTempView(table) {
+ val schema = StructType(Seq(StructField(col1, StringType), StructField(col2, DoubleType)))
+ // Empty input row
+ val rdd1 = spark.sparkContext.parallelize(Seq(Row(null, null)))
+ spark.createDataFrame(rdd1, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(Map.empty, Map.empty))
+
+ // Add some non-empty row
+ val rdd2 = spark.sparkContext.parallelize(Seq(Row(null, 3.0D), Row("a", null)))
+ spark.createDataFrame(rdd2, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(Map(("a", 1)), Map((3.0D, 1))))
+ }
+ }
+
+ test("returns empty result when ndv exceeds numBins") {
+ withTempView(table) {
+ spark.sparkContext.makeRDD(
+ Seq(("a", 4), ("d", 2), ("c", 4), ("b", 1), ("a", 3), ("a", 2)), 2).toDF(col1, col2)
+ .createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 4), Row(
+ Map(("a", 3), ("b", 1), ("c", 1), ("d", 1)),
+ Map((1.0D, 1), (2.0D, 2), (3.0D, 1), (4.0D, 2))))
+ // One partial exceeds numBins during update()
+ checkAnswer(query(numBins = 2), Row(Map.empty, Map.empty))
+ // Exceeding numBins during merge()
+ checkAnswer(query(numBins = 3), Row(Map.empty, Map.empty))
+ }
+ }
+
+ test("multiple columns of different types") {
+ def queryMultiColumns(numBins: Int): DataFrame = {
+ sql(
+ s"""
+ |SELECT
+ | map_aggregate(c1, $numBins),
+ | map_aggregate(c2, $numBins),
+ | map_aggregate(c3, $numBins),
+ | map_aggregate(c4, $numBins),
+ | map_aggregate(c5, $numBins),
+ | map_aggregate(c6, $numBins),
+ | map_aggregate(c7, $numBins),
+ | map_aggregate(c8, $numBins),
+ | map_aggregate(c9, $numBins),
+ | map_aggregate(c10, $numBins)
+ |FROM $table
+ """.stripMargin)
+ }
+
+ val ints = Seq(5, 3, 1)
+ val doubles = Seq(1.0D, 3.0D, 5.0D)
+ val dates = Seq("1970-01-01", "1970-02-02", "1970-03-03")
+ val timestamps = Seq("1970-01-01 00:00:00", "1970-01-01 00:00:05", "1970-01-01 00:00:10")
+ val strings = Seq("a", "bb", "ccc")
+
+ val data = ints.indices.map { i =>
+ (ints(i).toByte,
+ ints(i).toShort,
+ ints(i),
+ ints(i).toLong,
+ doubles(i).toFloat,
+ doubles(i),
+ Decimal(doubles(i)),
+ Date.valueOf(dates(i)),
+ Timestamp.valueOf(timestamps(i)),
+ strings(i))
+ }
+ withTempView(table) {
+ data.toDF("c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9", "c10")
--- End diff --
Can you use `def createDataFrame(rowRDD: RDD[Row], schema: StructType)`, instead of using implicit schema inference?
We need an explicit schema definition when we verifying different data types.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86411132
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In today's software development environment, speed is an important factor. One of the agile development methodology principles is to avoid over design. The over design can slow down the development and divert an engineer's attention on the unimportant cases. Since we develop MapAggregate.scala for CBO, let's focus on CBO's need. There is no need to develop NDV and histogram for binary data type. For Boolean data type, we already have number of trues and number of falses defined in ColumnStat. I think wxhfy's code is doing fine 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
So I don't get it - I thought you were only using the value, freq pairs (i.e. equi-width histogram with width = 1 - can we stop calling it equi-width histogram?) for string types. Why would the question you are asking matter for string types?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85645394
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
--- End diff --
Here, we need to document `distinct non-null values`. This function excludes NULL.
However, I am not sure whether this is expected. NULL is also a value, 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
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
/cc @hvanhovell @cloud-fan
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86410847
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In today's software development environment, speed is an important factor. One of the agile development methodology principles is to avoid over design. The over design can slow down the development and divert an engineer's attention on the unimportant cases. Since we develop MapAggregate.scala for CBO, let's focus on CBO's need. There is no need to develop NDV and histogram for binary data type. For Boolean data type, we already have number of trues and number of falses defined in ColumnStat. I think wxhfy's code is doing fine 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85645802
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/MapAggregateQuerySuite.scala ---
@@ -0,0 +1,124 @@
+/*
+ * 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
+
+import java.sql.{Date, Timestamp}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.test.SharedSQLContext
+import org.apache.spark.sql.types._
+
+
+class MapAggregateQuerySuite extends QueryTest with SharedSQLContext {
+ import testImplicits._
+
+ private val table = "map_aggregate_test"
+ private val col1 = "col1"
+ private val col2 = "col2"
+ private def query(numBins: Int): DataFrame = {
+ sql(s"SELECT map_aggregate($col1, $numBins), map_aggregate($col2, $numBins) FROM $table")
+ }
+
+ test("null handling") {
+ withTempView(table) {
+ val schema = StructType(Seq(StructField(col1, StringType), StructField(col2, DoubleType)))
+ // Empty input row
+ val rdd1 = spark.sparkContext.parallelize(Seq(Row(null, null)))
+ spark.createDataFrame(rdd1, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(Map.empty, Map.empty))
+
+ // Add some non-empty row
+ val rdd2 = spark.sparkContext.parallelize(Seq(Row(null, 3.0D), Row("a", null)))
+ spark.createDataFrame(rdd2, schema).createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 2), Row(Map(("a", 1)), Map((3.0D, 1))))
+ }
+ }
+
+ test("returns empty result when ndv exceeds numBins") {
+ withTempView(table) {
+ spark.sparkContext.makeRDD(
+ Seq(("a", 4), ("d", 2), ("c", 4), ("b", 1), ("a", 3), ("a", 2)), 2).toDF(col1, col2)
+ .createOrReplaceTempView(table)
+ checkAnswer(query(numBins = 4), Row(
+ Map(("a", 3), ("b", 1), ("c", 1), ("d", 1)),
+ Map((1.0D, 1), (2.0D, 2), (3.0D, 1), (4.0D, 2))))
+ // One partial exceeds numBins during update()
+ checkAnswer(query(numBins = 2), Row(Map.empty, Map.empty))
+ // Exceeding numBins during merge()
+ checkAnswer(query(numBins = 3), Row(Map.empty, Map.empty))
+ }
+ }
+
+ test("multiple columns of different types") {
+ def queryMultiColumns(numBins: Int): DataFrame = {
+ sql(
+ s"""
+ |SELECT
+ | map_aggregate(c1, $numBins),
+ | map_aggregate(c2, $numBins),
+ | map_aggregate(c3, $numBins),
+ | map_aggregate(c4, $numBins),
+ | map_aggregate(c5, $numBins),
+ | map_aggregate(c6, $numBins),
+ | map_aggregate(c7, $numBins),
+ | map_aggregate(c8, $numBins),
+ | map_aggregate(c9, $numBins),
+ | map_aggregate(c10, $numBins)
+ |FROM $table
+ """.stripMargin)
+ }
+
+ val ints = Seq(5, 3, 1)
+ val doubles = Seq(1.0D, 3.0D, 5.0D)
+ val dates = Seq("1970-01-01", "1970-02-02", "1970-03-03")
+ val timestamps = Seq("1970-01-01 00:00:00", "1970-01-01 00:00:05", "1970-01-01 00:00:10")
+ val strings = Seq("a", "bb", "ccc")
+
+ val data = ints.indices.map { i =>
+ (ints(i).toByte,
+ ints(i).toShort,
+ ints(i),
+ ints(i).toLong,
+ doubles(i).toFloat,
+ doubles(i),
+ Decimal(doubles(i)),
+ Date.valueOf(dates(i)),
+ Timestamp.valueOf(timestamps(i)),
+ strings(i))
+ }
+ withTempView(table) {
+ data.toDF("c1", "c2", "c3", "c4", "c5", "c6", "c7", "c8", "c9", "c10")
--- End diff --
In this example, below is the printed schema
```
+--------+--------------+-------+
|col_name| data_type|comment|
+--------+--------------+-------+
| c1| tinyint| null|
| c2| smallint| null|
| c3| int| null|
| c4| bigint| null|
| c5| float| null|
| c6| double| null|
| c7|decimal(38,18)| null|
| c8| date| null|
| c9| timestamp| null|
| c10| string| null|
+--------+--------------+-------+
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
Since the JIRA is closed, I'm closing this PR too.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86505390
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
+
+ // Update bins and clear it when its size exceeds numBins.
+ def updateMap(value: T, numBins: Int): Unit = {
+ mergeMap(mutable.HashMap(value -> 1L), numBins)
--- End diff --
i think you are doing this backwards by calling mergeMap in updateMap. If anything, it should be the other way around, i.e. mergeMap has a loop that calls updateMap. That way you will not need to create a whole new hash map for each record, or loop over some hash map of one entry for every record.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #68114 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68114/consoleFull)** for PR 15637 at commit [`b7fcaf4`](https://github.com/apache/spark/commit/b7fcaf46a865a1f48ffa1377f47d8f17ffe3e445).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86504961
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
+
+ // Update bins and clear it when its size exceeds numBins.
+ def updateMap(value: T, numBins: Int): Unit = {
+ mergeMap(mutable.HashMap(value -> 1L), numBins)
+ }
+
+ // Merge two maps and clear bins when its size exceeds numBins.
+ override def merge(otherDigest: MapDigest, numBins: Int): Unit = {
+ val otherMap = otherDigest.asInstanceOf[MapDigestBase[T]].bins
+ mergeMap(otherMap, numBins)
+ }
+
+ def mergeMap(otherMap: mutable.HashMap[T, Long], numBins: Int): Unit = {
+ otherMap.foreach { case (key, value) =>
+ if (bins.contains(key)) {
+ bins.update(key, bins(key) + value)
+ } else {
+ if (bins.size >= numBins) {
+ isInvalid = true
+ bins.clear()
+ return
+ } else {
+ bins.put(key, value)
+ }
+ }
+ }
+ }
+
+ override def clear(): Unit = bins.clear()
+
+ override def serialize(): Array[Byte] = {
--- End diff --
should we just serialize using Kryo rather than writing our own serializer 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
With (value, freq) pairs, i.e. equi-width histogram, given a predicate col<100 and col>50, we can know exactly how many records are in that range, which is the estimated rowcount after the filter. And we can also know the exact ndv after the filter, which is important for following estimation, e.g., in estimation of output row count of agg(group by).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
Thanks - that makes sense. I got really confused by "numBins".
Sorry for throwing a monkey wrench now: should we just use count min sketch and use that for string columns? For numeric columns it seems to me just using histograms will be sufficient.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
cc @srinathshankar
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
BTW we do have a count min sketch implementation already in Spark, and we can turn it into an aggregate function pretty easily.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86413803
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
--- End diff --
In today's software development environment, speed is an important factor. One of the agile development methodology principles is to avoid over design. The over design can slow down the development and divert an engineer's attention on the unimportant cases. Since we develop MapAggregate.scala for CBO, let's focus on CBO's need. There is no need to develop NDV and histogram for binary data type. For Boolean data type, we already have number of trues and number of falses defined in ColumnStat. I think wxhfy's code is doing fine 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67806 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67806/consoleFull)** for PR 15637 at commit [`98d502b`](https://github.com/apache/spark/commit/98d502b3c0f3dce2d743de7448f5af5ad4c419ae).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85931101
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,332 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns:
+ * 1. null if no non-null value exists.
+ * 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ * distinct non-null values is less than or equal to the specified maximum number of bins.
+ * 3. an empty map otherwise.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns 1. null if no non-null value exists.
+ 2. (distinct non-null value, frequency) pairs of equi-width histogram when the number of
+ distinct non-null values is less than or equal to the specified maximum number of bins.
+ 3. an empty map otherwise.
--- End diff --
ok
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/67758/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85657536
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
+ maximum number of bins. Otherwise, it returns an empty map.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $numBins)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (buffer.invalid) {
+ return
+ }
+ val evaluated = child.eval(input)
+ if (evaluated != null) {
+ buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (buffer.invalid) return
+ if (other.invalid) {
+ buffer.invalid = true
+ buffer.clear()
+ return
+ }
+ buffer.merge(other, numBins)
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.invalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer match {
+ case stringDigest: StringMapDigest => StringMapDigest.serialize(stringDigest)
+ case numericDigest: NumericMapDigest => NumericMapDigest.serialize(numericDigest)
+ }
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest.deserialize(bytes)
+ case _ => NumericMapDigest.deserialize(bytes)
+ }
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var invalid: Boolean = false
--- End diff --
yeah, that's better.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67551 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67551/consoleFull)** for PR 15637 at commit [`15eb372`](https://github.com/apache/spark/commit/15eb3721f56ac27bd90933ef7e66f3453eae4a75).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67551 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67551/consoleFull)** for PR 15637 at commit [`15eb372`](https://github.com/apache/spark/commit/15eb3721f56ac27bd90933ef7e66f3453eae4a75).
* This patch **fails Scala style tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
@rxin After looking at the `CountMinSketch`, I have a question:
It supports point queries, i.e., given a value, return its frequency. Can we get all distinct values from it? What we want is (value, frequency) pairs. If we only support point queries, how can we estimate selectivity of predicates like col < 100 and col > 50?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
To be clear, I'm suggesting
```
select count_min_sketch(hash(col1)), count_min_sketch(hash(col2)), ... from table
```
the count_min_sketch aggregate function only accepts a long value, and returns a byte array (binary)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/67769/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #68114 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/68114/consoleFull)** for PR 15637 at commit [`b7fcaf4`](https://github.com/apache/spark/commit/b7fcaf46a865a1f48ffa1377f47d8f17ffe3e445).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `abstract class MapDigestBase[T] extends MapDigest `
* `case class StringMapDigest(`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
How would this pr help you with equi-width histogram? This function just gives you the frequency count for keys when the cardinality is low.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86505614
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
+
+ // Update bins and clear it when its size exceeds numBins.
+ def updateMap(value: T, numBins: Int): Unit = {
+ mergeMap(mutable.HashMap(value -> 1L), numBins)
+ }
+
+ // Merge two maps and clear bins when its size exceeds numBins.
+ override def merge(otherDigest: MapDigest, numBins: Int): Unit = {
+ val otherMap = otherDigest.asInstanceOf[MapDigestBase[T]].bins
+ mergeMap(otherMap, numBins)
+ }
+
+ def mergeMap(otherMap: mutable.HashMap[T, Long], numBins: Int): Unit = {
+ otherMap.foreach { case (key, value) =>
+ if (bins.contains(key)) {
+ bins.update(key, bins(key) + value)
+ } else {
+ if (bins.size >= numBins) {
+ isInvalid = true
+ bins.clear()
+ return
+ } else {
+ bins.put(key, value)
+ }
+ }
+ }
+ }
+
+ override def clear(): Unit = bins.clear()
+
+ override def serialize(): Array[Byte] = {
--- End diff --
actually if you just use kryo maybe you don't need any of the keyLen, putKey, getKey functions and it will look a lot simpler.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/15637
How about the complex types? `Array`, `Map` and `Struct`? It sounds like the test cases do not cover these test cases. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/67555/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by ron8hu <gi...@git.apache.org>.
Github user ron8hu commented on the issue:
https://github.com/apache/spark/pull/15637
test
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
@rxin Moreover, with this pr we can compute accurate equi-width histogram, while count min sketch only gets an estimated result? I think it's better to have an accurate result given that both methods need a table scan.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
@rxin Originally this pr computed endpoints for equi-width (distinct values) or equi-height histograms (percentiles), then @srinathshankar gave a comment [here](https://github.com/apache/spark/pull/15637#pullrequestreview-6319075) and I think it was reasonable, so I adopted it. Now this pr computes distinct values and their counts for building equi-width histograms, and when ndv > threshold, it returns an empty result. I just updated description of the corresponding Jira.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
@wzhfy I'm pretty confused by this. I go ahead and read the JIRA -- it is about computing histograms, but the current aggregate function just computes counts for each value, up until a certain number of distinct elements. Where are we doing the binning?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85652094
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
--- End diff --
I see. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67786 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67786/consoleFull)** for PR 15637 at commit [`82de8df`](https://github.com/apache/spark/commit/82de8dfd7f3db3660ce329023f7371df2bda5685).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by srinathshankar <gi...@git.apache.org>.
Github user srinathshankar commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85609917
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HistogramEndpoints.scala ---
@@ -0,0 +1,465 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.ApproximatePercentile.{PercentileDigest, PercentileDigestSerializer}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, QuantileSummaries}
+import org.apache.spark.sql.catalyst.util.QuantileSummaries._
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The HistogramEndpoints function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, for column of string type, it returns an empty
+ * map; for column of numeric type, it returns endpoints of equi-height histogram - approximate
+ * percentiles at percentages 0.0, 1/numBins, 2/numBins, ..., (numBins-1)/numBins, 1.0.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ * @param accuracyExpression Accuracy used in computing approximate percentiles.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins [, accuracy]) - Returns bins - (distinct value, frequency) pairs
+ of equi-width histogram when the number of distinct values is less than or equal to the
+ specified maximum number of bins. Otherwise, for column of string type, it returns an empty
+ map; for column of numeric type, it returns endpoints of equi-height histogram - approximate
+ percentiles at percentages 0.0, 1/numBins, 2/numBins, ..., (numBins-1)/numBins, 1.0. The
+ `accuracy` parameter (default: 10000) is a positive integer literal which controls percentiles
+ approximation accuracy at the cost of memory. Higher value of `accuracy` yields better
+ accuracy, `1.0/accuracy` is the relative error of the approximation.
+ """)
+case class HistogramEndpoints(
+ child: Expression,
+ numBinsExpression: Expression,
+ accuracyExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[EndpointsDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression, accuracyExpression: Expression) = {
+ this(child, numBinsExpression, accuracyExpression, 0, 0)
+ }
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, Literal(ApproximatePercentile.DEFAULT_PERCENTILE_ACCURACY))
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ private lazy val percentages: Array[Double] = {
+ val array = new Array[Double](numBins + 1)
+ for (i <- 0 to numBins) {
+ array(i) = i / numBins.toDouble
+ }
+ array
+ }
+
+ private lazy val accuracy: Int = accuracyExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType, IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable || !accuracyExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins or accuracy provided must be a constant literal")
+ } else if (numBins < 2) {
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $numBins)")
+ } else if (accuracy <= 0) {
+ TypeCheckFailure(
+ s"The accuracy provided must be a positive integer literal (current value = $accuracy)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: EndpointsDigest, input: InternalRow): Unit = {
+ if (buffer.invalid) {
+ return
+ }
+ val evaluated = child.eval(input)
+ if (evaluated != null) {
+ buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: EndpointsDigest, other: EndpointsDigest): Unit = {
+ if (buffer.invalid) return
+ if (other.invalid) {
+ buffer.invalid = true
+ buffer.clear()
+ return
+ }
+ buffer.merge(other, numBins)
+ }
+
+ override def eval(buffer: EndpointsDigest): Any = {
+ if (buffer.invalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ buffer match {
+ case stringDigest: StringEndpointsDigest =>
+ // sort the result to make it more readable
+ val sorted = TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ case numericDigest: NumericEndpointsDigest =>
+ if (!numericDigest.mapInvalid) {
+ val sorted = TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ } else {
+ val percentiles = numericDigest.percentileDigest.getPercentiles(percentages)
+ // we only need percentiles, this is for constructing MapData
+ val padding = new Array[Long](percentiles.length)
+ ArrayBasedMapData(percentiles, padding)
+ }
+ }
+ }
+ }
+
+ override def serialize(buffer: EndpointsDigest): Array[Byte] = {
+ buffer match {
+ case stringDigest: StringEndpointsDigest => StringEndpointsDigest.serialize(stringDigest)
+ case numericDigest: NumericEndpointsDigest => NumericEndpointsDigest.serialize(numericDigest)
+ }
+ }
+
+ override def deserialize(bytes: Array[Byte]): EndpointsDigest = {
+ child.dataType match {
+ case StringType => StringEndpointsDigest.deserialize(bytes)
+ case _ => NumericEndpointsDigest.deserialize(bytes)
+ }
+ }
+
+ override def createAggregationBuffer(): EndpointsDigest = {
+ child.dataType match {
+ case StringType =>
+ StringEndpointsDigest()
+ case _ =>
+ NumericEndpointsDigest(new PercentileDigest(1.0D / accuracy))
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): HistogramEndpoints = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): HistogramEndpoints = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = false
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression, accuracyExpression)
+
+ override def prettyName: String = "histogram_endpoints"
+}
+
+trait EndpointsDigest {
+
+ // Mark this EndpointsDigest as invalid when:
+ // 1. for string type - the size of the hashmap (ndv of the column) exceeds numBins;
+ // 2. for numeric type - Some Decimal value out of the range of Double occurs.
+ var invalid: Boolean = false
+
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: EndpointsDigest, numBins: Int): Unit
+ def clear(): Unit
+
+ // Updates baseMap, and returns success or not.
+ def updateMap[T](baseMap: mutable.HashMap[T, Long], value: T, numBins: Int): Boolean = {
+ if (baseMap.contains(value)) {
--- End diff --
Isn't this just
`mergeMap(baseMap, (value -> 1), numBins)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85657509
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
+ maximum number of bins. Otherwise, it returns an empty map.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
--- End diff --
null is not int type, it will not pass type check in `checkInputDataTypes()`, so it is safe 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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/67806/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/15637
Yea good point - the underlying implementation really only needs a hash code, it would be trivial to support all types. But even easier, I think you can just compute the hash (using sql expression) and pass it in to support all data types.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/15637
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r86579702
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,310 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function Computes frequency for each distinct non-null value of a column.
+ * It returns: 1. null if the table is empty or all values of the column are null.
+ * 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values is
+ * less than or equal to the specified threshold.
+ * 3. an empty result if the number of distinct non-null values exceeds that threshold.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of pairs.
+ */
+@ExpressionDescription(
+ usage = """
+ _FUNC_(col, numBins) - Computes frequency for each distinct non-null value of column `col`.
+ It returns: 1. null if the table is empty or all values of column `col` are null.
+ 2. (distinct non-null value, frequency) pairs if the number of distinct non-null values
+ is less than or equal to the specified threshold `numBins`.
+ 3. an empty result if the number of distinct non-null values exceeds `numBins`.
+ """,
+ extended = """
+ Examples:
+ > SELECT map_aggregate(col, 3) FROM tbl;
+ 1. null - if `tbl` is empty or values of `col` are all nulls
+ 2. Map((10, 2), (20, 1)) - if values of `col` are (10, 20, 10)
+ 3. Map.empty - if values of `col` are (1, 2, 3, 4)
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
+
+ override def inputTypes: Seq[AbstractDataType] = {
+ Seq(TypeCollection(NumericType, TimestampType, DateType, StringType), IntegerType)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val defaultCheck = super.checkInputDataTypes()
+ if (defaultCheck.isFailure) {
+ defaultCheck
+ } else if (!numBinsExpression.foldable) {
+ TypeCheckFailure("The maximum number of bins provided must be a constant literal")
+ } else if (numBins < 2) {
+ val currentValue = if (numBinsExpression.eval() == null) null else numBins
+ TypeCheckFailure(
+ "The maximum number of bins provided must be a positive integer literal >= 2 " +
+ s"(current value = $currentValue)")
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def update(buffer: MapDigest, input: InternalRow): Unit = {
+ if (!buffer.isInvalid) {
+ val evaluated = child.eval(input)
+ if (evaluated != null) buffer.update(child.dataType, evaluated, numBins)
+ }
+ }
+
+ override def merge(buffer: MapDigest, other: MapDigest): Unit = {
+ if (!buffer.isInvalid) {
+ if (other.isInvalid) {
+ buffer.isInvalid = true
+ buffer.clear()
+ } else {
+ buffer.merge(other, numBins)
+ }
+ }
+ }
+
+ override def eval(buffer: MapDigest): Any = {
+ if (buffer.isInvalid) {
+ // return empty map
+ ArrayBasedMapData(Map.empty)
+ } else {
+ // sort the result to make it more readable
+ val sorted = buffer match {
+ case stringDigest: StringMapDigest => TreeMap[UTF8String, Long](stringDigest.bins.toSeq: _*)
+ case numericDigest: NumericMapDigest => TreeMap[Double, Long](numericDigest.bins.toSeq: _*)
+ }
+ if (sorted.isEmpty) {
+ // don't have non-null values
+ null
+ } else {
+ ArrayBasedMapData(sorted.keys.toArray, sorted.values.toArray)
+ }
+ }
+ }
+
+ override def serialize(buffer: MapDigest): Array[Byte] = {
+ buffer.serialize()
+ }
+
+ override def deserialize(bytes: Array[Byte]): MapDigest = {
+ MapDigest.deserialize(child.dataType, bytes)
+ }
+
+ override def createAggregationBuffer(): MapDigest = {
+ child.dataType match {
+ case StringType => StringMapDigest()
+ case _ => NumericMapDigest()
+ }
+ }
+
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): MapAggregate = {
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+ }
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): MapAggregate = {
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+ }
+
+ override def nullable: Boolean = true
+
+ override def dataType: DataType = {
+ child.dataType match {
+ case StringType => MapType(StringType, LongType)
+ case _ => MapType(DoubleType, LongType)
+ }
+ }
+
+ override def children: Seq[Expression] = Seq(child, numBinsExpression)
+
+ override def prettyName: String = "map_aggregate"
+}
+
+trait MapDigest {
+ // Mark this MapDigest invalid when the size of the hashmap (ndv of the column) exceeds numBins
+ var isInvalid: Boolean
+ def update(dataType: DataType, value: Any, numBins: Int): Unit
+ def merge(otherDigest: MapDigest, numBins: Int): Unit
+ def clear(): Unit
+ def serialize(): Array[Byte]
+}
+
+abstract class MapDigestBase[T] extends MapDigest {
+ val bins: mutable.HashMap[T, Long]
+
+ // Update bins and clear it when its size exceeds numBins.
+ def updateMap(value: T, numBins: Int): Unit = {
+ mergeMap(mutable.HashMap(value -> 1L), numBins)
+ }
+
+ // Merge two maps and clear bins when its size exceeds numBins.
+ override def merge(otherDigest: MapDigest, numBins: Int): Unit = {
+ val otherMap = otherDigest.asInstanceOf[MapDigestBase[T]].bins
+ mergeMap(otherMap, numBins)
+ }
+
+ def mergeMap(otherMap: mutable.HashMap[T, Long], numBins: Int): Unit = {
+ otherMap.foreach { case (key, value) =>
+ if (bins.contains(key)) {
+ bins.update(key, bins(key) + value)
+ } else {
+ if (bins.size >= numBins) {
+ isInvalid = true
+ bins.clear()
+ return
+ } else {
+ bins.put(key, value)
+ }
+ }
+ }
+ }
+
+ override def clear(): Unit = bins.clear()
+
+ override def serialize(): Array[Byte] = {
--- End diff --
I think why we are writing our own serializer here (and also in `ApproximatePercentile`) is because `SerializerInstance` is not thread-safe, 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
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #15637: [SPARK-18000] [SQL] Aggregation function for comp...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/15637#discussion_r85646119
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MapAggregate.scala ---
@@ -0,0 +1,324 @@
+/*
+ * 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.nio.ByteBuffer
+
+import scala.collection.immutable.TreeMap
+import scala.collection.mutable
+
+import com.google.common.primitives.{Doubles, Ints, Longs}
+
+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.{Expression, ExpressionDescription}
+import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
+import org.apache.spark.sql.types.{DataType, _}
+import org.apache.spark.unsafe.types.UTF8String
+
+/**
+ * The MapAggregate function for a column returns bins - (distinct value, frequency) pairs
+ * of equi-width histogram when the number of distinct values is less than or equal to the
+ * specified maximum number of bins. Otherwise, it returns an empty map.
+ *
+ * @param child child expression that can produce column value with `child.eval(inputRow)`
+ * @param numBinsExpression The maximum number of bins.
+ */
+@ExpressionDescription(
+ usage =
+ """
+ _FUNC_(col, numBins) - Returns bins - (distinct value, frequency) pairs of equi-width
+ histogram when the number of distinct values is less than or equal to the specified
+ maximum number of bins. Otherwise, it returns an empty map.
+ """)
+case class MapAggregate(
+ child: Expression,
+ numBinsExpression: Expression,
+ override val mutableAggBufferOffset: Int,
+ override val inputAggBufferOffset: Int) extends TypedImperativeAggregate[MapDigest] {
+
+ def this(child: Expression, numBinsExpression: Expression) = {
+ this(child, numBinsExpression, 0, 0)
+ }
+
+ // Mark as lazy so that numBinsExpression is not evaluated during tree transformation.
+ private lazy val numBins: Int = numBinsExpression.eval().asInstanceOf[Int]
--- End diff --
`asInstanceOf[Int]` is risky. When `eval()` returns null, it will be casted to 0.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing e...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/15637
**[Test build #67758 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/67758/consoleFull)** for PR 15637 at commit [`9d65f6f`](https://github.com/apache/spark/commit/9d65f6f174e6790ef3ea40dbd81175ab07c37401).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
Point query of countMinSketch can only help us on estimation for predicates like col=1.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #15637: [SPARK-18000] [SQL] Aggregation function for computing b...
Posted by wzhfy <gi...@git.apache.org>.
Github user wzhfy commented on the issue:
https://github.com/apache/spark/pull/15637
Oh, do you mean using count min sketch for equi-width histogram? Actually I don't know about the sketch, I need to look into it to see if it's easy to use as an agg function and also support float/double/decimal types. But I do think this pr takes much less effort to implement what we want :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org