feat(planner): Eliminate unnecessary sort (#14023)

eliminate unnecessary sort

src/query/sql/src/planner/optimizer/property/property.rs

@@ -16,6 +16,7 @@ use std::collections::HashSet;
 
 use super::column_stat::ColumnStatSet;
 use crate::plans::ScalarExpr;
+use crate::plans::SortItem;
 use crate::IndexType;
 
 pub type ColumnSet = HashSet<IndexType>;
@@ -59,6 +60,15 @@ pub struct RelationalProperty {
 
     /// Used columns of a relational expression
     pub used_columns: ColumnSet,
+
+    /// Ordering information of a relational expression
+    /// The sequence of sort items is important.
+    /// No ordering information is ensured if empty.
+    ///
+    /// TODO(leiysky): it's better to place ordering property
+    /// to the physical property, but at that time, we will have
+    /// to enforce the ordering property manually.
+    pub orderings: Vec<SortItem>,
 }
 
 #[derive(Default, Clone)]

src/query/sql/src/planner/optimizer/rule/factory.rs

@@ -28,6 +28,7 @@ use super::rewrite::RulePushDownLimitExpression;
 use super::rewrite::RulePushDownPrewhere;
 use super::rewrite::RuleTryApplyAggIndex;
 use crate::optimizer::rule::rewrite::RuleEliminateFilter;
+use crate::optimizer::rule::rewrite::RuleEliminateSort;
 use crate::optimizer::rule::rewrite::RuleMergeEvalScalar;
 use crate::optimizer::rule::rewrite::RuleMergeFilter;
 use crate::optimizer::rule::rewrite::RuleNormalizeAggregate;
@@ -87,6 +88,7 @@ impl RuleFactory {
             RuleID::EagerAggregation => Ok(Box::new(RuleEagerAggregation::new(metadata))),
             RuleID::PushDownPrewhere => Ok(Box::new(RulePushDownPrewhere::new(metadata))),
             RuleID::TryApplyAggIndex => Ok(Box::new(RuleTryApplyAggIndex::new(metadata))),
+            RuleID::EliminateSort => Ok(Box::new(RuleEliminateSort::new())),
         }
     }
 }

src/query/sql/src/planner/optimizer/rule/rewrite/mod.rs

@@ -17,6 +17,7 @@ mod filter_join;
 mod rule_commute_join;
 mod rule_eliminate_eval_scalar;
 mod rule_eliminate_filter;
+mod rule_eliminate_sort;
 mod rule_fold_count_aggregate;
 mod rule_infer_filter;
 mod rule_merge_eval_scalar;
@@ -45,6 +46,7 @@ mod rule_try_apply_agg_index;
 pub use rule_commute_join::RuleCommuteJoin;
 pub use rule_eliminate_eval_scalar::RuleEliminateEvalScalar;
 pub use rule_eliminate_filter::RuleEliminateFilter;
+pub use rule_eliminate_sort::RuleEliminateSort;
 pub use rule_fold_count_aggregate::RuleFoldCountAggregate;
 pub use rule_infer_filter::RuleInferFilter;
 pub use rule_merge_eval_scalar::RuleMergeEvalScalar;

src/query/sql/src/planner/optimizer/rule/rewrite/rule_eliminate_sort.rs

