(spark) branch branch-4.x updated: [SPARK-57727][SQL] Fix incorrect query result due to inferAdditionalConstraints incorrectly substituting attributes with non-binary-stable collations

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new 85ad6f3d797a [SPARK-57727][SQL] Fix incorrect query result due to inferAdditionalConstraints incorrectly substituting attributes with non-binary-stable collations
85ad6f3d797a is described below

commit 85ad6f3d797af0c7c3a58105f8d66a11e46c4315
Author: Eric Yang <ji...@gmail.com>
AuthorDate: Fri Jul 3 21:01:52 2026 +0800

    [SPARK-57727][SQL] Fix incorrect query result due to inferAdditionalConstraints incorrectly substituting attributes with non-binary-stable collations
    
    ### What changes were proposed in this pull request?
    Make attribute substitution in `QueryPlanConstraints.inferAdditionalConstraints` collation-aware. For a binary-stable `source`, substitution is unchanged (any position). For a non-binary-stable (e.g. collated) `source`, substitute only where it is a direct operand of a same-collation comparison, so it is never moved into a context that changes the effective collation.
    There is another draft PR https://github.com/apache/spark/pull/56836 fixes the issue in a different way in which the change is more scoped/safer but can be too conservative.
    
    This mirrors the existing guard in another rule `ConstantPropagation` (SPARK-55647).
    
    ### Why are the changes needed?
    `inferAdditionalConstraints` infers a predicate on `b` from `a = b` and a predicate on `a` by substituting `a` with `b`. This is only valid when `a` and `b` are byte-for-byte interchangeable. Under a non-binary-stable collation, `a = b` is a collation equality (e.g. `'hello' = 'HELLO'` under `UTF8_LCASE`), not byte equality, so substituting into a comparison evaluated in a different collation produces a wrong constraint and silently drops rows:
    ```sql
    CREATE TABLE t (a STRING COLLATE UTF8_LCASE, b STRING COLLATE UTF8_LCASE);
    INSERT INTO t VALUES ('hello', 'HELLO');
    SELECT a, b FROM t WHERE a = b AND a = 'hello' COLLATE UTF8_BINARY;
    ```
    
    ### Does this PR introduce _any_ user-facing change?
    Yes, fixed incorrect query result.
    
    ### How was this patch tested?
    Added UT.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    Yes. Claude Code
    
    Closes #56860 from jiwen624/SPARK-57727-perf.
    
    Authored-by: Eric Yang <ji...@gmail.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
    (cherry picked from commit bb9c363978a21f4294aa70d7e70cdaee50c66fa9)
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../plans/logical/QueryPlanConstraints.scala       | 21 ++++++++++++---
 .../spark/sql/collation/CollationSuite.scala       | 30 ++++++++++++++++++++++
 2 files changed, 48 insertions(+), 3 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
index 6a1aa54a54d0..b023d7320808 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
@@ -129,9 +129,24 @@ trait ConstraintHelper {
   private def replaceConstraints(
       constraints: ExpressionSet,
       source: Expression,
-      destination: Expression): ExpressionSet = constraints.map(_ transform {
-    case e: Expression if e.semanticEquals(source) => destination
-  })
+      destination: Expression): ExpressionSet = {
+    if (isBinaryStable(source.dataType)) {
+      constraints.map(_ transform {
+        case e: Expression if e.semanticEquals(source) => destination
+      })
+    } else {
+      constraints.map(_ transform {
+        case b: BinaryComparison if sameCollationOperand(b, source) =>
+          b.withNewChildren(b.children.map { c =>
+            if (c.semanticEquals(source)) destination else c
+          })
+      })
+    }
+  }
+
+  private def sameCollationOperand(b: BinaryComparison, source: Expression): Boolean =
+    (b.left.semanticEquals(source) || b.right.semanticEquals(source)) &&
+      b.left.dataType == source.dataType && b.right.dataType == source.dataType
 
   /**
    * Infers a set of `isNotNull` constraints from null intolerant expressions as well as
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
index 711e1e091a98..045748ad7cb1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
 import org.apache.spark.sql.catalyst.ExtendedAnalysisException
 import org.apache.spark.sql.catalyst.analysis.ResolvedIdentifier
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.logical.Filter
 import org.apache.spark.sql.catalyst.util.CollationFactory
 import org.apache.spark.sql.connector.{DatasourceV2SQLBase, FakeV2ProviderWithCustomSchema}
 import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, InMemoryTable}
@@ -2440,6 +2441,35 @@ class CollationSuite extends DatasourceV2SQLBase with AdaptiveSparkPlanHelper {
     }
   }
 
+  test("SPARK-57727: constraint inference does not substitute non-binary-stable attributes") {
+    withTable("t1") {
+      sql("CREATE TABLE t1 (a STRING COLLATE UTF8_LCASE, b STRING COLLATE UTF8_LCASE)")
+      sql("INSERT INTO t1 VALUES ('hello', 'HELLO')")
+
+      checkAnswer(
+        sql("SELECT a, b FROM t1 WHERE a = b AND a = 'hello' COLLATE UTF8_BINARY"),
+        Row("hello", "HELLO")
+      )
+    }
+  }
+
+  test("SPARK-57727: same-collation constraint inference is preserved") {
+    withTable("t1") {
+      sql("CREATE TABLE t1 (a STRING COLLATE UTF8_LCASE, b STRING COLLATE UTF8_LCASE)")
+
+      val optimized =
+        sql("SELECT a, b FROM t1 WHERE a = b AND a = 'hello'").queryExecution.optimizedPlan
+      val inferredOnB = optimized.exists {
+        case Filter(cond, _) => cond.exists {
+          case e: EqualTo => e.references.exists(_.name == "b") && e.toString.contains("hello")
+          case _ => false
+        }
+        case _ => false
+      }
+      assert(inferredOnB, s"expected inferred 'b = hello' filter to be preserved:\n$optimized")
+    }
+  }
+
   test("ConstantPropagation: replaces binary-stable attributes with contradicting predicates") {
     withTable("t1") {
       sql("CREATE TABLE t1 (c STRING)")


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org