From 5b79e9889b0e6867e89c30209059d4bfd10dcb40 Mon Sep 17 00:00:00 2001
From: Jay Zhan <jayzhan211@gmail.com>
Date: Sun, 1 Nov 2026 08:25:39 +0800
Subject: [PATCH 1/3] feat: physical optimizer rules declare whether they run
 before or after enforcement

---
 datafusion/core/src/physical_planner.rs       | 219 ++++++++++++++++--
 .../physical_optimizer/sanity_checker.rs      |  58 ++++-
 datafusion/ffi/src/physical_optimizer.rs      |  69 +++++-
 .../ffi/src/tests/physical_optimizer.rs       |  12 +-
 .../src/aggregate_statistics.rs               |   6 +-
 .../src/combine_partial_final_agg.rs          |   6 +-
 .../src/eliminate_constant_join.rs            |   6 +-
 .../physical-optimizer/src/ensure_coop.rs     |   6 +-
 .../src/ensure_requirements/mod.rs            |   6 +-
 .../physical-optimizer/src/filter_pushdown.rs |   9 +-
 .../src/hash_join_buffering.rs                |   6 +-
 .../physical-optimizer/src/join_selection.rs  |   6 +-
 datafusion/physical-optimizer/src/lib.rs      |   5 +-
 .../physical-optimizer/src/limit_pushdown.rs  |   6 +-
 .../src/limit_pushdown_past_window.rs         |   6 +-
 .../src/limited_distinct_aggregation.rs       |   6 +-
 .../physical-optimizer/src/optimizer.rs       |  53 ++++-
 .../src/output_requirements.rs                |   9 +-
 .../src/projection_pushdown.rs                |   6 +-
 .../physical-optimizer/src/pushdown_sort.rs   |   6 +-
 .../physical-optimizer/src/sanity_checker.rs  |   6 +-
 .../src/topk_aggregation.rs                   |   6 +-
 .../src/topk_repartition.rs                   |   6 +-
 .../src/update_aggr_exprs.rs                  |   6 +-
 .../physical-optimizer/src/window_topn.rs     |   6 +-
 datafusion/session/src/lib.rs                 |   3 +-
 datafusion/session/src/physical_optimizer.rs  |  30 +++
 .../src/test_context/custom_sort_pushdown.rs  |   7 +-
 28 files changed, 522 insertions(+), 54 deletions(-)

diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index 198e5bb7412b2..f7844dbb07da4 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -94,7 +94,6 @@ use datafusion_physical_expr::expressions::Literal;
 use datafusion_physical_expr::{
     LexOrdering, PhysicalSortExpr, create_physical_sort_exprs,
 };
-use datafusion_physical_optimizer::ensure_requirements::EnsureRequirements;
 use datafusion_physical_optimizer::sanity_checker::{
     check_plan_tree_sanity, input_requirements_hold,
 };
@@ -107,7 +106,9 @@ use datafusion_physical_plan::recursive_query::RecursiveQueryExec;
 use datafusion_physical_plan::scalar_subquery::{ScalarSubqueryExec, ScalarSubqueryLink};
 use datafusion_physical_plan::statistics::{StatisticsArgs, StatisticsContext};
 use datafusion_physical_plan::unnest::ListUnnest;
-use datafusion_session::{PhysicalOptimizerContext, PhysicalOptimizerRule, Session};
+use datafusion_session::{
+    PhysicalOptimizerContext, PhysicalOptimizerPhase, PhysicalOptimizerRule, Session,
+};
 
 use async_trait::async_trait;
 use datafusion_physical_plan::async_func::{AsyncFuncExec, AsyncMapper};
@@ -3450,6 +3451,7 @@ impl DefaultPhysicalPlanner {
     {
         let plan = bind_materialized_cte_scans(plan, session_state)?;
         let optimizers = session_state.physical_optimizers();
+        let analyzer_rules = analyzer_rule_count(optimizers)?;
         debug!(
             "Input physical plan:\n{}\n",
             displayable(plan.as_ref()).indent(false)
@@ -3494,6 +3496,7 @@ impl DefaultPhysicalPlanner {
                 &new_plan,
                 plan,
                 optimizers,
+                analyzer_rules,
                 &optimizer_context,
             ));
         }
@@ -3706,19 +3709,50 @@ fn tuple_err<T, R>(value: (Result<T>, Result<R>)) -> Result<(T, R)> {
     }
 }
 
+/// Returns how many rules at the start of `rules` are
+/// [`PhysicalOptimizerPhase::Analyzer`] rules. Fails when an analyzer rule
+/// comes after an optimizer rule: optimizer rules may assume that the plan
+/// meets its requirements, which only holds after the last analyzer rule.
+fn analyzer_rule_count(
+    rules: &[Arc<dyn PhysicalOptimizerRule + Send + Sync>],
+) -> Result<usize> {
+    let is_analyzer = |rule: &&Arc<dyn PhysicalOptimizerRule + Send + Sync>| {
+        rule.phase() == PhysicalOptimizerPhase::Analyzer
+    };
+    let count = rules.iter().take_while(is_analyzer).count();
+    if let Some(late) = rules[count..].iter().position(|rule| is_analyzer(&rule)) {
+        let early = rules[count].name();
+        return plan_err!(
+            "Physical optimizer rule '{early}' (position {}) is an optimizer rule \
+             but runs before analyzer rule '{}' (position {}). Optimizer rules \
+             assume the plan meets its distribution and ordering requirements, so \
+             they must run after every analyzer rule. Move '{early}' after the \
+             analyzer rules. Return PhysicalOptimizerPhase::Analyzer from its \
+             phase() only if it is written to run before the requirements are \
+             enforced",
+            count + 1,
+            rules[count + late].name(),
+            count + late + 1
+        );
+    }
+    Ok(count)
+}
+
 /// Adds context to an error from the final [`check_plan_tree_sanity`].
 ///
