[
https://issues.apache.org/jira/browse/SPARK-58696?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Wenchen Fan resolved SPARK-58696.
---------------------------------
Fix Version/s: 4.4.0
Resolution: Fixed
Issue resolved by pull request 57901
[https://github.com/apache/spark/pull/57901]
> [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.4.0, 3.5.0, 4.0.0
> Reporter: James Xu
> Assignee: James Xu
> Priority: Major
> Labels: CTE, optimizer, predicate-pushdown,
> pull-request-available
> Fix For: 4.4.0
>
>
> 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]
