You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by viirya <gi...@git.apache.org> on 2016/12/21 07:16:30 UTC
[GitHub] spark pull request #16371: [SPARK-18932][SQL] Support partial aggregation fo...
GitHub user viirya opened a pull request:
https://github.com/apache/spark/pull/16371
[SPARK-18932][SQL] Support partial aggregation for collect_set/collect_list
## What changes were proposed in this pull request?
Currently collect_set/collect_list aggregation expression don't support partial aggregation. This patch is to enable partial aggregation for them.
## How was this patch tested?
N/A
Please review http://spark.apache.org/contributing.html before opening a pull request.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/viirya/spark-1 collect-partial-support
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/16371.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 #16371
----
commit 6e8aa82d95e52c6c469aaa4a8e1cfc0105576e69
Author: Liang-Chi Hsieh <vi...@gmail.com>
Date: 2016-12-21T06:53:39Z
Support partial mode for collect.
----
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70464/
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94233881
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
We should remove the support for non-partial aggregation in that case.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r93388333
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala ---
@@ -33,10 +33,9 @@ import org.apache.spark.util.collection.OpenHashMap
* The Percentile aggregate function returns the exact percentile(s) of numeric column `expr` at
* the given percentage(s) with value range in [0.0, 1.0].
*
- * The operator is bound to the slower sort based aggregation path because the number of elements
--- End diff --
`TypedImperativeAggregate` is't bound to sort based aggregation. `ObjectHashAggregateExec` supports `TypedImperativeAggregate` in hash based aggregation 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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94102016
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +43,52 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
- }
-
- override def update(b: InternalRow, input: InternalRow): Unit = {
- // Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
- // See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
- if (value != null) {
- buffer += value
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
--- End diff --
Also, the serialized format (child.dataType, LongType) of `Percentile` is different than `CollectList` and `CollectSet` (i.e., child.dataType).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70801 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70801/testReport)** for PR 16371 at commit [`15a10ee`](https://github.com/apache/spark/commit/15a10eebe272428841772a58d06f2e889d70b75c).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r93677313
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +43,52 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
- }
-
- override def update(b: InternalRow, input: InternalRow): Unit = {
- // Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
- // See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
- if (value != null) {
- buffer += value
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
+ val projection = UnsafeProjection.create(Array[DataType](child.dataType))
+ obj.foreach { value =>
+ val row = InternalRow.apply(value)
+ val unsafeRow = projection.apply(row)
+ out.writeInt(unsafeRow.getSizeInBytes)
+ unsafeRow.writeToStream(out, buffer)
+ }
+ out.writeInt(-1)
+ out.flush()
+
+ bos.toByteArray
+ } finally {
+ out.close()
+ bos.close()
}
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
- }
-
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ protected def _deserialize(bytes: Array[Byte], updater: (Any) => Unit): Unit = {
--- End diff --
Same 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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r93677896
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -29,12 +30,10 @@ import org.apache.spark.sql.types._
/**
* The Collect aggregate function collects all seen expression values into a list of values.
*
- * The operator is bound to the slower sort based aggregation path because the number of
- * elements (and their memory usage) can not be determined in advance. This also means that the
- * collected elements are stored on heap, and that too many elements can cause GC pauses and
- * eventually Out of Memory Errors.
+ * We have to store all the collected elements in memory, and that too many elements can cause GC
+ * paused and eventually OutOfMemory Errors.
*/
-abstract class Collect extends ImperativeAggregate {
+abstract class Collect[T] extends TypedImperativeAggregate[T] {
--- End diff --
Why [T] and not T <: Growable[Any] with Iterable[Any]?
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16371
sounds good.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70709 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70709/testReport)** for PR 16371 at commit [`965a4a9`](https://github.com/apache/spark/commit/965a4a93ddc79f4f34a7a32a3b3ef560ebb08665).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94139716
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,78 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
+ }
+
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
--- End diff --
I think the effect is the same. Any benefit by doing this?
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70810 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70810/testReport)** for PR 16371 at commit [`15a10ee`](https://github.com/apache/spark/commit/15a10eebe272428841772a58d06f2e889d70b75c).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94221696
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,48 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
+ }
+
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val array = new GenericArrayData(obj.toArray)
+ val projection = UnsafeProjection.create(
--- End diff --
make this project a member variable, then we don't need to create it everytime we call `serialize`
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70459 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70459/testReport)** for PR 16371 at commit [`6e8aa82`](https://github.com/apache/spark/commit/6e8aa82d95e52c6c469aaa4a8e1cfc0105576e69).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70713 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70713/testReport)** for PR 16371 at commit [`965a4a9`](https://github.com/apache/spark/commit/965a4a93ddc79f4f34a7a32a3b3ef560ebb08665).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70459/
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94232541
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
i was thought about removing it. however, `RewriteDistinctAggregates` rule has logic for this case. this test is the only one for testing this logic. so i didn't remove it in the end.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94234459
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,48 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
--- End diff --
fixed.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70709/
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70810 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70810/testReport)** for PR 16371 at commit [`15a10ee`](https://github.com/apache/spark/commit/15a10eebe272428841772a58d06f2e889d70b75c).
* 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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94221390
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,48 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
--- End diff --
unnecessary `asInstanceOf`
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16371
@marmbrus I see that you filed the ticket. We can make this work with partial aggregation, but we shouldn't run them this way. Maybe it's time to define something as partial aggregateable vs should be partially aggregated.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on the issue:
https://github.com/apache/spark/pull/16371
LGTM
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70504/
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/16371
@cloud-fan I already did this in current PR. I am waiting for some consensus so I can push ahead to address the review comments from @hvanhovell.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r93677262
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/Percentile.scala ---
@@ -33,10 +33,9 @@ import org.apache.spark.util.collection.OpenHashMap
* The Percentile aggregate function returns the exact percentile(s) of numeric column `expr` at
* the given percentage(s) with value range in [0.0, 1.0].
*
- * The operator is bound to the slower sort based aggregation path because the number of elements
- * and their partial order cannot be determined in advance. Therefore we have to store all the
- * elements in memory, and that too many elements can cause GC paused and eventually OutOfMemory
- * Errors.
+ * Because the number of elements and their partial order cannot be determined in advance.
+ * Therefore we have to store all the elements in memory, and that too many elements can cause GC
--- End diff --
Could you fix the language? `..., and that too...` That sentence is weird. This is probably on me.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70707 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70707/testReport)** for PR 16371 at commit [`f60e824`](https://github.com/apache/spark/commit/f60e824e6913edb21ae5e0fa6207dcaccdf8de20).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94221353
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,48 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
--- End diff --
is it a behavior change?
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70504 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70504/testReport)** for PR 16371 at commit [`559db5b`](https://github.com/apache/spark/commit/559db5bafc58d4f4a1de1366f8e034cf1af8b994).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94363919
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -88,19 +92,19 @@ abstract class Collect extends ImperativeAggregate {
case class CollectList(
child: Expression,
mutableAggBufferOffset: Int = 0,
- inputAggBufferOffset: Int = 0) extends Collect {
+ inputAggBufferOffset: Int = 0) extends Collect[ArrayBuffer[Any]] {
def this(child: Expression) = this(child, 0, 0)
- override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): CollectList =
copy(mutableAggBufferOffset = newMutableAggBufferOffset)
override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
copy(inputAggBufferOffset = newInputAggBufferOffset)
- override def prettyName: String = "collect_list"
+ override def createAggregationBuffer(): ArrayBuffer[Any] = new ArrayBuffer[Any]()
- override protected[this] val buffer: mutable.ArrayBuffer[Any] = mutable.ArrayBuffer.empty
--- End diff --
`mutable.ArrayBuffer.empty` looks 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70810/
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94221486
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,48 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
+ }
+
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
--- End diff --
`override def serialize(buffer: T)`
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94101411
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +43,52 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
- }
-
- override def update(b: InternalRow, input: InternalRow): Unit = {
- // Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
- // See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
- if (value != null) {
- buffer += value
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
--- End diff --
`Percentile` takes `OpenHashMap`, which has a bit different behavior than `Growable[]`.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
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 issue #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70756 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70756/testReport)** for PR 16371 at commit [`5c6b02a`](https://github.com/apache/spark/commit/5c6b02af16ed1b960242af74932a050f1c390a6e).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94363844
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -16,16 +16,16 @@
*/
package org.apache.spark.sql.catalyst.optimizer
-import org.apache.spark.sql.catalyst.SimpleCatalystConf
+import org.apache.spark.sql.catalyst.{InternalRow, SimpleCatalystConf}
import org.apache.spark.sql.catalyst.analysis.{Analyzer, EmptyFunctionRegistry}
import org.apache.spark.sql.catalyst.catalog.{InMemoryCatalog, SessionCatalog}
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{If, Literal}
-import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectSet, Count}
+import org.apache.spark.sql.catalyst.expressions.{Expression, If, Literal}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{CollectSet, Count, ImperativeAggregate, TypedImperativeAggregate}
import org.apache.spark.sql.catalyst.plans.PlanTest
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Expand, LocalRelation, LogicalPlan}
-import org.apache.spark.sql.types.{IntegerType, StringType}
+import org.apache.spark.sql.types.{DataType, IntegerType, StringType}
--- End diff --
please revert unnecessary changes in `import`
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70801/
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the issue:
https://github.com/apache/spark/pull/16371
+1 I think we can move forward.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/16371
retest this please.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94234295
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
ok. make sense to me.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94234334
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
I will only remove this test in this pr.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94221742
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
shall we just remove the `supportsPartial` and this 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/16371
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70756/
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94363879
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -88,19 +92,19 @@ abstract class Collect extends ImperativeAggregate {
case class CollectList(
child: Expression,
mutableAggBufferOffset: Int = 0,
- inputAggBufferOffset: Int = 0) extends Collect {
+ inputAggBufferOffset: Int = 0) extends Collect[ArrayBuffer[Any]] {
def this(child: Expression) = this(child, 0, 0)
- override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): ImperativeAggregate =
+ override def withNewMutableAggBufferOffset(newMutableAggBufferOffset: Int): CollectList =
--- End diff --
unnecessary change
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70504 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70504/testReport)** for PR 16371 at commit [`559db5b`](https://github.com/apache/spark/commit/559db5bafc58d4f4a1de1366f8e034cf1af8b994).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class DummpAgg(`
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94233778
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
we can remove that logic in `RewriteDistinctAggregates` 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 issue #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94210128
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,78 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
+ }
+
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
--- End diff --
@cloud-fan I did the change, is this looks good for you?
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94221606
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,48 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
+ }
+
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val array = new GenericArrayData(obj.toArray)
+ val projection = UnsafeProjection.create(
+ Array[DataType](ArrayType(elementType = child.dataType, containsNull = false)))
+ projection.apply(InternalRow.apply(array)).getBytes()
+ }
+
+ protected def _deserialize(bytes: Array[Byte], buffer: T): Unit = {
--- End diff --
same here
```
override def deserialize(bytes: ArrayByte): Unit = {
val buffer = createAggregationBuffer()
...
}
```
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r93677104
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +43,52 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
- }
-
- override def update(b: InternalRow, input: InternalRow): Unit = {
- // Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
- // See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
- if (value != null) {
- buffer += value
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
--- End diff --
Should we generalize this code? It is also part the Percentile class.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94221462
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,48 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
+ }
+
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
--- End diff --
can't we just define `serialize` 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70467 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70467/testReport)** for PR 16371 at commit [`6677beb`](https://github.com/apache/spark/commit/6677beb59c16216e0ad9a25bb2adf655c8827df7).
* 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70713/
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/16371
I think we should support partial aggregation for collect, in the future we can define an interface to declare "should partial aggregate", so that a single collect function won't cause a 2-phase aggregation.
I think they are orthogonal and we can go ahead.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70464 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70464/testReport)** for PR 16371 at commit [`68d1e98`](https://github.com/apache/spark/commit/68d1e98a049da996feb202660e6c6b15f94183b7).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r93680880
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -29,12 +30,10 @@ import org.apache.spark.sql.types._
/**
* The Collect aggregate function collects all seen expression values into a list of values.
*
- * The operator is bound to the slower sort based aggregation path because the number of
- * elements (and their memory usage) can not be determined in advance. This also means that the
- * collected elements are stored on heap, and that too many elements can cause GC pauses and
- * eventually Out of Memory Errors.
+ * We have to store all the collected elements in memory, and that too many elements can cause GC
+ * paused and eventually OutOfMemory Errors.
*/
-abstract class Collect extends ImperativeAggregate {
+abstract class Collect[T] extends TypedImperativeAggregate[T] {
--- End diff --
If we can do that, we can make the PR smaller.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/16371
@cloud-fan @marmbrus ok. I will first address @hvanhovell's comments.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/16371
Why would we even want to support partial aggregation for collect_list?
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r93677654
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +43,52 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
- }
-
- override def update(b: InternalRow, input: InternalRow): Unit = {
- // Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
- // See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
- if (value != null) {
- buffer += value
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
+ val bos = new ByteArrayOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
+ val projection = UnsafeProjection.create(Array[DataType](child.dataType))
+ obj.foreach { value =>
+ val row = InternalRow.apply(value)
+ val unsafeRow = projection.apply(row)
+ out.writeInt(unsafeRow.getSizeInBytes)
+ unsafeRow.writeToStream(out, buffer)
+ }
+ out.writeInt(-1)
+ out.flush()
+
+ bos.toByteArray
+ } finally {
+ out.close()
+ bos.close()
}
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
- }
-
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ protected def _deserialize(bytes: Array[Byte], updater: (Any) => Unit): Unit = {
--- End diff --
You could also pass in a `Growable[]` instead of an anonymous 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70467/
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70467 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70467/testReport)** for PR 16371 at commit [`6677beb`](https://github.com/apache/spark/commit/6677beb59c16216e0ad9a25bb2adf655c8827df7).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94114431
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,78 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
+ }
+
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
--- End diff --
shall we create `new GenericArrayData(buffer.toArray)` and do unsafe projection?
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70733 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70733/testReport)** for PR 16371 at commit [`c2f3841`](https://github.com/apache/spark/commit/c2f384111d0b5d1f54355f2ac0189f2241e40fd3).
* 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70733 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70733/testReport)** for PR 16371 at commit [`c2f3841`](https://github.com/apache/spark/commit/c2f384111d0b5d1f54355f2ac0189f2241e40fd3).
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/16371
Just for reference. Hive's collect_list and collect_set seems supporting partial aggregation. Of course we don't necessarily follow Hive's.
https://github.com/apache/hive/blob/ff67cdda1c538dc65087878eeba3e165cf3230f4/ql/src/java/org/apache/hadoop/hive/ql/udf/generic/GenericUDAFMkCollectionEvaluator.java
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94104643
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -29,12 +30,10 @@ import org.apache.spark.sql.types._
/**
* The Collect aggregate function collects all seen expression values into a list of values.
*
- * The operator is bound to the slower sort based aggregation path because the number of
- * elements (and their memory usage) can not be determined in advance. This also means that the
- * collected elements are stored on heap, and that too many elements can cause GC pauses and
- * eventually Out of Memory Errors.
+ * We have to store all the collected elements in memory, and that too many elements can cause GC
+ * paused and eventually OutOfMemory Errors.
*/
-abstract class Collect extends ImperativeAggregate {
+abstract class Collect[T] extends TypedImperativeAggregate[T] {
--- End diff --
yeah, that's good idea.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94233910
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
because we don't have non-partial aggregation function 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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94234184
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteDistinctAggregatesSuite.scala ---
@@ -63,7 +63,7 @@ class RewriteDistinctAggregatesSuite extends PlanTest {
val input = testRelation
.groupBy('a, 'd)(
countDistinct('e, 'c).as('agg1),
- CollectSet('b).toAggregateExpression().as('agg2))
+ DummpAgg('b).toAggregateExpression().as('agg2))
--- End diff --
If we remove the logic from RewriteDistinctAggregates, we need to make sure that non-partial aggregates do not exist anymore. Lets do this in a follow-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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by hvanhovell <gi...@git.apache.org>.
Github user hvanhovell commented on the issue:
https://github.com/apache/spark/pull/16371
@viirya I left a few comments on the PR. Lets wait for some consensus before pushing ahead.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70733/
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/16371
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/16371
thanks, merging to master!
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/16371
I think we should implement collect with `TypedImperativeAggregate` and remove the `supportsPartual` flag, @viirya do you have time to work on 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
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 issue #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70713 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70713/testReport)** for PR 16371 at commit [`965a4a9`](https://github.com/apache/spark/commit/965a4a93ddc79f4f34a7a32a3b3ef560ebb08665).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `abstract class Collect[T <: Growable[Any] with Iterable[Any]] extends TypedImperativeAggregate[T] `
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/16371
@hvanhovell Got it. Thanks for review.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/16371
LGTM
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/16371
Thank you @cloud-fan @hvanhovell @rxin @marmbrus
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94141624
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +44,78 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
+ private def generateOutput(results: Iterable[Any]): Any = {
+ if (results.isEmpty) {
+ null
+ } else {
+ new GenericArrayData(results.toArray)
+ }
}
- override def update(b: InternalRow, input: InternalRow): Unit = {
+ override def update(buffer: T, input: InternalRow): T = {
+ val value = child.eval(input).asInstanceOf[Any]
+
// Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
// See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
if (value != null) {
buffer += value
}
+ buffer
+ }
+
+ override def merge(buffer: T, other: T): T = {
+ buffer ++= other
}
- override def merge(buffer: InternalRow, input: InternalRow): Unit = {
- sys.error("Collect cannot be used in partial aggregations.")
+ override def eval(buffer: T): Any = {
+ generateOutput(buffer)
}
- override def eval(input: InternalRow): Any = {
- new GenericArrayData(buffer.toArray)
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
+ val buffer = new Array[Byte](4 << 10) // 4K
--- End diff --
I just tried it locally. Looks like the code can be more compact.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation fo...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/16371#discussion_r94101455
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala ---
@@ -44,39 +43,52 @@ abstract class Collect extends ImperativeAggregate {
override def dataType: DataType = ArrayType(child.dataType)
- override def supportsPartial: Boolean = false
-
- override def aggBufferAttributes: Seq[AttributeReference] = Nil
-
- override def aggBufferSchema: StructType = StructType.fromAttributes(aggBufferAttributes)
-
- override def inputAggBufferAttributes: Seq[AttributeReference] = Nil
-
// Both `CollectList` and `CollectSet` are non-deterministic since their results depend on the
// actual order of input rows.
override def deterministic: Boolean = false
- protected[this] val buffer: Growable[Any] with Iterable[Any]
-
- override def initialize(b: InternalRow): Unit = {
- buffer.clear()
- }
-
- override def update(b: InternalRow, input: InternalRow): Unit = {
- // Do not allow null values. We follow the semantics of Hive's collect_list/collect_set here.
- // See: org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMkCollectionEvaluator
- val value = child.eval(input)
- if (value != null) {
- buffer += value
+ protected def _serialize(obj: Iterable[Any]): Array[Byte] = {
--- End diff --
We can generalize this to cover `Percentile`, but we may not use `T <: Growable[Any] with Iterable[Any]` but `T <: Iterable[Any]` then.
---
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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/16371
**[Test build #70756 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70756/testReport)** for PR 16371 at commit [`5c6b02a`](https://github.com/apache/spark/commit/5c6b02af16ed1b960242af74932a050f1c390a6e).
* 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 #16371: [SPARK-18932][SQL] Support partial aggregation for colle...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/16371
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70707/
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