pan3793 commented on code in PR #58735:
URL: https://github.com/apache/spark/pull/58735#discussion_r3988414207
##########
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:
Added in b5fb8a0d3d2, as a row count rather than row values: pre-fix the
query returns 8 rows instead of 10, and asserting `count()` avoids depending on
`Range` order surviving the single-partition shuffle. Kept the plan-shape
assertion alongside it, plus the non-inlining check.
--
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]
