Split physical optimizer rules into enforcement and optimization phases, with a convergence loop for the optimizers

[zhuqi-lucas]

Is your feature request related to a problem or challenge?

Follow-up to @alamb's proposal in #25355 (comment), filing the independent ticket discussed there.

The physical optimizer runs its rule list exactly once, in a hand-ordered sequence. Some passes in that list are not optimizations but enforcers of invariants (EnsureRequirements, OutputRequirements, the sanity check): they must hold before execution, and any rewrite can invalidate them. Under a once-through model the only way to keep a chain correct is to repeat the enforcement rules by hand after every rewrite that can invalidate them, and the repeats run unconditionally whether or not the rewrite fired.

On a production chain with custom rewrites this adds up: 35 physical rule passes of which 28 leave the plan untouched, including six EnsureRequirements passes of which four are no-ops, at roughly 80% of physical optimization time. The logical layer does not have this problem because it has both a phase split (AnalyzerRule vs OptimizerRule) and a convergence loop; the physical layer has neither.

Describe the solution you'd like

Mirror the logical layer, per #25355:

1. A new trait for enforcement passes (working name PhysicalAnalyzerRule), for rules that uphold invariants rather than improve the plan. The framework runs them at phase boundaries, so downstream chains stop hand-repeating them.
2. A convergence loop for the remaining optimizer rules, copying the shape of Optimizer::optimize: bounded by max_passes, terminated by plan identity.
3. A PhysicalPlanSignature for that loop. ExecutionPlan implements neither Hash nor Eq, so identity has to be derived; the fingerprint built in feat(core): skip a named rule handed a plan it has been seen to leave alone #25356 (the rendered plan plus the properties rules consult: partitioning, orderings, equivalences, with statistics deliberately excluded and the boundaries pinned by tests) is a working candidate and can be repurposed wholesale.

Design questions to settle

• Phase topology. Is the shape a fixed optimize-loop → enforce, or can a chain declare an alternation? Some downstream rules consume the operators enforcement materializes (for example, removing an enforced sort by reorganizing what runs beneath it) rather than the abstract requirement, and today they are deliberately scheduled after an enforcement pass. Either the split sanctions an enforce → optimize → enforce alternation, or it establishes the convention that optimizer rules read requirements, never materialized operators.
• Cycle detection, not just "unchanged since last pass". We have a measured case where an enforcement rule oscillates: two adjacent passes each undo the other's change, so the plan alternates between two forms without stabilizing. The logical loop already handles this by keeping a HashSet of every prior pass signature and stopping on first revisit; the physical loop should copy that, not compare only against the previous pass.
• Classification of the built-in rules into the two traits, and the compatibility story for downstream chains that currently splice enforcement rules in by position.

Describe alternatives you've considered

#25356 (skip a named rule handed a plan it was seen to leave alone) treats the symptom inside the once-through model. Review there converged on the same conclusion as #25355: the phase split is the better home, and the fingerprint work migrates into the convergence check.

Additional context

• Physical optimizer re-runs rules on plans they have already settled #25355
• feat(core): skip a named rule handed a plan it has been seen to leave alone #25356

[alamb]

Design questions to settle

• Phase topology. Is the shape a fixed optimize-loop → enforce, or can a chain declare an alternation? Some downstream rules consume the operators enforcement materializes (for example, removing an enforced sort by reorganizing what runs beneath it) rather than the abstract requirement, and today they are deliberately scheduled after an enforcement pass. Either the split sanctions an enforce → optimize → enforce alternation, or it establishes the convention that optimizer rules read requirements, never materialized operators.
• Classification of the built-in rules into the two traits, and the compatibility story for downstream chains that currently splice enforcement rules in by position.

I think this is saying that some downstream crates might rely on the fact they can add EnforceDistribution into their chain of operations and it would run.

I think we can handle this a few ways:

1. Leave a (deprecated) EnforceDistributon OptimizerRule in place that calls into the new EnforceRequirementsAnalyzer rule so they can keep using it for a while
2. Document what each pass should be doing / what rules it needs to follow to get the ExecutionPlans to work. For example, if downstream users currently splice in EnforceRequirements to make sure the output of their passes is valid, we should document that any pass that makes an invalid plan should run EnforceRequirements again itself (or not make the invalid plan)

@wiedld and I tried to add something like this with the Invariant API https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#method.check_invariants that could tell you if some invariant isn't being followed. However I don't think that idea got very far and is still only used a little

• Cycle detection, not just "unchanged since last pass". We have a measured case where an enforcement rule oscillates: two adjacent passes each undo the other's change, so the plan alternates between two forms without stabilizing. The logical loop already handles this by keeping a HashSet of every prior pass signature and stopping on first revisit; the physical loop should copy that, not compare only against the previous pass.

I agree this sounds like a goodplan

[zhuqi-lucas]

Thanks @alamb!

On the compatibility options: both work for the producer direction, and the deprecated shim (option 1) is what would keep a downstream chain like ours running unmodified through the migration. One clarification though, because the case I was describing is the opposite direction: not "my pass leaves the plan invalid, who cleans up", but "my pass only finds work on an enforced plan". Think of a rule that eliminates enforced sorts by reorganizing what runs beneath them: it pattern-matches the very operators enforcement materializes, produces nothing invalid itself, and simply finds no work at all if it runs before enforcement. "Any pass that makes an invalid plan should run EnforceRequirements itself" places the duty on producers, and consumers have no invalid plan to answer for. For that class I only see three shapes: a sanctioned enforce → optimize → enforce alternation, a convention that optimizer rules read requirements rather than materialized operators (which is a real rewrite of such rules, we have one and it is not small), or the consumer running enforcement itself, which re-derives requirements once per consumer and is the cost this whole effort wants to remove.

