[SPARK-59434][SQL] Non-deterministic predicates should not be pushed down to CTE definitions

### What changes were proposed in this pull request?

`PushdownPredicatesAndPruneColumnsForCTEDef` collects the predicates at each CTE reference and pushes their OR-merged combination into the shared CTE definition. Each reference keeps its own predicates, so this only works for deterministic predicates -- a non-deterministic one is then evaluated a second time in the definition.

This PR filters the collected predicates to deterministic ones before push-down. When that leaves a reference with nothing pushable, the combined predicate becomes `TRUE` and the definition gets no push-down at all, including for its other references; the scaladoc is updated to say so.

### Why are the changes needed?

A non-deterministic predicate evaluated a second time in the CTE definition can drop rows, which is a wrong-results bug:

```sql
create or replace temp view t as select * from values (0), (1), (2) as t(c1);

with v as (select c1, rand(1) r from t)
select c1 from v where rand(2) < 0.5
union all
select c1 from v where rand(3) < 0.5;
```

The definition is non-deterministic and referenced twice, so it survives `InlineCTE`. `(rand(2) < 0.5) OR (rand(3) < 0.5)` is pushed into the definition while both references keep their own filter, so the optimized plan holds four `rand` filters instead of two and rows are filtered twice.

The rule has had this behavior since it was added in SPARK-37670 (3.2.2). Its scaladoc claimed determinism was taken care of by `ScanOperation`, but that was never true -- the filter-collection guard admits the first filter whatever it is:

```scala
filters.isEmpty || (filters.forall(_.deterministic) && condition.deterministic)
```

`PhysicalOperation`, which SPARK-39764 (3.4.0) later swapped in, behaves the same way here -- it returns a single filter even when it is non-deterministic, since its `filters.length > 1` assert only binds when more than one filter was collected. The scaladoc is corrected along with the fix.

### Does this PR introduce _any_ user-facing change?

Yes, it fixes wrong results for the query shape above. Affects 3.2.2 and later.

### How was this patch tested?

Four new tests in `CTEInlineSuite`, all failing without the fix:

- non-deterministic predicates stay at the references, two `rand` filters instead of four;
- deterministic conjuncts are still pushed while the non-deterministic one stays up;
- a reference with only non-deterministic predicates blocks push-down for its deterministic sibling, which must not be pushed alone;
- a row-count assertion using `monotonically_increasing_id()`, 10 rows instead of 8.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 5

Closes #58735 from pan3793/cte-nondeterministic-pushdown.

Authored-by: Cheng Pan <pan3793@gmail.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: pan3793

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