-/// `EnsureRequirements` makes the order and distribution requirements hold,
-/// and every rule after it must keep them. When they do not hold for
-/// `final_plan`, the context names the rule that broke this contract: either
-/// `EnsureRequirements` itself, or the rule after which the requirements
-/// stopped holding. To find it, `optimizers` run again from `input_plan`, so
-/// planning that succeeds keeps no intermediate plans.
+/// The last of the first `analyzer_rules` rules, `EnsureRequirements` in the
+/// default list, makes the order and distribution requirements hold, and every
+/// rule after it must keep them. When they do not hold for `final_plan`, the
+/// context names the rule that broke this contract: either that last analyzer
+/// rule, or the rule after which the requirements stopped holding. To find
+/// it, `optimizers` run again from `input_plan`, so planning that succeeds
+/// keeps no intermediate plans.
 fn sanity_check_error(
     error: DataFusionError,
     final_plan: &Arc<dyn ExecutionPlan>,
     input_plan: Arc<dyn ExecutionPlan>,
     optimizers: &[Arc<dyn PhysicalOptimizerRule + Send + Sync>],
+    analyzer_rules: usize,
     optimizer_context: &dyn PhysicalOptimizerContext,
 ) -> DataFusionError {
     const SANITY_CHECK: &str = "SanityCheckPlan";
@@ -3729,13 +3763,10 @@ fn sanity_check_error(
         return plain_error(error);
     }
 
-    let ensure_requirements = EnsureRequirements::new();
-    let enforced = optimizers
-        .iter()
-        .rposition(|rule| rule.name() == ensure_requirements.name());
-    // Plans before `EnsureRequirements` need not satisfy the requirements, so
-    // only plans from its last run onwards are checked. Without it, every
-    // plan is checked, starting with the input plan.
+    let enforced = analyzer_rules.checked_sub(1);
+    // Plans before the last analyzer rule need not satisfy the requirements,
+    // so only plans from its output onwards are checked. Without analyzer
+    // rules, every plan is checked, starting with the input plan.
     let first_checked = enforced.unwrap_or(0);
     let mut held = enforced.is_none() && input_requirements_hold(&input_plan);
     let mut broken_by = None;
@@ -3760,10 +3791,10 @@ fn sanity_check_error(
              after physical optimizer rule '{}'",
             rule.name()
         ),
-        (None, Some(_)) => format!(
+        (None, Some(idx)) => format!(
             "{SANITY_CHECK}: the plan produced by physical optimizer rule '{}' \
              does not satisfy the order and distribution requirements",
-            ensure_requirements.name()
+            optimizers[idx].name()
         ),
         (None, None) => format!(
             "{SANITY_CHECK}: no physical optimizer rule produced a plan that \
@@ -3980,6 +4011,10 @@ mod tests {
         fn schema_check(&self) -> bool {
             true
         }
+
+        fn phase(&self) -> PhysicalOptimizerPhase {
+            PhysicalOptimizerPhase::Optimizer
+        }
     }
 
     #[derive(Debug)]
@@ -4250,6 +4285,152 @@ mod tests {
         Ok(())
     }
 
+    /// Leaves the plan unchanged and declares the phase it is given.
+    #[derive(Debug)]
+    struct NoopRule(PhysicalOptimizerPhase);
+
+    impl PhysicalOptimizerRule for NoopRule {
+        fn optimize(
+            &self,
+            plan: Arc<dyn ExecutionPlan>,
+            _config: &ConfigOptions,
+        ) -> Result<Arc<dyn ExecutionPlan>> {
+            Ok(plan)
+        }
+
+        fn name(&self) -> &str {
+            "noop"
+        }
+
+        fn schema_check(&self) -> bool {
+            true
+        }
+
+        fn phase(&self) -> PhysicalOptimizerPhase {
+            self.0
+        }
+    }
+
+    /// The default rules followed by `rule`.
+    async fn plan_with_rule_in_front(rule: NoopRule) -> Result<Arc<dyn ExecutionPlan>> {
+        let mut rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>> =
+            vec![Arc::new(rule)];
+        rules.extend(
+            datafusion_physical_optimizer::optimizer::PhysicalOptimizer::new().rules,
+        );
+        let session_state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_physical_optimizer_rules(rules)
+            .build();
+        let logical_plan = LogicalPlanBuilder::empty(false).build()?;
+        session_state.create_physical_plan(&logical_plan).await
+    }
+
+    /// An optimizer rule in front of the analyzer rules fails planning, and
+    /// the error names both rules.
+    #[tokio::test]
+    async fn analyzer_rule_after_optimizer_rule_fails_planning() {
+        let error = plan_with_rule_in_front(NoopRule(PhysicalOptimizerPhase::Optimizer))
+            .await
+            .unwrap_err()
+            .to_string();
+        assert!(
+            error.contains(
+                "Physical optimizer rule 'noop' (position 1) is an optimizer rule \
+                 but runs before analyzer rule 'OutputRequirements' (position 2)"
+            ),
+            "{error}"
+        );
+    }
+
+    /// The default rules with `LimitPushdown` moved to just before
+    /// `EnsureRequirements`, where it would see a plan that does not meet its
+    /// requirements yet, fail planning.
+    #[tokio::test]
+    async fn limit_pushdown_before_enforcement_fails_planning() -> Result<()> {
+        let mut rules =
+            datafusion_physical_optimizer::optimizer::PhysicalOptimizer::new().rules;
+        let limit_pushdown = rules
+            .iter()
+            .position(|rule| rule.name() == "LimitPushdown")
+            .unwrap();
+        let limit_pushdown = rules.remove(limit_pushdown);
+        let enforcement = rules
+            .iter()
+            .position(|rule| rule.name() == "EnsureRequirements")
+            .unwrap();
+        rules.insert(enforcement, limit_pushdown);
+        let session_state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_physical_optimizer_rules(rules)
+            .build();
+        let logical_plan = LogicalPlanBuilder::empty(false).build()?;
+        let error = session_state
+            .create_physical_plan(&logical_plan)
+            .await
+            .unwrap_err()
+            .to_string();
+        assert!(error.contains("'LimitPushdown' (position 8"), "{error}");
+        assert!(
+            error.contains("'EnsureRequirements' (position 9"),
+            "{error}"
+        );
+        Ok(())
+    }
+
+    /// Running `EnsureRequirements` again after the optimizer rules fails
+    /// planning.
+    #[tokio::test]
+    async fn enforcement_after_optimizer_rules_fails_planning() -> Result<()> {
+        let session_state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_physical_optimizer_rule(Arc::new(
+                datafusion_physical_optimizer::ensure_requirements::EnsureRequirements::new(),
+            ))
+            .build();
+        let logical_plan = LogicalPlanBuilder::empty(false).build()?;
+        let error = session_state
+            .create_physical_plan(&logical_plan)
+            .await
+            .unwrap_err()
+            .to_string();
+        assert!(
+            error.contains("'EnsureRequirements' (position 22"),
+            "{error}"
+        );
+        Ok(())
+    }
+
+    /// An analyzer rule appended after the default rules runs after the
+    /// optimizer rules, so planning fails.
+    #[tokio::test]
+    async fn appended_analyzer_rule_fails_planning() -> Result<()> {
+        let session_state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_physical_optimizer_rule(Arc::new(NoopRule(
+                PhysicalOptimizerPhase::Analyzer,
+            )))
+            .build();
+        let logical_plan = LogicalPlanBuilder::empty(false).build()?;
+        let error = session_state
+            .create_physical_plan(&logical_plan)
+            .await
+            .unwrap_err()
+            .to_string();
+        assert!(
+            error.contains("is an optimizer rule but runs before analyzer rule 'noop'"),
+            "{error}"
+        );
+        Ok(())
+    }
+
+    /// The same rule declared as an analyzer rule may run before enforcement.
+    #[tokio::test]
+    async fn analyzer_rule_before_enforcement_plans() -> Result<()> {
+        plan_with_rule_in_front(NoopRule(PhysicalOptimizerPhase::Analyzer)).await?;
+        Ok(())
+    }
+
     async fn aggregate_explain(logical_plan: &LogicalPlan) -> Result<String> {
         let physical_plan = plan(logical_plan).await?;
         Ok(displayable(physical_plan.as_ref()).indent(true).to_string())
@@ -5934,6 +6115,10 @@ digraph {
         fn schema_check(&self) -> bool {
             true
         }
+
+        fn phase(&self) -> PhysicalOptimizerPhase {
+            PhysicalOptimizerPhase::Optimizer
+        }
     }
 
     #[test]
diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
index 3fac55836c292..ac6acb4886eda 100644
--- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs
+++ b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
@@ -24,8 +24,9 @@ use datafusion_common::{JoinType, NullEquality, Result, ScalarValue};
 use datafusion_physical_expr::expressions::{Literal, col};
 use datafusion_physical_expr::{Partitioning, RangePartitioning, SplitPoint};
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
-use datafusion_physical_optimizer::PhysicalOptimizerRule;
+use datafusion_physical_optimizer::optimizer::PhysicalOptimizer;
 use datafusion_physical_optimizer::sanity_checker::check_plan_tree_sanity;
+use datafusion_physical_optimizer::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_physical_plan::joins::{StreamJoinPartitionMode, SymmetricHashJoinExec};
 use datafusion_physical_plan::repartition::RepartitionExec;
 use datafusion_physical_plan::sorts::sort::SortExec;
@@ -1033,6 +1034,36 @@ impl PhysicalOptimizerRule for RemoveSorts {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
+}
+
+/// [`RemoveSorts`] declared as an analyzer rule.
+#[derive(Debug)]
+struct RemoveSortsInAnalyzer;
+
+impl PhysicalOptimizerRule for RemoveSortsInAnalyzer {
+    fn optimize(
+        &self,
+        plan: Arc<dyn ExecutionPlan>,
+        config: &ConfigOptions,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        RemoveSorts.optimize(plan, config)
+    }
+
+    fn name(&self) -> &str {
+        "remove_sorts"
+    }
+
+    fn schema_check(&self) -> bool {
+        true
+    }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 async fn plan_window_query(state: SessionState) -> Result<Arc<dyn ExecutionPlan>> {
@@ -1087,3 +1118,28 @@ async fn test_final_check_without_ensure_requirements() -> Result<()> {
     );
     Ok(())
 }
+
+/// The analyzer rules must leave a plan whose requirements hold. When they do
+/// not, the error names the last analyzer rule.
+#[tokio::test]
+async fn test_final_check_names_last_analyzer_rule() -> Result<()> {
+    let mut rules = PhysicalOptimizer::new().rules;
+    let analyzer_rules = rules
+        .iter()
+        .take_while(|rule| rule.phase() == PhysicalOptimizerPhase::Analyzer)
+        .count();
+    rules.insert(analyzer_rules, Arc::new(RemoveSortsInAnalyzer));
+    let state = SessionStateBuilder::new()
+        .with_default_features()
+        .with_physical_optimizer_rules(rules)
+        .build();
+    let error = plan_window_query(state).await.unwrap_err().to_string();
+    assert!(
+        error.starts_with(
+            "SanityCheckPlan: the plan produced by physical optimizer rule \
+             'remove_sorts' does not satisfy the order and distribution requirements"
+        ),
+        "{error}"
+    );
+    Ok(())
+}
diff --git a/datafusion/ffi/src/physical_optimizer.rs b/datafusion/ffi/src/physical_optimizer.rs
index cc8ca4c6dbd16..6aa4c21cf3382 100644
--- a/datafusion/ffi/src/physical_optimizer.rs
+++ b/datafusion/ffi/src/physical_optimizer.rs
@@ -8,7 +8,9 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::error::Result;
-use datafusion_physical_optimizer::{PhysicalOptimizerContext, PhysicalOptimizerRule};
+use datafusion_physical_optimizer::{
+    PhysicalOptimizerContext, PhysicalOptimizerPhase, PhysicalOptimizerRule,
+};
 use datafusion_physical_plan::ExecutionPlan;
 use stabby::string::String as SString;
 use tokio::runtime::Handle;
@@ -265,6 +267,14 @@ impl FFI_PhysicalOptimizerRule {
 /// no guarantees about being able to access the data in `private_data`. Any functions
 /// defined on this struct must only use the stable functions provided in
 /// FFI_PhysicalOptimizerRule to interact with the foreign rule.
+///
+/// A foreign rule reports [`PhysicalOptimizerPhase::Optimizer`]:
+/// `FFI_PhysicalOptimizerRule` does not
+/// carry the phase, so that its layout stays compatible with DataFusion 55. A
+/// rule that comes back to the library that created it is unwrapped and keeps
+/// its own phase.
+///
+/// [`PhysicalOptimizerPhase::Optimizer`]: datafusion_physical_optimizer::PhysicalOptimizerPhase::Optimizer
 #[derive(Debug)]
 pub struct ForeignPhysicalOptimizerRule {
     name: String,
@@ -335,6 +345,11 @@ impl PhysicalOptimizerRule for ForeignPhysicalOptimizerRule {
     fn schema_check(&self) -> bool {
         unsafe { (self.rule.schema_check)(&self.rule) }
     }
+
+    // `FFI_PhysicalOptimizerRule` does not carry the phase; see the type docs.
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 #[cfg(test)]
@@ -374,6 +389,10 @@ mod tests {
         fn schema_check(&self) -> bool {
             self.schema_check
         }
+
+        fn phase(&self) -> PhysicalOptimizerPhase {
+            PhysicalOptimizerPhase::Optimizer
+        }
     }
 
     /// A rule that returns an error from `optimize` but succeeds when
@@ -407,6 +426,36 @@ mod tests {
         fn schema_check(&self) -> bool {
             true
         }
+
+        fn phase(&self) -> PhysicalOptimizerPhase {
+            PhysicalOptimizerPhase::Optimizer
+        }
+    }
+
+    /// A rule that leaves the plan unchanged and declares the analyzer phase.
+    #[derive(Debug)]
+    struct AnalyzerNoOpRule;
+
+    impl PhysicalOptimizerRule for AnalyzerNoOpRule {
+        fn optimize(
+            &self,
+            plan: Arc<dyn ExecutionPlan>,
+            _config: &ConfigOptions,
+        ) -> Result<Arc<dyn ExecutionPlan>> {
+            Ok(plan)
+        }
+
+        fn name(&self) -> &str {
+            "analyzer_no_op_rule"
+        }
+
+        fn schema_check(&self) -> bool {
+            true
+        }
+
+        fn phase(&self) -> PhysicalOptimizerPhase {
+            PhysicalOptimizerPhase::Analyzer
+        }
     }
 
     fn create_test_plan() -> Arc<dyn ExecutionPlan> {
@@ -483,6 +532,24 @@ mod tests {
         Ok(())
     }
 
+    /// `FFI_PhysicalOptimizerRule` does not carry the phase: a foreign rule
+    /// reports the optimizer phase, while a rule that comes back to the
+    /// library that created it keeps its own phase.
+    #[test]
+    fn test_phase_across_ffi() {
+        let rule: Arc<dyn PhysicalOptimizerRule + Send + Sync> =
+            Arc::new(AnalyzerNoOpRule);
+
+        let local_ffi = FFI_PhysicalOptimizerRule::new(Arc::clone(&rule), None);
+        let local: Arc<dyn PhysicalOptimizerRule + Send + Sync> = (&local_ffi).into();
+        assert_eq!(local.phase(), PhysicalOptimizerPhase::Analyzer);
+
+        let mut foreign_ffi = FFI_PhysicalOptimizerRule::new(rule, None);
+        foreign_ffi.library_marker_id = crate::mock_foreign_marker_id;
+        let foreign: Arc<dyn PhysicalOptimizerRule + Send + Sync> = (&foreign_ffi).into();
+        assert_eq!(foreign.phase(), PhysicalOptimizerPhase::Optimizer);
+    }
+
     #[test]
     fn test_clone() -> Result<()> {
         let rule: Arc<dyn PhysicalOptimizerRule + Send + Sync> =
diff --git a/datafusion/ffi/src/tests/physical_optimizer.rs b/datafusion/ffi/src/tests/physical_optimizer.rs
index 4e1c0102a3454..0063062abe961 100644
--- a/datafusion/ffi/src/tests/physical_optimizer.rs
+++ b/datafusion/ffi/src/tests/physical_optimizer.rs
@@ -6,7 +6,9 @@ use std::sync::Arc;
 
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::error::Result;
-use datafusion_physical_optimizer::{PhysicalOptimizerContext, PhysicalOptimizerRule};
+use datafusion_physical_optimizer::{
+    PhysicalOptimizerContext, PhysicalOptimizerPhase, PhysicalOptimizerRule,
+};
 use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::limit::GlobalLimitExec;
 
@@ -33,6 +35,10 @@ impl PhysicalOptimizerRule for AddLimitRule {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 pub(crate) extern "C" fn create_physical_optimizer_rule() -> FFI_PhysicalOptimizerRule {
@@ -73,6 +79,10 @@ impl PhysicalOptimizerRule for ContextAwareAddLimitRule {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 pub(crate) extern "C" fn create_context_aware_optimizer_rule() -> FFI_PhysicalOptimizerRule
diff --git a/datafusion/physical-optimizer/src/aggregate_statistics.rs b/datafusion/physical-optimizer/src/aggregate_statistics.rs
index ab6a0e75c5a60..9d6ae6fa1a7d1 100644
--- a/datafusion/physical-optimizer/src/aggregate_statistics.rs
+++ b/datafusion/physical-optimizer/src/aggregate_statistics.rs
@@ -19,8 +19,8 @@ use datafusion_physical_plan::udaf::{
 use datafusion_physical_plan::{ExecutionPlan, expressions};
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
 use crate::optimizer::{ConfigOnlyContext, PhysicalOptimizerContext};
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 /// Optimizer that uses available statistics for aggregate functions
 #[derive(Default, Debug)]
@@ -108,6 +108,10 @@ impl PhysicalOptimizerRule for AggregateStatistics {
     fn schema_check(&self) -> bool {
         false
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 /// Returns an `AggregateExec` whose statistics can replace the aggregate with
diff --git a/datafusion/physical-optimizer/src/combine_partial_final_agg.rs b/datafusion/physical-optimizer/src/combine_partial_final_agg.rs
index 8f4e14caba0c7..3b810111aa687 100644
--- a/datafusion/physical-optimizer/src/combine_partial_final_agg.rs
+++ b/datafusion/physical-optimizer/src/combine_partial_final_agg.rs
@@ -11,7 +11,7 @@ use datafusion_common::error::Result;
 use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::aggregates::AggregateExec;
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{TransformedResult, TreeNode};
 
@@ -85,6 +85,10 @@ impl PhysicalOptimizerRule for CombinePartialFinalAggregate {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 // See tests in datafusion/core/tests/physical_optimizer
diff --git a/datafusion/physical-optimizer/src/eliminate_constant_join.rs b/datafusion/physical-optimizer/src/eliminate_constant_join.rs
index c9e476b993711..d55e4c9c9e336 100644
--- a/datafusion/physical-optimizer/src/eliminate_constant_join.rs
+++ b/datafusion/physical-optimizer/src/eliminate_constant_join.rs
@@ -20,7 +20,7 @@ use datafusion_physical_plan::placeholder_row::PlaceholderRowExec;
 use datafusion_physical_plan::projection::{ProjectionExec, ProjectionExpr};
 use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 /// Rewrites an inner [`NestedLoopJoinExec`] or a [`CrossJoinExec`] whose input
 /// provably produces one row of literals into a [`FilterExec`] over the other
@@ -81,6 +81,10 @@ impl PhysicalOptimizerRule for EliminateConstantJoin {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 /// Rewrites `join` if it is an inner join against a constant row and its filter
diff --git a/datafusion/physical-optimizer/src/ensure_coop.rs b/datafusion/physical-optimizer/src/ensure_coop.rs
index 0e7a6bd31afd1..238093aa14df9 100644
--- a/datafusion/physical-optimizer/src/ensure_coop.rs
+++ b/datafusion/physical-optimizer/src/ensure_coop.rs
@@ -10,7 +10,7 @@
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
@@ -112,6 +112,10 @@ impl PhysicalOptimizerRule for EnsureCooperative {
         // Wrapping a leaf in YieldStreamExec preserves the schema, so it is safe.
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 #[cfg(test)]
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/mod.rs b/datafusion/physical-optimizer/src/ensure_requirements/mod.rs
index 4b0067f3510cd..23dc468c27d66 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/mod.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/mod.rs
@@ -140,10 +140,10 @@ pub mod enforce_sorting;
 
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
 use crate::optimizer::{
     ConfigOnlyContext, PhysicalOptimizerContext, with_statistics_context,
 };
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
@@ -272,6 +272,10 @@ impl PhysicalOptimizerRule for EnsureRequirements {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 // See tests in datafusion/core/tests/physical_optimizer/ensure_requirements.rs
diff --git a/datafusion/physical-optimizer/src/filter_pushdown.rs b/datafusion/physical-optimizer/src/filter_pushdown.rs
index 1a2bf3d7c46d9..9687c167f6abe 100644
--- a/datafusion/physical-optimizer/src/filter_pushdown.rs
+++ b/datafusion/physical-optimizer/src/filter_pushdown.rs
@@ -20,7 +20,7 @@
 
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 use datafusion_common::{Result, assert_eq_or_internal_err, config::ConfigOptions};
 use datafusion_physical_expr::PhysicalExpr;
@@ -422,6 +422,13 @@ impl PhysicalOptimizerRule for FilterPushdown {
     fn schema_check(&self) -> bool {
         true // Filter pushdown does not change the schema of the plan
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        match self.phase {
+            FilterPushdownPhase::Pre => PhysicalOptimizerPhase::Analyzer,
+            FilterPushdownPhase::Post => PhysicalOptimizerPhase::Optimizer,
+        }
+    }
 }
 
 fn push_down_filters(
diff --git a/datafusion/physical-optimizer/src/hash_join_buffering.rs b/datafusion/physical-optimizer/src/hash_join_buffering.rs
index 767b3b74c7399..0d6cf0b88fe18 100644
--- a/datafusion/physical-optimizer/src/hash_join_buffering.rs
+++ b/datafusion/physical-optimizer/src/hash_join_buffering.rs
@@ -2,7 +2,7 @@
 // SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership
 // SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_common::JoinSide;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
@@ -94,4 +94,8 @@ impl PhysicalOptimizerRule for HashJoinBuffering {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
diff --git a/datafusion/physical-optimizer/src/join_selection.rs b/datafusion/physical-optimizer/src/join_selection.rs
index b8a3aefbe8f6e..1f80a54a187ee 100644
--- a/datafusion/physical-optimizer/src/join_selection.rs
+++ b/datafusion/physical-optimizer/src/join_selection.rs
@@ -10,8 +10,8 @@
 //! pipeline-friendly ones. To achieve the second goal, it selects the proper
 //! `PartitionMode` and the build side using the available statistics for hash joins.
 
-use crate::PhysicalOptimizerRule;
 use crate::optimizer::{ConfigOnlyContext, PhysicalOptimizerContext};
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_common::Statistics;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::error::Result;
@@ -153,6 +153,10 @@ impl PhysicalOptimizerRule for JoinSelection {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 /// Determines whether it is possible to swap inputs of a hash join. Of the
diff --git a/datafusion/physical-optimizer/src/lib.rs b/datafusion/physical-optimizer/src/lib.rs
index 52fadbc73658e..eaada98e47bea 100644
--- a/datafusion/physical-optimizer/src/lib.rs
+++ b/datafusion/physical-optimizer/src/lib.rs
@@ -40,4 +40,7 @@ pub mod update_aggr_exprs;
 pub mod utils;
 pub mod window_topn;
 
-pub use optimizer::{ConfigOnlyContext, PhysicalOptimizerContext, PhysicalOptimizerRule};
+pub use optimizer::{
+    ConfigOnlyContext, PhysicalOptimizerContext, PhysicalOptimizerPhase,
+    PhysicalOptimizerRule,
+};
diff --git a/datafusion/physical-optimizer/src/limit_pushdown.rs b/datafusion/physical-optimizer/src/limit_pushdown.rs
index 740553280d72c..a8dc22d8c4202 100644
--- a/datafusion/physical-optimizer/src/limit_pushdown.rs
+++ b/datafusion/physical-optimizer/src/limit_pushdown.rs
@@ -50,11 +50,11 @@
 use std::fmt::Debug;
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
 use crate::optimizer::{
     ConfigOnlyContext, PhysicalOptimizerContext, with_statistics_context,
 };
 use crate::row_bound::{RowBound, limit_node, place};
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::error::Result;
@@ -104,6 +104,10 @@ impl PhysicalOptimizerRule for LimitPushdown {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 /// The row bound a node receives from its ancestors.
diff --git a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs
index 763f5d79c3dce..8deaf43131541 100644
--- a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs
+++ b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs
@@ -2,8 +2,8 @@
 // SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership
 // SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements
 
-use crate::PhysicalOptimizerRule;
 use crate::row_bound::RowBound;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_physical_plan::sorts::sort::SortExec;
@@ -117,6 +117,10 @@ impl PhysicalOptimizerRule for LimitPushPastWindows {
     fn schema_check(&self) -> bool {
         false // we don't change the schema
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 fn grow_limit(lookahead: Option<WindowLookahead>, ctx: &mut TraverseState) -> bool {
diff --git a/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs b/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs
index 46a09d4307c6f..55e25279011e3 100644
--- a/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs
+++ b/datafusion/physical-optimizer/src/limited_distinct_aggregation.rs
@@ -17,7 +17,7 @@ use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 /// Pushes a soft limit into unordered DISTINCT aggregations. For
 /// `SELECT DISTINCT a FROM t LIMIT 10`, any ten distinct values are sufficient:
@@ -167,6 +167,10 @@ impl PhysicalOptimizerRule for LimitedDistinctAggregation {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 // See tests in datafusion/core/tests/physical_optimizer
diff --git a/datafusion/physical-optimizer/src/optimizer.rs b/datafusion/physical-optimizer/src/optimizer.rs
index 7c54b3f02a8e0..83dbae9f575f4 100644
--- a/datafusion/physical-optimizer/src/optimizer.rs
+++ b/datafusion/physical-optimizer/src/optimizer.rs
@@ -32,7 +32,9 @@ use datafusion_common::config::ConfigOptions;
 use datafusion_physical_plan::statistics::StatisticsContext;
 
 // Re-export from this module for backwards compatibility.
-pub use datafusion_session::{PhysicalOptimizerContext, PhysicalOptimizerRule};
+pub use datafusion_session::{
+    PhysicalOptimizerContext, PhysicalOptimizerPhase, PhysicalOptimizerRule,
+};
 
 pub use datafusion_session::with_statistics_context;
 
@@ -94,19 +96,21 @@ impl PhysicalOptimizerContext for ConfigOnlyContext<'_> {
 ///    rules in a different order must still yield a valid plan that produces
 ///    the same results.
 ///
-/// ### Exceptions
+/// ### Analyzer rules
 ///
-/// Some built-in rules do not yet fully satisfy the invariants above because
-/// they are responsible for making an initial plan *executable* rather than for
-/// optimizing an existing plan:
+/// The invariants above hold for rules whose [`PhysicalOptimizerRule::phase`]
+/// is [`PhysicalOptimizerPhase::Optimizer`]. Rules that run
+/// before the plan meets its distribution and ordering requirements return
+/// [`PhysicalOptimizerPhase::Analyzer`]: in the default list, every rule up to
+/// and including [`EnsureRequirements`], which inserts the repartitioning and
+/// sorting the requirements need. An analyzer rule keeps the results unchanged
+/// but may leave requirements unmet, so analyzer rules cannot be reordered
+/// freely. The physical planner fails planning when a rule list puts an
+/// analyzer rule after an optimizer rule.
 ///
-/// - [`EnsureRequirements`] inserts the repartitioning and sorting needed to
-///   satisfy the distribution and ordering requirements.
-/// - [`OutputRequirements`] records output requirements.
-///
-/// Custom pipelines must keep these rules in a compatible relative order. We
-/// are working to remove these exceptions, so that all rules can be reordered
-/// freely.
+/// The post-optimization [`FilterPushdown`] is an exception to invariant 2: it
+/// runs after every other optimizer rule, because later rewrites can break the
+/// links between a dynamic filter and the operators that use it.
 ///
 /// After the last rule, including rules added by the user, the physical
 /// planner rejects the plan if its requirements are not satisfied (see
@@ -298,3 +302,28 @@ impl PhysicalOptimizer {
         Self { rules }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The rules that run before `EnsureRequirements` declare the analyzer
+    /// phase, and so does `EnsureRequirements` itself; the rest declare the
+    /// optimizer phase.
+    #[test]
+    fn default_rules_declare_their_phase() {
+        let rules = PhysicalOptimizer::new().rules;
+        let enforcement = rules
+            .iter()
+            .position(|rule| rule.name() == "EnsureRequirements")
+            .unwrap();
+        for (idx, rule) in rules.iter().enumerate() {
+            let expected = if idx <= enforcement {
+                PhysicalOptimizerPhase::Analyzer
+            } else {
+                PhysicalOptimizerPhase::Optimizer
+            };
+            assert_eq!(rule.phase(), expected, "{}", rule.name());
+        }
+    }
+}
diff --git a/datafusion/physical-optimizer/src/output_requirements.rs b/datafusion/physical-optimizer/src/output_requirements.rs
index bdacaec199a7d..972559c71237f 100644
--- a/datafusion/physical-optimizer/src/output_requirements.rs
+++ b/datafusion/physical-optimizer/src/output_requirements.rs
@@ -11,7 +11,7 @@
 
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{
@@ -365,6 +365,13 @@ impl PhysicalOptimizerRule for OutputRequirements {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        match self.mode {
+            RuleMode::Add => PhysicalOptimizerPhase::Analyzer,
+            RuleMode::Remove => PhysicalOptimizerPhase::Optimizer,
+        }
+    }
 }
 
 /// This functions adds ancillary `OutputRequirementExec` to the physical plan, so that
diff --git a/datafusion/physical-optimizer/src/projection_pushdown.rs b/datafusion/physical-optimizer/src/projection_pushdown.rs
index 7faa72c1d947e..c612f3fef5e49 100644
--- a/datafusion/physical-optimizer/src/projection_pushdown.rs
+++ b/datafusion/physical-optimizer/src/projection_pushdown.rs
@@ -7,7 +7,7 @@
 //! projections one by one if the operator below is amenable to this. If a
 //! projection reaches a source, it can even disappear from the plan entirely.
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use arrow::datatypes::{Fields, Schema, SchemaRef};
 use datafusion_common::alias::AliasGenerator;
 use std::collections::HashSet;
@@ -71,6 +71,10 @@ impl PhysicalOptimizerRule for ProjectionPushdown {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 /// Tries to push down parts of the filter.
diff --git a/datafusion/physical-optimizer/src/pushdown_sort.rs b/datafusion/physical-optimizer/src/pushdown_sort.rs
index 7e3268dd19c49..8bf5b4f9534db 100644
--- a/datafusion/physical-optimizer/src/pushdown_sort.rs
+++ b/datafusion/physical-optimizer/src/pushdown_sort.rs
@@ -41,7 +41,7 @@
 //!
 //! Related issue: <https://github.com/apache/datafusion/issues/17348>
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{
@@ -215,4 +215,8 @@ impl PhysicalOptimizerRule for PushdownSort {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
diff --git a/datafusion/physical-optimizer/src/sanity_checker.rs b/datafusion/physical-optimizer/src/sanity_checker.rs
index bdfb70b9bbcd3..b4f14c7dc5b43 100644
--- a/datafusion/physical-optimizer/src/sanity_checker.rs
+++ b/datafusion/physical-optimizer/src/sanity_checker.rs
@@ -32,7 +32,7 @@ use datafusion_physical_plan::{
     ChildSatisfactionOptions, ExecutionPlanProperties, get_plan_string,
 };
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_physical_expr_common::sort_expr::format_physical_sort_requirement_list;
 use itertools::izip;
 
@@ -74,6 +74,10 @@ impl PhysicalOptimizerRule for SanityCheckPlan {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 /// Checks every node of `plan` with [`check_plan_sanity`], children first.
diff --git a/datafusion/physical-optimizer/src/topk_aggregation.rs b/datafusion/physical-optimizer/src/topk_aggregation.rs
index 5a0fa7ababf5f..665b08fb8e087 100644
--- a/datafusion/physical-optimizer/src/topk_aggregation.rs
+++ b/datafusion/physical-optimizer/src/topk_aggregation.rs
@@ -6,7 +6,7 @@
 
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
@@ -129,6 +129,10 @@ impl PhysicalOptimizerRule for TopKAggregation {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 // see `aggregate.slt` for tests
diff --git a/datafusion/physical-optimizer/src/topk_repartition.rs b/datafusion/physical-optimizer/src/topk_repartition.rs
index 03538a4ce4141..3520a17ce40cc 100644
--- a/datafusion/physical-optimizer/src/topk_repartition.rs
+++ b/datafusion/physical-optimizer/src/topk_repartition.rs
@@ -31,7 +31,7 @@
 //!       DataSourceExec
 //! ```
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
@@ -162,6 +162,10 @@ impl PhysicalOptimizerRule for TopKRepartition {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 #[cfg(test)]
diff --git a/datafusion/physical-optimizer/src/update_aggr_exprs.rs b/datafusion/physical-optimizer/src/update_aggr_exprs.rs
index 70c4ec1694e76..ab7188f4aa39e 100644
--- a/datafusion/physical-optimizer/src/update_aggr_exprs.rs
+++ b/datafusion/physical-optimizer/src/update_aggr_exprs.rs
@@ -18,7 +18,7 @@ use datafusion_physical_plan::aggregates::{
 use datafusion_physical_plan::windows::get_ordered_partition_by_indices;
 use datafusion_physical_plan::{ExecutionPlan, ExecutionPlanProperties};
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 
 /// This optimizer rule checks ordering requirements of aggregate expressions.
 ///
@@ -129,6 +129,10 @@ impl PhysicalOptimizerRule for OptimizeAggregateOrder {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Optimizer
+    }
 }
 
 /// Tries to convert each aggregate expression to a potentially more efficient
diff --git a/datafusion/physical-optimizer/src/window_topn.rs b/datafusion/physical-optimizer/src/window_topn.rs
index 6b76f2101b0ec..6611511a3f5a5 100644
--- a/datafusion/physical-optimizer/src/window_topn.rs
+++ b/datafusion/physical-optimizer/src/window_topn.rs
@@ -39,7 +39,7 @@
 
 use std::sync::Arc;
 
-use crate::PhysicalOptimizerRule;
+use crate::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use arrow::datatypes::DataType;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
@@ -336,6 +336,10 @@ impl PhysicalOptimizerRule for WindowTopN {
     fn schema_check(&self) -> bool {
         true
     }
+
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 /// Extract a window limit from a predicate expression.
diff --git a/datafusion/session/src/lib.rs b/datafusion/session/src/lib.rs
index 31f0e0f2d10a2..5703ff5a9f22b 100644
--- a/datafusion/session/src/lib.rs
+++ b/datafusion/session/src/lib.rs
@@ -44,7 +44,8 @@ pub use crate::catalog::{
     EmptyCatalogProviderList,
 };
 pub use crate::physical_optimizer::{
-    PhysicalOptimizerContext, PhysicalOptimizerRule, with_statistics_context,
+    PhysicalOptimizerContext, PhysicalOptimizerPhase, PhysicalOptimizerRule,
+    with_statistics_context,
 };
 pub use crate::planner::{
     ExtensionPlanner, PhysicalPlanner, QueryPlanner, UnsupportedQueryPlanner,
diff --git a/datafusion/session/src/physical_optimizer.rs b/datafusion/session/src/physical_optimizer.rs
index aa276c9814db9..ce1796d1f8555 100644
--- a/datafusion/session/src/physical_optimizer.rs
+++ b/datafusion/session/src/physical_optimizer.rs
@@ -176,4 +176,34 @@ pub trait PhysicalOptimizerRule: Debug + std::any::Any {
     ///
     /// [`optimize`]: PhysicalOptimizerRule::optimize
     fn schema_check(&self) -> bool;
+
+    /// Whether this rule runs before or after the plan meets the distribution
+    /// and ordering requirements of its operators. See
+    /// [`PhysicalOptimizerPhase`].
+    ///
+    /// Most rules return [`PhysicalOptimizerPhase::Optimizer`]. A rule that
+    /// has to run before `EnsureRequirements`, because enforcement reads what
+    /// it changes or because it needs the plan before enforcement adds
+    /// repartitions and sorts, returns [`PhysicalOptimizerPhase::Analyzer`].
+    /// A wrapper that only observes another rule, for tracing for example,
+    /// returns that rule's phase.
+    fn phase(&self) -> PhysicalOptimizerPhase;
+}
+
+/// The two phases of physical optimization. The physical planner runs every
+/// [`Analyzer`](Self::Analyzer) rule before the first
+/// [`Optimizer`](Self::Optimizer) rule, and fails planning when a rule list
+/// puts an analyzer rule after an optimizer rule.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[non_exhaustive]
+pub enum PhysicalOptimizerPhase {
+    /// Runs while the requirements are only declared. The rule must keep the
+    /// results unchanged, may leave requirements unmet, and must stay correct
+    /// on a plan that already contains repartitions and sorts. The last
+    /// analyzer rule, `EnsureRequirements` in the default list, leaves a plan
+    /// that meets every requirement.
+    Analyzer,
+    /// Receives a plan that meets every requirement and must return one that
+    /// does, with the same results.
+    Optimizer,
 }
diff --git a/datafusion/sqllogictest/src/test_context/custom_sort_pushdown.rs b/datafusion/sqllogictest/src/test_context/custom_sort_pushdown.rs
index cc50db4974a6c..aaab01ddadf38 100644
--- a/datafusion/sqllogictest/src/test_context/custom_sort_pushdown.rs
+++ b/datafusion/sqllogictest/src/test_context/custom_sort_pushdown.rs
@@ -17,7 +17,7 @@ use datafusion::logical_expr::{Expr, Operator, TableType};
 use datafusion::physical_expr::expressions::{binary, col, lit, negative};
 use datafusion::physical_expr::projection::ProjectionExprs;
 use datafusion::physical_expr::{PhysicalExpr, PhysicalSortExpr};
-use datafusion::physical_optimizer::PhysicalOptimizerRule;
+use datafusion::physical_optimizer::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
 use datafusion::physical_plan::filter::{FilterExec, FilterExecBuilder};
 use datafusion::physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
 use datafusion::physical_plan::projection::ProjectionExec;
@@ -163,6 +163,11 @@ impl PhysicalOptimizerRule for CustomLimitRule {
     fn schema_check(&self) -> bool {
         true
     }
+
+    // `sort_pushdown.slt` inserts this rule before `EnsureRequirements`.
+    fn phase(&self) -> PhysicalOptimizerPhase {
+        PhysicalOptimizerPhase::Analyzer
+    }
 }
 
 /// Exposes a native plan's children and properties under a distinct plan type,

From 34301a6d8c30ff65519d6e7415005ca567964707 Mon Sep 17 00:00:00 2001
From: Jay Zhan <jayzhan211@gmail.com>
Date: Sun, 1 Nov 2026 08:26:05 +0800
Subject: [PATCH 2/3] docs: describe physical optimizer rule phases and who is
 affected

---
 .../core/src/optimizer_rule_reference.md      | 55 +++++++++++--------
 .../library-user-guide/upgrading/56.0.0.md    | 45 +++++++++++++++
 2 files changed, 76 insertions(+), 24 deletions(-)

diff --git a/datafusion/core/src/optimizer_rule_reference.md b/datafusion/core/src/optimizer_rule_reference.md
index a1393835c4555..6bf61794cc1ab 100644
--- a/datafusion/core/src/optimizer_rule_reference.md
+++ b/datafusion/core/src/optimizer_rule_reference.md
@@ -67,33 +67,40 @@ Rule order matters. The default pipeline may change between releases.
 ### Physical Optimizer Rules
 
 The same rule name may appear more than once when the default pipeline runs it
-in multiple phases.
+in multiple passes.
+
+Rules 1 to 8 are analyzer rules (`PhysicalOptimizerRule::phase` returns
+`PhysicalOptimizerPhase::Analyzer`): they run before the plan meets its
+distribution and ordering requirements, which `EnsureRequirements` makes hold.
+The other rules are optimizer rules: each receives a plan that meets the
+requirements and must return one. The physical planner fails planning when a
+rule list puts an analyzer rule after an optimizer rule.
 
 After the last rule, including any rule added by the user, the physical planner
 checks that the final plan meets its ordering, distribution, and infinite-input
 safety requirements (`SanityCheckPlan`). If the ordering or distribution
 requirements do not hold, the error names the rule that broke them.
 
-| order | rule                           | phase                   | summary                                                                                                  |
-| ----- | ------------------------------ | ----------------------- | -------------------------------------------------------------------------------------------------------- |
-| 1     | `OutputRequirements`           | add phase               | Adds helper nodes so output requirements survive later physical rewrites.                                |
-| 2     | `aggregate_statistics`         | -                       | Uses exact source statistics to answer some aggregates without scanning data.                            |
-| 3     | `EliminateConstantJoin`        | -                       | Replaces inner joins against a single row of literals with a filter that keeps the other input's order.  |
-| 4     | `join_selection`               | -                       | Chooses join implementation, build side, and partition mode from statistics and stream properties.       |
-| 5     | `LimitedDistinctAggregation`   | -                       | Pushes limit hints into grouped distinct-style aggregations when only a small result is needed.          |
-| 6     | `FilterPushdown`               | pre-optimization phase  | Pushes supported physical filters down toward data sources before distribution and sorting are enforced. |
-| 7     | `WindowTopN`                   | -                       | Replaces eligible row-number window and filter patterns with per-partition TopK execution.               |
-| 8     | `EnsureRequirements`           | -                       | Enforces both distribution and sorting requirements in a single idempotent rule.                         |
-| 9     | `CombinePartialFinalAggregate` | -                       | Collapses adjacent partial and final aggregates when the distributed shape makes them redundant.         |
-| 10    | `OptimizeAggregateOrder`       | -                       | Updates aggregate expressions to use the best ordering once sort requirements are known.                 |
-| 11    | `ProjectionPushdown`           | early pass              | Pushes projections toward inputs before later physical rewrites add more limit and TopK structure.       |
-| 12    | `OutputRequirements`           | remove phase            | Removes the temporary output-requirement helper nodes after requirement-sensitive planning is done.      |
-| 13    | `LimitAggregation`             | -                       | Passes a limit hint into eligible aggregations so they can keep fewer accumulator buckets.               |
-| 14    | `LimitPushPastWindows`         | -                       | Pushes fetch limits through bounded window operators when doing so keeps the result correct.             |
-| 15    | `HashJoinBuffering`            | -                       | Adds buffering on the probe side of hash joins so probing can start before build completion.             |
-| 16    | `LimitPushdown`                | -                       | Moves physical limits into child operators or fetch-enabled variants to cut data early.                  |
-| 17    | `TopKRepartition`              | -                       | Pushes TopK below hash repartition when the partition key is a prefix of the sort key.                   |
-| 18    | `ProjectionPushdown`           | late pass               | Runs projection pushdown again after limit and TopK rewrites expose new pruning opportunities.           |
-| 19    | `PushdownSort`                 | -                       | Pushes sort requirements into data sources that can already return sorted output.                        |
-| 20    | `EnsureCooperative`            | -                       | Wraps non-cooperative plan parts so long-running tasks yield fairly.                                     |
-| 21    | `FilterPushdown(Post)`         | post-optimization phase | Pushes dynamic filters at the end of optimization, after plan references stop moving.                    |
+| order | rule                           | pass              | summary                                                                                                  |
+| ----- | ------------------------------ | ----------------- | -------------------------------------------------------------------------------------------------------- |
+| 1     | `OutputRequirements`           | add               | Adds helper nodes so output requirements survive later physical rewrites.                                |
+| 2     | `aggregate_statistics`         | -                 | Uses exact source statistics to answer some aggregates without scanning data.                            |
+| 3     | `EliminateConstantJoin`        | -                 | Replaces inner joins against a single row of literals with a filter that keeps the other input's order.  |
+| 4     | `join_selection`               | -                 | Chooses join implementation, build side, and partition mode from statistics and stream properties.       |
+| 5     | `LimitedDistinctAggregation`   | -                 | Pushes limit hints into grouped distinct-style aggregations when only a small result is needed.          |
+| 6     | `FilterPushdown`               | pre-optimization  | Pushes supported physical filters down toward data sources before distribution and sorting are enforced. |
+| 7     | `WindowTopN`                   | -                 | Replaces eligible row-number window and filter patterns with per-partition TopK execution.               |
+| 8     | `EnsureRequirements`           | -                 | Enforces both distribution and sorting requirements in a single idempotent rule.                         |
+| 9     | `CombinePartialFinalAggregate` | -                 | Collapses adjacent partial and final aggregates when the distributed shape makes them redundant.         |
+| 10    | `OptimizeAggregateOrder`       | -                 | Updates aggregate expressions to use the best ordering once sort requirements are known.                 |
+| 11    | `ProjectionPushdown`           | early             | Pushes projections toward inputs before later physical rewrites add more limit and TopK structure.       |
+| 12    | `OutputRequirements`           | remove            | Removes the temporary output-requirement helper nodes after requirement-sensitive planning is done.      |
+| 13    | `LimitAggregation`             | -                 | Passes a limit hint into eligible aggregations so they can keep fewer accumulator buckets.               |
+| 14    | `LimitPushPastWindows`         | -                 | Pushes fetch limits through bounded window operators when doing so keeps the result correct.             |
+| 15    | `HashJoinBuffering`            | -                 | Adds buffering on the probe side of hash joins so probing can start before build completion.             |
+| 16    | `LimitPushdown`                | -                 | Moves physical limits into child operators or fetch-enabled variants to cut data early.                  |
+| 17    | `TopKRepartition`              | -                 | Pushes TopK below hash repartition when the partition key is a prefix of the sort key.                   |
+| 18    | `ProjectionPushdown`           | late              | Runs projection pushdown again after limit and TopK rewrites expose new pruning opportunities.           |
+| 19    | `PushdownSort`                 | -                 | Pushes sort requirements into data sources that can already return sorted output.                        |
+| 20    | `EnsureCooperative`            | -                 | Wraps non-cooperative plan parts so long-running tasks yield fairly.                                     |
+| 21    | `FilterPushdown(Post)`         | post-optimization | Pushes dynamic filters at the end of optimization, after plan references stop moving.                    |
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index 6b33b5ededd17..a52a7b462cb90 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -1454,6 +1454,51 @@ A custom rule list that left out `SanityCheckPlan` is now checked too. A plan
 whose requirements do not hold, such as a `SortMergeJoinExec` over unsorted
 input, now fails to plan instead of possibly returning wrong results.
 
+### Physical optimizer rules declare whether they run before or after enforcement
+
+`PhysicalOptimizerRule` has a new required method, `phase()`. It returns
+`PhysicalOptimizerPhase::Analyzer` for a rule that runs before the plan meets
+the distribution and ordering requirements of its operators, and
+`PhysicalOptimizerPhase::Optimizer` for a rule that receives a plan meeting
+them and must return one. In the default rule list, the rules up to and
+including `EnsureRequirements` are analyzer rules. Default plans, rule names
+and `EXPLAIN VERBOSE` output do not change.
+
+The physical planner now fails planning when a rule list puts an optimizer
+rule before an analyzer rule, with an error that names both rules and their
+positions.
+
+**Who is affected:**
+
+- Every type that implements `PhysicalOptimizerRule` must implement `phase()`.
+  Most custom rules run after `EnsureRequirements` and return the optimizer
+  phase:
+
+  ```rust,ignore
+  fn phase(&self) -> PhysicalOptimizerPhase {
+      PhysicalOptimizerPhase::Optimizer
+  }
+  ```
+
+  A rule that wraps another rule, for tracing for example, returns the wrapped
+  rule's `phase()`.
+
+- A custom rule placed before `EnsureRequirements`, for example in front of
+  `PhysicalOptimizer::new().rules`, fails planning if it returns the optimizer
+  phase. Return `PhysicalOptimizerPhase::Analyzer` only if the rule is written
+  to run before the requirements are enforced; otherwise move it after
+  `EnsureRequirements`.
+
+- A rule list that runs `EnsureRequirements`, or another analyzer rule, again
+  after optimizer rules fails planning for the same reason. A rule that breaks
+  requirements can repair them by running `EnsureRequirements` on its own
+  output.
+
+- A rule received through FFI reports `PhysicalOptimizerPhase::Optimizer`,
+  because `FFI_PhysicalOptimizerRule` does not carry the phase; its layout is
+  unchanged. To run such a rule before enforcement, wrap it in a local rule
+  that forwards to it and returns `PhysicalOptimizerPhase::Analyzer`.
+
 ### `AggregateUDFImpl::is_nullable` is deprecated
 
 `AggregateUDFImpl::is_nullable` is deprecated; set the nullability in the field

From a86d79d545352cb9a944fd237cb2f6324379840e Mon Sep 17 00:00:00 2001
From: Jay Zhan <jayzhan211@gmail.com>
Date: Sun, 1 Nov 2026 10:14:28 +0800
Subject: [PATCH 3/3] feat: separate the enforcement phase so analyzer rules
 cannot run after it

---
 .../core/src/optimizer_rule_reference.md      |  11 +-
 datafusion/core/src/physical_planner.rs       | 109 ++++++++++++------
 .../physical_optimizer/sanity_checker.rs      |  45 ++++++--
 .../src/ensure_requirements/mod.rs            |   2 +-
 .../physical-optimizer/src/optimizer.rs       |  25 ++--
 datafusion/session/src/physical_optimizer.rs  |  22 ++--
 .../library-user-guide/upgrading/56.0.0.md    |  38 +++---
 7 files changed, 162 insertions(+), 90 deletions(-)

diff --git a/datafusion/core/src/optimizer_rule_reference.md b/datafusion/core/src/optimizer_rule_reference.md
index 6bf61794cc1ab..13046c3f38ea5 100644
--- a/datafusion/core/src/optimizer_rule_reference.md
+++ b/datafusion/core/src/optimizer_rule_reference.md
@@ -69,12 +69,13 @@ Rule order matters. The default pipeline may change between releases.
 The same rule name may appear more than once when the default pipeline runs it
 in multiple passes.
 
-Rules 1 to 8 are analyzer rules (`PhysicalOptimizerRule::phase` returns
+Rules 1 to 7 are analyzer rules (`PhysicalOptimizerRule::phase` returns
 `PhysicalOptimizerPhase::Analyzer`): they run before the plan meets its
-distribution and ordering requirements, which `EnsureRequirements` makes hold.
-The other rules are optimizer rules: each receives a plan that meets the
-requirements and must return one. The physical planner fails planning when a
-rule list puts an analyzer rule after an optimizer rule.
+distribution and ordering requirements. Rule 8, `EnsureRequirements`, is the
+enforcement rule (`PhysicalOptimizerPhase::Enforcement`): it makes the
+requirements hold. The other rules are optimizer rules: each receives a plan
+that meets the requirements and must return one. The physical planner fails
+planning when a rule list puts a rule before a rule of an earlier phase.
 
 After the last rule, including any rule added by the user, the physical planner
 checks that the final plan meets its ordering, distribution, and infinite-input
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index f7844dbb07da4..9e160048a8bf3 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -3451,7 +3451,7 @@ impl DefaultPhysicalPlanner {
     {
         let plan = bind_materialized_cte_scans(plan, session_state)?;
         let optimizers = session_state.physical_optimizers();
-        let analyzer_rules = analyzer_rule_count(optimizers)?;
+        let enforced = check_phase_order(optimizers)?;
         debug!(
             "Input physical plan:\n{}\n",
             displayable(plan.as_ref()).indent(false)
@@ -3496,7 +3496,7 @@ impl DefaultPhysicalPlanner {
                 &new_plan,
                 plan,
                 optimizers,
-                analyzer_rules,
+                enforced,
                 &optimizer_context,
             ));
         }
@@ -3709,41 +3709,43 @@ fn tuple_err<T, R>(value: (Result<T>, Result<R>)) -> Result<(T, R)> {
     }
 }
 
-/// Returns how many rules at the start of `rules` are
-/// [`PhysicalOptimizerPhase::Analyzer`] rules. Fails when an analyzer rule
-/// comes after an optimizer rule: optimizer rules may assume that the plan
-/// meets its requirements, which only holds after the last analyzer rule.
-fn analyzer_rule_count(
+/// Checks that `rules` run in [`PhysicalOptimizerPhase`] order: analyzer
+/// rules, then enforcement rules, then optimizer rules. Returns the position
+/// of the last enforcement rule, after which the plan meets its requirements.
+fn check_phase_order(
     rules: &[Arc<dyn PhysicalOptimizerRule + Send + Sync>],
-) -> Result<usize> {
-    let is_analyzer = |rule: &&Arc<dyn PhysicalOptimizerRule + Send + Sync>| {
-        rule.phase() == PhysicalOptimizerPhase::Analyzer
-    };
-    let count = rules.iter().take_while(is_analyzer).count();
-    if let Some(late) = rules[count..].iter().position(|rule| is_analyzer(&rule)) {
-        let early = rules[count].name();
+) -> Result<Option<usize>> {
+    if let Some(idx) = rules
+        .windows(2)
+        .position(|pair| pair[1].phase() < pair[0].phase())
+    {
+        let (before, after) = (&rules[idx], &rules[idx + 1]);
         return plan_err!(
-            "Physical optimizer rule '{early}' (position {}) is an optimizer rule \
-             but runs before analyzer rule '{}' (position {}). Optimizer rules \
-             assume the plan meets its distribution and ordering requirements, so \
-             they must run after every analyzer rule. Move '{early}' after the \
-             analyzer rules. Return PhysicalOptimizerPhase::Analyzer from its \
-             phase() only if it is written to run before the requirements are \
-             enforced",
-            count + 1,
-            rules[count + late].name(),
-            count + late + 1
+            "Physical optimizer rules are out of phase order: '{}' (position {}, \
+             phase {:?}) runs before '{}' (position {}, phase {:?}). Rules must \
+             run in the order Analyzer, Enforcement, Optimizer, because each \
+             phase relies on what the earlier phases did. Move the rule that is \
+             out of place, and change a rule's phase() only if it is written for \
+             that phase",
+            before.name(),
+            idx + 1,
+            before.phase(),
+            after.name(),
+            idx + 2,
+            after.phase()
         );
     }
-    Ok(count)
+    Ok(rules
+        .iter()
+        .rposition(|rule| rule.phase() == PhysicalOptimizerPhase::Enforcement))
 }
 
 /// Adds context to an error from the final [`check_plan_tree_sanity`].
 ///
-/// The last of the first `analyzer_rules` rules, `EnsureRequirements` in the
+/// The enforcement rule at position `enforced`, `EnsureRequirements` in the
 /// default list, makes the order and distribution requirements hold, and every
 /// rule after it must keep them. When they do not hold for `final_plan`, the
-/// context names the rule that broke this contract: either that last analyzer
+/// context names the rule that broke this contract: either the enforcement
 /// rule, or the rule after which the requirements stopped holding. To find
 /// it, `optimizers` run again from `input_plan`, so planning that succeeds
 /// keeps no intermediate plans.
@@ -3752,7 +3754,7 @@ fn sanity_check_error(
     final_plan: &Arc<dyn ExecutionPlan>,
     input_plan: Arc<dyn ExecutionPlan>,
     optimizers: &[Arc<dyn PhysicalOptimizerRule + Send + Sync>],
-    analyzer_rules: usize,
+    enforced: Option<usize>,
     optimizer_context: &dyn PhysicalOptimizerContext,
 ) -> DataFusionError {
     const SANITY_CHECK: &str = "SanityCheckPlan";
@@ -3763,10 +3765,10 @@ fn sanity_check_error(
         return plain_error(error);
     }
 
-    let enforced = analyzer_rules.checked_sub(1);
-    // Plans before the last analyzer rule need not satisfy the requirements,
-    // so only plans from its output onwards are checked. Without analyzer
-    // rules, every plan is checked, starting with the input plan.
+    // Plans before the last enforcement rule need not satisfy the
+    // requirements, so only plans from its output onwards are checked.
+    // Without an enforcement rule, every plan is checked, starting with the
+    // input plan.
     let first_checked = enforced.unwrap_or(0);
     let mut held = enforced.is_none() && input_requirements_hold(&input_plan);
     let mut broken_by = None;
@@ -4336,8 +4338,8 @@ mod tests {
             .to_string();
         assert!(
             error.contains(
-                "Physical optimizer rule 'noop' (position 1) is an optimizer rule \
-                 but runs before analyzer rule 'OutputRequirements' (position 2)"
+                "'noop' (position 1, phase Optimizer) runs before \
+                 'OutputRequirements' (position 2, phase Analyzer)"
             ),
             "{error}"
         );
@@ -4401,6 +4403,40 @@ mod tests {
         Ok(())
     }
 
+    /// An analyzer rule right after enforcement fails planning: it would see
+    /// a plan whose requirements already hold.
+    #[tokio::test]
+    async fn analyzer_rule_after_enforcement_fails_planning() -> Result<()> {
+        let mut rules =
+            datafusion_physical_optimizer::optimizer::PhysicalOptimizer::new().rules;
+        let enforcement = rules
+            .iter()
+            .position(|rule| rule.phase() == PhysicalOptimizerPhase::Enforcement)
+            .unwrap();
+        rules.insert(
+            enforcement + 1,
+            Arc::new(NoopRule(PhysicalOptimizerPhase::Analyzer)),
+        );
+        let session_state = SessionStateBuilder::new()
+            .with_default_features()
+            .with_physical_optimizer_rules(rules)
+            .build();
+        let logical_plan = LogicalPlanBuilder::empty(false).build()?;
+        let error = session_state
+            .create_physical_plan(&logical_plan)
+            .await
+            .unwrap_err()
+            .to_string();
+        assert!(
+            error.contains(
+                "'EnsureRequirements' (position 8, phase Enforcement) runs before \
+                 'noop' (position 9, phase Analyzer)"
+            ),
+            "{error}"
+        );
+        Ok(())
+    }
+
     /// An analyzer rule appended after the default rules runs after the
     /// optimizer rules, so planning fails.
     #[tokio::test]
@@ -4418,7 +4454,10 @@ mod tests {
             .unwrap_err()
             .to_string();
         assert!(
-            error.contains("is an optimizer rule but runs before analyzer rule 'noop'"),
+            error.contains(
+                "'FilterPushdown(Post)' (position 21, phase Optimizer) runs before \
+                 'noop' (position 22, phase Analyzer)"
+            ),
             "{error}"
         );
         Ok(())
diff --git a/datafusion/core/tests/physical_optimizer/sanity_checker.rs b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
index ac6acb4886eda..cfe23daf46ae4 100644
--- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs
+++ b/datafusion/core/tests/physical_optimizer/sanity_checker.rs
@@ -24,6 +24,7 @@ use datafusion_common::{JoinType, NullEquality, Result, ScalarValue};
 use datafusion_physical_expr::expressions::{Literal, col};
 use datafusion_physical_expr::{Partitioning, RangePartitioning, SplitPoint};
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
+use datafusion_physical_optimizer::join_selection::JoinSelection;
 use datafusion_physical_optimizer::optimizer::PhysicalOptimizer;
 use datafusion_physical_optimizer::sanity_checker::check_plan_tree_sanity;
 use datafusion_physical_optimizer::{PhysicalOptimizerPhase, PhysicalOptimizerRule};
@@ -1040,11 +1041,11 @@ impl PhysicalOptimizerRule for RemoveSorts {
     }
 }
 
-/// [`RemoveSorts`] declared as an analyzer rule.
+/// [`RemoveSorts`] declared as an enforcement rule.
 #[derive(Debug)]
-struct RemoveSortsInAnalyzer;
+struct RemoveSortsInEnforcement;
 
-impl PhysicalOptimizerRule for RemoveSortsInAnalyzer {
+impl PhysicalOptimizerRule for RemoveSortsInEnforcement {
     fn optimize(
         &self,
         plan: Arc<dyn ExecutionPlan>,
@@ -1062,7 +1063,7 @@ impl PhysicalOptimizerRule for RemoveSortsInAnalyzer {
     }
 
     fn phase(&self) -> PhysicalOptimizerPhase {
-        PhysicalOptimizerPhase::Analyzer
+        PhysicalOptimizerPhase::Enforcement
     }
 }
 
@@ -1119,16 +1120,16 @@ async fn test_final_check_without_ensure_requirements() -> Result<()> {
     Ok(())
 }
 
-/// The analyzer rules must leave a plan whose requirements hold. When they do
-/// not, the error names the last analyzer rule.
+/// The enforcement rules must leave a plan whose requirements hold. When they
+/// do not, the error names the last enforcement rule.
 #[tokio::test]
-async fn test_final_check_names_last_analyzer_rule() -> Result<()> {
+async fn test_final_check_names_last_enforcement_rule() -> Result<()> {
     let mut rules = PhysicalOptimizer::new().rules;
-    let analyzer_rules = rules
+    let enforcement = rules
         .iter()
-        .take_while(|rule| rule.phase() == PhysicalOptimizerPhase::Analyzer)
-        .count();
-    rules.insert(analyzer_rules, Arc::new(RemoveSortsInAnalyzer));
+        .rposition(|rule| rule.phase() == PhysicalOptimizerPhase::Enforcement)
+        .unwrap();
+    rules.insert(enforcement + 1, Arc::new(RemoveSortsInEnforcement));
     let state = SessionStateBuilder::new()
         .with_default_features()
         .with_physical_optimizer_rules(rules)
@@ -1143,3 +1144,25 @@ async fn test_final_check_names_last_analyzer_rule() -> Result<()> {
     );
     Ok(())
 }
+
+/// Analyzer rules without an enforcement rule never make the requirements
+/// hold, so no rule is named.
+#[tokio::test]
+async fn test_final_check_without_enforcement_names_no_rule() -> Result<()> {
+    let state = SessionStateBuilder::new()
+        .with_default_features()
+        .with_physical_optimizer_rules(vec![
+            Arc::new(JoinSelection::new()),
+            Arc::new(RemoveSorts),
+        ])
+        .build();
+    let error = plan_window_query(state).await.unwrap_err().to_string();
+    assert!(
+        error.starts_with(
+            "SanityCheckPlan: no physical optimizer rule produced a plan that \
+             satisfies the order and distribution requirements"
+        ),
+        "{error}"
+    );
+    Ok(())
+}
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/mod.rs b/datafusion/physical-optimizer/src/ensure_requirements/mod.rs
index 23dc468c27d66..03aeb9a29892f 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/mod.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/mod.rs
@@ -274,7 +274,7 @@ impl PhysicalOptimizerRule for EnsureRequirements {
     }
 
     fn phase(&self) -> PhysicalOptimizerPhase {
-        PhysicalOptimizerPhase::Analyzer
+        PhysicalOptimizerPhase::Enforcement
     }
 }
 
diff --git a/datafusion/physical-optimizer/src/optimizer.rs b/datafusion/physical-optimizer/src/optimizer.rs
index 83dbae9f575f4..0a45bd94b7aae 100644
--- a/datafusion/physical-optimizer/src/optimizer.rs
+++ b/datafusion/physical-optimizer/src/optimizer.rs
@@ -101,12 +101,13 @@ impl PhysicalOptimizerContext for ConfigOnlyContext<'_> {
 /// The invariants above hold for rules whose [`PhysicalOptimizerRule::phase`]
 /// is [`PhysicalOptimizerPhase::Optimizer`]. Rules that run
 /// before the plan meets its distribution and ordering requirements return
-/// [`PhysicalOptimizerPhase::Analyzer`]: in the default list, every rule up to
-/// and including [`EnsureRequirements`], which inserts the repartitioning and
-/// sorting the requirements need. An analyzer rule keeps the results unchanged
-/// but may leave requirements unmet, so analyzer rules cannot be reordered
-/// freely. The physical planner fails planning when a rule list puts an
-/// analyzer rule after an optimizer rule.
+/// [`PhysicalOptimizerPhase::Analyzer`]: in the default list, every rule before
+/// [`EnsureRequirements`]. An analyzer rule keeps the results unchanged but
+/// may leave requirements unmet, so analyzer rules cannot be reordered freely.
+/// [`EnsureRequirements`] returns [`PhysicalOptimizerPhase::Enforcement`]: it
+/// inserts the repartitioning and sorting the requirements need. The physical
+/// planner fails planning when a rule list puts a rule before a rule of an
+/// earlier phase.
 ///
 /// The post-optimization [`FilterPushdown`] is an exception to invariant 2: it
 /// runs after every other optimizer rule, because later rewrites can break the
@@ -308,8 +309,8 @@ mod tests {
     use super::*;
 
     /// The rules that run before `EnsureRequirements` declare the analyzer
-    /// phase, and so does `EnsureRequirements` itself; the rest declare the
-    /// optimizer phase.
+    /// phase, `EnsureRequirements` declares the enforcement phase, and the
+    /// rest declare the optimizer phase.
     #[test]
     fn default_rules_declare_their_phase() {
         let rules = PhysicalOptimizer::new().rules;
@@ -318,10 +319,10 @@ mod tests {
             .position(|rule| rule.name() == "EnsureRequirements")
             .unwrap();
         for (idx, rule) in rules.iter().enumerate() {
-            let expected = if idx <= enforcement {
-                PhysicalOptimizerPhase::Analyzer
-            } else {
-                PhysicalOptimizerPhase::Optimizer
+            let expected = match idx.cmp(&enforcement) {
+                std::cmp::Ordering::Less => PhysicalOptimizerPhase::Analyzer,
+                std::cmp::Ordering::Equal => PhysicalOptimizerPhase::Enforcement,
+                std::cmp::Ordering::Greater => PhysicalOptimizerPhase::Optimizer,
             };
             assert_eq!(rule.phase(), expected, "{}", rule.name());
         }
diff --git a/datafusion/session/src/physical_optimizer.rs b/datafusion/session/src/physical_optimizer.rs
index ce1796d1f8555..a81beecf96ef3 100644
--- a/datafusion/session/src/physical_optimizer.rs
+++ b/datafusion/session/src/physical_optimizer.rs
@@ -185,24 +185,26 @@ pub trait PhysicalOptimizerRule: Debug + std::any::Any {
     /// has to run before `EnsureRequirements`, because enforcement reads what
     /// it changes or because it needs the plan before enforcement adds
     /// repartitions and sorts, returns [`PhysicalOptimizerPhase::Analyzer`].
-    /// A wrapper that only observes another rule, for tracing for example,
-    /// returns that rule's phase.
+    /// A rule that makes the requirements hold, such as `EnsureRequirements`,
+    /// returns [`PhysicalOptimizerPhase::Enforcement`]. A wrapper that only
+    /// observes another rule, for tracing for example, returns that rule's
+    /// phase.
     fn phase(&self) -> PhysicalOptimizerPhase;
 }
 
-/// The two phases of physical optimization. The physical planner runs every
-/// [`Analyzer`](Self::Analyzer) rule before the first
-/// [`Optimizer`](Self::Optimizer) rule, and fails planning when a rule list
-/// puts an analyzer rule after an optimizer rule.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+/// The phases of physical optimization, in the order they run. The physical
+/// planner fails planning when a rule list puts a rule before a rule of an
+/// earlier phase.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 #[non_exhaustive]
 pub enum PhysicalOptimizerPhase {
     /// Runs while the requirements are only declared. The rule must keep the
     /// results unchanged, may leave requirements unmet, and must stay correct
-    /// on a plan that already contains repartitions and sorts. The last
-    /// analyzer rule, `EnsureRequirements` in the default list, leaves a plan
-    /// that meets every requirement.
+    /// on a plan that already contains repartitions and sorts.
     Analyzer,
+    /// Makes every distribution and ordering requirement hold, as
+    /// `EnsureRequirements` does in the default list.
+    Enforcement,
     /// Receives a plan that meets every requirement and must return one that
     /// does, with the same results.
     Optimizer,
diff --git a/docs/source/library-user-guide/upgrading/56.0.0.md b/docs/source/library-user-guide/upgrading/56.0.0.md
index a52a7b462cb90..94e992f86effb 100644
--- a/docs/source/library-user-guide/upgrading/56.0.0.md
+++ b/docs/source/library-user-guide/upgrading/56.0.0.md
@@ -1456,17 +1456,22 @@ input, now fails to plan instead of possibly returning wrong results.
 
 ### Physical optimizer rules declare whether they run before or after enforcement
 
-`PhysicalOptimizerRule` has a new required method, `phase()`. It returns
-`PhysicalOptimizerPhase::Analyzer` for a rule that runs before the plan meets
-the distribution and ordering requirements of its operators, and
-`PhysicalOptimizerPhase::Optimizer` for a rule that receives a plan meeting
-them and must return one. In the default rule list, the rules up to and
-including `EnsureRequirements` are analyzer rules. Default plans, rule names
-and `EXPLAIN VERBOSE` output do not change.
-
-The physical planner now fails planning when a rule list puts an optimizer
-rule before an analyzer rule, with an error that names both rules and their
-positions.
+`PhysicalOptimizerRule` has a new required method, `phase()`. It returns one
+of the three phases of physical optimization, in the order they run:
+
+- `PhysicalOptimizerPhase::Analyzer` for a rule that runs before the plan meets
+  the distribution and ordering requirements of its operators,
+- `PhysicalOptimizerPhase::Enforcement` for a rule that makes them hold, such as
+  `EnsureRequirements`,
+- `PhysicalOptimizerPhase::Optimizer` for a rule that receives a plan meeting
+  them and must return one.
+
+In the default rule list, the rules before `EnsureRequirements` are analyzer
+rules. Default plans, rule names and `EXPLAIN VERBOSE` output do not change.
+
+The physical planner now fails planning when a rule list puts a rule before a
+rule of an earlier phase, with an error that names both rules, their positions
+and their phases.
 
 **Who is affected:**
 
@@ -1485,12 +1490,13 @@ positions.
 
 - A custom rule placed before `EnsureRequirements`, for example in front of
   `PhysicalOptimizer::new().rules`, fails planning if it returns the optimizer
-  phase. Return `PhysicalOptimizerPhase::Analyzer` only if the rule is written
-  to run before the requirements are enforced; otherwise move it after
-  `EnsureRequirements`.
+  phase. A rule placed after `EnsureRequirements` fails planning if it returns
+  the analyzer phase. Return `PhysicalOptimizerPhase::Analyzer` only if the
+  rule is written to run before the requirements are enforced; otherwise move
+  it after `EnsureRequirements`.
 
-- A rule list that runs `EnsureRequirements`, or another analyzer rule, again
-  after optimizer rules fails planning for the same reason. A rule that breaks
+- A rule list that runs `EnsureRequirements`, or an analyzer rule, again after
+  optimizer rules fails planning for the same reason. A rule that breaks
   requirements can repair them by running `EnsureRequirements` on its own
   output.