[SPARK-49695][SQL] Postgres fix xor push-down

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala

@@ -986,4 +986,13 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
   test("scan with filter push-down with date time functions") {
     testDatetime(s"$catalogAndNamespace.${caseConvert("datetime")}")
   }
+
+  test("xor operator push-down") {
+    val df1 = spark.sql(
+      s"""SELECT * FROM $catalogAndNamespace.${caseConvert("pattern_testing_table")}
+         |WHERE id ^ 3 = 0""".stripMargin)
+    val rows1 = df1.collect()
+    assert(rows1.length === 1)
+    assert(rows1(0).getInt(0) === 3)
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala

@@ -23,14 +23,15 @@ import java.util
 import java.util.Locale
 
 import scala.util.Using
+import scala.util.control.NonFatal
 
 import org.apache.spark.internal.LogKeys.COLUMN_NAME
 import org.apache.spark.internal.MDC
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.SQLConfHelper
 import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NonEmptyNamespaceException, NoSuchIndexException}
 import org.apache.spark.sql.connector.catalog.Identifier
-import org.apache.spark.sql.connector.expressions.NamedReference
+import org.apache.spark.sql.connector.expressions.{Expression, GeneralScalarExpression, NamedReference}
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
 import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
@@ -379,4 +380,26 @@ private case class PostgresDialect()
       case _ =>
     }
   }
+
+  override def compileExpression(expr: Expression): Option[String] = {
+    val builder = new PostgresSQLBuilder()
+    try {
+      Some(builder.build(expr))
+    } catch {
+      case NonFatal(e) =>
+        logWarning("Error occurs while compiling V2 expression", e)
+        None
+    }
+  }
+
+  private class PostgresSQLBuilder extends JDBCSQLBuilder {
+    override def build(expr: Expression): String = {
+      expr match {
+        // Postgres uses '#' for xor, rather then '^'.
+        case e: GeneralScalarExpression if e.name() == "^" =>
+          visitBinaryArithmetic("#", inputToSQL(e.children().head), inputToSQL(e.children()(1)))
+        case _ => super.build(expr)
+      }
+    }
+  }
 }