You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "amaliujia (via GitHub)" <gi...@apache.org> on 2023/02/17 20:37:38 UTC
[GitHub] [spark] amaliujia opened a new pull request, #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
amaliujia opened a new pull request, #40070:
URL: https://github.com/apache/spark/pull/40070
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Adding more API to `agg` including max,min,mean,count,avg,sum.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
API coverage
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
NO
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
UT
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110380640
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -149,4 +149,111 @@ class RelationalGroupedDataset protected[sql] (
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
+ c
+ // TODO: deal with typed columns.
+ })
+ }
+
+ /**
+ * Count the number of rows for each group. The resulting `DataFrame` will also contain the
+ * grouping columns.
+ *
+ * @since 3.4.0
+ */
+ def count(): DataFrame = toDF(Seq(functions.count(functions.lit(1)).alias("count")))
+
+ /**
+ * Compute the average value for each numeric columns for each group. This is an alias for
+ * `avg`. The resulting `DataFrame` will also contain the grouping columns. When specified
+ * columns are given, only compute the average values for them.
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def mean(colNames: String*): DataFrame = {
+ toDF(colNames.map(colName => functions.mean(colName)).toSeq)
Review Comment:
hmmm I see. Removing those `toSeq`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110378209
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
Review Comment:
I think I did right replacement and hit a proto -> plan test generation failure.
I am planing look into that separately. I am gonna need some time to learn how to debug `org.apache.spark.sql.connect.ProtoToParsedPlanTestSuite`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110372332
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
.setIsDistinct(false)
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
Review Comment:
Makes sense. Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110388531
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2065,6 +2065,15 @@ class Dataset[T] private[sql] (val session: SparkSession, private[sql] val plan:
collectResult().iterator.asInstanceOf[java.util.Iterator[T]]
}
+ /**
+ * Returns the number of rows in the Dataset.
+ * @group action
+ * @since 3.4.0
+ */
+ def count(): Long = {
+ groupBy().count().collect().head.getLong(0)
Review Comment:
Well you are right... I thought I added it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110348196
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
Review Comment:
Use `Column.fn` instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110375466
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2065,6 +2065,15 @@ class Dataset[T] private[sql] (val session: SparkSession, private[sql] val plan:
collectResult().iterator.asInstanceOf[java.util.Iterator[T]]
}
+ /**
+ * Returns the number of rows in the Dataset.
+ * @group action
+ * @since 3.4.0
+ */
+ def count(): Long = {
+ groupBy().count().collect().head.getLong(0)
Review Comment:
In my local branch that I rebased today, there is no this API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110359583
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
.setIsDistinct(false)
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
+ c.expr
+ // TODO: deal with typed columns.
+ })
+ }
+
+ /**
+ * Count the number of rows for each group. The resulting `DataFrame` will also contain the
+ * grouping columns.
+ *
+ * @since 3.4.0
+ */
+ def count(): DataFrame = toDF(Seq(functions.count(functions.lit(1)).alias("count").expr))
Review Comment:
Wait I think we should do reverse way right? So the Dataset.count = groupby().count().collect()?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on PR #40070:
URL: https://github.com/apache/spark/pull/40070#issuecomment-1435241324
@hvanhovell
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110349585
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
Review Comment:
Hold on that I will revert this part. It seems hit an issue somewhere if I switch to use functions API. I will debug that separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110355264
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
Review Comment:
See my earlier comment, and also the tests are broken.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110347846
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
Review Comment:
You currently don't return this function, but the result of builder.build(). If you do, it should be `functions.avg(columnName).expr`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110350177
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
.setIsDistinct(false)
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
+ c.expr
+ // TODO: deal with typed columns.
+ })
+ }
+
+ /**
+ * Count the number of rows for each group. The resulting `DataFrame` will also contain the
+ * grouping columns.
+ *
+ * @since 3.4.0
+ */
+ def count(): DataFrame = toDF(Seq(functions.count(functions.lit(1)).alias("count").expr))
Review Comment:
Can you replace `Dataset.count()` with this code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on PR #40070:
URL: https://github.com/apache/spark/pull/40070#issuecomment-1435257024
I need to update tests in this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110360552
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
.setIsDistinct(false)
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
+ c.expr
+ // TODO: deal with typed columns.
+ })
+ }
+
+ /**
+ * Count the number of rows for each group. The resulting `DataFrame` will also contain the
+ * grouping columns.
+ *
+ * @since 3.4.0
+ */
+ def count(): DataFrame = toDF(Seq(functions.count(functions.lit(1)).alias("count").expr))
Review Comment:
I see what you mean now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110349585
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
Review Comment:
Hold on that I will revert this part. It seems hit an issue somewhere if I switch to use functions API. I need to understand more on the functions implementation.
I will debug this separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110352767
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala:
##########
@@ -1199,6 +1199,34 @@ class PlanGenerationTestSuite extends ConnectFunSuite with BeforeAndAfterAll wit
"b" -> "avg",
"*" -> "size",
"a" -> "count")
+
+ simple
Review Comment:
This doesn't work. You need to construct a single `Dataset` per test, now you are just returning the last one. Can you split this into a bunch of tests?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110374514
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2065,6 +2065,15 @@ class Dataset[T] private[sql] (val session: SparkSession, private[sql] val plan:
collectResult().iterator.asInstanceOf[java.util.Iterator[T]]
}
+ /**
+ * Returns the number of rows in the Dataset.
+ * @group action
+ * @since 3.4.0
+ */
+ def count(): Long = {
+ groupBy().count().collect().head.getLong(0)
Review Comment:
Didn't I implement that?
lol...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110352041
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
.setIsDistinct(false)
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
+ c.expr
+ // TODO: deal with typed columns.
+ })
+ }
+
+ /**
+ * Count the number of rows for each group. The resulting `DataFrame` will also contain the
+ * grouping columns.
+ *
+ * @since 3.4.0
+ */
+ def count(): DataFrame = toDF(Seq(functions.count(functions.lit(1)).alias("count").expr))
Review Comment:
That is different. Dataset.count() returns a long: `def count(): Long`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110346693
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -24,7 +24,7 @@ import scala.collection.JavaConverters._
import org.apache.spark.connect.proto
/**
- * A set of methods for aggregations on a `DataFrame`, created by [[Dataset#groupBy groupBy]],
+ * A set of methods for aggregations on a `DataFrame`, creted by [[Dataset#groupBy groupBy]],
Review Comment:
Created?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on PR #40070:
URL: https://github.com/apache/spark/pull/40070#issuecomment-1435428773
Merging.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110376126
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -149,4 +149,111 @@ class RelationalGroupedDataset protected[sql] (
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
+ c
+ // TODO: deal with typed columns.
+ })
+ }
+
+ /**
+ * Count the number of rows for each group. The resulting `DataFrame` will also contain the
+ * grouping columns.
+ *
+ * @since 3.4.0
+ */
+ def count(): DataFrame = toDF(Seq(functions.count(functions.lit(1)).alias("count")))
+
+ /**
+ * Compute the average value for each numeric columns for each group. This is an alias for
+ * `avg`. The resulting `DataFrame` will also contain the grouping columns. When specified
+ * columns are given, only compute the average values for them.
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def mean(colNames: String*): DataFrame = {
+ toDF(colNames.map(colName => functions.mean(colName)).toSeq)
Review Comment:
do we need `toSeq` here? I though scala varags are always a Seq...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110385162
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,7 +109,7 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToColumn(expr: String, inputExpr: proto.Expression): Column = {
Review Comment:
How about:
```scala
private[this] def strToColumn(expr: String, inputExpr: Column): Column = {
expr.toLowerCase(Locale.ROOT) match {
case "avg" | "average" | "mean" => functions.avg(inputExpr)
case "stddev" | "std" => functions.avg(inputExpr)
case "count" | "size" => functions.count(inputExpr) // Analyzer will take care of * expansion
case name => Column.fn(name, inputExpr)
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110400229
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,7 +109,7 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToColumn(expr: String, inputExpr: proto.Expression): Column = {
Review Comment:
I see what you are suggesting now. Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] amaliujia commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110374201
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala:
##########
@@ -1199,6 +1199,34 @@ class PlanGenerationTestSuite extends ConnectFunSuite with BeforeAndAfterAll wit
"b" -> "avg",
"*" -> "size",
"a" -> "count")
+
+ simple
Review Comment:
done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110354619
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
.setIsDistinct(false)
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
+ c.expr
+ // TODO: deal with typed columns.
+ })
+ }
+
+ /**
+ * Count the number of rows for each group. The resulting `DataFrame` will also contain the
+ * grouping columns.
+ *
+ * @since 3.4.0
+ */
+ def count(): DataFrame = toDF(Seq(functions.count(functions.lit(1)).alias("count").expr))
Review Comment:
You can replace the dataframe generation code there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell commented on a diff in pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #40070:
URL: https://github.com/apache/spark/pull/40070#discussion_r1110349605
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/RelationalGroupedDataset.scala:
##########
@@ -109,44 +109,131 @@ class RelationalGroupedDataset protected[sql] (
agg(exprs.asScala.toMap)
}
- private[this] def strToExpr(expr: String, inputExpr: proto.Expression): proto.Expression = {
+ private[this] def strToExpr(expr: String, columnName: String): proto.Expression = {
val builder = proto.Expression.newBuilder()
expr.toLowerCase(Locale.ROOT) match {
// We special handle a few cases that have alias that are not in function registry.
case "avg" | "average" | "mean" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("avg")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.avg(columnName)
case "stddev" | "std" =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("stddev")
- .addArguments(inputExpr)
- .setIsDistinct(false)
+ functions.stddev(columnName)
// Also special handle count because we need to take care count(*).
case "count" | "size" =>
- // Turn count(*) into count(1)
- inputExpr match {
- case s if s.hasUnresolvedStar =>
- val exprBuilder = proto.Expression.newBuilder
- exprBuilder.getLiteralBuilder.setInteger(1)
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(exprBuilder)
- .setIsDistinct(false)
- case _ =>
- builder.getUnresolvedFunctionBuilder
- .setFunctionName("count")
- .addArguments(inputExpr)
- .setIsDistinct(false)
- }
+ functions.col(columnName)
case name =>
builder.getUnresolvedFunctionBuilder
.setFunctionName(name)
- .addArguments(inputExpr)
+ .addArguments(df(columnName).expr)
.setIsDistinct(false)
}
builder.build()
}
+
+ /**
+ * Compute aggregates by specifying a series of aggregate columns. Note that this function by
+ * default retains the grouping columns in its output. To not retain grouping columns, set
+ * `spark.sql.retainGroupColumns` to false.
+ *
+ * The available aggregate methods are defined in [[org.apache.spark.sql.functions]].
+ *
+ * {{{
+ * // Selects the age of the oldest employee and the aggregate expense for each department
+ *
+ * // Scala:
+ * import org.apache.spark.sql.functions._
+ * df.groupBy("department").agg(max("age"), sum("expense"))
+ *
+ * // Java:
+ * import static org.apache.spark.sql.functions.*;
+ * df.groupBy("department").agg(max("age"), sum("expense"));
+ * }}}
+ *
+ * Note that before Spark 1.4, the default behavior is to NOT retain grouping columns. To change
+ * to that behavior, set config variable `spark.sql.retainGroupColumns` to `false`.
+ * {{{
+ * // Scala, 1.3.x:
+ * df.groupBy("department").agg($"department", max("age"), sum("expense"))
+ *
+ * // Java, 1.3.x:
+ * df.groupBy("department").agg(col("department"), max("age"), sum("expense"));
+ * }}}
+ *
+ * @since 3.4.0
+ */
+ @scala.annotation.varargs
+ def agg(expr: Column, exprs: Column*): DataFrame = {
+ toDF((expr +: exprs).map { case c =>
Review Comment:
TBH I would just make `toDF` take `Column`s instead of `Expression`s, that saves you from doing conversions everywhere.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] hvanhovell closed pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell closed pull request #40070: [SPARK-42481][CONNECT] Implement agg.{max,min,mean,count,avg,sum}
URL: https://github.com/apache/spark/pull/40070
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org