EnsureRequirements: distribution and sorting phases undo each other on most calls

[zhuqi-lucas]

Describe the bug

Inside one EnsureRequirements::optimize call, the distribution phase and the sorting phase frequently both rewrite the plan and arrive back where they started. The call does two full tree rewrites and returns an equivalent plan.

I instrumented every phase of EnsureRequirements::optimize (physical-optimizer/src/ensure_requirements/mod.rs) to report whether it changed the rendered plan, and ran a real 34-node query plan through a chain that enforces requirements six times. Over 31 calls:

  16x  net=unchanged  phases that fired: (2a ensure_distribution, 2b ensure_sorting)
   7x  net=unchanged  phases that fired: ()
   4x  net=CHANGED    phases that fired: (2a, 2b, 3b replace_with_order_preserving_variants)
   4x  net=CHANGED    phases that fired: (2a, 2b)

So of 23 calls that changed nothing overall, 16 got there by having 2a rewrite the plan and 2b rewrite it back. Only 7 were genuinely quiet.

A second, independent measurement agrees. Comparing each call's input against its output by pointer as well as by content:

   8x  content=diff  pointer=diff     real work
  16x  content=SAME  pointer=diff     changed nothing, still rebuilt the tree
   7x  content=SAME  pointer=SAME     genuinely untouched

The 16 and the 7 line up exactly with the phase data.

To Reproduce

Reproduces on main with built-in types only. Drops straight into datafusion/core/tests/physical_optimizer/ensure_requirements.rs, which already has MockMultiPartitionExec and sort_expr_on:

#[test]
fn phases_cancel() {
    let config = ConfigOptions::default();
    let rule = EnsureRequirements::new();

    // SinglePartition + ordering required, over a source with 8 partitions
    // that already claims that ordering.
    let plan = Arc::new(OutputRequirementExec::new(
        Arc::new(MockMultiPartitionExec::new(8)),
        Some(OrderingRequirements::from(sort_expr_on("a", 0, true, true))),
        Distribution::SinglePartition,
        None,
    )) as Arc<dyn ExecutionPlan>;

    let settled = rule.optimize(plan, &config).unwrap();
    let again = rule.optimize(Arc::clone(&settled), &config).unwrap();

    // The second call describes the same plan ...
    assert_eq!(
        displayable(settled.as_ref()).indent(true).to_string(),
        displayable(again.as_ref()).indent(true).to_string(),
    );
    // ... but it is a different object: the tree was rebuilt to get back
    // where it started.
    assert!(!Arc::ptr_eq(&settled, &again));
}

Both assertions hold today. Instrumenting the phase boundaries inside optimize shows why: on each of those calls,

PHASE 2a-ensure_distribution CHANGED
PHASE 2b-ensure_sorting      CHANGED
→ net: plan byte-identical

The same shape without the OutputRequirementExec, or with a single-partition source, settles to pointer identity, so this is specific to the two phases interacting.

Expected behavior

A call that changes nothing should not rewrite the plan twice to get there. Beyond the wasted work, this is what makes the rule's output a fresh object every time, which costs every caller that wants to know whether anything happened.

Relatedly, the rule does not reach its fixpoint in one pass: two consecutive applications both changed the plan in these measurements, the second moving a RepartitionExec below a SortExec and switching that sort to per-partition. A chain that enforces requirements once after its own rewrites therefore may not be getting a settled plan.

Additional context

Worth noting what this is not: it is not the rule misreporting whether it transformed. replace_children_if_necessary returns the original plan when the child pointers are unchanged, so a pass that truly does nothing already keeps its input. These 16 lost identity because the tree really was rewritten twice.

Found while measuring #25355 / #25356. In that plan, four of the chain's six enforcement passes leave it byte-identical, about 30ms each in a debug build, so roughly 122ms of a 209ms physical optimization phase is spent arriving back at the same plan. Most of that is this issue rather than the passes being unnecessary: the passes themselves are there because each follows a rewrite that may or may not have fired, but a pass that finds nothing to do should not cost two full tree rewrites to establish it.