Skip to content

[SPARK-58554][SQL] Block MySQL double cast pushdown #57756

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
alekjarmov wants to merge 1 commit into from
Closed

[SPARK-58554][SQL] Block MySQL double cast pushdown #57756

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion ...integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MySQLIntegrationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down Expand Up @@ -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")))
}
}

/**
Expand Down
10 changes: 10 additions & 0 deletions sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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] = {
Expand Down
9 changes: 8 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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)
Expand Down