@@ -0,0 +1,81 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use common_exception::Result;
+
+use crate::optimizer::rule::Rule;
+use crate::optimizer::rule::RuleID;
+use crate::optimizer::rule::TransformResult;
+use crate::optimizer::RelExpr;
+use crate::optimizer::SExpr;
+use crate::plans::PatternPlan;
+use crate::plans::RelOp;
+use crate::plans::Sort;
+
+pub struct RuleEliminateSort {
+    id: RuleID,
+    patterns: Vec<SExpr>,
+}
+
+impl RuleEliminateSort {
+    pub fn new() -> Self {
+        Self {
+            id: RuleID::EliminateSort,
+            // Sort
+            //  \
+            //   *
+            patterns: vec![SExpr::create_unary(
+                Arc::new(
+                    PatternPlan {
+                        plan_type: RelOp::Sort,
+                    }
+                    .into(),
+                ),
+                Arc::new(SExpr::create_leaf(Arc::new(
+                    PatternPlan {
+                        plan_type: RelOp::Pattern,
+                    }
+                    .into(),
+                ))),
+            )],
+        }
+    }
+}
+
+impl Rule for RuleEliminateSort {
+    fn id(&self) -> RuleID {
+        self.id
+    }
+
+    fn apply(&self, s_expr: &SExpr, state: &mut TransformResult) -> Result<()> {
+        let sort: Sort = s_expr.plan().clone().try_into()?;
+        let input = s_expr.child(0)?;
+        let rel_expr = RelExpr::with_s_expr(input);
+        let prop = rel_expr.derive_relational_prop()?;
+
+        if prop.orderings == sort.items {
+            // If the derived ordering is completely equal to
+            // the sort operator, we can eliminate the sort.
+            state.add_result(input.clone());
+        }
+
+        Ok(())
+    }
+
+    fn patterns(&self) -> &Vec<SExpr> {
+        &self.patterns
+    }
+}

src/query/sql/src/planner/optimizer/rule/rule.rs

@@ -29,6 +29,7 @@ pub static DEFAULT_REWRITE_RULES: LazyLock<Vec<RuleID>> = LazyLock::new(|| {
         RuleID::NormalizeScalarFilter,
         RuleID::NormalizeAggregate,
         RuleID::EliminateFilter,
+        RuleID::EliminateSort,
         RuleID::MergeFilter,
         RuleID::InferFilter,
         RuleID::MergeEvalScalar,
@@ -95,6 +96,7 @@ pub enum RuleID {
     PushDownSortScan,
     EliminateEvalScalar,
     EliminateFilter,
+    EliminateSort,
     MergeEvalScalar,
     MergeFilter,
     SplitAggregate,
@@ -128,6 +130,7 @@ impl Display for RuleID {
             RuleID::PushDownSortScan => write!(f, "PushDownSortScan"),
             RuleID::EliminateEvalScalar => write!(f, "EliminateEvalScalar"),
             RuleID::EliminateFilter => write!(f, "EliminateFilter"),
+            RuleID::EliminateSort => write!(f, "EliminateSort"),
             RuleID::MergeEvalScalar => write!(f, "MergeEvalScalar"),
             RuleID::MergeFilter => write!(f, "MergeFilter"),
             RuleID::NormalizeScalarFilter => write!(f, "NormalizeScalarFilter"),

src/query/sql/src/planner/plans/aggregate.rs

@@ -183,6 +183,7 @@ impl Operator for Aggregate {
             output_columns,
             outer_columns,
             used_columns,
+            orderings: vec![],
         }))
     }
 

src/query/sql/src/planner/plans/constant_table_scan.rs

@@ -110,6 +110,7 @@ impl Operator for ConstantTableScan {
             output_columns: self.columns.clone(),
             outer_columns: Default::default(),
             used_columns: self.columns.clone(),
+            orderings: vec![],
         }))
     }
 

src/query/sql/src/planner/plans/cte_scan.rs

@@ -72,6 +72,7 @@ impl Operator for CteScan {
             output_columns: self.used_columns()?,
             outer_columns: ColumnSet::new(),
             used_columns: self.used_columns()?,
+            orderings: vec![],
         }))
     }
 

src/query/sql/src/planner/plans/dummy_table_scan.rs

@@ -48,6 +48,7 @@ impl Operator for DummyTableScan {
             output_columns: ColumnSet::from([DUMMY_COLUMN_INDEX]),
             outer_columns: ColumnSet::new(),
             used_columns: ColumnSet::new(),
+            orderings: vec![],
         }))
     }
 

