This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.3 by this push:
new b3cd07b236f [SPARK-38761][SQL] DS V2 supports push down misc
non-aggregate functions
b3cd07b236f is described below
commit b3cd07b236f46e8c402b06820d6f3a25fe608593
Author: Jiaan Geng <belie...@163.com>
AuthorDate: Mon Apr 11 13:50:57 2022 +0800
[SPARK-38761][SQL] DS V2 supports push down misc non-aggregate functions
### What changes were proposed in this pull request?
Currently, Spark have some misc non-aggregate functions of ANSI standard.
Please refer
https://github.com/apache/spark/blob/2f8613f22c0750c00cf1dcfb2f31c431d8dc1be7/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala#L362.
These functions show below:
`abs`,
`coalesce`,
`nullif`,
`CASE WHEN`
DS V2 should supports push down these misc non-aggregate functions.
Because DS V2 already support push down `CASE WHEN`, so this PR no need do
the job again.
Because `nullif` extends `RuntimeReplaceable`, so this PR no need do the
job too.
### Why are the changes needed?
DS V2 supports push down misc non-aggregate functions
### Does this PR introduce _any_ user-facing change?
'No'.
New feature.
### How was this patch tested?
New tests.
Closes #36039 from beliefer/SPARK-38761.
Authored-by: Jiaan Geng <belie...@163.com>
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
(cherry picked from commit 9ce4ba02d3f67116a4a9786af453d869596fb3ec)
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
---
.../sql/connector/util/V2ExpressionSQLBuilder.java | 8 ++++
.../sql/catalyst/util/V2ExpressionBuilder.scala | 11 ++++-
.../org/apache/spark/sql/jdbc/JDBCV2Suite.scala | 50 +++++++++++-----------
3 files changed, 44 insertions(+), 25 deletions(-)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java
index c8d924db75a..a7d1ed7f85e 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/util/V2ExpressionSQLBuilder.java
@@ -93,6 +93,10 @@ public class V2ExpressionSQLBuilder {
return visitNot(build(e.children()[0]));
case "~":
return visitUnaryArithmetic(name, inputToSQL(e.children()[0]));
+ case "ABS":
+ case "COALESCE":
+ return visitSQLFunction(name,
+ Arrays.stream(e.children()).map(c ->
build(c)).toArray(String[]::new));
case "CASE_WHEN": {
List<String> children =
Arrays.stream(e.children()).map(c ->
build(c)).collect(Collectors.toList());
@@ -210,6 +214,10 @@ public class V2ExpressionSQLBuilder {
return sb.toString();
}
+ protected String visitSQLFunction(String funcName, String[] inputs) {
+ return funcName + "(" +
Arrays.stream(inputs).collect(Collectors.joining(", ")) + ")";
+ }
+
protected String visitUnexpectedExpr(Expression expr) throws
IllegalArgumentException {
throw new IllegalArgumentException("Unexpected V2 expression: " + expr);
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
index 5fd01ac5636..37db499470a 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.catalyst.util
-import org.apache.spark.sql.catalyst.expressions.{Add, And, BinaryComparison,
BinaryOperator, BitwiseAnd, BitwiseNot, BitwiseOr, BitwiseXor, CaseWhen, Cast,
Contains, Divide, EndsWith, EqualTo, Expression, In, InSet, IsNotNull, IsNull,
Literal, Multiply, Not, Or, Predicate, Remainder, StartsWith, StringPredicate,
Subtract, UnaryMinus}
+import org.apache.spark.sql.catalyst.expressions.{Abs, Add, And,
BinaryComparison, BinaryOperator, BitwiseAnd, BitwiseNot, BitwiseOr,
BitwiseXor, CaseWhen, Cast, Coalesce, Contains, Divide, EndsWith, EqualTo,
Expression, In, InSet, IsNotNull, IsNull, Literal, Multiply, Not, Or,
Predicate, Remainder, StartsWith, StringPredicate, Subtract, UnaryMinus}
import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression
=> V2Expression, FieldReference, GeneralScalarExpression, LiteralValue}
import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse,
AlwaysTrue, And => V2And, Not => V2Not, Or => V2Or, Predicate => V2Predicate}
import org.apache.spark.sql.execution.datasources.PushableColumn
@@ -95,6 +95,15 @@ class V2ExpressionBuilder(
}
case Cast(child, dataType, _, true) =>
generateExpression(child).map(v => new V2Cast(v, dataType))
+ case Abs(child, true) => generateExpression(child)
+ .map(v => new GeneralScalarExpression("ABS", Array[V2Expression](v)))
+ case Coalesce(children) =>
+ val childrenExpressions = children.flatMap(generateExpression(_))
+ if (children.length == childrenExpressions.length) {
+ Some(new GeneralScalarExpression("COALESCE",
childrenExpressions.toArray[V2Expression]))
+ } else {
+ None
+ }
case and: And =>
// AND expects predicate
val l = generateExpression(and.left, true)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
index 23138b8899b..858781f2cde 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala
@@ -26,7 +26,7 @@ import
org.apache.spark.sql.catalyst.analysis.CannotReplaceMissingTableException
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Sort}
import
org.apache.spark.sql.execution.datasources.v2.{DataSourceV2ScanRelation,
V1ScanWrapper}
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
-import org.apache.spark.sql.functions.{avg, count, count_distinct, lit, not,
sum, udf, when}
+import org.apache.spark.sql.functions.{abs, avg, coalesce, count,
count_distinct, lit, not, sum, udf, when}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.util.Utils
@@ -381,19 +381,13 @@ class JDBCV2Suite extends QueryTest with
SharedSparkSession with ExplainSuiteHel
checkAnswer(df, Seq(Row("fred", 1), Row("mary", 2)))
val df2 = spark.table("h2.test.people").filter($"id" + Int.MaxValue >
1)
-
checkFiltersRemoved(df2, ansiMode)
-
- df2.queryExecution.optimizedPlan.collect {
- case _: DataSourceV2ScanRelation =>
- val expected_plan_fragment = if (ansiMode) {
- "PushedFilters: [ID IS NOT NULL, (ID + 2147483647) > 1], "
- } else {
- "PushedFilters: [ID IS NOT NULL], "
- }
- checkKeywordsExistsInExplain(df2, expected_plan_fragment)
+ val expectedPlanFragment2 = if (ansiMode) {
+ "PushedFilters: [ID IS NOT NULL, (ID + 2147483647) > 1], "
+ } else {
+ "PushedFilters: [ID IS NOT NULL], "
}
-
+ checkPushedInfo(df2, expectedPlanFragment2)
if (ansiMode) {
val e = intercept[SparkException] {
checkAnswer(df2, Seq.empty)
@@ -422,22 +416,30 @@ class JDBCV2Suite extends QueryTest with
SharedSparkSession with ExplainSuiteHel
val df4 = spark.table("h2.test.employee")
.filter(($"salary" > 1000d).and($"salary" < 12000d))
-
checkFiltersRemoved(df4, ansiMode)
-
- df4.queryExecution.optimizedPlan.collect {
- case _: DataSourceV2ScanRelation =>
- val expected_plan_fragment = if (ansiMode) {
- "PushedFilters: [SALARY IS NOT NULL, " +
- "CAST(SALARY AS double) > 1000.0, CAST(SALARY AS double) <
12000.0], "
- } else {
- "PushedFilters: [SALARY IS NOT NULL], "
- }
- checkKeywordsExistsInExplain(df4, expected_plan_fragment)
+ val expectedPlanFragment4 = if (ansiMode) {
+ "PushedFilters: [SALARY IS NOT NULL, " +
+ "CAST(SALARY AS double) > 1000.0, CAST(SALARY AS double) <
12000.0], "
+ } else {
+ "PushedFilters: [SALARY IS NOT NULL], "
}
-
+ checkPushedInfo(df4, expectedPlanFragment4)
checkAnswer(df4, Seq(Row(1, "amy", 10000, 1000, true),
Row(1, "cathy", 9000, 1200, false), Row(2, "david", 10000, 1300,
true)))
+
+ val df5 = spark.table("h2.test.employee")
+ .filter(abs($"dept" - 3) > 1)
+ .filter(coalesce($"salary", $"bonus") > 2000)
+ checkFiltersRemoved(df5, ansiMode)
+ val expectedPlanFragment5 = if (ansiMode) {
+ "PushedFilters: [DEPT IS NOT NULL, ABS(DEPT - 3) > 1, " +
+ "(COALESCE(CAST(SALARY AS double), BONUS)) > 2000.0]"
+ } else {
+ "PushedFilters: [DEPT IS NOT NULL]"
+ }
+ checkPushedInfo(df5, expectedPlanFragment5)
+ checkAnswer(df5, Seq(Row(1, "amy", 10000, 1000, true),
+ Row(1, "cathy", 9000, 1200, false), Row(6, "jen", 12000, 1200,
true)))
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
