Physical optimizer re-runs rules on plans they have already settled #25355

Is your feature request related to a problem or challenge?

DataFusion's two optimizers take opposite positions on repeated rule execution, and the physical side is the one without a mechanism.

The logical optimizer iterates (max_passes, default 3) and detects convergence: it records a LogicalPlanSignature (node count + plan hash) per pass and stops once a plan repeats (datafusion/optimizer/src/optimizer.rs).

The physical optimizer runs its rule list exactly once, in a hand-ordered sequence, with no iteration, no convergence detection, and no way to recognize that a rule is being handed a plan it has already settled. That works for the default chain, which is carefully ordered so that everything able to invalidate distribution/ordering requirements runs before the single EnsureRequirements — the per-rule comments in physical-optimizer/src/optimizer.rs say so explicitly.

It stops working as soon as a chain is assembled with rules after that point. DataFusion is a library, with_physical_optimizer_rules is a supported extension point, and a rule inserted late (a custom scan rewrite, a distributed-execution boundary, an MV substitution) invalidates requirements again and needs another enforcement pass behind it.

Measured on a chain that enforces requirements six times, over a 34-node plan:

• Of the six enforcement passes in that chain, four leave the plan byte-identical and still pay the full tree walk, about 30ms each in a debug build. That is roughly 122ms of a 209ms physical optimization phase spent arriving back at the same plan.
• Across 31 EnsureRequirements invocations observed in the process, covering several queries, 23 left the plan byte-identical.
• A mechanism like the one below recovers two of the four, 61ms measured, not all four: after anything changes the plan, the first pass that finds nothing to do still has to run in order to establish that the plan is a fixpoint.

Describe the solution you'd like

Let the chain's owner name the rules whose repeats may be answered from what was already observed, and have the optimizer remember, per planning run, the plans each named rule has been seen to leave untouched.

datafusion.optimizer.skip_unchanged_physical_rules = "EnsureRequirements"

Per pass, for a named rule: if the plan it is about to be given is one this rule has already returned unchanged, return it; otherwise run the rule, and record the plan only if the rule gave that same plan back.

Three properties matter, and two of them were learned the hard way:

1. Key on the plan the rule was given, not the plan it last returned. The tempting version skips when a rule is handed back its own output, on the assumption that a rule cannot improve on itself. That assumption is false: a rule is not required to reach its fixpoint in one pass, and EnsureRequirements routinely does not. Two consecutive applications both changed the plan in my measurements, the second moving a RepartitionExec below a SortExec and switching that sort to per-partition. Recording only proven fixpoints makes a skip a replay of an observed outcome rather than a prediction.

2. Compare plans by content, not by pointer. A rule that changes nothing still commonly rebuilds the tree: of the 23 no-op calls above, only 7 also returned the input object. Pointer identity cannot see a fixpoint. This is not a reporting problem that could be fixed in the rule: replace_children_if_necessary already returns the original plan when the child pointers are unchanged, so the passes that really did nothing already preserve identity. The other 16 lost it because the phases inside the rule genuinely rewrote the tree and then rewrote it back, which is EnsureRequirements: distribution and sorting phases undo each other on most calls #25360.

3. Scope it to one planning run. Rule instances are shared between queries, so this must not live on the rule; and keeping it per-run keeps the config out of the key, since it cannot change midway through a run.

Debug builds re-run a skipped rule and assert it still leaves the plan alone, so a rule that does not depend only on the plan and the config fails a test rather than a query. Spark is the only surveyed engine that checks this (RuleExecutor.checkBatchIdempotence under Utils.isTesting); engines relying on counters or an Optional instead have public incidents from non-idempotent rules looping (trinodb/trino#11559, prestodb/presto#9362).

Default behaviour is unchanged, since the option is empty.

This is not a DataFusion-specific shape. DuckDB's pipeline runs CTE_INLINING, UNUSED_COLUMNS and COLUMN_LIFETIME twice with no dedup at all. Spark Catalyst runs the same ~30-rule set twice around "Infer Filters". The difference is that the others grew a mechanism for it:

Describe alternatives you've considered

• A defaulted trait method, fn skip_if_unchanged(&self) -> bool { false }, letting each rule declare for itself. Wrong object: whether a chain repeats a rule is a property of the chain, not of the rule, and upstream cannot know what a downstream chain and node types will do to a rule's behaviour. It also carries a trap, since the optimizer asks only the outermost rule, so a rule wrapped for timing or tracing must forward the answer and a wrapper that leaves the default in place disables the feature silently. That is the same shape as Wrapper PhysicalOptimizerRules silently drop the schema_check() of the rules they wrap #25316, where a wrapper drops schema_check(). Matching on names needs no cooperation from a wrapper, which already reports the name it wraps.
• Iterating the whole chain to a fixpoint, like the logical optimizer. Closer to the logical side's shape, but it changes when every existing rule runs, which is a much larger behavioural change than making redundant repeats cheap.
• ClickHouse's update_depth. Strictly more information and it bounds re-traversal precisely, but it changes the signature of every rule.

Additional context

Implemented in #25356.

Three problems found in EnsureRequirements while measuring this are filed separately, since each stands on its own:

• EnsureRequirements does not reach a fixpoint in one application #25362 — it does not reach a fixpoint in one application, which is what makes "skip when a rule is handed back its own output" unsafe.
• EnsureRequirements: distribution and sorting phases undo each other on most calls #25360 — its distribution and sorting phases undo each other on 16 of 23 no-op calls, which is where most of the redundant work comes from.

Fixing #25360 would remove much of the need for this feature, and would also let the comparison here fall back to pointers for the cases it covers. It is not a prerequisite for this PR.