[SPARK-53454][SQL] Fix AlwaysTrue/AlwaysFalse predicate translation in JDBC connectors

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)
 }

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)",

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)
+      }
+  }
 }

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)
 

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 = {