This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new f9f1b679f0b0 [SPARK-58554][SQL] Block MySQL double cast pushdown
f9f1b679f0b0 is described below
commit f9f1b679f0b0f961e4a7f7f940f54dadf2739b6a
Author: Alek Jarmov <[email protected]>
AuthorDate: Tue Aug 4 19:20:56 2026 +0800
[SPARK-58554][SQL] Block MySQL double cast pushdown
MariaDB rejects CAST(... AS DOUBLE PRECISION). Stop pushing DoubleType
casts in MySQLDialect so Spark evaluates them locally instead.
### What changes were proposed in this pull request?
Block pushdown of cast to DOUBLE. Without this fix I got the error
```
[JDBC_EXTERNAL_ENGINE_SYNTAX_ERROR.DURING_QUERY_EXECUTION] JDBC external
engine syntax error. The error was caused by the query SELECT `name` FROM
`employee` WHERE (`salary` IS NOT NULL) AND (CAST(`salary` AS DOUBLE
PRECISION) > 10000.5) . (conn=584) You have an error in your SQL syntax;
check the manual that corresponds to your MariaDB server version for the right
syntax to use near 'PRECISION) > 10000.5)' at line 1. The error occurred during
query execution. SQLSTATE: 42000
```
### Why are the changes needed?
Fix a wrong query.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Opus 5
Closes #57756 from alekjarmov/alekjarmov/block-mysql-double-cast-pushdown.
Authored-by: Alek Jarmov <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 00ea1abbbf09c56b63ea5bea61fb3c0ec7a304dc)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala | 11 ++++++++++-
.../main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala | 10 ++++++++++
.../src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala | 9 ++++++++-
3 files changed, 28 insertions(+), 2 deletions(-)
diff --git
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
index c2714587e2d2..fc2cad1b7098 100644
---
a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
+++
b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.jdbc.v2
import java.sql.{Connection, SQLFeatureNotSupportedException}
import org.apache.spark.{SparkConf, SparkSQLFeatureNotSupportedException}
-import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog
import org.apache.spark.sql.jdbc.MySQLDatabaseOnDocker
import org.apache.spark.sql.types._
@@ -313,6 +313,15 @@ class MySQLIntegrationSuite extends
DockerJDBCIntegrationV2Suite with V2JDBCTest
assert(rows10(0).getString(0) === "amy")
assert(rows10(1).getString(0) === "alex")
}
+
+ test("do not push down casts to double") {
+ val df = sql(
+ s"SELECT name FROM $catalogName.employee " +
+ "WHERE CAST(salary AS DOUBLE) > 10000.5")
+
+ checkFilterPushed(df, pushed = false)
+ checkAnswer(df, Seq(Row("alex"), Row("jen")))
+ }
}
/**
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
index b301c0c0bd5b..113429f81dc5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
@@ -118,6 +118,16 @@ private case class MySQLDialect() extends JdbcDialect with
SQLConfHelper with No
} else {
super.visitAggregateFunction(funcName, isDistinct, inputs)
}
+
+ override def visitCast(expr: String, exprDataType: DataType, dataType:
DataType): String = {
+ dataType match {
+ case DoubleType =>
+ // The common JDBC mapping is DOUBLE PRECISION, which is not a
portable CAST target for
+ // MySQL-compatible databases. In particular, MariaDB rejects it
with a syntax error.
+ throw new UnsupportedOperationException("Cannot cast to double type")
+ case _ => super.visitCast(expr, exprDataType, dataType)
+ }
+ }
}
override def compileExpression(expr: Expression): Option[String] = {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index b654deb12ad4..0213d18922b7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.plans.logical.ShowCreateTable
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap,
CharVarcharUtils, DateTimeTestUtils}
import org.apache.spark.sql.connector.catalog.Identifier
-import org.apache.spark.sql.connector.expressions.{Expression => V2Expression,
FieldReference, GeneralScalarExpression, LiteralValue}
+import org.apache.spark.sql.connector.expressions.{Cast => V2Cast, Expression
=> V2Expression, FieldReference, GeneralScalarExpression, LiteralValue}
import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse,
AlwaysTrue, Predicate}
import org.apache.spark.sql.execution.{DataSourceScanExec, ExtendedMode,
ProjectExec}
import org.apache.spark.sql.execution.command.{ExplainCommand,
ShowCreateTableCommand}
@@ -1454,6 +1454,13 @@ class JDBCSuite extends SharedSparkSession {
assert(mySqlDialect.getJDBCType(FloatType).map(_.databaseTypeDefinition).get ==
"FLOAT")
}
+ test("MySQL blocks casts to double") {
+ val dialect = MySQLDialect()
+ val cast = new V2Cast(FieldReference("value"), IntegerType, DoubleType)
+
+ assert(dialect.compileExpression(cast).isEmpty)
+ }
+
test("PostgresDialect type mapping") {
val Postgres = JdbcDialects.get("jdbc:postgresql://127.0.0.1/db")
val md = new MetadataBuilder().putLong("scale",
0).putBoolean("isTimestampNTZ", false)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
