diff --git a/datafusion/core/src/optimizer_rule_reference.md b/datafusion/core/src/optimizer_rule_reference.md index a1393835c4555..13046c3f38ea5 100644 --- a/datafusion/core/src/optimizer_rule_reference.md +++ b/datafusion/core/src/optimizer_rule_reference.md @@ -67,33 +67,41 @@ 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 7 are analyzer rules (`PhysicalOptimizerRule::phase` returns +`PhysicalOptimizerPhase::Analyzer`): they run before the plan meets its +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 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/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 198e5bb7412b2..9e160048a8bf3 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 enforced = check_phase_order(optimizers)?; debug!( "Input physical plan:\n{}\n", displayable(plan.as_ref()).indent(false) @@ -3494,6 +3496,7 @@ impl DefaultPhysicalPlanner { &new_plan, plan, optimizers, + enforced, &optimizer_context, )); } @@ -3706,19 +3709,52 @@ fn tuple_err(value: (Result, Result)) -> Result<(T, R)> { } } +/// 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], +) -> Result> { + 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 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(rules + .iter() + .rposition(|rule| rule.phase() == PhysicalOptimizerPhase::Enforcement)) +} + /// 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 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 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. fn sanity_check_error( error: DataFusionError, final_plan: &Arc, input_plan: Arc, optimizers: &[Arc], + enforced: Option, optimizer_context: &dyn PhysicalOptimizerContext, ) -> DataFusionError { const SANITY_CHECK: &str = "SanityCheckPlan"; @@ -3729,13 +3765,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. + // 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; @@ -3760,10 +3793,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 +4013,10 @@ mod tests { fn schema_check(&self) -> bool { true } + + fn phase(&self) -> PhysicalOptimizerPhase { + PhysicalOptimizerPhase::Optimizer + } } #[derive(Debug)] @@ -4250,6 +4287,189 @@ 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, + _config: &ConfigOptions, + ) -> Result> { + 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> { + let mut rules: Vec> = + 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( + "'noop' (position 1, phase Optimizer) runs before \ + 'OutputRequirements' (position 2, phase Analyzer)" + ), + "{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 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] + 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( + "'FilterPushdown(Post)' (position 21, phase Optimizer) runs before \ + 'noop' (position 22, phase Analyzer)" + ), + "{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 { let physical_plan = plan(logical_plan).await?; Ok(displayable(physical_plan.as_ref()).indent(true).to_string()) @@ -5934,6 +6154,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..cfe23daf46ae4 100644 --- a/datafusion/core/tests/physical_optimizer/sanity_checker.rs +++ b/datafusion/core/tests/physical_optimizer/sanity_checker.rs @@ -24,8 +24,10 @@ 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::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}; use datafusion_physical_plan::joins::{StreamJoinPartitionMode, SymmetricHashJoinExec}; use datafusion_physical_plan::repartition::RepartitionExec; use datafusion_physical_plan::sorts::sort::SortExec; @@ -1033,6 +1035,36 @@ impl PhysicalOptimizerRule for RemoveSorts { fn schema_check(&self) -> bool { true } + + fn phase(&self) -> PhysicalOptimizerPhase { + PhysicalOptimizerPhase::Optimizer + } +} + +/// [`RemoveSorts`] declared as an enforcement rule. +#[derive(Debug)] +struct RemoveSortsInEnforcement; + +impl PhysicalOptimizerRule for RemoveSortsInEnforcement { + fn optimize( + &self, + plan: Arc, + config: &ConfigOptions, + ) -> Result> { + RemoveSorts.optimize(plan, config) + } + + fn name(&self) -> &str { + "remove_sorts" + } + + fn schema_check(&self) -> bool { + true + } + + fn phase(&self) -> PhysicalOptimizerPhase { + PhysicalOptimizerPhase::Enforcement + } } async fn plan_window_query(state: SessionState) -> Result> { @@ -1087,3 +1119,50 @@ async fn test_final_check_without_ensure_requirements() -> Result<()> { ); Ok(()) } + +/// 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_enforcement_rule() -> Result<()> { + let mut rules = PhysicalOptimizer::new().rules; + let enforcement = rules + .iter() + .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) + .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(()) +} + +/// 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/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, + _config: &ConfigOptions, + ) -> Result> { + 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 { @@ -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 = + Arc::new(AnalyzerNoOpRule); + + let local_ffi = FFI_PhysicalOptimizerRule::new(Arc::clone(&rule), None); + let local: Arc = (&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 = (&foreign_ffi).into(); + assert_eq!(foreign.phase(), PhysicalOptimizerPhase::Optimizer); + } + #[test] fn test_clone() -> Result<()> { let rule: Arc = 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..03aeb9a29892f 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::Enforcement + } } // 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, 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..0a45bd94b7aae 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,22 @@ 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 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. /// -/// - [`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 +303,28 @@ impl PhysicalOptimizer { Self { rules } } } + +#[cfg(test)] +mod tests { + use super::*; + + /// The rules that run before `EnsureRequirements` declare the analyzer + /// 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; + let enforcement = rules + .iter() + .position(|rule| rule.name() == "EnsureRequirements") + .unwrap(); + for (idx, rule) in rules.iter().enumerate() { + 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/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: -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..a81beecf96ef3 100644 --- a/datafusion/session/src/physical_optimizer.rs +++ b/datafusion/session/src/physical_optimizer.rs @@ -176,4 +176,36 @@ 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 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 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. + 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/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, 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..94e992f86effb 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,57 @@ 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 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:** + +- 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. 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 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. + +- 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