peter-toth commented on code in PR #58735:
URL: https://github.com/apache/spark/pull/58735#discussion_r3988077738
##########
sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala:
##########
@@ -451,6 +451,29 @@ abstract class CTEInlineSuiteBase
}
}
+ test("SPARK-59434: non-deterministic predicates are not pushed into a CTE
def") {
Review Comment:
**Finding 2.** The new filter has two outcomes and this test covers one of
them.
- Every predicate at a reference is non-deterministic. `filteredPredicates`
ends up empty, the combined predicate becomes `Literal.TrueLiteral`, and the
definition loses push-down entirely. That is this test.
- Some conjuncts are deterministic. Those are still pushed, while the
non-deterministic ones stay at the reference. Nothing covers this.
I ran the second shape on this branch:
```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
```
The definition gets `Filter ((_1 > 0) OR (_1 < 2))` and one `rand` filter
stays at the reference. On `master` the definition gets `Filter (((_1 > 0) AND
(rand(2) < 0.5)) OR (_1 < 2))`, so an assertion on the pushed condition fails
there.
Worth a second test. Without it, narrowing the fix to "skip the whole
reference when any predicate is non-deterministic" would pass the suite while
quietly losing this push-down.
##########
sql/core/src/test/scala/org/apache/spark/sql/CTEInlineSuite.scala:
##########
@@ -451,6 +451,29 @@ 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 and pushed into the shared def.
Each reference
+ // keeps its own predicate, so pushing a non-deterministic one down
evaluates it twice.
+ 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)
+ assert(
+
df.queryExecution.optimizedPlan.exists(_.isInstanceOf[RepartitionOperation]),
+ "Non-deterministic With-CTE with multiple references should be not
inlined.")
+ val randFilters = df.queryExecution.optimizedPlan.collect {
+ case f: Filter if f.condition.exists(_.isInstanceOf[Rand]) => f
+ }
+ assert(randFilters.length == 2,
Review Comment:
**Finding 3.** This is a wrong-results fix, but nothing in the test looks at
a row. The filter count is a proxy, and it would still pass if a later change
kept two filters and dropped rows another way.
A deterministic row assertion is available for this shape.
`monotonically_increasing_id()` is non-deterministic and counts rows inside a
partition, so the second evaluation is directly visible:
```scala
withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") {
withTempView("t") {
spark.range(0, 6, 1, 1).selectExpr("cast(id as int)
c1").createOrReplaceTempView("t")
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)
checkAnswer(df, ((1 to 5) ++ (1 to 5)).map(Row(_)))
}
}
```
I ran it in both suites, AQE on and off. It passes on this branch and fails
on `master`, which returns 8 rows instead of 10. `rand(1)` is only there to
stop `InlineCTE` from inlining the definition. Column pruning then removes it,
so it never reaches a filter.
Worth adding alongside the plan-shape assertion.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala:
##########
@@ -57,8 +57,8 @@ 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.
+ * using pattern `PhysicalOperation` and combine those predicates and
attributes that belong
+ * to the same CTE definition.
Review Comment:
**Finding 1.** The scaladoc correction is right, but the story behind it in
the description is not. `ScanOperation` never rejected a lone non-deterministic
filter either, so SPARK-39764 did not regress anything here.
At `175e429cca2`, the SPARK-37670 commit that added this rule,
`ScanOperation` ran with `legacyMode = false`, and that path admits the first
filter whatever it is:
```scala
fields.forall(_.forall(_.deterministic)) && {
filters.isEmpty || (filters.forall(_.deterministic) &&
condition.deterministic)
} && canCollapseExpressions(Seq(condition), aliases, alwaysInline)
```
Today's `ScanOperation` still hands a lone non-deterministic filter back as
pushable, at
`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala:152`:
```scala
} else {
val filtersCanPushDown = splitConjunctivePredicates(filters.head)
val filtersStayUp = filters.drop(1)
```
I measured this. Swapping the rule back to `ScanOperation(projects, _,
predicates, ref)` and removing the new `filter(_.deterministic)` leaves your
test failing exactly as on `master`, with four `rand` filters.
SPARK-37670 also first shipped in 3.4.0, so "Affects 3.4.0 and later" stays
correct. What changes is the cause. The rule was born with this bug, and the
old scaladoc's "which takes care of determinism" was never true. Worth
rewording, since the description becomes the commit message.
The scaladoc could carry the reason too:
```suggestion
* to the same CTE definition. `PhysicalOperation` still returns a single
filter even when it
* is non-deterministic, so such predicates are excluded below.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
