peter-toth commented on code in PR #58661:
URL: https://github.com/apache/spark/pull/58661#discussion_r3987281465
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushdownPredicatesAndPruneColumnsForCTEDef.scala:
##########
@@ -77,8 +77,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.
+ // The reference keeps its predicates, so a non-deterministic one
pushed into the
+ // definition as well would be evaluated twice.
+ val deterministicPredicates = predicates.filter(_.deterministic)
Review Comment:
**Finding 9.** This is a wrong-results fix that stands on its own, and the
shape it fixes needs no `MATERIALIZED`. Any definition that survives
`InlineCTE` gets its references' predicates OR-merged and pushed into the
shared definition while the references keep them. That is idempotent for a
deterministic predicate. For `rand() < 0.5` it filters twice. A
non-deterministic definition referenced more than once already survives
`InlineCTE` on master.
Measured on this head with `predicates.filter(_.deterministic)` put back to
`predicates`, `t` being a 3-row temp view:
```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
```
With the filter: two `rand` filters in the optimized plan, one per
reference, and 3 rows. Without it: four filters, because `(rand(2) < 0.5) OR
(rand(3) < 0.5)` sits in the definition as well, and 2 rows.
The comment change above names the cause correctly. This rule has matched
`PhysicalOperation` since
[SPARK-39764](https://github.com/apache/spark/commit/b02316ccf3e5af545f6e5444761761ddd73fb931)
in 3.4.0, and `PhysicalOperation` hands back a single filter even when it is
non-deterministic. The assert at
`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/planning/patterns.scala:117`
only binds when it collected more than one.
So the maintenance branches likely carry this. Please give it its own JIRA,
and pin it with a test that does not use the option, e.g. the query above.
`CTEInlineSuite`'s "non-deterministic predicates are not pushed into a
MATERIALIZED CTE" reaches the code only through `MATERIALIZED`, so a backport
of the two lines would arrive untested.
##########
docs/sql-ref-syntax-qry-select-cte.md:
##########
@@ -40,6 +40,19 @@ expression_name [ ( column_name [ , ... ] ) ] [ AS ] ( query
)
Specifies a name for the common table expression.
+* **MATERIALIZED**, **NOT MATERIALIZED**
+
+ Optionally specifies how the common table expression is evaluated.
`MATERIALIZED` forces it
+ to be evaluated once and shared by all references. `NOT MATERIALIZED`
forces it to be inlined,
+ so that each reference is planned and evaluated independently, and
non-deterministic
+ expressions such as `rand()` may yield different values per reference. A
`MATERIALIZED`
+ common table expression cannot reference columns of an outer query.
`MATERIALIZED` is not
+ supported in a statement whose common table expressions are always
inlined, such as a
+ multi-insert statement, nor in a subquery whose query, after the WITH
clause, references
Review Comment:
**Finding 10.** The check rejects on
`SubExprUtils.hasOuterReferences(withCTE)`, which covers the whole `WithCTE`,
definitions included. So the error also fires when the query after the WITH
clause references nothing outer and another CTE of the same clause carries the
correlation. Measured on this head:
```sql
select * from t o where exists (
with c as (select c1 from t2 where t2.c1 = o.c1),
v as materialized (select c1 from t2)
select * from v join c on v.c1 = c.c1
)
```
`UNSUPPORTED_FEATURE.MATERIALIZED_CTE_IN_CORRELATED_SUBQUERY` naming `v`.
The rejection is right here, since `c` is inlined into the body and the
`WithCTE` ends up on the correlated path. It is the sentence that is narrower
than the code.
```suggestion
multi-insert statement, nor in a subquery whose WITH clause or query
references
```
One shape the wording change does not cover: with `c` unreferenced (`select
* from v` as the body) the error still fires, though `InlineCTE` drops `c` and
the statement would run. A plain CTE in that shape returns rows on this head.
That matches how SPARK-45752 treats errors in unreferenced CTEs, so I am not
asking for a behaviour change.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/MaterializedCTECheck.scala:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis
+
+import scala.collection.mutable
+
+import org.apache.spark.sql.catalyst.expressions.{OuterReference,
OuterScopeReference, SubExprUtils, SubqueryExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{CTERelationDef,
CTERelationRef, LogicalPlan, SubqueryAlias, WithCTE}
+import org.apache.spark.sql.catalyst.trees.TreePattern.CTE
+import org.apache.spark.sql.errors.QueryCompilationErrors
+
+/**
+ * Checks that a MATERIALIZED CTE does not reference the query enclosing it,
as it is evaluated
+ * once on its own. The CTEs it references, transitively, are checked with it,
since they are
+ * inlined into it, unless they are MATERIALIZED themselves and form their own
boundary. Each
+ * definition is scanned at its own operator level and never inside its
subquery plans: a
+ * correlation a definition keeps inside its own subquery targets that
definition, not the
+ * enclosing query. A MATERIALIZED CTE is also rejected in a correlated
subquery when the query
+ * of its WITH clause references the outer query, as a `WithCTE` that is not
inlined cannot be
+ * decorrelated. The check covers the given plan and all its subqueries.
+ */
+object MaterializedCTECheck extends (LogicalPlan => Unit) {
+ override def apply(plan: LogicalPlan): Unit = {
+ if (plan.containsPattern(CTE)) {
+ // All CTE definitions, including those of nested subqueries, so that
references from a
+ // MATERIALIZED CTE can be followed across subquery boundaries.
+ val cteDefs = mutable.LinkedHashMap.empty[Long, CTERelationDef]
+ plan.foreachWithSubqueries {
+ case cteDef: CTERelationDef => cteDefs(cteDef.id) = cteDef
+ case _ =>
+ }
+ cteDefs.values.filter(_.materialized.contains(true)).foreach { cteDef =>
+ (cteDef +: collectReferencedDefs(cteDef,
cteDefs)).foreach(checkDefinition)
+ }
+ // Decorrelation stops at a subtree without outer references, so only a
`WithCTE` whose
+ // own subtree is correlated is on its path. A correlation above the
WITH clause, e.g. on a
+ // derived table holding it, is fine.
+ plan.subqueriesAll.foreach(_.foreach {
Review Comment:
**Finding 11.** This pass runs for every plan that contains a CTE, whether
or not any definition is `MATERIALIZED`. `subqueriesAll` builds the list of
every subquery plan recursively, and each one is then walked in full.
`CheckAnalysis` runs the check on every analyzed plan, and single-pass runs it
once per subquery plan on top (`ResolutionCheckRunner.runWithSubqueries`), so
the ungated pass is paid several times over on a subquery-heavy query.
The traversal above already collected the answer, so both passes can hang
off it:
```scala
val materializedDefs =
cteDefs.values.filter(_.materialized.contains(true)).toSeq
if (materializedDefs.nonEmpty) {
materializedDefs.foreach { cteDef =>
(cteDef +: collectReferencedDefs(cteDef,
cteDefs)).foreach(checkDefinition)
}
// Decorrelation stops at a subtree without outer references, so
only a `WithCTE` whose
// ...
plan.subqueriesAll.foreach(...)
}
```
##########
sql/core/src/test/resources/sql-tests/inputs/cte-command.sql:
##########
@@ -29,5 +29,20 @@ INSERT INTO cte_tbl2 SELECT col;
SELECT * FROM cte_tbl;
SELECT * FROM cte_tbl2;
+-- MATERIALIZED CTE in a Multi-INSERT, should fail
+WITH s AS MATERIALIZED (SELECT 46 AS col)
+FROM s
+INSERT INTO cte_tbl SELECT col
+INSERT INTO cte_tbl2 SELECT col;
+
+-- NOT MATERIALIZED CTE in a Multi-INSERT
+WITH s AS NOT MATERIALIZED (SELECT 46 AS col)
+FROM s
+INSERT INTO cte_tbl SELECT col
+INSERT INTO cte_tbl2 SELECT col;
Review Comment:
**Finding 12.** This file pins both rejections, and nothing pins the command
path that does keep the definition. A future widening of the `alwaysInline`
condition in `CTESubstitution` would turn a working statement into
`MATERIALIZED_CTE_ALWAYS_INLINED` with every test still green.
It works today. Measured on this head with a counting UDF over a 3-row
source and two references, the body produced 3 rows, so the definition is
evaluated once inside the command:
```sql
insert into a
with v as materialized (select counted(c1) c1 from t)
select c1 from v union all select c1 from v
```
One more block here covers it:
```sql
-- MATERIALIZED CTE in a single-INSERT statement
INSERT INTO cte_tbl WITH s AS MATERIALIZED (SELECT 47 AS col) SELECT col
FROM s UNION ALL SELECT col FROM s;
```
--
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]
