You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2022/05/18 12:29:20 UTC
[spark] branch branch-3.3 updated: [SPARK-39162][SQL][3.3] Jdbc dialect should decide which function could be pushed down
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 b5ce32f41f9 [SPARK-39162][SQL][3.3] Jdbc dialect should decide which function could be pushed down
b5ce32f41f9 is described below
commit b5ce32f41f9e4aecb02cc383184ac0a6dfabc4dd
Author: Jiaan Geng <be...@163.com>
AuthorDate: Wed May 18 20:29:01 2022 +0800
[SPARK-39162][SQL][3.3] Jdbc dialect should decide which function could be pushed down
### What changes were proposed in this pull request?
This PR used to back port https://github.com/apache/spark/pull/36521 to 3.3
### Why are the changes needed?
Let function push-down more flexible.
### Does this PR introduce _any_ user-facing change?
'No'.
New feature.
### How was this patch tested?
Exists tests.
Closes #36556 from beliefer/SPARK-39162_3.3.
Authored-by: Jiaan Geng <be...@163.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/errors/QueryCompilationErrors.scala | 4 ----
.../org/apache/spark/sql/jdbc/H2Dialect.scala | 28 ++++------------------
.../org/apache/spark/sql/jdbc/JdbcDialects.scala | 19 +++++++++++++++
3 files changed, 23 insertions(+), 28 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 8925d4f5317..2d2dba63e3a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -2411,8 +2411,4 @@ object QueryCompilationErrors extends QueryErrorsBase {
new AnalysisException(
"Sinks cannot request distribution and ordering in continuous execution mode")
}
-
- def noSuchFunctionError(database: String, funcInfo: String): Throwable = {
- new AnalysisException(s"$database does not support function: $funcInfo")
- }
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
index 56cadbe8e2c..4a88203ec59 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
@@ -20,13 +20,9 @@ package org.apache.spark.sql.jdbc
import java.sql.{SQLException, Types}
import java.util.Locale
-import scala.util.control.NonFatal
-
import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
-import org.apache.spark.sql.connector.expressions.Expression
import org.apache.spark.sql.connector.expressions.aggregate.{AggregateFunc, GeneralAggregateFunc}
-import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils
import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DecimalType, ShortType, StringType}
@@ -34,27 +30,11 @@ private object H2Dialect extends JdbcDialect {
override def canHandle(url: String): Boolean =
url.toLowerCase(Locale.ROOT).startsWith("jdbc:h2")
- class H2SQLBuilder extends JDBCSQLBuilder {
- override def visitSQLFunction(funcName: String, inputs: Array[String]): String = {
- funcName match {
- case "WIDTH_BUCKET" =>
- val functionInfo = super.visitSQLFunction(funcName, inputs)
- throw QueryCompilationErrors.noSuchFunctionError("H2", functionInfo)
- case _ => super.visitSQLFunction(funcName, inputs)
- }
- }
- }
+ private val supportedFunctions =
+ Set("ABS", "COALESCE", "LN", "EXP", "POWER", "SQRT", "FLOOR", "CEIL")
- override def compileExpression(expr: Expression): Option[String] = {
- val h2SQLBuilder = new H2SQLBuilder()
- try {
- Some(h2SQLBuilder.build(expr))
- } catch {
- case NonFatal(e) =>
- logWarning("Error occurs while compiling V2 expression", e)
- None
- }
- }
+ override def isSupportedFunction(funcName: String): Boolean =
+ supportedFunctions.contains(funcName)
override def compileAggregate(aggFunction: AggregateFunc): Option[String] = {
super.compileAggregate(aggFunction).orElse(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
index 0ef23073a27..e1883e4e7f4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
@@ -240,8 +240,27 @@ abstract class JdbcDialect extends Serializable with Logging{
getJDBCType(dataType).map(_.databaseTypeDefinition).getOrElse(dataType.typeName)
s"CAST($l AS $databaseTypeDefinition)"
}
+
+ override def visitSQLFunction(funcName: String, inputs: Array[String]): String = {
+ if (isSupportedFunction(funcName)) {
+ s"""$funcName(${inputs.mkString(", ")})"""
+ } else {
+ // The framework will catch the error and give up the push-down.
+ // Please see `JdbcDialect.compileExpression(expr: Expression)` for more details.
+ throw new UnsupportedOperationException(
+ s"${this.getClass.getSimpleName} does not support function: $funcName")
+ }
+ }
}
+ /**
+ * Returns whether the database supports function.
+ * @param funcName Upper-cased function name
+ * @return True if the database supports function.
+ */
+ @Since("3.3.0")
+ def isSupportedFunction(funcName: String): Boolean = false
+
/**
* Converts V2 expression to String representing a SQL expression.
* @param expr The V2 expression to be converted.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org