On check_invariants: I think this is the piece that makes the enforcement phase cheap, and maybe its second life. An enforcement pass that finds nothing to do still pays a full-tree requirement re-derivation today, which is most of what our six passes spend (4 of 6 are byte-identical no-ops). If the enforcement phase is gated on check_invariants (invariants hold, skip the pass entirely), the no-op case drops from "re-derive everything" to one linear walk, and it is a semantic check rather than the textual fingerprint #25356 had to use for the same purpose. That would also give the Invariant API the adoption it did not get the first time: the phase boundary is a natural, always-on call site. It likely needs the invariant coverage audited first (distribution and ordering requirements would have to be fully expressible as invariants), which could be its own preparatory issue.

On sequencing: since the cycle-detection shape is agreed, I would propose implementing in three steps so each lands reviewable:

1. PhysicalPlanSignature as a type plus its boundary tests (what identity sees and deliberately does not see: partitioning/ordering/equivalences in, statistics out; the tests from feat(core): skip a named rule handed a plan it has been seen to leave alone #25356 port over wholesale);
2. the convergence loop over the existing optimizer list behind a config, using the signature HashSet;
3. the trait split with the deprecated shims from your option 1, plus the pass-contract documentation from your option 2.

I am working on the POC first.

[zhuqi-lucas]

Since my last comment I built the POC end to end, and the findings changed the picture enough that this supersedes the sequencing I proposed above. Short version: I think there are three separable problems here, and the empirical findings assign each one a different status. Draft PR with everything below: #25585.

Problem A: chains hand-repeat enforcement because there is no phase structure. Your diagnosis, and the POC supports it: the phase split removes the reason most of those repeats exist. This is the structural track and it stays the main body of this issue.

Problem B: the once-through list depends on hand-ordering, and a loop would fix that. This is the convergence idea, and the POC says it is premature, not wrong. Rerunning rules speculatively means f(f(x)) in an ecosystem that has only ever exercised f(g(x)) (re-enforce after someone else's rewrite; even the default chain's own duplicated ProjectionPushdown/FilterPushdown entries are that pattern). Testing the idempotence this requires falsified every rule it touched:

1. FilterPushdown re-application conjuncts the same predicate again (k < 10 AND k < 10 AND k < 10; 63 slt failures, one signature).
2. A second PushdownSort application returns wrong rows (sort_pushdown.slt:2487: 4,5,6 for an ORDER BY id ASC LIMIT 3 whose answer is 1,2,3). Probably reachable today by any chain listing the rule twice; filing separately with the repro.
3. Even EnsureRequirements, the one rule whose specification promises idempotence, changed 35 plans across eight slt files when declared: a second application rewrites sort-preserving merges to other members of the ordering equivalence class, once replacing a CoalescePartitionsExec with a costlier merge.

So in the PR the loop exists per call site behind fn idempotent(&self) -> bool { false } on the rule trait, and nothing declares it: the machinery is built, mutation-tested, and dormant, and each future declaration is gated on a fix plus a regression test. The declaration being the rule's own method follows the same logic that (rightly) rejected the config key in #25356: how often a rule runs is the chain's property, but whether re-application is safe is the rule's.

Problem C: authored repeats that survive both A and B still mostly re-derive settled plans. Some downstream rules consume the operators enforcement materializes, so at least one enforce-optimize-enforce alternation outlives the phase split, and its enforcement calls are usually no-ops (on the chain that motivated this: 4 of 6 calls byte-identical, 79% of physical optimization time). The property this needs is not idempotence but determinism, which survives everything above: a rule that oscillates is deterministic, records no fixpoint, and is simply never skipped. Hence the second trait method, fn deterministic(&self) -> bool { false }: a declaring rule handed a plan it has been observed to leave unchanged in this run is skipped outright (pointer fast path, then a full-fingerprint compare; full rather than hashed, since a collision would skip an enforcement pass that had work to do). EnsureRequirements declares this one, and it holds at every level. Upstream, the full sqllogictest suite is green. Downstream, a live A/B on the very endpoint that motivated this work, two server builds side by side on the same host and the same data at the same moment, one with the declaration and one without: the six authored enforcement calls drop to three executed, physical optimization wall goes from 87.0 ms to 48.5 ms (-44%), and the two builds' rendered plans are byte-identical (74,964 bytes each). A unit test over the same chain pins the mechanics and one integration lesson worth writing down: instrumentation wrappers must delegate deterministic(), or the skip dies silently. This is #25356's mechanism re-homed onto the rule, where I now think it always belonged, and it is the piece that solves the production latency problem that started this whole thread for us: no longer a projection, measured live.

Sequencing I would propose: (1) PhysicalPlanSignature + deterministic() + the skip as the first real PR, since it has a user today and zero contract exposure; (2) the dormant idempotent()/convergence machinery lands with the phase-split design work, waking rule by rule as fixes land (FilterPushdown predicate dedup, the PushdownSort bug, canonicalizing ER's equivalence-class rewrite); (3) the PhysicalAnalyzerRule split continues here as the structural track.

@alamb what do you think of the deterministic() approach? If the shape looks right to you I will polish #25585 and mark it ready for review.