Improve performance of collapse, from @akurmustafa

alamb#26

fix formatting

datafusion/physical-expr-common/Cargo.toml

@@ -41,4 +41,5 @@ arrow = { workspace = true }
 datafusion-common = { workspace = true, default-features = true }
 datafusion-expr-common = { workspace = true }
 hashbrown = { workspace = true }
+indexmap = { workspace = true }
 itertools = { workspace = true }

datafusion/physical-expr-common/src/sort_expr.rs

@@ -30,7 +30,8 @@ use arrow::datatypes::Schema;
 use arrow::record_batch::RecordBatch;
 use datafusion_common::Result;
 use datafusion_expr_common::columnar_value::ColumnarValue;
-use itertools::Itertools;
+use indexmap::IndexSet;
+use itertools::{izip, Itertools};
 
 /// Represents Sort operation for a column in a RecordBatch
 ///
@@ -568,11 +569,19 @@ impl LexRequirement {
     /// `vec![a Some(ASC)]`.
     pub fn collapse(self) -> Self {
         let mut output = Vec::<PhysicalSortRequirement>::new();
+        let mut exprs = IndexSet::new();
+        let mut reqs = vec![];
         for item in self {
-            if !output.iter().any(|req| req.expr.eq(&item.expr)) {
-                output.push(item);
+            let PhysicalSortRequirement { expr, options: req } = item;
+            // new insertion
+            if exprs.insert(expr) {
+                reqs.push(req);
             }
         }
+        debug_assert_eq!(reqs.len(), exprs.len());
+        for (expr, req) in izip!(exprs, reqs) {
+            output.push(PhysicalSortRequirement::new(expr, req));
+        }
         LexRequirement::new(output)
     }
 }