[SPARK-43095][SQL] Avoid Once strategy's idempotence is broken for batch: `Infer Filters` #40742

wangyum · 2023-04-11T14:04:03Z

What changes were proposed in this pull request?

This PR makes it also remove EqualNullSafe when removing EqualTo if their children are same when constructing candidate constraints in ConstraintHelper.inferAdditionalConstraints.
For example: l = r and l <=> r. Before this PR, l = r and l <=> r can infer l <=> l and r <=> r which is useless. After This PR, it can't infer anything.

Why are the changes needed?

Avoid Once strategy's idempotence is broken for batch: Infer Filters:

export SPARK_TESTING=1

CREATE TABLE t1 (i INT, j INT, k STRING) USING parquet;
CREATE TABLE t2 (i INT, j INT, k STRING) USING parquet;
CREATE TABLE t3 (i INT, j INT, k STRING) USING parquet;

SELECT *
FROM   (SELECT t1.i, t1.i as t1i
        FROM t1 JOIN t3 ON t1.i = t3.i) t
       JOIN t2 ON t.i = t2.i;

Before this PR:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints ===
 Join Inner, (i#72 = i#78)                                               Join Inner, (i#72 = i#78)
!:- Project [i#72, i#72 AS t1i#71]                                       :- Filter ((i#72 <=> i#72) AND (t1i#71 <=> t1i#71))
!:  +- Join Inner, (i#72 = i#75)                                         :  +- Project [i#72, i#72 AS t1i#71]
!:     :- Project [i#72]                                                 :     +- Join Inner, (i#72 = i#75)
!:     :  +- Relation spark_catalog.default.t1[i#72,j#73,k#74] parquet   :        :- Filter isnotnull(i#72)
!:     +- Project [i#75]                                                 :        :  +- Project [i#72]
!:        +- Relation spark_catalog.default.t3[i#75,j#76,k#77] parquet   :        :     +- Relation spark_catalog.default.t1[i#72,j#73,k#74] parquet
!+- Relation spark_catalog.default.t2[i#78,j#79,k#80] parquet            :        +- Filter isnotnull(i#75)
!                                                                        :           +- Project [i#75]
!                                                                        :              +- Relation spark_catalog.default.t3[i#75,j#76,k#77] parquet
!                                                                        +- Filter isnotnull(i#78)
!                                                                           +- Relation spark_catalog.default.t2[i#78,j#79,k#80] parquet


org.apache.spark.SparkRuntimeException: Once strategy's idempotence is broken for batch Infer Filters
 Join Inner, (i#72 = i#78)                                                     Join Inner, (i#72 = i#78)
 :- Filter ((i#72 <=> i#72) AND (t1i#71 <=> t1i#71))                           :- Filter ((i#72 <=> i#72) AND (t1i#71 <=> t1i#71))
 :  +- Project [i#72, i#72 AS t1i#71]                                          :  +- Project [i#72, i#72 AS t1i#71]
 :     +- Join Inner, (i#72 = i#75)                                            :     +- Join Inner, (i#72 = i#75)
 :        :- Filter isnotnull(i#72)                                            :        :- Filter isnotnull(i#72)
 :        :  +- Project [i#72]                                                 :        :  +- Project [i#72]
 :        :     +- Relation spark_catalog.default.t1[i#72,j#73,k#74] parquet   :        :     +- Relation spark_catalog.default.t1[i#72,j#73,k#74] parquet
 :        +- Filter isnotnull(i#75)                                            :        +- Filter isnotnull(i#75)
 :           +- Project [i#75]                                                 :           +- Project [i#75]
 :              +- Relation spark_catalog.default.t3[i#75,j#76,k#77] parquet   :              +- Relation spark_catalog.default.t3[i#75,j#76,k#77] parquet
!+- Filter isnotnull(i#78)                                                     +- Filter (i#78 <=> i#78)
!   +- Relation spark_catalog.default.t2[i#78,j#79,k#80] parquet                  +- Filter isnotnull(i#78)
!                                                                                    +- Relation spark_catalog.default.t2[i#78,j#79,k#80] parquet.

After this PR:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints ===
 Join Inner, (i#72 = i#78)                                               Join Inner, (i#72 = i#78)
 :- Project [i#72, i#72 AS t1i#71]                                       :- Project [i#72, i#72 AS t1i#71]
 :  +- Join Inner, (i#72 = i#75)                                         :  +- Join Inner, (i#72 = i#75)
!:     :- Project [i#72]                                                 :     :- Filter isnotnull(i#72)
!:     :  +- Relation spark_catalog.default.t1[i#72,j#73,k#74] parquet   :     :  +- Project [i#72]
!:     +- Project [i#75]                                                 :     :     +- Relation spark_catalog.default.t1[i#72,j#73,k#74] parquet
!:        +- Relation spark_catalog.default.t3[i#75,j#76,k#77] parquet   :     +- Filter isnotnull(i#75)
!+- Relation spark_catalog.default.t2[i#78,j#79,k#80] parquet            :        +- Project [i#75]
!                                                                        :           +- Relation spark_catalog.default.t3[i#75,j#76,k#77] parquet
!                                                                        +- Filter isnotnull(i#78)
!                                                                           +- Relation spark_catalog.default.t2[i#78,j#79,k#80] parquet

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Unit test.

wangyum · 2023-04-11T23:45:24Z

cc @cloud-fan

cloud-fan · 2023-04-12T02:32:25Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -1430,6 +1430,10 @@ object InferFiltersFromConstraints extends Rule[LogicalPlan]
      .union(constructIsNotNullConstraints(constraints, plan.output))
      .filter { c =>
        c.references.nonEmpty && c.references.subsetOf(plan.outputSet) && c.deterministic
+      }.filterNot {
+        // Avoid once strategy idempotence is broken.
+        case a EqualNullSafe b => a.semanticEquals(b)


can we fix the place that generates this useless predicate?

Yes. Moved it to ConstraintHelper.inferAdditionalConstraints.

wangyum · 2023-04-13T07:19:52Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala

@@ -75,7 +75,10 @@ trait ConstraintHelper {
        inferredConstraints ++= replaceConstraints(predicates - eq, l, r)
      case _ => // No inference
    }
-    inferredConstraints -- constraints
+    (inferredConstraints -- constraints).filterNot {
+      case a EqualNullSafe b => a.semanticEquals(b)


Before this PR. the generated EqualNullSafe will be optimized by SimplifyBinaryComparison.

cloud-fan · 2023-04-13T07:20:58Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala

@@ -75,7 +75,10 @@ trait ConstraintHelper {
        inferredConstraints ++= replaceConstraints(predicates - eq, l, r)
      case _ => // No inference
    }
-    inferredConstraints -- constraints
+    (inferredConstraints -- constraints).filterNot {


shall we update the pattern match above to add if !l.semanticEquals(r)?

cloud-fan · 2023-04-13T08:55:44Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala

@@ -66,13 +66,13 @@ trait ConstraintHelper {
    val predicates = constraints.filterNot(_.isInstanceOf[IsNotNull])
    predicates.foreach {
      case eq @ EqualTo(l: Attribute, r: Attribute) =>
-        val candidateConstraints = predicates - eq
+        val candidateConstraints = predicates - eq - EqualNullSafe(l, r)


if l.semanticEquals(r), we don't need to infer predicates at all, right?

Yes. But the current situation is like this: l <=> l and r <=> r is inferred from l === r and l <=> r. At this time, l.semanticEquals(r) is false.

ah got it. Let's add a code comment to explain it with this example.

can you explain it a bit more about why it makes the optimizer not idempotent?

For example:

Join Inner, (a#0 = a#7) :- Project [a#0, a#0 AS xa#3] : +- Join Inner, (a#0 = a#4) : :- LocalRelation <empty>, [a#0, b#1, c#2] : +- LocalRelation <empty>, [a#4, b#5, c#6] +- LocalRelation <empty>, [a#7, b#8, c#9]

After InferFiltersFromConstraints:

Join Inner, (a#0 = a#7) :- Filter ((a#0 <=> a#0) AND (xa#3 <=> xa#3)) : +- Project [a#0, a#0 AS xa#3] : +- Join Inner, (a#0 = a#4) : :- Filter isnotnull(a#0) : : +- LocalRelation <empty>, [a#0, b#1, c#2] : +- Filter isnotnull(a#4) : +- LocalRelation <empty>, [a#4, b#5, c#6] +- Filter isnotnull(a#7) +- LocalRelation <empty>, [a#7, b#8, c#9]

And run InferFiltersFromConstraints again. The left side constraints:

(a#0 <=> xa#3) (xa#3 = a#0) (a#0 <=> a#0) (xa#3 <=> xa#3)

The right side constraints:

(isnotnull(a#7))

Join condition is:

(a#0 = a#7)

Based on these constraints, the inferred result is:

(a#7 <=> xa#3) (xa#3 = a#7) (a#7 <=> a#7)

(a#7 <=> a#7) is a valid constraints for right side. The result:

Join Inner, (a#0 = a#7) :- Filter ((a#0 <=> a#0) AND (xa#3 <=> xa#3)) : +- Project [a#0, a#0 AS xa#3] : +- Join Inner, (a#0 = a#4) : :- Filter isnotnull(a#0) : : +- LocalRelation <empty>, [a#0, b#1, c#2] : +- Filter isnotnull(a#4) : +- LocalRelation <empty>, [a#4, b#5, c#6] +- Filter (a#7 <=> a#7) +- Filter isnotnull(a#7) +- LocalRelation <empty>, [a#7, b#8, c#9]

cloud-fan · 2023-04-13T12:16:04Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala

@@ -66,13 +66,13 @@ trait ConstraintHelper {
    val predicates = constraints.filterNot(_.isInstanceOf[IsNotNull])
    predicates.foreach {
      case eq @ EqualTo(l: Attribute, r: Attribute) =>


not related to this PR, but shall we match Equality here?

Yes. It seems ok.

cloud-fan · 2023-04-14T06:11:13Z

...talyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala

@@ -66,13 +66,15 @@ trait ConstraintHelper {
    val predicates = constraints.filterNot(_.isInstanceOf[IsNotNull])
    predicates.foreach {
      case eq @ EqualTo(l: Attribute, r: Attribute) =>
-        val candidateConstraints = predicates - eq
+        // Also remove EqualNullSafe with the same l and r to avoid Once strategy's idempotence
+        // is broken.


let's mention l === r and l <=> r can infer l <=> l and r <=> r which is useless.

wangyum · 2023-04-15T01:25:36Z

Merged to master.

Avoid Once strategy's idempotence is broken for batch: Infer Filters

a5cdd58

github-actions bot added the SQL label Apr 11, 2023

cloud-fan reviewed Apr 12, 2023

View reviewed changes

Address comments

0181764

wangyum commented Apr 13, 2023

View reviewed changes

cloud-fan reviewed Apr 13, 2023

View reviewed changes

wangyum force-pushed the SPARK-43095 branch from 2b3f866 to e9a80e9 Compare April 13, 2023 09:00

fix

842faec

wangyum force-pushed the SPARK-43095 branch from e9a80e9 to 842faec Compare April 13, 2023 09:10

cloud-fan reviewed Apr 13, 2023

View reviewed changes

Add comments

f1ae941

cloud-fan reviewed Apr 14, 2023

View reviewed changes

fix

917fa3e

cloud-fan approved these changes Apr 14, 2023

View reviewed changes

wangyum closed this in 7796028 Apr 15, 2023

wangyum deleted the SPARK-43095 branch April 15, 2023 01:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-43095][SQL] Avoid Once strategy's idempotence is broken for batch: `Infer Filters` #40742

[SPARK-43095][SQL] Avoid Once strategy's idempotence is broken for batch: `Infer Filters` #40742

wangyum commented Apr 11, 2023 •

edited

Loading

wangyum commented Apr 11, 2023

cloud-fan Apr 12, 2023

wangyum Apr 13, 2023

wangyum Apr 13, 2023

cloud-fan Apr 13, 2023 •

edited

Loading

cloud-fan Apr 13, 2023

wangyum Apr 13, 2023 •

edited

Loading

cloud-fan Apr 13, 2023

cloud-fan Apr 13, 2023 •

edited

Loading

wangyum Apr 14, 2023

cloud-fan Apr 13, 2023

wangyum Apr 14, 2023

cloud-fan Apr 14, 2023

wangyum commented Apr 15, 2023

[SPARK-43095][SQL] Avoid Once strategy's idempotence is broken for batch: Infer Filters #40742

[SPARK-43095][SQL] Avoid Once strategy's idempotence is broken for batch: Infer Filters #40742

Conversation

wangyum commented Apr 11, 2023 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

wangyum commented Apr 11, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Apr 13, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wangyum Apr 13, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Apr 13, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wangyum commented Apr 15, 2023

[SPARK-43095][SQL] Avoid Once strategy's idempotence is broken for batch: `Infer Filters` #40742

[SPARK-43095][SQL] Avoid Once strategy's idempotence is broken for batch: `Infer Filters` #40742

wangyum commented Apr 11, 2023 •

edited

Loading

cloud-fan Apr 13, 2023 •

edited

Loading

wangyum Apr 13, 2023 •

edited

Loading

cloud-fan Apr 13, 2023 •

edited

Loading