Add Optimizer Sanity Checker, improve sortedness equivalence properties

datafusion/core/src/physical_optimizer/mod.rs

@@ -30,10 +30,10 @@ pub mod join_selection;
 pub mod limited_distinct_aggregation;
 pub mod optimizer;
 pub mod output_requirements;
-pub mod pipeline_checker;
 pub mod projection_pushdown;
 pub mod pruning;
 pub mod replace_with_order_preserving_variants;
+pub mod sanity_checker;
 mod sort_pushdown;
 pub mod topk_aggregation;
 pub mod update_aggr_exprs;

datafusion/core/src/physical_optimizer/optimizer.rs

@@ -30,7 +30,7 @@ use crate::physical_optimizer::enforce_sorting::EnforceSorting;
 use crate::physical_optimizer::join_selection::JoinSelection;
 use crate::physical_optimizer::limited_distinct_aggregation::LimitedDistinctAggregation;
 use crate::physical_optimizer::output_requirements::OutputRequirements;
-use crate::physical_optimizer::pipeline_checker::PipelineChecker;
+use crate::physical_optimizer::sanity_checker::SanityCheckPlan;
 use crate::physical_optimizer::topk_aggregation::TopKAggregation;
 use crate::{error::Result, physical_plan::ExecutionPlan};
 
@@ -124,11 +124,15 @@ impl PhysicalOptimizer {
             // are not present, the load of executors such as join or union will be
             // reduced by narrowing their input tables.
             Arc::new(ProjectionPushdown::new()),
-            // The PipelineChecker rule will reject non-runnable query plans that use
-            // pipeline-breaking operators on infinite input(s). The rule generates a
-            // diagnostic error message when this happens. It makes no changes to the
-            // given query plan; i.e. it only acts as a final gatekeeping rule.
-            Arc::new(PipelineChecker::new()),
+            // The SanityCheckPlan rule checks whether the order and
+            // distribution requirements of each node in the plan
+            // is satisfied. It will also reject non-runnable query
+            // plans that use pipeline-breaking operators on infinite
+            // input(s). The rule generates a diagnostic error
+            // message for invalid plans. It makes no changes to the
+            // given query plan; i.e. it only acts as a final
+            // gatekeeping rule.
+            Arc::new(SanityCheckPlan::new()),
         ];
 
         Self::with_rules(rules)

datafusion/core/src/physical_optimizer/output_requirements.rs

@@ -248,7 +248,9 @@ fn require_top_ordering_helper(
     if children.len() != 1 {
         Ok((plan, false))
     } else if let Some(sort_exec) = plan.as_any().downcast_ref::<SortExec>() {
-        let req_ordering = sort_exec.properties().output_ordering().unwrap_or(&[]);
+        // In case of constant columns, output ordering of SortExec would give an empty set.
+        // Therefore; we check the sort expression field of the SortExec to assign the requirements.
+        let req_ordering = sort_exec.expr();
         let req_dist = sort_exec.required_input_distribution()[0].clone();
         let reqs = PhysicalSortRequirement::from_sort_exprs(req_ordering);
         Ok((