[jira] [Created] (SPARK-58696) [SQL] CTE predicate push-down discards filters inferred by other optimizer rules

["James Xu (Jira)" <ji...@apache.org>]

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: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org