src/query/sql/src/planner/plans/eval_scalar.rs

@@ -95,10 +95,14 @@ impl Operator for EvalScalar {
         let mut used_columns = self.used_columns()?;
         used_columns.extend(input_prop.used_columns.clone());
 
+        // Derive orderings
+        let orderings = input_prop.orderings.clone();
+
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }
 

src/query/sql/src/planner/plans/filter.rs

@@ -86,10 +86,14 @@ impl Operator for Filter {
         let mut used_columns = self.used_columns()?;
         used_columns.extend(input_prop.used_columns.clone());
 
+        // Derive orderings
+        let orderings = input_prop.orderings.clone();
+
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }
 

src/query/sql/src/planner/plans/join.rs

@@ -390,10 +390,14 @@ impl Operator for Join {
         used_columns.extend(left_prop.used_columns.clone());
         used_columns.extend(right_prop.used_columns.clone());
 
+        // Derive orderings
+        let orderings = vec![];
+
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }
 

src/query/sql/src/planner/plans/materialized_cte.rs

@@ -45,11 +45,13 @@ impl Operator for MaterializedCte {
         let output_columns = right_prop.output_columns.clone();
         let outer_columns = right_prop.outer_columns.clone();
         let used_columns = right_prop.used_columns.clone();
+        let orderings = right_prop.orderings.clone();
 
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }
 

src/query/sql/src/planner/plans/project_set.rs

@@ -77,10 +77,14 @@ impl Operator for ProjectSet {
             );
         }
 
+        // Derive orderings
+        let orderings = vec![];
+
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }
 

src/query/sql/src/planner/plans/scan.rs

@@ -169,6 +169,7 @@ impl Operator for Scan {
             output_columns: self.columns.clone(),
             outer_columns: Default::default(),
             used_columns: self.used_columns(),
+            orderings: vec![],
         }))
     }
 

src/query/sql/src/planner/plans/sort.rs

@@ -70,7 +70,21 @@ impl Operator for Sort {
     }
 
     fn derive_relational_prop(&self, rel_expr: &RelExpr) -> Result<Arc<RelationalProperty>> {
-        rel_expr.derive_relational_prop_child(0)
+        let input_prop = rel_expr.derive_relational_prop_child(0)?;
+
+        let output_columns = input_prop.output_columns.clone();
+        let outer_columns = input_prop.outer_columns.clone();
+        let used_columns = input_prop.used_columns.clone();
+
+        // Derive orderings
+        let orderings = self.items.clone();
+
+        Ok(Arc::new(RelationalProperty {
+            output_columns,
+            outer_columns,
+            used_columns,
+            orderings,
+        }))
     }
 
     fn derive_cardinality(&self, rel_expr: &RelExpr) -> Result<Arc<StatInfo>> {

src/query/sql/src/planner/plans/udf.rs

@@ -74,10 +74,14 @@ impl Operator for Udf {
         let mut used_columns = self.used_columns()?;
         used_columns.extend(input_prop.used_columns.clone());
 
+        // Derive orderings
+        let orderings = input_prop.orderings.clone();
+
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }
 

src/query/sql/src/planner/plans/union_all.rs

@@ -74,10 +74,14 @@ impl Operator for UnionAll {
         used_columns.extend(left_prop.used_columns.clone());
         used_columns.extend(right_prop.used_columns.clone());
 
+        // Derive orderings
+        let orderings = vec![];
+
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }
 

src/query/sql/src/planner/plans/window.rs

@@ -131,10 +131,14 @@ impl Operator for Window {
         let mut used_columns = self.used_columns()?;
         used_columns.extend(input_prop.used_columns.clone());
 
+        // Derive orderings
+        let orderings = input_prop.orderings.clone();
+
         Ok(Arc::new(RelationalProperty {
             output_columns,
             outer_columns,
             used_columns,
+            orderings,
         }))
     }