@@ -57,10 +57,12 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] with
 
   /**
    * Gather all the predicates and referenced attributes on different points of CTE references
-   * using pattern `ScanOperation` (which takes care of determinism) and combine those predicates
-   * and attributes that belong to the same CTE definition.
-   * For the same CTE definition, if any of its references does not have predicates, the combined
-   * predicate will be a TRUE literal, which means there will be no predicate push-down.
+   * using pattern `PhysicalOperation` and combine those predicates and attributes that belong
+   * to the same CTE definition. `PhysicalOperation` still returns a single filter even when it
+   * is non-deterministic, so such predicates are excluded below.
+   * For the same CTE definition, if any of its references does not have pushable predicates, the
+   * combined predicate will be a TRUE literal, which means there will be no predicate push-down
+   * for that definition at all, including for its other references.
    */
   private def gatherPredicatesAndAttributes(plan: LogicalPlan, cteMap: CTEMap): Unit = {
     plan match {
@@ -77,8 +79,11 @@ object PushdownPredicatesAndPruneColumnsForCTEDef extends Rule[LogicalPlan] with
         val newPredicates = if (isTruePredicate(preds)) {
           preds
         } else {
-          // Make sure we only push down predicates that do not contain forward CTE references.
-          val filteredPredicates = restoreCTEDefAttrs(predicates.filter(_.find {
+          // Only push down deterministic predicates that do not contain forward CTE references.
+          // A reference keeps its own predicates, so a non-deterministic one pushed into the
+          // shared definition as well would be evaluated a second time.
+          val deterministicPredicates = predicates.filter(_.deterministic)
+          val filteredPredicates = restoreCTEDefAttrs(deterministicPredicates.filter(_.find {
             case s: SubqueryExpression => s.plan.find {
               case r: CTERelationRef =>
                 // If the ref's ID does not exist in the map or if ref's corresponding precedence

sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala

@@ -451,6 +451,111 @@ abstract class CTEInlineSuiteBase
     }
   }
 
+  test("SPARK-59434: non-deterministic predicates are not pushed into a CTE def") {
+    withTempView("t") {
+      Seq(0, 1, 2).toDF("c1").createOrReplaceTempView("t")
+      // The CTE def is non-deterministic and referenced twice, so it is not inlined and the
+      // references' predicates get OR-merged into the shared def. A reference keeps its own
+      // predicate, so a non-deterministic one must not be pushed down as well.
+      val df = sql(
+        """with v as (select c1, rand(1) r from t)
+          |select c1 from v where rand(2) < 0.5
+          |union all
+          |select c1 from v where rand(3) < 0.5
+          |""".stripMargin)
+      val cteRepartitions = df.queryExecution.optimizedPlan.collect {
+        case r: RepartitionOperation => r
+      }
+      assert(cteRepartitions.nonEmpty,
+        "Non-deterministic With-CTE with multiple references should not be inlined.")
+      assert(
+        cteRepartitions.forall(_.collectFirst {
+          case f: Filter if f.condition.exists(_.isInstanceOf[Rand]) => f
+        }.isEmpty),
+        "Non-deterministic predicate should not be pushed down to the CTE def 'v'.")
+      val randFilters = df.queryExecution.optimizedPlan.collect {
+        case f: Filter if f.condition.exists(_.isInstanceOf[Rand]) => f
+      }
+      assert(randFilters.length == 2,
+        "Each reference's non-deterministic predicate should be evaluated once.")
+    }
+  }
+
+  test("SPARK-59434: deterministic conjuncts are still pushed into a CTE def") {
+    withTempView("t") {
+      Seq((0, 1), (1, 2), (2, 3)).toDF("c1", "c2").createOrReplaceTempView("t")
+      val df = sql(
+        """with v as (select c1, c2, rand(1) r from t)
+          |select c1 from v where c1 > 0 and rand(2) < 0.5
+          |union all
+          |select c1 from v where c1 < 2
+          |""".stripMargin)
+      val cteRepartitions = df.queryExecution.optimizedPlan.collect {
+        case r: RepartitionOperation => r
+      }
+      assert(cteRepartitions.nonEmpty, "CTE should not be inlined after optimization.")
+      // The non-deterministic conjunct stays at the reference, the deterministic ones are
+      // still OR-merged and pushed into the definition.
+      val distinctCteRepartitions = cteRepartitions.map(_.canonicalized).distinct
+      assert(distinctCteRepartitions.length == 1)
+      assert(
+        distinctCteRepartitions.head.collectFirst {
+          case f: Filter if f.condition.semanticEquals(
+            Or(GreaterThan(f.output(0), Literal(0)), LessThan(f.output(0), Literal(2)))) => f
+        }.isDefined,
+        "Predicate 'c1 > 0 OR c1 < 2' should be pushed down to the CTE def 'v'.")
+      assert(
+        distinctCteRepartitions.head.collectFirst {
+          case f: Filter if f.condition.exists(_.isInstanceOf[Rand]) => f
+        }.isEmpty,
+        "Non-deterministic predicate should not be pushed down to the CTE def 'v'.")
+    }
+  }
+
+  test("SPARK-59434: a non-deterministic reference blocks push-down for its siblings") {
+    withTempView("t") {
+      Seq((0, 1), (1, 2), (2, 3)).toDF("c1", "c2").createOrReplaceTempView("t")
+      val df = sql(
+        """with v as (select c1, c2, rand(1) r from t)
+          |select c1 from v where rand(2) < 0.5
+          |union all
+          |select c1 from v where c1 > 0
+          |""".stripMargin)
+      val cteRepartitions = df.queryExecution.optimizedPlan.collect {
+        case r: RepartitionOperation => r
+      }
+      assert(cteRepartitions.nonEmpty, "CTE should not be inlined after optimization.")
+      // The first reference has no pushable predicate, so the combined predicate is TRUE and
+      // the definition gets no filter. The sibling's 'c1 > 0' must not be pushed on its own,
+      // which would drop rows the first reference needs.
+      assert(
+        cteRepartitions.forall(_.collectFirst { case f: Filter => f }.isEmpty),
+        "CTE def 'v' should get no pushed-down filter.")
+    }
+  }
+
+  test("SPARK-59434: a non-deterministic predicate is evaluated once per CTE reference") {
+    withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
+      withTempView("t") {
+        spark.range(0, 6, 1, 1).selectExpr("cast(id as int) c1").createOrReplaceTempView("t")
+        // `monotonically_increasing_id` counts the rows it sees within a partition, so each
+        // reference drops exactly its own first row. A second evaluation in the shared def
+        // would drop a row there as well, leaving fewer. `rand(1)` only keeps the def from
+        // being inlined; column pruning removes it, so it never reaches a filter.
+        val df = sql(
+          """with v as (select c1, rand(1) r from t)
+            |select c1 from v where monotonically_increasing_id() > 0
+            |union all
+            |select c1 from v where monotonically_increasing_id() > 0
+            |""".stripMargin)
+        assert(
+          df.queryExecution.optimizedPlan.exists(_.isInstanceOf[RepartitionOperation]),
+          "Non-deterministic With-CTE with multiple references should not be inlined.")
+        assert(df.count() === 10, "Each reference should drop only its own first row.")
+      }
+    }
+  }
+
   test("Views with CTEs - 1 temp view") {
     withTempView("t", "t2") {
       Seq((0, 1), (1, 2)).toDF("c1", "c2").createOrReplaceTempView("t")