feat: eliminate the duplicated sort keys in Order By clause

datafusion/optimizer/src/eliminate_duplicated_expr.rs

@@ -0,0 +1,103 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 crate::optimizer::ApplyOrder;
+use crate::{OptimizerConfig, OptimizerRule};
+use datafusion_common::Result;
+use datafusion_expr::logical_plan::LogicalPlan;
+use datafusion_expr::Sort;
+use hashbrown::HashSet;
+
+/// Optimization rule that eliminate duplicated expr.
+#[derive(Default)]
+pub struct EliminateDuplicatedExpr;
+
+impl EliminateDuplicatedExpr {
+    #[allow(missing_docs)]
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl OptimizerRule for EliminateDuplicatedExpr {
+    fn try_optimize(
+        &self,
+        plan: &LogicalPlan,
+        _config: &dyn OptimizerConfig,
+    ) -> Result<Option<LogicalPlan>> {
+        match plan {
+            LogicalPlan::Sort(sort) => {
+                // dedup sort.expr and keep order
+                let mut dedup_expr = Vec::new();
+                let mut dedup_set = HashSet::new();
+                for expr in &sort.expr {
+                    if !dedup_set.contains(expr) {
+                        dedup_expr.push(expr);
+                        dedup_set.insert(expr.clone());
+                    }
+                }
+                if dedup_expr.len() == sort.expr.len() {
+                    Ok(None)
+                } else {
+                    Ok(Some(LogicalPlan::Sort(Sort {
+                        expr: dedup_expr.into_iter().cloned().collect::<Vec<_>>(),
+                        input: sort.input.clone(),
+                        fetch: sort.fetch,
+                    })))
+                }
+            }
+            _ => Ok(None),
+        }
+    }
+
+    fn name(&self) -> &str {
+        "eliminate_duplicated_expr"
+    }
+
+    fn apply_order(&self) -> Option<ApplyOrder> {
+        Some(ApplyOrder::TopDown)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test::*;
+    use datafusion_expr::{col, logical_plan::builder::LogicalPlanBuilder};
+    use std::sync::Arc;
+
+    fn assert_optimized_plan_eq(plan: &LogicalPlan, expected: &str) -> Result<()> {
+        crate::test::assert_optimized_plan_eq(
+            Arc::new(EliminateDuplicatedExpr::new()),
+            plan,
+            expected,
+        )
+    }
+
+    #[test]
+    fn eliminate_sort_expr() -> Result<()> {
+        let table_scan = test_table_scan().unwrap();
+        let plan = LogicalPlanBuilder::from(table_scan)
+            .sort(vec![col("a"), col("a"), col("b"), col("c")])?
+            .limit(5, Some(10))?
+            .build()?;
+        let expected = "Limit: skip=5, fetch=10\
+        \n  Sort: test.a, test.b, test.c\
+        \n    TableScan: test";
+        assert_optimized_plan_eq(&plan, expected)
+    }
+}

datafusion/optimizer/src/lib.rs

@@ -20,6 +20,7 @@ pub mod common_subexpr_eliminate;
 pub mod decorrelate_where_exists;
 pub mod decorrelate_where_in;
 pub mod eliminate_cross_join;
+pub mod eliminate_duplicated_expr;
 pub mod eliminate_filter;
 pub mod eliminate_limit;
 pub mod eliminate_outer_join;
@@ -33,17 +34,17 @@ pub mod propagate_empty_relation;
 pub mod push_down_filter;
 pub mod push_down_limit;
 pub mod push_down_projection;
+pub mod replace_distinct_aggregate;
+pub mod rewrite_disjunctive_predicate;
 pub mod scalar_subquery_to_join;
 pub mod simplify_expressions;
 pub mod single_distinct_to_groupby;
 pub mod type_coercion;
+pub mod unwrap_cast_in_comparison;
 pub mod utils;
 
-pub mod replace_distinct_aggregate;
-pub mod rewrite_disjunctive_predicate;
 #[cfg(test)]
 pub mod test;
-pub mod unwrap_cast_in_comparison;
 
 pub use optimizer::{OptimizerConfig, OptimizerContext, OptimizerRule};
 pub use utils::optimize_children;