diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/DB2IntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/DB2IntegrationSuite.scala index ecc02f7051787..60516e95e6d15 100644 --- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/DB2IntegrationSuite.scala +++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/DB2IntegrationSuite.scala @@ -18,7 +18,6 @@ package org.apache.spark.sql.jdbc.v2 import java.sql.Connection -import java.util.Locale import org.apache.spark.SparkConf import org.apache.spark.sql.AnalysisException @@ -120,6 +119,4 @@ class DB2IntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCTest { .add("ID", IntegerType, true, defaultMetadata(IntegerType, JdbcClientTypes.INTEGER)) assert(t.schema === expectedSchema) } - - override def caseConvert(tableName: String): String = tableName.toUpperCase(Locale.ROOT) } diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala index 2845f3a5f19f7..6574a63d7a103 100644 --- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala +++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/MsSqlServerIntegrationSuite.scala @@ -20,11 +20,7 @@ package org.apache.spark.sql.jdbc.v2 import java.sql.Connection import org.apache.spark.{SparkConf, SparkSQLFeatureNotSupportedException} -import org.apache.spark.rdd.RDD import org.apache.spark.sql.AnalysisException -import org.apache.spark.sql.catalyst.InternalRow -import org.apache.spark.sql.execution.{RowDataSourceScanExec, SparkPlan} -import org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog import org.apache.spark.sql.jdbc.MsSQLServerDatabaseOnDocker import org.apache.spark.sql.types._ @@ -46,17 +42,6 @@ class MsSqlServerIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JD val STRING = "nvarchar" } - def getExternalEngineQuery(executedPlan: SparkPlan): String = { - getExternalEngineRdd(executedPlan).asInstanceOf[JDBCRDD].getExternalEngineQuery - } - - def getExternalEngineRdd(executedPlan: SparkPlan): RDD[InternalRow] = { - val queryNode = executedPlan.collect { case r: RowDataSourceScanExec => - r - }.head - queryNode.rdd - } - // Following tests are disabled for both single and multiple partition read override def excluded: Seq[String] = Seq( "simple scan with OFFSET (false)", diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala index fb18dca97994e..9777fe91d5ecd 100644 --- a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala +++ b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala @@ -17,16 +17,21 @@ package org.apache.spark.sql.jdbc.v2 +import java.util.Locale + import org.apache.logging.log4j.Level import org.apache.spark.rdd.RDD import org.apache.spark.sql.{AnalysisException, DataFrame} +import org.apache.spark.sql.catalyst.InternalRow import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException} import org.apache.spark.sql.connector.DataSourcePushdownTestUtils import org.apache.spark.sql.connector.catalog.{Catalogs, Identifier, TableCatalog} import org.apache.spark.sql.connector.catalog.index.SupportsIndex import org.apache.spark.sql.connector.expressions.NullOrdering +import org.apache.spark.sql.execution.{RowDataSourceScanExec, SparkPlan} import org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.jdbc.DockerIntegrationFunSuite import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ @@ -46,6 +51,17 @@ private[v2] trait V2JDBCTest private def catalogAndNamespace = namespaceOpt.map(namespace => s"$catalogName.$namespace").getOrElse(catalogName) + def getExternalEngineQuery(executedPlan: SparkPlan): String = { + getExternalEngineRdd(executedPlan).asInstanceOf[JDBCRDD].getExternalEngineQuery + } + + def getExternalEngineRdd(executedPlan: SparkPlan): RDD[InternalRow] = { + val queryNode = executedPlan.collect { case r: RowDataSourceScanExec => + r + }.head + queryNode.rdd + } + // dialect specific update column type test def testUpdateColumnType(tbl: String): Unit @@ -1281,4 +1297,41 @@ private[v2] trait V2JDBCTest assert(e.getCondition !== "FAILED_JDBC.TABLE_EXISTS") } } + + val alwaysTrueFalseTestCases = Seq(("true", "1=1"), ("false", "1=0")) + + gridTest("Test boolean literal pushdown")(alwaysTrueFalseTestCases) { + case (condition, expected) => + import org.apache.spark.sql._ + val rows: Seq[Row] = withSQLConf( + SQLConf.LEGACY_JDBC_CONNECTOR_BOOLEAN_LITERAL_TRANSLATION.key -> "false" + ) { + val df = spark.sql( + s""" + |SELECT * + |FROM $catalogAndNamespace + |WHERE $condition + |""".stripMargin) + + df.collect().toIndexedSeq + } + + withSQLConf( + (SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> + "org.apache.spark.sql.catalyst.optimizer.PruneFilters"), + (SQLConf.LEGACY_JDBC_CONNECTOR_BOOLEAN_LITERAL_TRANSLATION.key -> "true") + ) { + val df = sql( + s""" + |SELECT * + |FROM $catalogAndNamespace + |WHERE $condition + |""".stripMargin) + + val sqlText = getExternalEngineQuery(df.queryExecution.sparkPlan) + + assert(sqlText.toLowerCase(Locale.ROOT).contains(s"where ($expected)")) + checkAnswer(df, rows) + } + } } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 86807914c2362..7cf68520ba38f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -5141,6 +5141,16 @@ object SQLConf { .booleanConf .createWithDefault(false) + val LEGACY_JDBC_CONNECTOR_BOOLEAN_LITERAL_TRANSLATION = + buildConf("spark.sql.legacy.jdbc.connectorBooleanLiteralTranslation") + .internal() + .doc("When set to true, boolean literals (true, false) will be translated as " + + "such (true, false). When false, boolean 'true' will be translated to '1=1' and" + + " boolean 'false' will be translated to 'false'. Some JDBC drivers don't support" + + " 'WHERE true' syntax that can be the result of AQE.") + .booleanConf + .createWithDefault(false) + val LEGACY_MSSQLSERVER_NUMERIC_MAPPING_ENABLED = buildConf("spark.sql.legacy.mssqlserver.numericMapping.enabled") .internal() @@ -6760,6 +6770,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf { def addSingleFileInAddFile: Boolean = getConf(LEGACY_ADD_SINGLE_FILE_IN_ADD_FILE) + def legacyJdbcConnectorBooleanLiteralTranslation: Boolean = + getConf(LEGACY_JDBC_CONNECTOR_BOOLEAN_LITERAL_TRANSLATION) + def legacyMsSqlServerNumericMappingEnabled: Boolean = getConf(LEGACY_MSSQLSERVER_NUMERIC_MAPPING_ENABLED) 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 ce4c347cad349..63cb0224eb648 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 @@ -40,7 +40,7 @@ import org.apache.spark.sql.connector.catalog.functions.UnboundFunction import org.apache.spark.sql.connector.catalog.index.TableIndex import org.apache.spark.sql.connector.expressions.{Expression, Literal, NamedReference} import org.apache.spark.sql.connector.expressions.aggregate.AggregateFunc -import org.apache.spark.sql.connector.expressions.filter.Predicate +import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse, AlwaysTrue, Predicate} import org.apache.spark.sql.connector.util.V2ExpressionSQLBuilder import org.apache.spark.sql.errors.QueryCompilationErrors import org.apache.spark.sql.execution.datasources.jdbc.{DriverRegistry, JDBCOptions, JdbcOptionsInWrite, JdbcUtils} @@ -392,9 +392,16 @@ abstract class JdbcDialect extends Serializable with Logging { "CASE WHEN " + input + " THEN 1 ELSE 0 END" override def visitLiteral(literal: Literal[_]): String = { - Option(literal.value()).map(v => - compileValue(CatalystTypeConverters.convertToScala(v, literal.dataType())).toString) - .getOrElse(super.visitLiteral(literal)) + literal match { + case _: AlwaysTrue if !SQLConf.get.legacyJdbcConnectorBooleanLiteralTranslation => + "1=1" + case _: AlwaysFalse if !SQLConf.get.legacyJdbcConnectorBooleanLiteralTranslation => + "1=0" + case _ => + Option(literal.value()).map(v => + compileValue(CatalystTypeConverters.convertToScala(v, literal.dataType())).toString) + .getOrElse(super.visitLiteral(literal)) + } } override def visitNamedReference(namedRef: NamedReference): String = {