Add EnsureRequirements: merged EnforceDistribution + EnforceSorting with idempotent pushdown_sorts

[zhuqi-lucas]

Summary

Replace the separate EnforceDistribution and EnforceSorting optimizer rules with a single EnsureRequirements rule in the default optimizer chain. Fix pushdown_sorts to be distribution-aware and fix EnforceDistribution fetch preservation (#14150), making the composition idempotent.

Epic: #21973

Problem

EnforceDistribution and EnforceSorting run as separate rules, but sorting and distribution are coupled through SortExec.preserve_partitioning. This caused:

1. Production 502 errors: pushdown_sorts set preserve_partitioning=true on multi-partition input without inserting SortPreservingMergeExec, violating SinglePartition requirements from GlobalLimitExec → SanityCheckPlan failure.

2. Non-idempotent composition: Running the rules multiple times produced different (sometimes invalid) plans.

3. Lost fetch values (Bug: applying multiple times EnforceDistribution generates invalid plan #14150): EnforceDistribution dropped fetch from SortPreservingMergeExec/CoalescePartitionsExec when stripping and re-adding distribution operators.

DataFusion was the only major query engine with separate rules — Spark (EnsureRequirements) and Presto/Trino (AddExchanges) handle both in a single rule.

Changes

1. EnsureRequirements rule (new, replaces default chain)

• Composes EnforceDistribution::optimize() + EnforceSorting::optimize() in a single rule
• Replaces Arc::new(EnforceDistribution) + Arc::new(EnforceSorting) in the default optimizer chain
• 53 comprehensive tests covering all known bug topologies + idempotency verification

2. Distribution-aware pushdown_sorts (sort_pushdown.rs)

• Add distribution_requirement: Distribution field to ParentRequirements
• New add_sort_above_with_distribution() in utils.rs — inserts SortPreservingMergeExec when parent requires SinglePartition and input has multiple partitions
• Switch both add_sort_above call sites to distribution-aware variant
• Propagate distribution through recursion with stronger_distribution() helper
• Reset distribution below partition-merging nodes (SPM, single-partition outputs)

3. Fix EnforceDistribution fetch preservation (#14150)

• remove_dist_changing_operators() now saves fetch from removed SPM/Coalesce
• add_merge_on_top() re-applies saved fetch to re-created operators

4. Updated SLT

• explain.slt: Two optimizer rule names (EnforceDistribution, EnforceSorting) become one (EnsureRequirements) in EXPLAIN VERBOSE output

Testing

Suite	Result
EnsureRequirements (new)	53 passed
enforce_sorting (existing)	124 passed, 0 regressions
enforce_distribution (existing)	66 passed, 0 regressions
SLT (465 files)	1 pre-existing failure only
Total	243 unit + 464 SLT = 0 new failures

Idempotency coverage

Scenario	Verified
Multi-partition sort + limit (2,4,8,16,32,64 partitions)	✅
Union with mixed partition counts	✅
Projection over multi-partition	✅
HashJoin (Partitioned)	✅
SortMergeJoin	✅
Window function partitioning + ordering	✅
Aggregate (Partial + FinalPartitioned)	✅
Nested sort + limit	✅
Hash repartition + sort	✅
CoalescePartitions + sort (parallelize_sorts)	✅
SPM → Sort → multi-partition	✅
PR #53 scenario (OutputRequirementExec + SinglePartition)	✅
PR #54 scenario (ProjectionExec + multi-partition)	✅
#14150 scenario (fetch preservation across passes)	✅
All partition counts 1-64 sweep	✅
Triple optimization convergence	✅
10x consecutive optimization stability	✅
EnforceDistribution::optimize twice (sort+limit)	✅
EnforceDistribution::optimize twice (HashJoin)	✅

Architecture

EnsureRequirements::optimize(plan)
  Step 1: EnforceDistribution::optimize(plan)
    - Join key reordering (top-down)
    - Distribution enforcement (bottom-up)
    - Fetch preservation on SPM/Coalesce removal (#14150 fix)
  Step 2: EnforceSorting::optimize(plan)
    - ensure_sorting (bottom-up)
    - parallelize_sorts (bottom-up)
    - replace_with_order_preserving_variants (bottom-up)
    - pushdown_sorts (top-down, distribution-aware)
    - replace_with_partial_sort (bottom-up)

Idempotent because:

• pushdown_sorts now carries distribution_requirement and uses add_sort_above_with_distribution
• EnforceDistribution preserves fetch across strip/re-add cycles
• Running EnsureRequirements twice produces identical plans (verified by 53 tests)

Next Steps (future PRs)

• Gradually implement pushdown_sorts optimizations in the bottom-up ensure_sorting pass
• Eliminate pushdown_sorts top-down pass entirely
• Single-pass architecture (one transform_up for both distribution + sorting, like Spark)

Closes: #14150