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/19 09:00:31 UTC
[GitHub] [spark] beliefer opened a new pull request #35248: [SPARK-37960][SQL] Support aggregate push down `SUM(CASE ... WHEN ... ELSE ... END)`
beliefer opened a new pull request #35248:
URL: https://github.com/apache/spark/pull/35248
### What changes were proposed in this pull request?
Currently, Spark supports aggregate push down `SUM(column)` into JDBC data source.
`SUM(CASE ... WHEN ... ELSE ... END)` is very useful for users.
case when could help users to finish many works.
### Why are the changes needed?
Support push down `SUM(CASE ... WHEN ... ELSE ... END)`
### Does this PR introduce _any_ user-facing change?
Yes. Users could use `SUM(CASE ... WHEN ... ELSE ... END)` with aggregate push down.
### 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] beliefer commented on a change in pull request #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r790079978
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
##########
@@ -860,3 +870,30 @@ object PushableColumnAndNestedColumn extends PushableColumnBase {
object PushableColumnWithoutNestedColumn extends PushableColumnBase {
override val nestedPredicatePushdownEnabled = false
}
+
+/**
+ * Get the SQL string of an expression that can be pushed down.
+ */
+object PushableExpression {
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r801712556
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, EqualTo, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to build expression and generate sql string.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"${l.get} ${b.sqlOperator} ${r.get}")
Review comment:
is it safer to do `(${l.get}) ${b.sqlOperator} (${r.get)`?
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r794390939
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{CaseWhen, Expression, Literal}
+import org.apache.spark.sql.connector.expressions.{FieldReference, GeneralSQLExpression, LiteralValue}
+import org.apache.spark.sql.execution.datasources.PushableColumnWithoutNestedColumn
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy
+
+/**
+ * The builder to build {@link GeneralSQLExpression}.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[GeneralSQLExpression] = e match {
+ case CaseWhen(branches, elseValue) =>
+ val newBranches = branches.collect {
+ case (predicate: Expression, Literal(value, dataType)) =>
Review comment:
We should implement this SQL string builder in a more flexible way. Something like
```
def build = generateSQL(e)
private def generateSQL(expr: Expression): Option[String] = expr match {
case l: Literal => Some(...)
case a: Attribute => Some(quoteIfNeeded(a.name))
case ... // other basic expressions, like predicates?
case CaseWhen(branches, elseValue) =>
val conditionsSQL = branches.map(_._1).flatMap(generateSQL)
val valuesSQL = branches.map(_._2).flatMap(generateSQL)
...
case _ => None
}
```
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802530842
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
##########
@@ -487,18 +488,18 @@ object OrcUtils extends Logging {
val aggORCValues: Seq[WritableComparable[_]] =
aggregation.aggregateExpressions.zipWithIndex.map {
- case (max: Max, index) =>
- val columnName = max.column.fieldNames.head
+ case (max: Max, index) if max.column.isInstanceOf[NamedReference] =>
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802694708
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -806,17 +806,124 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(query, Seq(Row(29000.0)))
}
- test("scan with aggregate push-down: SUM(CASE WHEN) with group by") {
- val df =
- sql("SELECT SUM(CASE WHEN SALARY > 0 THEN 1 ELSE 0 END) FROM h2.test.employee GROUP BY DEPT")
- checkAggregateRemoved(df, false)
+ test("scan with aggregate push-down: aggregate function with CASE WHEN") {
+ val df = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | COUNT(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | AVG(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MAX(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN SALARY is null THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df)
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [], "
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NULL THEN 0.00 ELSE SALARY END), " +
+ "COUNT(CASE WHEN SALARY IS NULL THEN 0.0..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
- checkAnswer(df, Seq(Row(1), Row(2), Row(2)))
+ checkAnswer(df, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df2 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | AVG(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MIN(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df2)
+ df2.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NOT NULL THEN SALARY ELSE 0.00 END), " +
+ "COUNT(CASE WHEN SALARY IS NOT NULL ..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ }
+ checkAnswer(df2, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df3 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY > 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY >= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY < 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY <= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY = 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY > 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY >= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY < 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY <= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY = 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY != 0 THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df3)
+ df3.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN (SALARY) > (0.00) THEN 0.00 ELSE SALARY END), " +
+ "SUM(CASE WHEN (SALARY) >= (0.00) THE..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df3, expected_plan_fragment)
+ }
+ checkAnswer(df3, Seq(Row(0d, 0d, 12000d, 12000d, 12000d, 12000d, 12000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 19000d, 19000d, 19000d, 19000d, 19000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 22000d, 22000d, 22000d, 22000d, 22000d, 0d, 0d, 0d, 0d)))
+
+ val df4 = sql(
+ """
+ |SELECT
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END),
Review comment:
I think this test covers everything. We don't need the above 3 test queries.
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r801713720
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, EqualTo, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to build expression and generate sql string.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"${l.get} ${b.sqlOperator} ${r.get}")
+ } else {
+ None
+ }
+ case Not(EqualTo(left, right)) =>
Review comment:
can we also support `Not` in general? something like `NOT (generateSQL(child))`
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #35248:
URL: https://github.com/apache/spark/pull/35248#issuecomment-1034920088
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r794383849
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/aggregate/Avg.java
##########
@@ -27,15 +27,15 @@
*/
@Evolving
public final class Avg implements AggregateFunc {
- private final NamedReference column;
+ private final Expression column;
Review comment:
shall we rename `column` to `input`?
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802315733
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
##########
@@ -860,3 +856,13 @@ object PushableColumnAndNestedColumn extends PushableColumnBase {
object PushableColumnWithoutNestedColumn extends PushableColumnBase {
override val nestedPredicatePushdownEnabled = false
}
+
+/**
+ * Get the SQL string of an expression that can be pushed down.
Review comment:
need to update the comment now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] beliefer commented on a change in pull request #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r803213008
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
##########
@@ -52,10 +52,11 @@ object AggregatePushDownUtils {
def processMinOrMax(agg: AggregateFunc): Boolean = {
val (column, aggType) = agg match {
- case max: Max => (max.column, "max")
- case min: Min => (min.column, "min")
- case _ =>
- throw new IllegalArgumentException(s"Unexpected type of AggregateFunc ${agg.describe}")
+ case max: Max if max.column().isInstanceOf[NamedReference] =>
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 closed pull request #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #35248:
URL: https://github.com/apache/spark/pull/35248
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802332175
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
##########
@@ -860,3 +856,13 @@ object PushableColumnAndNestedColumn extends PushableColumnBase {
object PushableColumnWithoutNestedColumn extends PushableColumnBase {
override val nestedPredicatePushdownEnabled = false
}
+
+/**
+ * Get the SQL string of an expression that can be pushed down.
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r803289518
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -806,17 +806,51 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(query, Seq(Row(29000.0)))
}
- test("scan with aggregate push-down: SUM(CASE WHEN) with group by") {
- val df =
- sql("SELECT SUM(CASE WHEN SALARY > 0 THEN 1 ELSE 0 END) FROM h2.test.employee GROUP BY DEPT")
- checkAggregateRemoved(df, false)
+ test("scan with aggregate push-down: aggregate function with CASE WHEN") {
+ val df = sql(
+ """
+ |SELECT
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY <= 13000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 11000 OR SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR SALARY < 9000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR NOT(SALARY >= 9000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND SALARY >= 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) OR SALARY > 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY != 0) OR NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000 AND SALARY > 8000) THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN NOT(SALARY > 8000 OR SALARY IS NULL) THEN SALARY ELSE 0 END),
+ | SUM(CASE WHEN NOT(SALARY > 8000 AND SALARY IS NOT NULL) THEN SALARY ELSE 0 END),
+ | SUM(CASE WHEN SALARY > 10000 THEN 2 WHEN SALARY > 8000 THEN 1 END),
+ | AVG(CASE WHEN NOT(SALARY > 8000 OR SALARY IS NOT NULL) THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df)
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [], "
+ "PushedAggregates: [COUNT(CASE WHEN ((SALARY) > (8000.00)) AND ((SALARY) < (10000.00))" +
+ " THEN SALARY ELSE 0.00 END), C..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
- checkAnswer(df, Seq(Row(1), Row(2), Row(2)))
+ checkAnswer(df, Seq(Row(1, 1, 1, 1, 1, 0d, 12000d, 0d, 12000d, 12000d, 0d, 0d, 2, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 10000d, 0d, 10000d, 10000d, 0d, 0d, 2, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 12000d, 0d, 12000d, 12000d, 0d, 0d, 3, 0d)))
+ }
+
+ test("scan with aggregate push-down: aggregate function with NVL") {
Review comment:
can we test with UDF? something that we will never support SQL generation.
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r789499565
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
##########
@@ -860,3 +870,30 @@ object PushableColumnAndNestedColumn extends PushableColumnBase {
object PushableColumnWithoutNestedColumn extends PushableColumnBase {
override val nestedPredicatePushdownEnabled = false
}
+
+/**
+ * Get the SQL string of an expression that can be pushed down.
+ */
+object PushableExpression {
Review comment:
The code size here can go very big very fast. I'd like to have a new class `ExpressionSQLBuilder` in a new file, probably under the package `org.apache.spark.sql.catalyst.utils`, in the `sql/core` module. It can be used like `new ExpressionSQLBuilder(catalystExpr).build()`
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r794385170
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{CaseWhen, Expression, Literal}
+import org.apache.spark.sql.connector.expressions.{FieldReference, GeneralSQLExpression, LiteralValue}
+import org.apache.spark.sql.execution.datasources.PushableColumnWithoutNestedColumn
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy
+
+/**
+ * The builder to build {@link GeneralSQLExpression}.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[GeneralSQLExpression] = e match {
Review comment:
let's return `String` here, as this `ExpressionSQLBuilder` doesn't need to be dedicated to 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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802392529
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, EqualTo, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to generate SQL string from catalyst expressions.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"(${l.get}) ${b.sqlOperator} (${r.get})")
+ } else {
+ None
+ }
+ case Not(EqualTo(left, right)) =>
+ val l = generateSQL(left)
+ val r = generateSQL(right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"${l.get} != ${r.get}")
+ } else {
+ None
+ }
+ case Not(child) => generateSQL(child).map(v => s"NOT $v")
Review comment:
```suggestion
case Not(child) => generateSQL(child).map(v => s"NOT ($v)")
```
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r801710904
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralSQLExpression.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * The general SQL string corresponding to expression.
+ * <p>
+ * The currently supported SQL expressions:
Review comment:
I think we can omit the internal details in the classdoc, as the users don't care. They just take the SQL string anyway.
This is different from `GeneralAggregateFunction`, which user needs to know all the supported functions and match them.
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r803213746
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -806,17 +806,124 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(query, Seq(Row(29000.0)))
}
- test("scan with aggregate push-down: SUM(CASE WHEN) with group by") {
- val df =
- sql("SELECT SUM(CASE WHEN SALARY > 0 THEN 1 ELSE 0 END) FROM h2.test.employee GROUP BY DEPT")
- checkAggregateRemoved(df, false)
+ test("scan with aggregate push-down: aggregate function with CASE WHEN") {
+ val df = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | COUNT(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | AVG(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MAX(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN SALARY is null THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df)
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [], "
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NULL THEN 0.00 ELSE SALARY END), " +
+ "COUNT(CASE WHEN SALARY IS NULL THEN 0.0..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
- checkAnswer(df, Seq(Row(1), Row(2), Row(2)))
+ checkAnswer(df, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df2 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | AVG(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MIN(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df2)
+ df2.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NOT NULL THEN SALARY ELSE 0.00 END), " +
+ "COUNT(CASE WHEN SALARY IS NOT NULL ..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ }
+ checkAnswer(df2, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df3 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY > 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY >= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY < 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY <= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY = 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY > 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY >= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY < 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY <= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY = 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY != 0 THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df3)
+ df3.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN (SALARY) > (0.00) THEN 0.00 ELSE SALARY END), " +
+ "SUM(CASE WHEN (SALARY) >= (0.00) THE..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df3, expected_plan_fragment)
+ }
+ checkAnswer(df3, Seq(Row(0d, 0d, 12000d, 12000d, 12000d, 12000d, 12000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 19000d, 19000d, 19000d, 19000d, 19000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 22000d, 22000d, 22000d, 22000d, 22000d, 0d, 0d, 0d, 0d)))
+
+ val df4 = sql(
+ """
+ |SELECT
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY <= 13000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 11000 OR SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR SALARY < 9000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR NOT(SALARY >= 9000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND SALARY >= 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) OR SALARY > 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY != 0) OR NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000 AND SALARY > 8000) THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN NOT(SALARY > 8000 OR SALARY IS NULL) THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df4)
+ df4.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [COUNT(CASE WHEN ((SALARY) > (8000.00)) AND ((SALARY) < (10000.00))" +
+ " THEN SALARY ELSE 0.00 END), C..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df4, expected_plan_fragment)
+ }
+ checkAnswer(df4, Seq(Row(1, 1, 1, 1, 1, 0d, 12000d, 0d, 12000d, 12000d, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 10000d, 0d, 10000d, 10000d, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 12000d, 0d, 12000d, 12000d, 0d)))
+ }
+
+ test("scan with aggregate push-down: aggregate function with CAST") {
Review comment:
The test case used to test cannot pushdown aggregate function with `CAST`.
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r803217925
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -806,17 +806,124 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(query, Seq(Row(29000.0)))
}
- test("scan with aggregate push-down: SUM(CASE WHEN) with group by") {
- val df =
- sql("SELECT SUM(CASE WHEN SALARY > 0 THEN 1 ELSE 0 END) FROM h2.test.employee GROUP BY DEPT")
- checkAggregateRemoved(df, false)
+ test("scan with aggregate push-down: aggregate function with CASE WHEN") {
+ val df = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | COUNT(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | AVG(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MAX(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN SALARY is null THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df)
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [], "
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NULL THEN 0.00 ELSE SALARY END), " +
+ "COUNT(CASE WHEN SALARY IS NULL THEN 0.0..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
- checkAnswer(df, Seq(Row(1), Row(2), Row(2)))
+ checkAnswer(df, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df2 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | AVG(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MIN(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df2)
+ df2.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NOT NULL THEN SALARY ELSE 0.00 END), " +
+ "COUNT(CASE WHEN SALARY IS NOT NULL ..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ }
+ checkAnswer(df2, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df3 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY > 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY >= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY < 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY <= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY = 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY > 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY >= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY < 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY <= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY = 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY != 0 THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df3)
+ df3.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN (SALARY) > (0.00) THEN 0.00 ELSE SALARY END), " +
+ "SUM(CASE WHEN (SALARY) >= (0.00) THE..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df3, expected_plan_fragment)
+ }
+ checkAnswer(df3, Seq(Row(0d, 0d, 12000d, 12000d, 12000d, 12000d, 12000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 19000d, 19000d, 19000d, 19000d, 19000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 22000d, 22000d, 22000d, 22000d, 22000d, 0d, 0d, 0d, 0d)))
+
+ val df4 = sql(
+ """
+ |SELECT
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END),
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on pull request #35248:
URL: https://github.com/apache/spark/pull/35248#issuecomment-1035673217
@cloud-fan Thank you for your review. @HyukjinKwon Thank you too.
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r795114429
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralSQLExpression.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * The general SQL string corresponding to expression.
+ * <p>
+ * The currently supported SQL expressions:
+ * <ol>
+ * <li><pre>Literal</pre> Since 3.3.0</li>
+ * <li><pre>Attribute</pre> Since 3.3.0</li>
+ * <li><pre>IsNull</pre> Since 3.3.0</li>
+ * <li><pre>IsNotNull</pre> Since 3.3.0</li>
+ * <li><pre>BinaryOperator</pre> Since 3.3.0</li>
+ * <li><pre>Not</pre> Since 3.3.0</li>
+ * <li><pre>CaseWhen</pre> Since 3.3.0</li>
+ * </ol>
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public class GeneralSQLExpression implements Expression, Serializable {
+ private String sql;
+
+ public GeneralSQLExpression(String sql) {
Review comment:
The V2 expression holds the V1 expression, which looks strange.
cc @cloud-fan
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on pull request #35248:
URL: https://github.com/apache/spark/pull/35248#issuecomment-1018071857
ping @huaxingao cc @cloud-fan
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r795114429
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralSQLExpression.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * The general SQL string corresponding to expression.
+ * <p>
+ * The currently supported SQL expressions:
+ * <ol>
+ * <li><pre>Literal</pre> Since 3.3.0</li>
+ * <li><pre>Attribute</pre> Since 3.3.0</li>
+ * <li><pre>IsNull</pre> Since 3.3.0</li>
+ * <li><pre>IsNotNull</pre> Since 3.3.0</li>
+ * <li><pre>BinaryOperator</pre> Since 3.3.0</li>
+ * <li><pre>Not</pre> Since 3.3.0</li>
+ * <li><pre>CaseWhen</pre> Since 3.3.0</li>
+ * </ol>
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public class GeneralSQLExpression implements Expression, Serializable {
+ private String sql;
+
+ public GeneralSQLExpression(String sql) {
Review comment:
The V2 expression holds the V1 expression, which looks strange
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802206354
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralSQLExpression.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * The general SQL string corresponding to expression.
+ * <p>
+ * The currently supported SQL expressions:
Review comment:
I got 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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802318146
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
##########
@@ -543,3 +544,10 @@ object OrcUtils extends Logging {
result
}
}
+
+object V2Column {
+ def unapply(e: org.apache.spark.sql.connector.expressions.Expression): Option[String] = e match {
Review comment:
Can you show the errors you hit? It's not clear to me what's wrong with this object.
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802686510
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, EqualTo, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to generate SQL string from catalyst expressions.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"(${l.get}) ${b.sqlOperator} (${r.get})")
+ } else {
+ None
+ }
+ case Not(EqualTo(left, right)) =>
+ val l = generateSQL(left)
+ val r = generateSQL(right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"${l.get} != ${r.get}")
+ } else {
+ None
+ }
+ case Not(child) => generateSQL(child).map(v => s"NOT ($v)")
+ case CaseWhen(branches, elseValue) =>
+ val conditionsSQL = branches.map(_._1).flatMap(generateSQL)
+ val valuesSQL = branches.map(_._2).flatMap(generateSQL)
+ if (conditionsSQL.length == branches.length && valuesSQL.length == branches.length) {
+ val branchSQL =
+ conditionsSQL.zip(valuesSQL).map { case (c, v) => s" WHEN $c THEN $v" }.mkString
+ val elseSQL = elseValue.flatMap(generateSQL).map(v => s" ELSE $v").getOrElse("")
Review comment:
We need to return None if we can't generate SQL for the else value.
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802689573
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
##########
@@ -52,10 +52,11 @@ object AggregatePushDownUtils {
def processMinOrMax(agg: AggregateFunc): Boolean = {
val (column, aggType) = agg match {
- case max: Max => (max.column, "max")
- case min: Min => (min.column, "min")
- case _ =>
- throw new IllegalArgumentException(s"Unexpected type of AggregateFunc ${agg.describe}")
+ case max: Max if max.column().isInstanceOf[NamedReference] =>
Review comment:
shall we use `V2ColumnUtils` 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] cloud-fan commented on a change in pull request #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802315103
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to generate SQL string from catalyst expressions.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"(${l.get}) ${b.sqlOperator} (${r.get})")
+ } else {
+ None
+ }
+ case Not(child) => generateSQL(child).map(v => s"NOT $v")
Review comment:
We can keep matching `Not(EqualTo)` which can generate simpler SQL
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802327687
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to generate SQL string from catalyst expressions.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"(${l.get}) ${b.sqlOperator} (${r.get})")
+ } else {
+ None
+ }
+ case Not(child) => generateSQL(child).map(v => s"NOT $v")
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802399880
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
##########
@@ -487,18 +488,18 @@ object OrcUtils extends Logging {
val aggORCValues: Seq[WritableComparable[_]] =
aggregation.aggregateExpressions.zipWithIndex.map {
- case (max: Max, index) =>
- val columnName = max.column.fieldNames.head
+ case (max: Max, index) if max.column.isInstanceOf[NamedReference] =>
Review comment:
OK at least we can make a util function
```
object V2ColumnUtils {
def extractV2Column(expr: V2Expression): Option[String] = ...
}
```
and here we can write
```
case (max: Max, index) if V2ColumnUtils.extractV2Column(max.column).isDefined =>
val columnName = V2ColumnUtils.extractV2Column(max.column).get
...
```
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #35248:
URL: https://github.com/apache/spark/pull/35248#issuecomment-1032681495
@beliefer Please enrich the PR description to explain the motivation of this approach. It's a new framework to represent catalyst expressions in DS v2 APIs, and CASE WHEN is just the first use case.
To @HyukjinKwon , I proposed this idea because it's too much work to duplicate all the catalyst expressions in DS v2 with Java classes. In fact, the current basic DS v2 predicate/aggregate expressions should be sufficient for simple data sources that are file-based. It turns out that the sources which need advanced pushdown support are almost all SQL systems like Oracle, Clickhouse, etc., and it's actually better for them to get a SQL string directly, instead of building SQL string manually from DS v2 expressions.
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r803257890
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, EqualTo, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to generate SQL string from catalyst expressions.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"(${l.get}) ${b.sqlOperator} (${r.get})")
+ } else {
+ None
+ }
+ case Not(EqualTo(left, right)) =>
+ val l = generateSQL(left)
+ val r = generateSQL(right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"${l.get} != ${r.get}")
+ } else {
+ None
+ }
+ case Not(child) => generateSQL(child).map(v => s"NOT ($v)")
+ case CaseWhen(branches, elseValue) =>
+ val conditionsSQL = branches.map(_._1).flatMap(generateSQL)
+ val valuesSQL = branches.map(_._2).flatMap(generateSQL)
+ if (conditionsSQL.length == branches.length && valuesSQL.length == branches.length) {
+ val branchSQL =
+ conditionsSQL.zip(valuesSQL).map { case (c, v) => s" WHEN $c THEN $v" }.mkString
+ val elseSQL = elseValue.flatMap(generateSQL).map(v => s" ELSE $v").getOrElse("")
Review comment:
I got 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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802694039
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -806,17 +806,124 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(query, Seq(Row(29000.0)))
}
- test("scan with aggregate push-down: SUM(CASE WHEN) with group by") {
- val df =
- sql("SELECT SUM(CASE WHEN SALARY > 0 THEN 1 ELSE 0 END) FROM h2.test.employee GROUP BY DEPT")
- checkAggregateRemoved(df, false)
+ test("scan with aggregate push-down: aggregate function with CASE WHEN") {
+ val df = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | COUNT(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | AVG(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MAX(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN SALARY is null THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df)
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [], "
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NULL THEN 0.00 ELSE SALARY END), " +
+ "COUNT(CASE WHEN SALARY IS NULL THEN 0.0..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
- checkAnswer(df, Seq(Row(1), Row(2), Row(2)))
+ checkAnswer(df, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df2 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | AVG(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MIN(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df2)
+ df2.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NOT NULL THEN SALARY ELSE 0.00 END), " +
+ "COUNT(CASE WHEN SALARY IS NOT NULL ..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ }
+ checkAnswer(df2, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df3 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY > 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY >= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY < 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY <= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY = 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY > 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY >= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY < 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY <= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY = 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY != 0 THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df3)
+ df3.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN (SALARY) > (0.00) THEN 0.00 ELSE SALARY END), " +
+ "SUM(CASE WHEN (SALARY) >= (0.00) THE..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df3, expected_plan_fragment)
+ }
+ checkAnswer(df3, Seq(Row(0d, 0d, 12000d, 12000d, 12000d, 12000d, 12000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 19000d, 19000d, 19000d, 19000d, 19000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 22000d, 22000d, 22000d, 22000d, 22000d, 0d, 0d, 0d, 0d)))
+
+ val df4 = sql(
+ """
+ |SELECT
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY <= 13000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 11000 OR SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR SALARY < 9000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR NOT(SALARY >= 9000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND SALARY >= 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) OR SALARY > 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY != 0) OR NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000 AND SALARY > 8000) THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN NOT(SALARY > 8000 OR SALARY IS NULL) THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df4)
+ df4.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [COUNT(CASE WHEN ((SALARY) > (8000.00)) AND ((SALARY) < (10000.00))" +
+ " THEN SALARY ELSE 0.00 END), C..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df4, expected_plan_fragment)
+ }
+ checkAnswer(df4, Seq(Row(1, 1, 1, 1, 1, 0d, 12000d, 0d, 12000d, 12000d, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 10000d, 0d, 10000d, 10000d, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 12000d, 0d, 12000d, 12000d, 0d)))
+ }
+
+ test("scan with aggregate push-down: aggregate function with CAST") {
Review comment:
I think we will support CAST soon. Do we really need this test?
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r801720807
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
##########
@@ -487,18 +488,18 @@ object OrcUtils extends Logging {
val aggORCValues: Seq[WritableComparable[_]] =
aggregation.aggregateExpressions.zipWithIndex.map {
- case (max: Max, index) =>
- val columnName = max.column.fieldNames.head
+ case (max: Max, index) if max.column.isInstanceOf[NamedReference] =>
Review comment:
this is not caused by this PR, but can we also check the length of field names to be future proof? We can even create a util object to save duplicated code:
```
object V2Column {
def unapply(e: org.apache.spark.sql.connector.expressions.Expression): Option[String] = e match {
case r: NamedReference if r. fieldNames.length == 1 => Some(r.fieldNames.head)
case _ => None
}
}
```
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r789495389
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/SQLExpression.java
##########
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+import org.apache.spark.annotation.Evolving;
+
+import java.io.Serializable;
+
+/**
+ * Base class of the SQL Expressions.
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public interface SQLExpression extends Expression, Serializable {
Review comment:
why do we need it? I think a single `GeneralSQLExpression` should be sufficient.
--
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] HyukjinKwon commented on a change in pull request #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r795110916
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralSQLExpression.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * The general SQL string corresponding to expression.
+ * <p>
+ * The currently supported SQL expressions:
+ * <ol>
+ * <li><pre>Literal</pre> Since 3.3.0</li>
+ * <li><pre>Attribute</pre> Since 3.3.0</li>
+ * <li><pre>IsNull</pre> Since 3.3.0</li>
+ * <li><pre>IsNotNull</pre> Since 3.3.0</li>
+ * <li><pre>BinaryOperator</pre> Since 3.3.0</li>
+ * <li><pre>Not</pre> Since 3.3.0</li>
+ * <li><pre>CaseWhen</pre> Since 3.3.0</li>
+ * </ol>
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public class GeneralSQLExpression implements Expression, Serializable {
+ private String sql;
+
+ public GeneralSQLExpression(String sql) {
Review comment:
Hm, shouldn't we keep the expression tree here, and let the JDBC source side to translate into SQL string? I think we can't assume that all downstream source implementation can understand SQL expression in string
--
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] HyukjinKwon commented on a change in pull request #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r795110993
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/GeneralSQLExpression.java
##########
@@ -0,0 +1,52 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * The general SQL string corresponding to expression.
+ * <p>
+ * The currently supported SQL expressions:
+ * <ol>
+ * <li><pre>Literal</pre> Since 3.3.0</li>
+ * <li><pre>Attribute</pre> Since 3.3.0</li>
+ * <li><pre>IsNull</pre> Since 3.3.0</li>
+ * <li><pre>IsNotNull</pre> Since 3.3.0</li>
+ * <li><pre>BinaryOperator</pre> Since 3.3.0</li>
+ * <li><pre>Not</pre> Since 3.3.0</li>
+ * <li><pre>CaseWhen</pre> Since 3.3.0</li>
+ * </ol>
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public class GeneralSQLExpression implements Expression, Serializable {
Review comment:
I would also come up with another name instead of `GeneralSQLExpression` (e.g., `PushedExpression`)
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r794390939
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{CaseWhen, Expression, Literal}
+import org.apache.spark.sql.connector.expressions.{FieldReference, GeneralSQLExpression, LiteralValue}
+import org.apache.spark.sql.execution.datasources.PushableColumnWithoutNestedColumn
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy
+
+/**
+ * The builder to build {@link GeneralSQLExpression}.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[GeneralSQLExpression] = e match {
+ case CaseWhen(branches, elseValue) =>
+ val newBranches = branches.collect {
+ case (predicate: Expression, Literal(value, dataType)) =>
Review comment:
We should implement this SQL string builder in a more flexible way. Something like
```
def build = generateSQL(e)
private def generateSQL(expr: Expression): Option[String] = expr match {
case l: Literal => Some(...)
case a: Attribute => Some(quoteIfNeeded(a.name))
// other basic expressions, like predicates?
case CaseWhen(branches, elseValue) =>
val conditionsSQL = branches.map(_._1).flatMap(generateSQL)
val valuesSQL = branches.map(_._2).flatMap(generateSQL)
...
case _ => None
}
```
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r789554783
##########
File path: sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/SQLExpression.java
##########
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+import org.apache.spark.annotation.Evolving;
+
+import java.io.Serializable;
+
+/**
+ * Base class of the SQL Expressions.
+ *
+ * @since 3.3.0
+ */
+@Evolving
+public interface SQLExpression extends Expression, Serializable {
Review comment:
I'm worried about exceptions.
It seems no need for the time being.
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r794390939
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{CaseWhen, Expression, Literal}
+import org.apache.spark.sql.connector.expressions.{FieldReference, GeneralSQLExpression, LiteralValue}
+import org.apache.spark.sql.execution.datasources.PushableColumnWithoutNestedColumn
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy
+
+/**
+ * The builder to build {@link GeneralSQLExpression}.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[GeneralSQLExpression] = e match {
+ case CaseWhen(branches, elseValue) =>
+ val newBranches = branches.collect {
+ case (predicate: Expression, Literal(value, dataType)) =>
Review comment:
We should implement this SQL string builder in a more flexible way. Something like
```
def build = generateSQL(e)
private def generateSQL(expr: Expression): Option[String] = expr match {
case l: Literal => Some(...)
case PushableColumnWithoutNestedColumn(name) => Some(name)
// other basic expressions, like predicates?
case CaseWhen(branches, elseValue) =>
val conditionsSQL = branches.map(_._1).flatMap(generateSQL)
val valuesSQL = branches.map(_._2).flatMap(generateSQL)
...
}
```
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r794394138
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
##########
@@ -52,8 +52,10 @@ object AggregatePushDownUtils {
def processMinOrMax(agg: AggregateFunc): Boolean = {
val (column, aggType) = agg match {
- case max: Max => (max.column, "max")
- case min: Min => (min.column, "min")
+ case max: Max if max.column().isInstanceOf[NamedReference] =>
+ (max.column.asInstanceOf[NamedReference], "max")
+ case min: Min if min.column().isInstanceOf[NamedReference] =>
+ (min.column.asInstanceOf[NamedReference], "min")
case _ =>
throw new IllegalArgumentException(s"Unexpected type of AggregateFunc ${agg.describe}")
Review comment:
does the caller side guarantee to pass an `AggregateFunc` with only `Min`/`Max` and `NamedReference`?
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r794444947
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/AggregatePushDownUtils.scala
##########
@@ -52,8 +52,10 @@ object AggregatePushDownUtils {
def processMinOrMax(agg: AggregateFunc): Boolean = {
val (column, aggType) = agg match {
- case max: Max => (max.column, "max")
- case min: Min => (min.column, "min")
+ case max: Max if max.column().isInstanceOf[NamedReference] =>
+ (max.column.asInstanceOf[NamedReference], "max")
+ case min: Min if min.column().isInstanceOf[NamedReference] =>
+ (min.column.asInstanceOf[NamedReference], "min")
case _ =>
throw new IllegalArgumentException(s"Unexpected type of AggregateFunc ${agg.describe}")
Review comment:
According to the code, it seems Orc and parquet only support the push down of max, min and 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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r801717547
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
##########
@@ -707,20 +708,30 @@ object DataSourceStrategy
agg.aggregateFunction match {
case aggregate.Min(PushableColumnWithoutNestedColumn(name)) =>
Some(new Min(FieldReference.column(name)))
+ case aggregate.Min(PushableExpression(expr)) =>
Review comment:
To simplify the code, can we make `PushableExpression` handle field reference as well? e.g. it returns `FieldReference` if possible, or return `GeneralSQLExpression`. Then here we can just use whatever v2 expression returned by `PushableExpression`.
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r801707729
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, EqualTo, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to build expression and generate sql string.
Review comment:
```suggestion
* The builder to generate SQL string from catalyst expressions.
```
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r803208261
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/ExpressionSQLBuilder.scala
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryOperator, CaseWhen, EqualTo, Expression, IsNotNull, IsNull, Literal, Not}
+import org.apache.spark.sql.connector.expressions.LiteralValue
+
+/**
+ * The builder to generate SQL string from catalyst expressions.
+ */
+class ExpressionSQLBuilder(e: Expression) {
+
+ def build(): Option[String] = generateSQL(e)
+
+ private def generateSQL(expr: Expression): Option[String] = expr match {
+ case Literal(value, dataType) => Some(LiteralValue(value, dataType).toString)
+ case a: Attribute => Some(quoteIfNeeded(a.name))
+ case IsNull(col) => generateSQL(col).map(c => s"$c IS NULL")
+ case IsNotNull(col) => generateSQL(col).map(c => s"$c IS NOT NULL")
+ case b: BinaryOperator =>
+ val l = generateSQL(b.left)
+ val r = generateSQL(b.right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"(${l.get}) ${b.sqlOperator} (${r.get})")
+ } else {
+ None
+ }
+ case Not(EqualTo(left, right)) =>
+ val l = generateSQL(left)
+ val r = generateSQL(right)
+ if (l.isDefined && r.isDefined) {
+ Some(s"${l.get} != ${r.get}")
+ } else {
+ None
+ }
+ case Not(child) => generateSQL(child).map(v => s"NOT ($v)")
+ case CaseWhen(branches, elseValue) =>
+ val conditionsSQL = branches.map(_._1).flatMap(generateSQL)
+ val valuesSQL = branches.map(_._2).flatMap(generateSQL)
+ if (conditionsSQL.length == branches.length && valuesSQL.length == branches.length) {
+ val branchSQL =
+ conditionsSQL.zip(valuesSQL).map { case (c, v) => s" WHEN $c THEN $v" }.mkString
+ val elseSQL = elseValue.flatMap(generateSQL).map(v => s" ELSE $v").getOrElse("")
Review comment:
`CASE whenClause+ (ELSE elseExpression=expression)? END`
--
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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802222257
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala
##########
@@ -707,20 +708,30 @@ object DataSourceStrategy
agg.aggregateFunction match {
case aggregate.Min(PushableColumnWithoutNestedColumn(name)) =>
Some(new Min(FieldReference.column(name)))
+ case aggregate.Min(PushableExpression(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 #35248: [SPARK-37960][SQL] Support aggregate push down with `CASE ... WHEN ... ELSE ... END`
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r802254240
##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
##########
@@ -543,3 +544,10 @@ object OrcUtils extends Logging {
result
}
}
+
+object V2Column {
+ def unapply(e: org.apache.spark.sql.connector.expressions.Expression): Option[String] = e match {
Review comment:
It seems cannot use this method.
--
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 #35248: [SPARK-37960][SQL] A new framework to represent catalyst expressions in DS v2 APIs
Posted by GitBox <gi...@apache.org>.
beliefer commented on a change in pull request #35248:
URL: https://github.com/apache/spark/pull/35248#discussion_r803213746
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
##########
@@ -806,17 +806,124 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
checkAnswer(query, Seq(Row(29000.0)))
}
- test("scan with aggregate push-down: SUM(CASE WHEN) with group by") {
- val df =
- sql("SELECT SUM(CASE WHEN SALARY > 0 THEN 1 ELSE 0 END) FROM h2.test.employee GROUP BY DEPT")
- checkAggregateRemoved(df, false)
+ test("scan with aggregate push-down: aggregate function with CASE WHEN") {
+ val df = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | COUNT(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | AVG(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MAX(CASE WHEN SALARY is null THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN SALARY is null THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df)
df.queryExecution.optimizedPlan.collect {
case _: DataSourceV2ScanRelation =>
val expected_plan_fragment =
- "PushedFilters: [], "
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NULL THEN 0.00 ELSE SALARY END), " +
+ "COUNT(CASE WHEN SALARY IS NULL THEN 0.0..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
checkKeywordsExistsInExplain(df, expected_plan_fragment)
}
- checkAnswer(df, Seq(Row(1), Row(2), Row(2)))
+ checkAnswer(df, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df2 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | AVG(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END),
+ | MIN(CASE WHEN SALARY is not null THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df2)
+ df2.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN SALARY IS NOT NULL THEN SALARY ELSE 0.00 END), " +
+ "COUNT(CASE WHEN SALARY IS NOT NULL ..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ }
+ checkAnswer(df2, Seq(Row(12000d, 1, 12000d, 12000d, 12000d),
+ Row(19000d, 2, 9500d, 10000d, 9000d), Row(22000d, 2, 11000d, 12000d, 10000d)))
+
+ val df3 = sql(
+ """
+ |SELECT
+ | SUM(CASE WHEN SALARY > 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY >= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY < 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY <= 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY = 0 THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY > 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY >= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY < 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY <= 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN NOT(SALARY = 0) THEN 0 ELSE SALARY END),
+ | SUM(CASE WHEN SALARY != 0 THEN 0 ELSE SALARY END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df3)
+ df3.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [SUM(CASE WHEN (SALARY) > (0.00) THEN 0.00 ELSE SALARY END), " +
+ "SUM(CASE WHEN (SALARY) >= (0.00) THE..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df3, expected_plan_fragment)
+ }
+ checkAnswer(df3, Seq(Row(0d, 0d, 12000d, 12000d, 12000d, 12000d, 12000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 19000d, 19000d, 19000d, 19000d, 19000d, 0d, 0d, 0d, 0d),
+ Row(0d, 0d, 22000d, 22000d, 22000d, 22000d, 22000d, 0d, 0d, 0d, 0d)))
+
+ val df4 = sql(
+ """
+ |SELECT
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 8000 AND SALARY <= 13000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY > 11000 OR SALARY < 10000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR SALARY < 9000 THEN SALARY ELSE 0 END),
+ | COUNT(CASE WHEN SALARY >= 12000 OR NOT(SALARY >= 9000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND SALARY >= 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) OR SALARY > 8000 THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000) AND NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY != 0) OR NOT(SALARY < 8000) THEN SALARY ELSE 0 END),
+ | MAX(CASE WHEN NOT(SALARY > 8000 AND SALARY > 8000) THEN 0 ELSE SALARY END),
+ | MIN(CASE WHEN NOT(SALARY > 8000 OR SALARY IS NULL) THEN SALARY ELSE 0 END)
+ |FROM h2.test.employee GROUP BY DEPT
+ """.stripMargin)
+ checkAggregateRemoved(df4)
+ df4.queryExecution.optimizedPlan.collect {
+ case _: DataSourceV2ScanRelation =>
+ val expected_plan_fragment =
+ "PushedAggregates: [COUNT(CASE WHEN ((SALARY) > (8000.00)) AND ((SALARY) < (10000.00))" +
+ " THEN SALARY ELSE 0.00 END), C..., " +
+ "PushedFilters: [], " +
+ "PushedGroupByColumns: [DEPT]"
+ checkKeywordsExistsInExplain(df4, expected_plan_fragment)
+ }
+ checkAnswer(df4, Seq(Row(1, 1, 1, 1, 1, 0d, 12000d, 0d, 12000d, 12000d, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 10000d, 0d, 10000d, 10000d, 0d),
+ Row(2, 2, 2, 2, 2, 0d, 12000d, 0d, 12000d, 12000d, 0d)))
+ }
+
+ test("scan with aggregate push-down: aggregate function with CAST") {
Review comment:
The test case used to test cannot pushdown aggregate function with `CAST`.
I use `NVL` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org