James Xu created SPARK-58696:
--------------------------------
Summary: [SQL] CTE predicate push-down discards filters inferred
by other optimizer rules
Key: SPARK-58696
URL: https://issues.apache.org/jira/browse/SPARK-58696
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 3.3.0, 3.5.0, 4.0.0, 3.4.0
Reporter: James Xu
h3. Problem
When a non-inlined CTE (or a view referenced from multiple call sites) is read
with different filter predicates at each reference, predicate push-down into
the shared CTE definition is what makes partition pruning and data source
filter push-down effective for all references. Between this rule's two
applications during optimization, other rules may inject additional inferred
filters into the CTE definition — and those filters are silently discarded by
the rule's second application, producing a plan with strictly weaker scan-level
filtering than the optimizer had already derived.
This was first observed as a production incident: a query reading a
daily-partitioned table through a multiply-referenced view failed with a hard
"full table scan is not allowed" guard. The view's scan listed 41,545
partitions instead of the intended 44 daily partitions — the inferred per-day
partition filter (derived from the call-site date predicate and the view's join
condition via constraint propagation) was present mid-optimization and was
deleted by the push-down rule's second pass.
On stock Spark the query results remain correct (the discarded filters are
logically redundant with the call-site filters), but partition pruning / scan
push-down is silently lost, which can turn a pruned read into a full scan.
Minimal repro (stock Apache Spark 3.3.1; the defective mechanism is unchanged
on master):
{code}
SET spark.sql.planChangeLog.level=WARN;
SET
spark.sql.planChangeLog.rules=org.apache.spark.sql.catalyst.optimizer.PushdownPredicatesAndPruneColumnsForCTEDef,org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints;
CREATE OR REPLACE TEMP VIEW t1 AS
SELECT id AS a FROM RANGE(10) UNION ALL SELECT CAST(NULL AS BIGINT);
CREATE OR REPLACE TEMP VIEW t2 AS SELECT id AS a FROM RANGE(10);
EXPLAIN EXTENDED
WITH c AS (
SELECT t1.a AS a, t2.a AS b, rand(0) AS r
FROM t1 JOIN t2 ON t1.a = t2.a
)
SELECT a, b FROM c WHERE a = 5
UNION ALL
SELECT a, b FROM c WHERE a = 7;
{code}
Observed: the {{InferFiltersFromConstraints}} plan-change log injects
{{Filter((isnotnull(b) AND ((b = 5) OR (b = 7))) ...)}} above {{t2}}'s
{{Range}} inside the CTE definition (propagated through the join condition),
and adds {{isnotnull}} conjuncts at the reference sites. The next application
of the push-down rule rebuilds the definition and the injected filter is gone.
In the final optimized plan, {{t1}}'s {{Range}} keeps {{((id = 5) OR (id =
7))}} while {{t2}}'s {{Range}} has no filter at all.
h3. Root Cause
The rule keeps a snapshot of the pre-push-down CTE definition plan in
{{CTERelationDef.originalPlanWithPredicates}} to stay idempotent across
fixedPoint iterations, and rebuilds the definition from that frozen snapshot
whenever newly gathered reference predicates re-arm its guard:
{code}
CTERelationDef(Filter(newCombinedPred, originalPlan), id, ...)
{code}
The rule runs in both "Operator Optimization before Inferring Filters" and
"Operator Optimization after Inferring Filters" ({{fixedPoint}}), with the
{{Once}} batch "Infer Filters" ({{InferFiltersFromConstraints}}) sandwiched
between them. The sequence:
# Pass 1 pushes the combined reference predicates and snapshots the
pre-push-down child.
# {{InferFiltersFromConstraints}} injects new filters into the definition's
child (e.g. propagating a pushed predicate through a join condition to the
other side) and enriches the reference-site filters (e.g. adding {{IsNotNull}}).
# The enrichment makes the re-gathered predicates differ from the snapshotted
ones, so pass 2 rebuilds from the frozen snapshot — discarding everything other
rules did to the child in between. Since the {{Once}} batch never re-runs, the
injected filters are lost permanently.
Structurally, this rule is the only rule in the shared fixedPoint set that
_regenerates_ content in a subtree it does not own (it synthesizes a filter
from predicates gathered at distant reference sites and stamps it onto the
definition's child). The snapshot rebuild effectively assumes sole ownership of
that subtree, which the interleaved {{Once}} batch violates.
h3. Solution
Rebuild from the current child instead of the frozen snapshot:
* On a re-push, remove the push-down filter the rule itself placed in the
previous pass, then wrap the result with the latest combined predicate.
* The removal locates the rule's own previous push wherever predicate push-down
may have moved it: it mirrors the cases of the predicate push-down rules,
translating the stored predicate back through projection and grouping-key
aliases (using the same alias-substitution helpers as push-down, so the two
cannot drift apart), positionally into each {{Union}} branch, and unchanged
through {{Join}}/{{Window}} and output-preserving unary nodes. Comparison uses
canonicalized forms once attribute ids are unified, which also absorbs operand
reordering by other rules.
* If the previous push can no longer be located (another rule rewrote or merged
it with other filters), the current child is used as-is: re-pushing the
disjunction of the reference predicates is redundant but always
semantics-preserving, since every reference re-applies its own predicates above
the reference. The worst case is a bounded redundant conjunct — never a
discarded foreign filter.
Key safety invariant: a CTE definition is consumed only through its references,
and every reference re-applies its own filters, so replacing or removing
filters inside the definition can only affect plan quality, never correctness.
The existing "push only when every reference has predicates" guard is
unchanged, and no optimizer batch ordering changes.
Also add the first direct tests of this rule's idempotency guarantee (the
mechanism shipped with SPARK-37670 was never covered by a dedicated test): the
new suite covers the staleness regression (fails on unmodified master),
idempotency under foreign mutation, and removal of the previous push after it
was moved through {{Project}}/{{Join}}/{{Union}}/{{Aggregate}}/{{Window}}.
h3. Expected Impact
* Production query (Spark 3.3, measured): the view scan regains the inferred
daily partition filter — 44 partitions read instead of 41,545 — eliminating the
hard full-scan failure.
* Upstream: queries over multiply-referenced CTEs/views whose reference
predicates are enriched between the two optimization batches (join-constraint
propagation, {{IsNotNull}} inference) keep their inferred scan-level filters,
preserving partition pruning and data source push-down.
* No plan regressions: all existing CTE push-down plan shapes asserted by
{{CTEInlineSuite}} are unchanged after the fix (63/63 CTE tests pass; full
{{sql/catalyst}} optimizer package: 1401/1401 pass).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
