You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/01/07 08:43:38 UTC
[GitHub] [spark] beliefer opened a new pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
beliefer opened a new pull request #35130:
URL: https://github.com/apache/spark/pull/35130
### What changes were proposed in this pull request?
Currently, DS V2 supports complete aggregate push-down `AVG`. But, supports partial aggregate push-down `AVG` is very useful.
This PR convert `AVG` as `SUM/COUNT` to implement partial aggregate push-down.
### Why are the changes needed?
DS V2 supports partial aggregate push-down `AVG`
### Does this PR introduce _any_ user-facing change?
'Yes'. DS V2 could partial aggregate push-down `AVG`
### How was this patch tested?
New 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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785764681
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -210,16 +250,33 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
}
}
+ private def collectAggregates(resultExpressions: Seq[NamedExpression],
+ aggExprToOutputOrdinal: mutable.HashMap[Expression, Int]): Seq[AggregateExpression] = {
+ var ordinal = 0
+ resultExpressions.flatMap { expr =>
+ expr.collect {
+ // Do not push down duplicated aggregate expressions. For example,
+ // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
+ // `max(a)` to the data source.
+ case agg: AggregateExpression
+ if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
+ aggExprToOutputOrdinal(agg.canonicalized) = ordinal
+ ordinal += 1
+ agg
+ }
+ }
+ }
+
private def supportPartialAggPushDown(agg: Aggregation): Boolean = {
// We don't know the agg buffer of `GeneralAggregateFunc`, so can't do partial agg push down.
agg.aggregateExpressions().forall(!_.isInstanceOf[GeneralAggregateFunc])
}
- private def addCastIfNeeded(aggAttribute: AttributeReference, aggDataType: DataType) =
- if (aggAttribute.dataType == aggDataType) {
- aggAttribute
+ private def addCastIfNeeded(expression: Expression, aggDataType: DataType) =
Review comment:
```suggestion
private def addCastIfNeeded(expression: Expression, expectedDataType: DataType) =
```
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r784021016
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -164,25 +188,25 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
Project(projectExpressions, scanRelation)
} else {
val plan = Aggregate(
- output.take(groupingExpressions.length), resultExpressions, scanRelation)
+ output.take(groupingExpressions.length), finalResultExpressions, scanRelation)
// scalastyle:off
// Change the optimized logical plan to reflect the pushed down aggregate
// e.g. TABLE t (c1 INT, c2 INT, c3 INT)
- // SELECT min(c1), max(c1) FROM t GROUP BY c2;
+ // SELECT min(c1), max(c1), avg(c1) FROM t GROUP BY c2;
Review comment:
When we reach here, the AVG translation is already done and it's weird to see we mention it in the comment.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r786046107
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ // scalastyle:on
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = aggregate.Sum(avg.child).toAggregateExpression(isDistinct)
Review comment:
In normal cases, `sum(c_long)` should be OK, as the final result should be the same after division. However, in the case of overflow, `sum(c_long)` fails and `sum(cast(c_long as double)` returns Double.Infinity.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782221165
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
##########
@@ -220,10 +222,16 @@ abstract class JdbcDialect extends Serializable with Logging{
Some(s"SUM($distinct$column)")
case _: CountStar =>
Some("COUNT(*)")
- case f: GeneralAggregateFunc if f.name() == "AVG" =>
- assert(f.inputs().length == 1)
- val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"AVG($distinct${f.inputs().head})")
+ case avg: Avg =>
+ if (avg.column.fieldNames.length != 1) return None
+ val distinct = if (avg.isDistinct) "DISTINCT " else ""
+ val column = quoteIdentifier(avg.column.fieldNames.head)
+ if (supportCompletePushDown) {
+ Some(s"AVG($distinct$column)")
+ } else {
+ // For simplify code, we not reuse exists `SUM` or `COUNT`.
+ Some(s"SUM($distinct$column), COUNT($distinct$column)")
Review comment:
It's super weird to see this translation at the data source side, not the Spark side.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r780295489
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
##########
@@ -223,7 +225,11 @@ abstract class JdbcDialect extends Serializable with Logging{
case f: GeneralAggregateFunc if f.name() == "AVG" =>
assert(f.inputs().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"AVG($distinct${f.inputs().head})")
+ if (supportCompletePushDown) {
+ Some(s"AVG($distinct${f.inputs().head})")
+ } else {
+ Some(s"Sum($distinct${f.inputs().head}), Count($distinct${f.inputs().head})")
Review comment:
why do we need it? It's Spark's responsibility to rewrite avg to sum/count, not the data source.
--
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] beliefer commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
beliefer commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1007849173
> What's the new algorithm? It's simple before: if agg funcs contain `GeneralAggregateFunc`, don't try partial pushdown. Othwewise, try complete pushdown first, then partial pushdown.
Yes. But AVG is common used aggregate function, we hope try my best to pushdown. Maybe we could construct Arerage in DS V2, so we keep the simply.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782782172
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -87,9 +87,26 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
if filters.isEmpty && project.forall(_.isInstanceOf[AttributeReference]) =>
sHolder.builder match {
case r: SupportsPushDownAggregates =>
+ val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
+ groupingExpressions, sHolder.relation.output)
+ val translatedGroupBys =
+ PushDownUtils.translateAggregation(Seq.empty, normalizedGroupingExpressions)
Review comment:
This assumes the v2 source only need the group by columns to decide if it supports complete pushdown or not. Do you have strong evidence to indicate it's true for all v2 sources?
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r784483389
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,49 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ var findAverage = false
+ val newResultExpressions = resultExpressions.map { expr =>
Review comment:
OK
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785765189
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.expressions.aggregate;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+
+/**
+ * An aggregate function that returns the mean of all the values in a group.
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public final class Avg implements AggregateFunc {
Review comment:
can we update the classdoc of `GeneralAggregateFunc` to exclude avg?
--
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] cloud-fan commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1007449447
What's the new algorithm? It's simple before: if agg funcs contain `GeneralAggregateFunc`, don't try partial pushdown. Othwewise, try complete pushdown first, then partial pushdown.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782762488
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -87,9 +87,26 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
if filters.isEmpty && project.forall(_.isInstanceOf[AttributeReference]) =>
sHolder.builder match {
case r: SupportsPushDownAggregates =>
+ val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
+ groupingExpressions, sHolder.relation.output)
+ val translatedGroupBys =
+ PushDownUtils.translateAggregation(Seq.empty, normalizedGroupingExpressions)
Review comment:
Because the aggregates are collected from Aggregate. I want call `r.supportCompletePushDown` and split `AVG` into 2 functions: `SUM` and `COUNT` first, so we collect `AggregateExpression` just once.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782686444
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
##########
@@ -220,10 +222,16 @@ abstract class JdbcDialect extends Serializable with Logging{
Some(s"SUM($distinct$column)")
case _: CountStar =>
Some("COUNT(*)")
- case f: GeneralAggregateFunc if f.name() == "AVG" =>
- assert(f.inputs().length == 1)
- val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"AVG($distinct${f.inputs().head})")
+ case avg: Avg =>
+ if (avg.column.fieldNames.length != 1) return None
+ val distinct = if (avg.isDistinct) "DISTINCT " else ""
+ val column = quoteIdentifier(avg.column.fieldNames.head)
+ if (supportCompletePushDown) {
+ Some(s"AVG($distinct$column)")
+ } else {
+ // For simplify code, we not reuse exists `SUM` or `COUNT`.
+ Some(s"SUM($distinct$column), COUNT($distinct$column)")
Review comment:
OK. Let me have a try.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785762233
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = addCastIfNeeded(aggregate.Sum(avg.child)
+ .toAggregateExpression(isDistinct), avg.sumDataType)
+ val count = aggregate.Count(avg.child).toAggregateExpression(isDistinct)
+ val aggExpr = avg.sumDataType match {
+ case _: YearMonthIntervalType =>
+ If(EqualTo(count, Literal(0L)),
+ Literal(null, YearMonthIntervalType()), DivideYMInterval(sum, count))
+ case _: DayTimeIntervalType =>
+ If(EqualTo(count, Literal(0L)),
+ Literal(null, DayTimeIntervalType()), DivideDTInterval(sum, count))
+ case _ =>
+ Divide(sum, addCastIfNeeded(count, avg.sumDataType), false)
Review comment:
why do you cast the count? It always returns long type.
--
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] beliefer commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1008887243
> Can you describe the algorithm in more detail? What if the data source support complete aggregate pushdown and can take AVG directly?
This PR used to partial aggregate pushdown when data source doesn't support complete aggregate pushdown.
--
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] cloud-fan commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1007449447
What's the new algorithm? It's simple before: if agg funcs contain `GeneralAggregateFunc`, don't try partial pushdown. Othwewise, try complete pushdown first, then partial pushdown.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782759892
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
##########
@@ -107,14 +107,12 @@ object PushDownUtils extends PredicateHelper {
}
/**
- * Pushes down aggregates to the data source reader
+ * Translate aggregate expressions and group by expressions.
*
- * @return pushed aggregation.
+ * @return translated aggregation.
*/
- def pushAggregates(
- scanBuilder: SupportsPushDownAggregates,
- aggregates: Seq[AggregateExpression],
- groupBy: Seq[Expression]): Option[Aggregation] = {
+ def translateAggregation(
Review comment:
shall we put it in `DataSourceStrategy`?
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r786047569
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ // scalastyle:on
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = aggregate.Sum(avg.child).toAggregateExpression(isDistinct)
Review comment:
Maybe we can leave a TODO and deal with the overflow issue later.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785814514
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
Review comment:
OK
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782760381
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
##########
@@ -107,14 +107,12 @@ object PushDownUtils extends PredicateHelper {
}
/**
- * Pushes down aggregates to the data source reader
+ * Translate aggregate expressions and group by expressions.
*
- * @return pushed aggregation.
+ * @return translated aggregation.
*/
- def pushAggregates(
- scanBuilder: SupportsPushDownAggregates,
- aggregates: Seq[AggregateExpression],
- groupBy: Seq[Expression]): Option[Aggregation] = {
+ def translateAggregation(
Review comment:
OK
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r783002684
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -87,9 +87,26 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
if filters.isEmpty && project.forall(_.isInstanceOf[AttributeReference]) =>
sHolder.builder match {
case r: SupportsPushDownAggregates =>
+ val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
+ groupingExpressions, sHolder.relation.output)
+ val translatedGroupBys =
+ PushDownUtils.translateAggregation(Seq.empty, normalizedGroupingExpressions)
Review comment:
Let's pass aggregate expressions too for all the DS V2.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r783609060
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,48 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // Spark can't do complete agg push down to data source and try to do partial agg
+ // push down by convert `AVG` to: `SUM / COUNT`.
Review comment:
I'd put it this way
```
// The data source doesn't support the complete push-down of this aggregation.
// Here we translate AVG to SUM / COUNT, so that it's more likely to be pushed,
// completely or partially.
```
--
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] huaxingao commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
huaxingao commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r784389060
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -874,4 +874,32 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df, Seq(Row(2)))
// scalastyle:on
}
+
+ test("scan with aggregate push-down: partial push-down SUM, AVG, COUNT") {
+ val df = spark.read
+ .option("partitionColumn", "dept")
+ .option("lowerBound", "0")
+ .option("upperBound", "2")
+ .option("numPartitions", "2")
+ .table("h2.test.employee")
+ .agg(sum($"SALARY").as("sum"), avg($"SALARY").as("avg"), count($"SALARY").as("count"))
+ checkAggregateRemoved(df, false)
Review comment:
This only verifies the aggregates are not completely pushed down. We still need to check if the partial push down works, right?
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r784022191
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,49 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ var findAverage = false
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ findAverage = true
+ val left = addCastIfNeeded(aggregate.Sum(avg.child)
+ .toAggregateExpression(isDistinct), avg.dataType)
+ val right = addCastIfNeeded(aggregate.Count(avg.child)
+ .toAggregateExpression(isDistinct), avg.dataType)
+ Divide(left, right)
Review comment:
This issue is not resolved yet, I don't think here we return the same data type as AVG before.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785761473
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = addCastIfNeeded(aggregate.Sum(avg.child)
+ .toAggregateExpression(isDistinct), avg.sumDataType)
Review comment:
This cast seems not necessary, as `Sum.dataType` is exactly the same with `Average.sumDataType`
--
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] cloud-fan commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1017102913
thanks, merging to master!
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r786033834
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ // scalastyle:on
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = aggregate.Sum(avg.child).toAggregateExpression(isDistinct)
Review comment:
to make the `Sum` work properly, we should cast the input to a wider type:
`aggregate.Sum(addCastIfNeeded(avg.child, avg.sumDataType)).toAgg...`
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r783566245
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,40 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ val newResultExpressions = resultExpressions.map { expr =>
Review comment:
OK
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782762488
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -87,9 +87,26 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
if filters.isEmpty && project.forall(_.isInstanceOf[AttributeReference]) =>
sHolder.builder match {
case r: SupportsPushDownAggregates =>
+ val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
+ groupingExpressions, sHolder.relation.output)
+ val translatedGroupBys =
+ PushDownUtils.translateAggregation(Seq.empty, normalizedGroupingExpressions)
Review comment:
Because the aggregates are collected from Aggregate. I want call r.supportCompletePushDown and split AVG into 2 functions: SUM and COUNT first, so we collect AggregateExpression just once.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782760379
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
##########
@@ -130,8 +128,17 @@ object PushDownUtils extends PredicateHelper {
return None
}
- val agg = new Aggregation(translatedAggregates.toArray, translatedGroupBys.toArray)
- Some(agg).filter(scanBuilder.pushAggregation)
+ Some(new Aggregation(translatedAggregates.toArray, translatedGroupBys.toArray))
+ }
+
+ /**
+ * Pushes down aggregates to the data source reader
+ *
+ * @return pushed aggregation.
+ */
+ def pushAggregates(
+ scanBuilder: SupportsPushDownAggregates, aggOpt: Option[Aggregation]): Option[Aggregation] = {
+ aggOpt.filter(scanBuilder.pushAggregation)
Review comment:
Doesn't seem necessary to create a method for only one line of 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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r783130063
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,40 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val left = addCastIfNeeded(aggregate.Sum(avg.child)
+ .toAggregateExpression(isDistinct), avg.dataType)
Review comment:
I don't think we can retain the original data type of AVG by doing so. Please check the implementation of Spark AVG.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r783128769
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,40 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ val newResultExpressions = resultExpressions.map { expr =>
Review comment:
nit: add some code comments to explain 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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782761438
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -103,10 +120,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
}
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
- val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
- groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = PushDownUtils.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+
+ val pushedAggregates = PushDownUtils.pushAggregates(r, translatedAggregates)
if (pushedAggregates.isEmpty) {
aggNode // return original plan node
} else if (!supportPartialAggPushDown(pushedAggregates.get) &&
Review comment:
do we still need to call `supportCompletePushDown` here?
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r780603780
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
##########
@@ -223,7 +225,11 @@ abstract class JdbcDialect extends Serializable with Logging{
case f: GeneralAggregateFunc if f.name() == "AVG" =>
assert(f.inputs().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"AVG($distinct${f.inputs().head})")
+ if (supportCompletePushDown) {
+ Some(s"AVG($distinct${f.inputs().head})")
+ } else {
+ Some(s"Sum($distinct${f.inputs().head}), Count($distinct${f.inputs().head})")
Review comment:
If Spark cannot complete pushdown AVG to database, we still want partial pushdown.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r780603780
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
##########
@@ -223,7 +225,11 @@ abstract class JdbcDialect extends Serializable with Logging{
case f: GeneralAggregateFunc if f.name() == "AVG" =>
assert(f.inputs().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"AVG($distinct${f.inputs().head})")
+ if (supportCompletePushDown) {
+ Some(s"AVG($distinct${f.inputs().head})")
+ } else {
+ Some(s"Sum($distinct${f.inputs().head}), Count($distinct${f.inputs().head})")
Review comment:
If Spark cannot complete pushdown AVG to database, we still want partial pushdown.
--
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] cloud-fan closed pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #35130:
URL: https://github.com/apache/spark/pull/35130
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785767123
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -874,4 +876,47 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df, Seq(Row(2)))
// scalastyle:on
}
+
+ private def checkAverageConversion(df: DataFrame): Unit = {
Review comment:
this does not check the partial agg pushdown. I think we need to test two cases:
1. the JDBC table has only 1 partition and supports complete pushdown, so there is no agg operator at Spark side (optimized plan)
2. the JDBC table doesn't support complete pushdown, there is still an agg operator at Spark side (optimized plan) but the scan relation output contains aggregated result.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r786360517
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ // scalastyle:on
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = aggregate.Sum(avg.child).toAggregateExpression(isDistinct)
Review comment:
OK
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785764169
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = addCastIfNeeded(aggregate.Sum(avg.child)
+ .toAggregateExpression(isDistinct), avg.sumDataType)
+ val count = aggregate.Count(avg.child).toAggregateExpression(isDistinct)
+ val aggExpr = avg.sumDataType match {
+ case _: YearMonthIntervalType =>
+ If(EqualTo(count, Literal(0L)),
+ Literal(null, YearMonthIntervalType()), DivideYMInterval(sum, count))
+ case _: DayTimeIntervalType =>
+ If(EqualTo(count, Literal(0L)),
+ Literal(null, DayTimeIntervalType()), DivideDTInterval(sum, count))
+ case _ =>
+ Divide(sum, addCastIfNeeded(count, avg.sumDataType), false)
+ }
+
+ Cast(aggExpr, avg.dataType)
Review comment:
to fully match the implementation of `Average`, we should not add cast here, but to the inputs of `Divide`. The reason is that, `Divide.dataType` is the same as `left.DataType`, so casting the inputs of `Divide` can guarantee the final result data type as well.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r783563380
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,40 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val left = addCastIfNeeded(aggregate.Sum(avg.child)
+ .toAggregateExpression(isDistinct), avg.dataType)
Review comment:
If `AVG.child.dataType` is `LongType`, `AVG.dataType` will be `DoubleType`, but `Sum.dataType` will be `LongType `. Because `Count.dataType` is `LongType` and `Div.dataType` is `Sum.dataType`, so `Sum/Count` returns 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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782762975
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -103,10 +120,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
}
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
- val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
- groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = PushDownUtils.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+
+ val pushedAggregates = PushDownUtils.pushAggregates(r, translatedAggregates)
if (pushedAggregates.isEmpty) {
aggNode // return original plan node
} else if (!supportPartialAggPushDown(pushedAggregates.get) &&
Review comment:
Yes.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782760821
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
##########
@@ -130,8 +128,17 @@ object PushDownUtils extends PredicateHelper {
return None
}
- val agg = new Aggregation(translatedAggregates.toArray, translatedGroupBys.toArray)
- Some(agg).filter(scanBuilder.pushAggregation)
+ Some(new Aggregation(translatedAggregates.toArray, translatedGroupBys.toArray))
+ }
+
+ /**
+ * Pushes down aggregates to the data source reader
+ *
+ * @return pushed aggregation.
+ */
+ def pushAggregates(
+ scanBuilder: SupportsPushDownAggregates, aggOpt: Option[Aggregation]): Option[Aggregation] = {
+ aggOpt.filter(scanBuilder.pushAggregation)
Review comment:
Yeah. I'm a little uncertain, so I'm waiting for your comment.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785758951
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
Review comment:
please turn on scalastyle back after the comment block
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785767123
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -874,4 +876,47 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(df, Seq(Row(2)))
// scalastyle:on
}
+
+ private def checkAverageConversion(df: DataFrame): Unit = {
Review comment:
this does not check the partial agg pushdown. I think we need to test two cases:
1. the JDBC table has only 1 partition and supports complete pushdown, so there is no agg operator at Spark side
2. the JDBC table doesn't support complete pushdown, there is only one agg operator at Spark side as the partial agg has been pushed.
--
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] beliefer commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r785779225
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector.expressions.aggregate;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.expressions.NamedReference;
+
+/**
+ * An aggregate function that returns the mean of all the values in a group.
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public final class Avg implements AggregateFunc {
Review comment:
OK
--
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] beliefer commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
beliefer commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1017304510
@cloud-fan @huaxingao Thank you for your review.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r784020031
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,49 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get)) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ var findAverage = false
+ val newResultExpressions = resultExpressions.map { expr =>
Review comment:
we can avoid this transformation at all if `translatedAggregates.get.aggregateExpressions` contains `Avg`
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r780294560
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -212,7 +224,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
private def supportPartialAggPushDown(agg: Aggregation): Boolean = {
// We don't know the agg buffer of `GeneralAggregateFunc`, so can't do partial agg push down.
- agg.aggregateExpressions().forall(!_.isInstanceOf[GeneralAggregateFunc])
+ agg.aggregateExpressions().forall { aggregateFunc =>
+ !aggregateFunc.isInstanceOf[GeneralAggregateFunc] ||
+ aggregateFunc.asInstanceOf[GeneralAggregateFunc].name() == "AVG"
Review comment:
If `AVG` can be partially pushed, I'd like to create a dedicated class for it, instead of using `GeneralAggregateFunc`.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782227932
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
##########
@@ -220,10 +222,16 @@ abstract class JdbcDialect extends Serializable with Logging{
Some(s"SUM($distinct$column)")
case _: CountStar =>
Some("COUNT(*)")
- case f: GeneralAggregateFunc if f.name() == "AVG" =>
- assert(f.inputs().length == 1)
- val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"AVG($distinct${f.inputs().head})")
+ case avg: Avg =>
+ if (avg.column.fieldNames.length != 1) return None
+ val distinct = if (avg.isDistinct) "DISTINCT " else ""
+ val column = quoteIdentifier(avg.column.fieldNames.head)
+ if (supportCompletePushDown) {
+ Some(s"AVG($distinct$column)")
+ } else {
+ // For simplify code, we not reuse exists `SUM` or `COUNT`.
+ Some(s"SUM($distinct$column), COUNT($distinct$column)")
Review comment:
Since partial agg pushdown needs Spark to do final agg, Spark must be fully aware of the AVG translation, to make the final agg match the data source scan with partial agg pushed.
I think the process should be
1. Spark translates catalyst Aggregate operator to a DS V2 `Aggregation`.
2. Spark calls `supportCompletePushDown` to check if it can completely push down agg
3. JDBC source returns false in `supportCompletePushDown` if AVG is present
4. Spark gives up complete agg push down, and starts to try partial agg push down
5. Spark splits AVG into 2 functions: SUM and COUNT, and pushes the `Aggregation` to JDBC source
6. Spark constructs the final agg and calculates AVG by SUM / COUNT.
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r786036810
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -88,25 +88,65 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
sHolder.builder match {
case r: SupportsPushDownAggregates =>
val aggExprToOutputOrdinal = mutable.HashMap.empty[Expression, Int]
- var ordinal = 0
- val aggregates = resultExpressions.flatMap { expr =>
- expr.collect {
- // Do not push down duplicated aggregate expressions. For example,
- // `SELECT max(a) + 1, max(a) + 2 FROM ...`, we should only push down one
- // `max(a)` to the data source.
- case agg: AggregateExpression
- if !aggExprToOutputOrdinal.contains(agg.canonicalized) =>
- aggExprToOutputOrdinal(agg.canonicalized) = ordinal
- ordinal += 1
- agg
- }
- }
+ val aggregates = collectAggregates(resultExpressions, aggExprToOutputOrdinal)
val normalizedAggregates = DataSourceStrategy.normalizeExprs(
aggregates, sHolder.relation.output).asInstanceOf[Seq[AggregateExpression]]
val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
groupingExpressions, sHolder.relation.output)
- val pushedAggregates = PushDownUtils.pushAggregates(
- r, normalizedAggregates, normalizedGroupingExpressions)
+ val translatedAggregates = DataSourceStrategy.translateAggregation(
+ normalizedAggregates, normalizedGroupingExpressions)
+ val (finalResultExpressions, finalAggregates, finalTranslatedAggregates) = {
+ if (translatedAggregates.isEmpty ||
+ r.supportCompletePushDown(translatedAggregates.get) ||
+ translatedAggregates.get.aggregateExpressions().forall(!_.isInstanceOf[Avg])) {
+ (resultExpressions, aggregates, translatedAggregates)
+ } else {
+ // scalastyle:off
+ // The data source doesn't support the complete push-down of this aggregation.
+ // Here we translate `AVG` to `SUM / COUNT`, so that it's more likely to be
+ // pushed, completely or partially.
+ // e.g. TABLE t (c1 INT, c2 INT, c3 INT)
+ // SELECT avg(c1) FROM t GROUP BY c2;
+ // The original logical plan is
+ // Aggregate [c2#10],[avg(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ //
+ // After convert avg(c1#9) to sum(c1#9)/count(c1#9)
+ // we have the following
+ // Aggregate [c2#10],[sum(c1#9)/count(c1#9) AS avg(c1)#19]
+ // +- ScanOperation[...]
+ // scalastyle:on
+ val newResultExpressions = resultExpressions.map { expr =>
+ expr.transform {
+ case AggregateExpression(avg: aggregate.Average, _, isDistinct, _, _) =>
+ val sum = aggregate.Sum(avg.child).toAggregateExpression(isDistinct)
Review comment:
OK I see the problem here. For `avg(c_long)`, we want to evaluate `sum(cast(c_long as double)) / cast(count(c_long) as double)`. However, `sum(cast(c_long as double))` can't be pushed down as the input is not a simple attribute.
--
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] cloud-fan commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1015090389
LGTM if tests pass
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r780294560
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -212,7 +224,10 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
private def supportPartialAggPushDown(agg: Aggregation): Boolean = {
// We don't know the agg buffer of `GeneralAggregateFunc`, so can't do partial agg push down.
- agg.aggregateExpressions().forall(!_.isInstanceOf[GeneralAggregateFunc])
+ agg.aggregateExpressions().forall { aggregateFunc =>
+ !aggregateFunc.isInstanceOf[GeneralAggregateFunc] ||
+ aggregateFunc.asInstanceOf[GeneralAggregateFunc].name() == "AVG"
Review comment:
If `AVG` can be partially pushed, I'd like to create a dedicated class for it, instead of using `GeneralAggregateFunc`.
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
##########
@@ -223,7 +225,11 @@ abstract class JdbcDialect extends Serializable with Logging{
case f: GeneralAggregateFunc if f.name() == "AVG" =>
assert(f.inputs().length == 1)
val distinct = if (f.isDistinct) "DISTINCT " else ""
- Some(s"AVG($distinct${f.inputs().head})")
+ if (supportCompletePushDown) {
+ Some(s"AVG($distinct${f.inputs().head})")
+ } else {
+ Some(s"Sum($distinct${f.inputs().head}), Count($distinct${f.inputs().head})")
Review comment:
why do we need it? It's Spark's responsibility to rewrite avg to sum/count, not the data source.
--
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] beliefer commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down AVG
Posted by GitBox <gi...@apache.org>.
beliefer commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1007849173
> What's the new algorithm? It's simple before: if agg funcs contain `GeneralAggregateFunc`, don't try partial pushdown. Othwewise, try complete pushdown first, then partial pushdown.
Yes. But AVG is common used aggregate function, we hope try my best to pushdown. Maybe we could construct Arerage in DS V2, so we keep the simply.
--
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] cloud-fan commented on pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #35130:
URL: https://github.com/apache/spark/pull/35130#issuecomment-1008817825
Can you describe the algorithm in more detail? What if the data source support complete aggregate pushdown and can take AVG directly?
--
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] cloud-fan commented on a change in pull request #35130: [SPARK-37839][SQL] DS V2 supports partial aggregate push-down `AVG`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35130:
URL: https://github.com/apache/spark/pull/35130#discussion_r782760705
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
##########
@@ -87,9 +87,26 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
if filters.isEmpty && project.forall(_.isInstanceOf[AttributeReference]) =>
sHolder.builder match {
case r: SupportsPushDownAggregates =>
+ val normalizedGroupingExpressions = DataSourceStrategy.normalizeExprs(
+ groupingExpressions, sHolder.relation.output)
+ val translatedGroupBys =
+ PushDownUtils.translateAggregation(Seq.empty, normalizedGroupingExpressions)
Review comment:
why can't we put the real aggregate functions here?
--
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