From 7a1e03019457073c1bc6919de84b9d14dc802887 Mon Sep 17 00:00:00 2001
From: Mustafa Akur <106137913+mustafasrepo@users.noreply.github.com>
Date: Fri, 17 Nov 2023 08:59:00 +0300
Subject: [PATCH 1/7] Complex exprs requirement support (#215)

* Discover ordering of complex expressions in group by and window partition by

* Remove unnecessary tests

* Update comments

* Minor changes

* Better projection support complex expression support

* Fix failing test

* Simplifications

* Simplifications

* Add is end flag

* Simplifications

* Simplifications

* Simplifications

* Minor changes

* Minor changes

* Minor changes

* All tests pass

* Change implementation of find_longest_permutation

* Minor changes

* Minor changes

* Remove projection section

* Remove projection implementation

* Fix linter errors

* Remove projection sections

* Minor changes

* Add docstring comments

* Add comments

* Minor changes

* Minor changes

* Add comments

* simplifications
---
 .../enforce_distribution.rs                   |    8 +-
 datafusion/physical-expr/src/equivalence.rs   | 1039 +++++++++++++----
 .../physical-expr/src/sort_properties.rs      |    1 -
 datafusion/physical-plan/src/projection.rs    |    2 +-
 .../sqllogictest/test_files/groupby.slt       |   45 +
 5 files changed, 893 insertions(+), 202 deletions(-)

diff --git a/datafusion/core/src/physical_optimizer/enforce_distribution.rs b/datafusion/core/src/physical_optimizer/enforce_distribution.rs
index 4aedc3b0d1a9..a34958a6c96d 100644
--- a/datafusion/core/src/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/src/physical_optimizer/enforce_distribution.rs
@@ -3787,7 +3787,7 @@ pub(crate) mod tests {
     fn repartition_transitively_past_sort_with_projection_and_filter() -> Result<()> {
         let schema = schema();
         let sort_key = vec![PhysicalSortExpr {
-            expr: col("c", &schema).unwrap(),
+            expr: col("a", &schema).unwrap(),
             options: SortOptions::default(),
         }];
         let plan = sort_exec(
@@ -3804,9 +3804,9 @@ pub(crate) mod tests {
         );
 
         let expected = &[
-            "SortPreservingMergeExec: [c@2 ASC]",
+            "SortPreservingMergeExec: [a@0 ASC]",
             // Expect repartition on the input to the sort (as it can benefit from additional parallelism)
-            "SortExec: expr=[c@2 ASC]",
+            "SortExec: expr=[a@0 ASC]",
             "ProjectionExec: expr=[a@0 as a, b@1 as b, c@2 as c]",
             "FilterExec: c@2 = 0",
             // repartition is lowest down
@@ -3817,7 +3817,7 @@ pub(crate) mod tests {
         assert_optimized!(expected, plan.clone(), true);
 
         let expected_first_sort_enforcement = &[
-            "SortExec: expr=[c@2 ASC]",
+            "SortExec: expr=[a@0 ASC]",
             "CoalescePartitionsExec",
             "ProjectionExec: expr=[a@0 as a, b@1 as b, c@2 as c]",
             "FilterExec: c@2 = 0",
diff --git a/datafusion/physical-expr/src/equivalence.rs b/datafusion/physical-expr/src/equivalence.rs
index f3bfe4961622..87031b12a949 100644
--- a/datafusion/physical-expr/src/equivalence.rs
+++ b/datafusion/physical-expr/src/equivalence.rs
@@ -15,26 +15,23 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::HashSet;
 use std::hash::Hash;
 use std::sync::Arc;
 
-use crate::expressions::Column;
+use crate::expressions::{Column, Literal};
+use crate::physical_expr::deduplicate_physical_exprs;
 use crate::sort_properties::{ExprOrdering, SortProperties};
 use crate::{
     physical_exprs_bag_equal, physical_exprs_contains, LexOrdering, LexOrderingRef,
     LexRequirement, LexRequirementRef, PhysicalExpr, PhysicalSortExpr,
     PhysicalSortRequirement,
 };
-
 use arrow::datatypes::SchemaRef;
 use arrow_schema::SortOptions;
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_common::{JoinSide, JoinType, Result};
 
-use crate::physical_expr::deduplicate_physical_exprs;
-use indexmap::map::Entry;
-use indexmap::IndexMap;
+use indexmap::IndexSet;
 
 /// An `EquivalenceClass` is a set of [`Arc<dyn PhysicalExpr>`]s that are known
 /// to have the same value for all tuples in a relation. These are generated by
@@ -196,6 +193,32 @@ impl ProjectionMapping {
     ) -> impl Iterator<Item = &(Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)> + '_ {
         self.inner.iter()
     }
+
+    /// This function returns target value for given expression
+    ///
+    /// # Arguments
+    ///
+    /// * `expr` - Source (e.g key) physical expression
+    ///
+    /// # Returns
+    ///
+    /// An `Option` containing a the target (e.g value) for the source expression.
+    /// `None` means that source is not found inside the mapping.
+    pub fn target_expr(
+        &self,
+        expr: &Arc<dyn PhysicalExpr>,
+    ) -> Option<Arc<dyn PhysicalExpr>> {
+        if let Some(idx) = self
+            .inner
+            .iter()
+            .position(|(source, _target)| source.eq(expr))
+        {
+            let (_source, target) = &self.inner[idx];
+            Some(target.clone())
+        } else {
+            None
+        }
+    }
 }
 
 /// An `EquivalenceGroup` is a collection of `EquivalenceClass`es where each
@@ -413,24 +436,32 @@ impl EquivalenceGroup {
         mapping: &ProjectionMapping,
         expr: &Arc<dyn PhysicalExpr>,
     ) -> Option<Arc<dyn PhysicalExpr>> {
-        let children = expr.children();
-        if children.is_empty() {
+        // First, we try to project expressions with an exact match. If we are
+        // unable to do this, we consult equivalence classes.
+        if let Some(target) = mapping.target_expr(expr) {
+            // If we match the source, we can project directly:
+            return Some(target);
+        } else {
+            // If the given expression is not inside the mapping, try to project
+            // expressions considering the equivalence classes.
             for (source, target) in mapping.iter() {
-                // If we match the source, or an equivalent expression to source,
+                // If we match an equivalent expression to source,
                 // then we can project. For example, if we have the mapping
-                // (a as a1, a + c) and the equivalence class (a, b), expression
-                // b also projects to a1.
-                if source.eq(expr)
-                    || self
-                        .get_equivalence_class(source)
-                        .map_or(false, |group| group.contains(expr))
+                // (a as a1, a + c) and the equivalence class (a, b), expression `b` projects to `a1`.
+                if self
+                    .get_equivalence_class(source)
+                    .map_or(false, |group| group.contains(expr))
                 {
                     return Some(target.clone());
                 }
             }
         }
         // Project a non-leaf expression by projecting its children.
-        else if let Some(children) = children
+        let children = expr.children();
+        if children.is_empty() {
+            // Leaf expression should be inside mapping.
+            return None;
+        } else if let Some(children) = children
             .into_iter()
             .map(|child| self.project_expr(mapping, &child))
             .collect::<Option<Vec<_>>>()
@@ -660,9 +691,27 @@ impl OrderingEquivalenceClass {
     /// For instance, If we already have the ordering [a ASC, b ASC, c DESC],
     /// then there is no need to keep ordering [a ASC, b ASC] in the state.
     fn remove_redundant_entries(&mut self) {
+        let leading_ordering_exprs = self
+            .orderings
+            .iter()
+            .flat_map(|ordering| ordering.first().map(|sort_expr| sort_expr.expr.clone()))
+            .collect::<Vec<_>>();
+
+        // Remove leading orderings that are beyond index 0, to simplify ordering.
+        self.orderings.iter_mut().for_each(|ordering| {
+            let mut counter = 0;
+            ordering.retain(|sort_expr| {
+                // Either first entry or is not leading ordering
+                let should_retain = counter == 0
+                    || !physical_exprs_contains(&leading_ordering_exprs, &sort_expr.expr);
+                counter += 1;
+                should_retain
+            });
+        });
+
         let mut idx = 0;
         while idx < self.orderings.len() {
-            let mut removal = false;
+            let mut removal = self.orderings[idx].is_empty();
             for (ordering_idx, ordering) in self.orderings[0..idx].iter().enumerate() {
                 if let Some(right_finer) = finer_side(ordering, &self.orderings[idx]) {
                     if right_finer {
@@ -985,36 +1034,56 @@ impl EquivalenceProperties {
     /// Checks whether the given sort requirements are satisfied by any of the
     /// existing orderings.
     pub fn ordering_satisfy_requirement(&self, reqs: LexRequirementRef) -> bool {
+        let mut eq_properties = self.clone();
         // First, standardize the given requirement:
-        let normalized_reqs = self.normalize_sort_requirements(reqs);
-        if normalized_reqs.is_empty() {
-            // Requirements are tautologically satisfied if empty.
-            return true;
+        let normalized_reqs = eq_properties.normalize_sort_requirements(reqs);
+        for normalized_req in normalized_reqs {
+            // Check whether given ordering is satisfied
+            if !eq_properties.ordering_satisfy_single_req(&normalized_req) {
+                return false;
+            }
+            // - Treat satisfied ordering as constant in the next iterations. Since in lexicographical ordering
+            //   next orderings are only considered as long as their left side have same values
+            //   (e.g for them their left side is constant).
+            //
+            // Please note that, these expressions are not properly constant. This is just implementation
+            // and this interpretation doesn't effect outside users in anyway.
+            //
+            // As an example:
+            // If the requirement is `[a ASC, b + c ASC]`.
+            // and existing orderings are `[a ASC, b ASC], [c ASC]`.
+            // From the analysis above, we know that `[a ASC]` is satisfied.
+            // Then here, we add column `a` as constant to the state.
+            // This enables us to deduce that `b + c ASC` is satisfied, given `a` is constant.
+            eq_properties =
+                eq_properties.add_constants(std::iter::once(normalized_req.expr));
         }
-        let mut indices = HashSet::new();
-        for ordering in self.normalized_oeq_class().iter() {
-            let match_indices = ordering
-                .iter()
-                .map(|sort_expr| {
-                    normalized_reqs
-                        .iter()
-                        .position(|sort_req| sort_expr.satisfy(sort_req, &self.schema))
-                })
-                .collect::<Vec<_>>();
-            // Find the largest contiguous increasing sequence starting from the first index:
-            if let Some(&Some(first)) = match_indices.first() {
-                indices.insert(first);
-                let mut iter = match_indices.windows(2);
-                while let Some([Some(current), Some(next)]) = iter.next() {
-                    if next > current {
-                        indices.insert(*next);
-                    } else {
-                        break;
-                    }
-                }
+        true
+    }
+
+    /// Determines whether the ordering specified by a given `PhysicalSortRequirement` is satisfied
+    /// based on the internal orderings, equivalent classes, and constant expressions.
+    ///
+    /// # Arguments
+    ///
+    /// - `req`: A reference to a `PhysicalSortRequirement` for which the ordering satisfaction
+    ///   needs to be determined.
+    ///
+    /// # Returns
+    ///
+    /// Returns `true` if the specified ordering is satisfied; otherwise, returns `false`.
+    fn ordering_satisfy_single_req(&self, req: &PhysicalSortRequirement) -> bool {
+        let expr_ordering = self.get_expr_ordering(req.expr.clone());
+        let ExprOrdering { expr, state, .. } = expr_ordering;
+        match state {
+            SortProperties::Ordered(options) => {
+                let sort_expr = PhysicalSortExpr { expr, options };
+                sort_expr.satisfy(req, self.schema())
             }
+            // Singleton expressions satisfies any ordering.
+            SortProperties::Singleton => true,
+            SortProperties::Unordered => false,
         }
-        indices.len() == normalized_reqs.len()
     }
 
     /// Checks whether the `given`` sort requirements are equal or more specific
@@ -1138,6 +1207,43 @@ impl EquivalenceProperties {
         self.eq_group.project_expr(projection_mapping, expr)
     }
 
+    /// Projects constants based on the provided `ProjectionMapping`.
+    ///
+    /// This function takes a `ProjectionMapping` and identifies and projects constants based on
+    /// the existing constants and the mapping. It ensures that constants are appropriately
+    /// propagated through the projection expressions.
+    ///
+    /// # Arguments
+    ///
+    /// - `mapping`: A reference to a `ProjectionMapping` representing the mapping of source
+    ///   expressions to target expressions in the projection.
+    ///
+    /// # Returns
+    ///
+    /// Returns a vector of `Arc<dyn PhysicalExpr>` containing the projected constants.
+    fn projected_constants(
+        &self,
+        mapping: &ProjectionMapping,
+    ) -> Vec<Arc<dyn PhysicalExpr>> {
+        // Project existing constants
+        // As an example assume that a+b is known to be constant. If projection were:
+        // `a as a_new`, `b as b_new`; we would project constant `a+b` as `a_new+b_new`
+        let mut projected_constants = self
+            .constants
+            .iter()
+            .flat_map(|expr| self.eq_group.project_expr(mapping, expr))
+            .collect::<Vec<_>>();
+        // Add projection expressions that are known to be constant.
+        for (source, target) in mapping.iter() {
+            if self.is_expr_constant(source)
+                && !physical_exprs_contains(&projected_constants, target)
+            {
+                projected_constants.push(target.clone());
+            }
+        }
+        projected_constants
+    }
+
     /// Projects the equivalences within according to `projection_mapping`
     /// and `output_schema`.
     pub fn project(
@@ -1145,6 +1251,9 @@ impl EquivalenceProperties {
         projection_mapping: &ProjectionMapping,
         output_schema: SchemaRef,
     ) -> Self {
+        let projected_constants = self.projected_constants(projection_mapping);
+        let projected_eq_group = self.eq_group.project(projection_mapping);
+
         let mut projected_orderings = self
             .oeq_class
             .iter()
@@ -1152,7 +1261,8 @@ impl EquivalenceProperties {
             .collect::<Vec<_>>();
         for (source, target) in projection_mapping.iter() {
             let expr_ordering = ExprOrdering::new(source.clone())
-                .transform_up(&|expr| update_ordering(expr, self))
+                .transform_up(&|expr| Ok(update_ordering(expr, self)))
+                // It is guaranteed to always return Ok.
                 .unwrap();
             if let SortProperties::Ordered(options) = expr_ordering.state {
                 // Push new ordering to the state.
@@ -1163,9 +1273,9 @@ impl EquivalenceProperties {
             }
         }
         Self {
-            eq_group: self.eq_group.project(projection_mapping),
+            eq_group: projected_eq_group,
             oeq_class: OrderingEquivalenceClass::new(projected_orderings),
-            constants: vec![],
+            constants: projected_constants,
             schema: output_schema,
         }
     }
@@ -1184,38 +1294,125 @@ impl EquivalenceProperties {
         &self,
         exprs: &[Arc<dyn PhysicalExpr>],
     ) -> (LexOrdering, Vec<usize>) {
-        let normalized_exprs = self.eq_group.normalize_exprs(exprs.to_vec());
-        // Use a map to associate expression indices with sort options:
-        let mut ordered_exprs = IndexMap::<usize, SortOptions>::new();
-        for ordering in self.normalized_oeq_class().iter() {
-            for sort_expr in ordering {
-                if let Some(idx) = normalized_exprs
-                    .iter()
-                    .position(|expr| sort_expr.expr.eq(expr))
-                {
-                    if let Entry::Vacant(e) = ordered_exprs.entry(idx) {
-                        e.insert(sort_expr.options);
+        let mut eq_properties = self.clone();
+        let mut result = vec![];
+        // Algorithm is as follows:
+        // - 1. Iterate over all expressions and insert the expressions that are known to be ordered
+        //      into result_ordering.
+        // - 2. Treat inserted expressions as constants (add them as constant to the state)
+        // - 3. Go back to step 1.
+        // - Continue the above iteration as long as no new expression is inserted (Algorithm reached a fixed point).
+
+        let mut search_indices = (0..exprs.len()).collect::<IndexSet<_>>();
+        // Algorithm above should reach a fixed point at-most `exprs.len()` number of iterations.
+        // We could have use loop{}, However, to guarantee we exit anyway (in-case of bugs).
+        // We use upper limit number of iterations (worst-case).
+        for _idx in 0..exprs.len() {
+            // Get ordered expressions with their indices.
+            let ordered_exprs = search_indices
+                .iter()
+                .flat_map(|&idx| {
+                    let expr = exprs[idx].clone();
+                    let ExprOrdering { expr, state, .. } =
+                        eq_properties.get_expr_ordering(expr);
+                    if let SortProperties::Ordered(options) = state {
+                        let sort_expr = PhysicalSortExpr { expr, options };
+                        Some((sort_expr, idx))
+                    } else {
+                        None
                     }
-                } else {
-                    // We only consider expressions that correspond to a prefix
-                    // of one of the equivalent orderings we have.
-                    break;
-                }
+                })
+                .collect::<Vec<_>>();
+
+            // We reached a fixed point, exit.
+            if ordered_exprs.is_empty() {
+                break;
             }
+
+            // - Remove indices that have an ordering from search_indices
+            // - Treat ordered expressions as constant in the next iterations. Since in lexicographical ordering
+            //   next orderings are only considered as long as their left side have same values
+            //   (e.g for them their left side is constant).
+            //
+            // Please note that, these expressions are not properly constant. This is just implementation
+            // and this interpretation doesn't effect outside users in anyway.
+            for (sort_expr, idx) in &ordered_exprs {
+                eq_properties =
+                    eq_properties.add_constants(std::iter::once(sort_expr.expr.clone()));
+                search_indices.remove(idx);
+            }
+
+            // Add new ordered section to the state.
+            result.extend(ordered_exprs);
         }
-        // Construct the lexicographical ordering according to the permutation:
-        ordered_exprs
-            .into_iter()
-            .map(|(idx, options)| {
-                (
-                    PhysicalSortExpr {
-                        expr: exprs[idx].clone(),
-                        options,
-                    },
-                    idx,
-                )
-            })
-            .unzip()
+        result.into_iter().unzip()
+    }
+
+    /// Checks whether a given expression is constant.
+    ///
+    /// This function determines whether the provided expression is constant based on the known constants.
+    ///
+    /// # Arguments
+    ///
+    /// - `expr`: A reference to a `Arc<dyn PhysicalExpr>` representing the expression to be checked.
+    ///
+    /// # Returns
+    ///
+    /// Returns `true` if the expression is constant within the equivalence group; otherwise, returns `false`.
+    fn is_expr_constant(&self, expr: &Arc<dyn PhysicalExpr>) -> bool {
+        // As an example, assume that we know columns [a, b] are constant.
+        // `a`, `b`, `a+b` will all return `true`, whereas `c` will return `false`.
+        let normalized_constants = self.eq_group.normalize_exprs(self.constants.to_vec());
+        let normalized_expr = self.eq_group.normalize_expr(expr.clone());
+        is_expr_constant_util(&normalized_constants, &normalized_expr)
+    }
+
+    /// Retrieves the ordering information for a given physical expression.
+    ///
+    /// This function constructs an `ExprOrdering` object for the provided expression, which encapsulates
+    /// information about the expression's ordering, including its state and associated options.
+    ///
+    /// # Arguments
+    ///
+    /// - `expr`: An `Arc<dyn PhysicalExpr>` representing the physical expression for which ordering information is sought.
+    ///
+    /// # Returns
+    ///
+    /// Returns an `ExprOrdering` object containing the ordering information for the given expression.
+    pub fn get_expr_ordering(&self, expr: Arc<dyn PhysicalExpr>) -> ExprOrdering {
+        let expr_ordering = ExprOrdering::new(expr.clone());
+        expr_ordering
+            .transform_up(&|expr| Ok(update_ordering(expr, self)))
+            // It is guaranteed to always return Ok.
+            .unwrap()
+    }
+}
+
+/// Checks whether a given expression is constant. This function determines whether the
+/// provided expression is constant based on the known constants.
+///
+/// # Arguments
+///
+/// - `constants`: A reference to a `[Arc<dyn PhysicalExpr>]` containing expressions known to be a constant.
+/// - `expr`: A reference to a `Arc<dyn PhysicalExpr>` representing the expression to be checked.
+///
+/// # Returns
+///
+/// Returns `true` if the expression is constant within the equivalence group; otherwise, returns `false`.
+fn is_expr_constant_util(
+    constants: &[Arc<dyn PhysicalExpr>],
+    expr: &Arc<dyn PhysicalExpr>,
+) -> bool {
+    if physical_exprs_contains(constants, expr) {
+        return true;
+    }
+    let children = expr.children();
+    if children.is_empty() {
+        false
+    } else {
+        children
+            .iter()
+            .all(|child| is_expr_constant_util(constants, child))
     }
 }
 
@@ -1330,26 +1527,28 @@ fn updated_right_ordering_equivalence_class(
 fn update_ordering(
     mut node: ExprOrdering,
     eq_properties: &EquivalenceProperties,
-) -> Result<Transformed<ExprOrdering>> {
+) -> Transformed<ExprOrdering> {
+    // We have a Column, which is one of the two possible leaf node types:
+    let eq_group = &eq_properties.eq_group;
+    let normalized_expr = eq_group.normalize_expr(node.expr.clone());
+    let oeq_class = &eq_properties.normalized_oeq_class();
+    if eq_properties.is_expr_constant(&normalized_expr) {
+        node.state = SortProperties::Singleton;
+        return Transformed::Yes(node);
+    } else if let Some(options) = oeq_class.get_options(&normalized_expr) {
+        node.state = SortProperties::Ordered(options);
+        return Transformed::Yes(node);
+    }
     if !node.expr.children().is_empty() {
         // We have an intermediate (non-leaf) node, account for its children:
         node.state = node.expr.get_ordering(&node.children_states);
-        Ok(Transformed::Yes(node))
-    } else if node.expr.as_any().is::<Column>() {
-        // We have a Column, which is one of the two possible leaf node types:
-        let eq_group = &eq_properties.eq_group;
-        let normalized_expr = eq_group.normalize_expr(node.expr.clone());
-        let oeq_class = &eq_properties.oeq_class;
-        if let Some(options) = oeq_class.get_options(&normalized_expr) {
-            node.state = SortProperties::Ordered(options);
-            Ok(Transformed::Yes(node))
-        } else {
-            Ok(Transformed::No(node))
-        }
-    } else {
+        Transformed::Yes(node)
+    } else if node.expr.as_any().is::<Literal>() {
         // We have a Literal, which is the other possible leaf node type:
         node.state = node.expr.get_ordering(&[]);
-        Ok(Transformed::Yes(node))
+        Transformed::Yes(node)
+    } else {
+        Transformed::No(node)
     }
 }
 
@@ -1359,14 +1558,16 @@ mod tests {
     use std::sync::Arc;
 
     use super::*;
+    use crate::execution_props::ExecutionProps;
     use crate::expressions::{col, lit, BinaryExpr, Column, Literal};
+    use crate::functions::create_physical_expr;
 
     use arrow::compute::{lexsort_to_indices, SortColumn};
     use arrow::datatypes::{DataType, Field, Schema};
-    use arrow_array::{ArrayRef, RecordBatch, UInt32Array, UInt64Array};
-    use arrow_schema::{Fields, SortOptions};
+    use arrow_array::{ArrayRef, Float64Array, RecordBatch, UInt32Array};
+    use arrow_schema::{Fields, SortOptions, TimeUnit};
     use datafusion_common::{Result, ScalarValue};
-    use datafusion_expr::Operator;
+    use datafusion_expr::{BuiltinScalarFunction, Operator};
 
     use itertools::{izip, Itertools};
     use rand::rngs::StdRng;
@@ -1432,12 +1633,12 @@ mod tests {
 
     // Generate a schema which consists of 6 columns (a, b, c, d, e, f)
     fn create_test_schema_2() -> Result<SchemaRef> {
-        let a = Field::new("a", DataType::Int32, true);
-        let b = Field::new("b", DataType::Int32, true);
-        let c = Field::new("c", DataType::Int32, true);
-        let d = Field::new("d", DataType::Int32, true);
-        let e = Field::new("e", DataType::Int32, true);
-        let f = Field::new("f", DataType::Int32, true);
+        let a = Field::new("a", DataType::Float64, true);
+        let b = Field::new("b", DataType::Float64, true);
+        let c = Field::new("c", DataType::Float64, true);
+        let d = Field::new("d", DataType::Float64, true);
+        let e = Field::new("e", DataType::Float64, true);
+        let f = Field::new("f", DataType::Float64, true);
         let schema = Arc::new(Schema::new(vec![a, b, c, d, e, f]));
 
         Ok(schema)
@@ -1602,19 +1803,20 @@ mod tests {
             Field::new("a4", DataType::Int64, true),
         ]));
 
+        // a as a1, a as a2, a as a3, a as a3
+        let proj_exprs = vec![
+            (col_a.clone(), "a1".to_string()),
+            (col_a.clone(), "a2".to_string()),
+            (col_a.clone(), "a3".to_string()),
+            (col_a.clone(), "a4".to_string()),
+        ];
+        let projection_mapping = ProjectionMapping::try_new(&proj_exprs, &input_schema)?;
+
         // a as a1, a as a2, a as a3, a as a3
         let col_a1 = &col("a1", &out_schema)?;
         let col_a2 = &col("a2", &out_schema)?;
         let col_a3 = &col("a3", &out_schema)?;
         let col_a4 = &col("a4", &out_schema)?;
-        let projection_mapping = ProjectionMapping {
-            inner: vec![
-                (col_a.clone(), col_a1.clone()),
-                (col_a.clone(), col_a2.clone()),
-                (col_a.clone(), col_a3.clone()),
-                (col_a.clone(), col_a4.clone()),
-            ],
-        };
         let out_properties = input_properties.project(&projection_mapping, out_schema);
 
         // At the output a1=a2=a3=a4
@@ -1631,6 +1833,10 @@ mod tests {
 
     #[test]
     fn test_ordering_satisfy() -> Result<()> {
+        let input_schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int64, true),
+            Field::new("b", DataType::Int64, true),
+        ]));
         let crude = vec![PhysicalSortExpr {
             expr: Arc::new(Column::new("a", 0)),
             options: SortOptions::default(),
@@ -1646,13 +1852,12 @@ mod tests {
             },
         ];
         // finer ordering satisfies, crude ordering should return true
-        let empty_schema = &Arc::new(Schema::empty());
-        let mut eq_properties_finer = EquivalenceProperties::new(empty_schema.clone());
+        let mut eq_properties_finer = EquivalenceProperties::new(input_schema.clone());
         eq_properties_finer.oeq_class.push(finer.clone());
         assert!(eq_properties_finer.ordering_satisfy(&crude));
 
         // Crude ordering doesn't satisfy finer ordering. should return false
-        let mut eq_properties_crude = EquivalenceProperties::new(empty_schema.clone());
+        let mut eq_properties_crude = EquivalenceProperties::new(input_schema.clone());
         eq_properties_crude.oeq_class.push(crude.clone());
         assert!(!eq_properties_crude.ordering_satisfy(&finer));
         Ok(())
@@ -1826,6 +2031,296 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_ordering_satisfy_with_equivalence2() -> Result<()> {
+        let test_schema = create_test_schema()?;
+        let col_a = &col("a", &test_schema)?;
+        let col_b = &col("b", &test_schema)?;
+        let col_c = &col("c", &test_schema)?;
+        let col_d = &col("d", &test_schema)?;
+        let col_e = &col("e", &test_schema)?;
+        let col_f = &col("f", &test_schema)?;
+        let floor_a = &create_physical_expr(
+            &BuiltinScalarFunction::Floor,
+            &[col("a", &test_schema)?],
+            &test_schema,
+            &ExecutionProps::default(),
+        )?;
+        let floor_f = &create_physical_expr(
+            &BuiltinScalarFunction::Floor,
+            &[col("f", &test_schema)?],
+            &test_schema,
+            &ExecutionProps::default(),
+        )?;
+        let exp_a = &create_physical_expr(
+            &BuiltinScalarFunction::Exp,
+            &[col("a", &test_schema)?],
+            &test_schema,
+            &ExecutionProps::default(),
+        )?;
+        let a_plus_b = Arc::new(BinaryExpr::new(
+            col_a.clone(),
+            Operator::Plus,
+            col_b.clone(),
+        )) as Arc<dyn PhysicalExpr>;
+        let options = SortOptions {
+            descending: false,
+            nulls_first: false,
+        };
+
+        let test_cases = vec![
+            // ------------ TEST CASE 1 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, d ASC, b ASC]
+                    vec![(col_a, options), (col_d, options), (col_b, options)],
+                    // [c ASC]
+                    vec![(col_c, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [a ASC, b ASC], requirement is not satisfied.
+                vec![(col_a, options), (col_b, options)],
+                // expected: requirement is not satisfied.
+                false,
+            ),
+            // ------------ TEST CASE 2 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, c ASC, b ASC]
+                    vec![(col_a, options), (col_c, options), (col_b, options)],
+                    // [d ASC]
+                    vec![(col_d, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [floor(a) ASC],
+                vec![(floor_a, options)],
+                // expected: requirement is satisfied.
+                true,
+            ),
+            // ------------ TEST CASE 2.1 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, c ASC, b ASC]
+                    vec![(col_a, options), (col_c, options), (col_b, options)],
+                    // [d ASC]
+                    vec![(col_d, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [floor(f) ASC], (Please note that a=f)
+                vec![(floor_f, options)],
+                // expected: requirement is satisfied.
+                true,
+            ),
+            // ------------ TEST CASE 3 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, c ASC, b ASC]
+                    vec![(col_a, options), (col_c, options), (col_b, options)],
+                    // [d ASC]
+                    vec![(col_d, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [a ASC, c ASC, a+b ASC],
+                vec![(col_a, options), (col_c, options), (&a_plus_b, options)],
+                // expected: requirement is satisfied.
+                true,
+            ),
+            // ------------ TEST CASE 4 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, b ASC, c ASC, d ASC]
+                    vec![
+                        (col_a, options),
+                        (col_b, options),
+                        (col_c, options),
+                        (col_d, options),
+                    ],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [floor(a) ASC, a+b ASC],
+                vec![(floor_a, options), (&a_plus_b, options)],
+                // expected: requirement is satisfied.
+                false,
+            ),
+            // ------------ TEST CASE 5 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, b ASC, c ASC, d ASC]
+                    vec![
+                        (col_a, options),
+                        (col_b, options),
+                        (col_c, options),
+                        (col_d, options),
+                    ],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [exp(a) ASC, a+b ASC],
+                vec![(exp_a, options), (&a_plus_b, options)],
+                // expected: requirement is not satisfied.
+                // TODO: If we know that exp function is 1-to-1 function.
+                //  we could have deduced that above requirement is satisfied.
+                false,
+            ),
+            // ------------ TEST CASE 6 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, d ASC, b ASC]
+                    vec![(col_a, options), (col_d, options), (col_b, options)],
+                    // [c ASC]
+                    vec![(col_c, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [a ASC, d ASC, floor(a) ASC],
+                vec![(col_a, options), (col_d, options), (floor_a, options)],
+                // expected: requirement is satisfied.
+                true,
+            ),
+            // ------------ TEST CASE 7 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, c ASC, b ASC]
+                    vec![(col_a, options), (col_c, options), (col_b, options)],
+                    // [d ASC]
+                    vec![(col_d, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [a ASC, floor(a) ASC, a + b ASC],
+                vec![(col_a, options), (floor_a, options), (&a_plus_b, options)],
+                // expected: requirement is not satisfied.
+                false,
+            ),
+            // ------------ TEST CASE 8 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, b ASC, c ASC]
+                    vec![(col_a, options), (col_b, options), (col_c, options)],
+                    // [d ASC]
+                    vec![(col_d, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [a ASC, c ASC, floor(a) ASC, a + b ASC],
+                vec![
+                    (col_a, options),
+                    (col_c, options),
+                    (&floor_a, options),
+                    (&a_plus_b, options),
+                ],
+                // expected: requirement is not satisfied.
+                false,
+            ),
+            // ------------ TEST CASE 9 ------------
+            (
+                // orderings
+                vec![
+                    // [a ASC, b ASC, c ASC, d ASC]
+                    vec![
+                        (col_a, options),
+                        (col_b, options),
+                        (col_c, options),
+                        (col_d, options),
+                    ],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [a ASC, b ASC, c ASC, floor(a) ASC],
+                vec![
+                    (col_a, options),
+                    (col_b, options),
+                    (&col_c, options),
+                    (&floor_a, options),
+                ],
+                // expected: requirement is satisfied.
+                true,
+            ),
+            // ------------ TEST CASE 10 ------------
+            (
+                // orderings
+                vec![
+                    // [d ASC, b ASC]
+                    vec![(col_d, options), (col_b, options)],
+                    // [c ASC, a ASC]
+                    vec![(col_c, options), (col_a, options)],
+                ],
+                // equivalence classes
+                vec![vec![col_a, col_f]],
+                // constants
+                vec![col_e],
+                // requirement [c ASC, d ASC, a + b ASC],
+                vec![(col_c, options), (col_d, options), (&a_plus_b, options)],
+                // expected: requirement is satisfied.
+                true,
+            ),
+        ];
+
+        for (orderings, eq_group, constants, reqs, expected) in test_cases {
+            let err_msg =
+                format!("error in test orderings: {orderings:?}, eq_group: {eq_group:?}, constants: {constants:?}, reqs: {reqs:?}, expected: {expected:?}");
+            let mut eq_properties = EquivalenceProperties::new(test_schema.clone());
+            let orderings = convert_to_orderings(&orderings);
+            eq_properties.add_new_orderings(orderings);
+            let eq_group = eq_group
+                .into_iter()
+                .map(|eq_class| {
+                    let eq_classes = eq_class.into_iter().cloned().collect::<Vec<_>>();
+                    EquivalenceClass::new(eq_classes)
+                })
+                .collect::<Vec<_>>();
+            let eq_group = EquivalenceGroup::new(eq_group);
+            eq_properties.add_equivalence_group(eq_group);
+
+            let constants = constants.into_iter().cloned();
+            eq_properties = eq_properties.add_constants(constants);
+
+            let reqs = convert_to_sort_exprs(&reqs);
+            assert_eq!(
+                eq_properties.ordering_satisfy(&reqs),
+                expected,
+                "{}",
+                err_msg
+            );
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_ordering_satisfy_with_equivalence_random() -> Result<()> {
         const N_RANDOM_SCHEMA: usize = 5;
@@ -1865,8 +2360,8 @@ mod tests {
                         table_data_with_properties.clone(),
                     )?;
                     let err_msg = format!(
-                        "Error in test case requirement:{:?}, expected: {:?}",
-                        requirement, expected
+                        "Error in test case requirement:{:?}, expected: {:?}, eq_properties.oeq_class: {:?}, eq_properties.eq_group: {:?}, eq_properties.constants: {:?}",
+                        requirement, expected, eq_properties.oeq_class, eq_properties.eq_group, eq_properties.constants
                     );
                     // Check whether ordering_satisfy API result and
                     // experimental result matches.
@@ -1883,6 +2378,78 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_ordering_satisfy_with_equivalence_complex_random() -> Result<()> {
+        const N_RANDOM_SCHEMA: usize = 100;
+        const N_ELEMENTS: usize = 125;
+        const N_DISTINCT: usize = 5;
+        const SORT_OPTIONS: SortOptions = SortOptions {
+            descending: false,
+            nulls_first: false,
+        };
+
+        for seed in 0..N_RANDOM_SCHEMA {
+            // Create a random schema with random properties
+            let (test_schema, eq_properties) = create_random_schema(seed as u64)?;
+            // Generate a data that satisfies properties given
+            let table_data_with_properties =
+                generate_table_for_eq_properties(&eq_properties, N_ELEMENTS, N_DISTINCT)?;
+
+            let exp_fn = create_physical_expr(
+                &BuiltinScalarFunction::Floor,
+                &[col("a", &test_schema)?],
+                &test_schema,
+                &ExecutionProps::default(),
+            )?;
+            let a_plus_b = Arc::new(BinaryExpr::new(
+                col("a", &test_schema)?,
+                Operator::Plus,
+                col("b", &test_schema)?,
+            )) as Arc<dyn PhysicalExpr>;
+            let exprs = vec![
+                col("a", &test_schema)?,
+                col("b", &test_schema)?,
+                col("c", &test_schema)?,
+                col("d", &test_schema)?,
+                col("e", &test_schema)?,
+                col("f", &test_schema)?,
+                exp_fn,
+                a_plus_b,
+            ];
+
+            for n_req in 0..=exprs.len() {
+                for exprs in exprs.iter().combinations(n_req) {
+                    let requirement = exprs
+                        .into_iter()
+                        .map(|expr| PhysicalSortExpr {
+                            expr: expr.clone(),
+                            options: SORT_OPTIONS,
+                        })
+                        .collect::<Vec<_>>();
+                    let expected = is_table_same_after_sort(
+                        requirement.clone(),
+                        table_data_with_properties.clone(),
+                    )?;
+                    let err_msg = format!(
+                        "Error in test case requirement:{:?}, expected: {:?}, eq_properties.oeq_class: {:?}, eq_properties.eq_group: {:?}, eq_properties.constants: {:?}",
+                        requirement, expected, eq_properties.oeq_class, eq_properties.eq_group, eq_properties.constants
+                    );
+                    // Check whether ordering_satisfy API result and
+                    // experimental result matches.
+
+                    assert_eq!(
+                        eq_properties.ordering_satisfy(&requirement),
+                        (expected | false),
+                        "{}",
+                        err_msg
+                    );
+                }
+            }
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_ordering_satisfy_different_lengths() -> Result<()> {
         let test_schema = create_test_schema()?;
@@ -2111,6 +2678,30 @@ mod tests {
                     ],
                 ],
             ),
+            // ------- TEST CASE 5 ---------
+            // Empty ordering
+            (
+                vec![vec![]],
+                // No ordering in the state (empty ordering is ignored).
+                vec![],
+            ),
+            // ------- TEST CASE 6 ---------
+            (
+                // ORDERINGS GIVEN
+                vec![
+                    // [a ASC, b ASC]
+                    vec![(col_a, option_asc), (col_b, option_asc)],
+                    // [b ASC]
+                    vec![(col_b, option_asc)],
+                ],
+                // EXPECTED orderings that is succinct.
+                vec![
+                    // [a ASC]
+                    vec![(col_a, option_asc)],
+                    // [b ASC]
+                    vec![(col_b, option_asc)],
+                ],
+            ),
         ];
         for (orderings, expected) in test_cases {
             let orderings = convert_to_orderings(&orderings);
@@ -2216,13 +2807,16 @@ mod tests {
         let mut columns = batch.columns().to_vec();
 
         // Create a new unique column
-        let n_row = batch.num_rows() as u64;
-        let unique_col = Arc::new(UInt64Array::from_iter_values(0..n_row)) as ArrayRef;
+        let n_row = batch.num_rows();
+        let vals: Vec<usize> = (0..n_row).collect::<Vec<_>>();
+        let vals: Vec<f64> = vals.into_iter().map(|val| val as f64).collect();
+        let unique_col = Arc::new(Float64Array::from_iter_values(vals)) as ArrayRef;
         columns.push(unique_col.clone());
 
         // Create a new schema with the added unique column
         let unique_col_name = "unique";
-        let unique_field = Arc::new(Field::new(unique_col_name, DataType::UInt64, false));
+        let unique_field =
+            Arc::new(Field::new(unique_col_name, DataType::Float64, false));
         let fields: Vec<_> = original_schema
             .fields()
             .iter()
@@ -2241,17 +2835,17 @@ mod tests {
         });
 
         // Convert the required ordering to a list of SortColumn
-        let sort_columns: Vec<_> = required_ordering
+        let sort_columns = required_ordering
             .iter()
-            .filter_map(|order_expr| {
-                let col = order_expr.expr.as_any().downcast_ref::<Column>()?;
-                let col_index = schema.column_with_name(col.name())?.0;
-                Some(SortColumn {
-                    values: new_batch.column(col_index).clone(),
+            .map(|order_expr| {
+                let expr_result = order_expr.expr.evaluate(&new_batch)?;
+                let values = expr_result.into_array(new_batch.num_rows())?;
+                Ok(SortColumn {
+                    values,
                     options: Some(order_expr.options),
                 })
             })
-            .collect();
+            .collect::<Result<Vec<_>>>()?;
 
         // Check if the indices after sorting match the initial ordering
         let sorted_indices = lexsort_to_indices(&sort_columns, None)?;
@@ -2292,18 +2886,18 @@ mod tests {
 
         // Utility closure to generate random array
         let mut generate_random_array = |num_elems: usize, max_val: usize| -> ArrayRef {
-            let values: Vec<u64> = (0..num_elems)
-                .map(|_| rng.gen_range(0..max_val) as u64)
+            let values: Vec<f64> = (0..num_elems)
+                .map(|_| rng.gen_range(0..max_val) as f64 / 2.0)
                 .collect();
-            Arc::new(UInt64Array::from_iter_values(values))
+            Arc::new(Float64Array::from_iter_values(values))
         };
 
         // Fill constant columns
         for constant in &eq_properties.constants {
             let col = constant.as_any().downcast_ref::<Column>().unwrap();
             let (idx, _field) = schema.column_with_name(col.name()).unwrap();
-            let arr =
-                Arc::new(UInt64Array::from_iter_values(vec![0; n_elem])) as ArrayRef;
+            let arr = Arc::new(Float64Array::from_iter_values(vec![0 as f64; n_elem]))
+                as ArrayRef;
             schema_vec[idx] = Some(arr);
         }
 
@@ -2620,6 +3214,12 @@ mod tests {
         let col_d = &col("d", &test_schema)?;
         let col_e = &col("e", &test_schema)?;
         let col_h = &col("h", &test_schema)?;
+        // a + d
+        let a_plus_d = Arc::new(BinaryExpr::new(
+            col_a.clone(),
+            Operator::Plus,
+            col_d.clone(),
+        )) as Arc<dyn PhysicalExpr>;
 
         let option_asc = SortOptions {
             descending: false,
@@ -2650,14 +3250,26 @@ mod tests {
                 vec![col_d, col_e, col_b],
                 vec![
                     (col_d, option_asc),
-                    (col_b, option_asc),
                     (col_e, option_desc),
+                    (col_b, option_asc),
                 ],
             ),
             // TEST CASE 4
             (vec![col_b], vec![]),
             // TEST CASE 5
             (vec![col_d], vec![(col_d, option_asc)]),
+            // TEST CASE 5
+            (vec![&a_plus_d], vec![(&a_plus_d, option_asc)]),
+            // TEST CASE 6
+            (
+                vec![col_b, col_d],
+                vec![(col_d, option_asc), (col_b, option_asc)],
+            ),
+            // TEST CASE 6
+            (
+                vec![col_c, col_e],
+                vec![(col_c, option_asc), (col_e, option_desc)],
+            ),
         ];
         for (exprs, expected) in test_cases {
             let exprs = exprs.into_iter().cloned().collect::<Vec<_>>();
@@ -2669,6 +3281,82 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_find_longest_permutation_random() -> Result<()> {
+        const N_RANDOM_SCHEMA: usize = 100;
+        const N_ELEMENTS: usize = 125;
+        const N_DISTINCT: usize = 5;
+
+        for seed in 0..N_RANDOM_SCHEMA {
+            // Create a random schema with random properties
+            let (test_schema, eq_properties) = create_random_schema(seed as u64)?;
+            // Generate a data that satisfies properties given
+            let table_data_with_properties =
+                generate_table_for_eq_properties(&eq_properties, N_ELEMENTS, N_DISTINCT)?;
+
+            let exp_fn = create_physical_expr(
+                &BuiltinScalarFunction::Floor,
+                &[col("a", &test_schema)?],
+                &test_schema,
+                &ExecutionProps::default(),
+            )?;
+            let a_plus_b = Arc::new(BinaryExpr::new(
+                col("a", &test_schema)?,
+                Operator::Plus,
+                col("b", &test_schema)?,
+            )) as Arc<dyn PhysicalExpr>;
+            let exprs = vec![
+                col("a", &test_schema)?,
+                col("b", &test_schema)?,
+                col("c", &test_schema)?,
+                col("d", &test_schema)?,
+                col("e", &test_schema)?,
+                col("f", &test_schema)?,
+                exp_fn,
+                a_plus_b,
+            ];
+
+            for n_req in 0..=exprs.len() {
+                for exprs in exprs.iter().combinations(n_req) {
+                    let exprs = exprs.into_iter().cloned().collect::<Vec<_>>();
+                    let (ordering, indices) =
+                        eq_properties.find_longest_permutation(&exprs);
+                    // Make sure that find_longest_permutation return values are consistent
+                    let ordering2 = indices
+                        .iter()
+                        .zip(ordering.iter())
+                        .map(|(&idx, sort_expr)| PhysicalSortExpr {
+                            expr: exprs[idx].clone(),
+                            options: sort_expr.options,
+                        })
+                        .collect::<Vec<_>>();
+                    assert_eq!(
+                        ordering, ordering2,
+                        "indices and lexicographical ordering do not match"
+                    );
+
+                    let err_msg = format!(
+                        "Error in test case ordering:{:?}, eq_properties.oeq_class: {:?}, eq_properties.eq_group: {:?}, eq_properties.constants: {:?}",
+                        ordering, eq_properties.oeq_class, eq_properties.eq_group, eq_properties.constants
+                    );
+                    assert_eq!(ordering.len(), indices.len(), "{}", err_msg);
+                    // Since ordered section satisfies schema, we expect
+                    // that result will be same after sort (e.g sort was unnecessary).
+                    assert!(
+                        is_table_same_after_sort(
+                            ordering.clone(),
+                            table_data_with_properties.clone(),
+                        )?,
+                        "{}",
+                        err_msg
+                    );
+                }
+            }
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_update_ordering() -> Result<()> {
         let schema = Schema::new(vec![
@@ -2726,11 +3414,14 @@ mod tests {
             ),
         ];
         for (expr, expected) in test_cases {
-            let expr_ordering = ExprOrdering::new(expr.clone());
-            let expr_ordering = expr_ordering
-                .transform_up(&|expr| update_ordering(expr, &eq_properties))?;
+            let leading_orderings = eq_properties
+                .oeq_class()
+                .iter()
+                .flat_map(|ordering| ordering.first().cloned())
+                .collect::<Vec<_>>();
+            let expr_ordering = eq_properties.get_expr_ordering(expr.clone());
             let err_msg = format!(
-                "expr:{:?}, expected: {:?}, actual: {:?}",
+                "expr:{:?}, expected: {:?}, actual: {:?}, leading_orderings: {leading_orderings:?}",
                 expr, expected, expr_ordering.state
             );
             assert_eq!(expr_ordering.state, expected, "{}", err_msg);
@@ -2921,74 +3612,30 @@ mod tests {
     }
 
     #[test]
-    fn project_empty_output_ordering() -> Result<()> {
-        let schema = Schema::new(vec![
-            Field::new("a", DataType::Int32, true),
-            Field::new("b", DataType::Int32, true),
-            Field::new("c", DataType::Int32, true),
-        ]);
-        let mut eq_properties = EquivalenceProperties::new(Arc::new(schema.clone()));
-        let ordering = vec![PhysicalSortExpr {
-            expr: Arc::new(Column::new("b", 1)),
-            options: SortOptions::default(),
-        }];
-        eq_properties.add_new_orderings([ordering]);
-        let projection_mapping = ProjectionMapping {
-            inner: vec![
-                (
-                    Arc::new(Column::new("b", 1)) as _,
-                    Arc::new(Column::new("b_new", 0)) as _,
-                ),
-                (
-                    Arc::new(Column::new("a", 0)) as _,
-                    Arc::new(Column::new("a_new", 1)) as _,
-                ),
-            ],
-        };
-        let projection_schema = Arc::new(Schema::new(vec![
-            Field::new("b_new", DataType::Int32, true),
-            Field::new("a_new", DataType::Int32, true),
-        ]));
-        let orderings = eq_properties
-            .project(&projection_mapping, projection_schema)
-            .oeq_class()
-            .output_ordering()
-            .unwrap_or_default();
-
-        assert_eq!(
-            vec![PhysicalSortExpr {
-                expr: Arc::new(Column::new("b_new", 0)),
-                options: SortOptions::default(),
-            }],
-            orderings
-        );
-
-        let schema = Schema::new(vec![
+    fn test_expr_consists_of_constants() -> Result<()> {
+        let schema = Arc::new(Schema::new(vec![
             Field::new("a", DataType::Int32, true),
             Field::new("b", DataType::Int32, true),
             Field::new("c", DataType::Int32, true),
-        ]);
-        let eq_properties = EquivalenceProperties::new(Arc::new(schema));
-        let projection_mapping = ProjectionMapping {
-            inner: vec![
-                (
-                    Arc::new(Column::new("c", 2)) as _,
-                    Arc::new(Column::new("c_new", 0)) as _,
-                ),
-                (
-                    Arc::new(Column::new("b", 1)) as _,
-                    Arc::new(Column::new("b_new", 1)) as _,
-                ),
-            ],
-        };
-        let projection_schema = Arc::new(Schema::new(vec![
-            Field::new("c_new", DataType::Int32, true),
-            Field::new("b_new", DataType::Int32, true),
+            Field::new("d", DataType::Int32, true),
+            Field::new("ts", DataType::Timestamp(TimeUnit::Nanosecond, None), true),
         ]));
-        let projected = eq_properties.project(&projection_mapping, projection_schema);
-        // After projection there is no ordering.
-        assert!(projected.oeq_class().output_ordering().is_none());
-
+        let col_a = col("a", &schema)?;
+        let col_b = col("b", &schema)?;
+        let col_d = col("d", &schema)?;
+        let b_plus_d = Arc::new(BinaryExpr::new(
+            col_b.clone(),
+            Operator::Plus,
+            col_d.clone(),
+        )) as Arc<dyn PhysicalExpr>;
+
+        let constants = vec![col_a.clone(), col_b.clone()];
+        let expr = b_plus_d.clone();
+        assert!(!is_expr_constant_util(&constants, &expr));
+
+        let constants = vec![col_a.clone(), col_b.clone(), col_d.clone()];
+        let expr = b_plus_d.clone();
+        assert!(is_expr_constant_util(&constants, &expr));
         Ok(())
     }
 }
diff --git a/datafusion/physical-expr/src/sort_properties.rs b/datafusion/physical-expr/src/sort_properties.rs
index a3b201f84e9d..2e72edc07c37 100644
--- a/datafusion/physical-expr/src/sort_properties.rs
+++ b/datafusion/physical-expr/src/sort_properties.rs
@@ -18,7 +18,6 @@
 use std::{ops::Neg, sync::Arc};
 
 use crate::PhysicalExpr;
-
 use arrow_schema::SortOptions;
 use datafusion_common::tree_node::{TreeNode, VisitRecursion};
 use datafusion_common::Result;
diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs
index b8e2d0e425d4..06ff6bde9db3 100644
--- a/datafusion/physical-plan/src/projection.rs
+++ b/datafusion/physical-plan/src/projection.rs
@@ -38,10 +38,10 @@ use arrow::record_batch::{RecordBatch, RecordBatchOptions};
 use datafusion_common::stats::Precision;
 use datafusion_common::Result;
 use datafusion_execution::TaskContext;
+use datafusion_physical_expr::equivalence::ProjectionMapping;
 use datafusion_physical_expr::expressions::{Literal, UnKnownColumn};
 use datafusion_physical_expr::EquivalenceProperties;
 
-use datafusion_physical_expr::equivalence::ProjectionMapping;
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
diff --git a/datafusion/sqllogictest/test_files/groupby.slt b/datafusion/sqllogictest/test_files/groupby.slt
index 4438d69af306..192f87cc7007 100644
--- a/datafusion/sqllogictest/test_files/groupby.slt
+++ b/datafusion/sqllogictest/test_files/groupby.slt
@@ -3842,6 +3842,51 @@ ProjectionExec: expr=[SUM(alias1)@1 as SUM(DISTINCT t1.x), MAX(alias1)@2 as MAX(
 --------------------ProjectionExec: expr=[CAST(x@0 AS Float64) as CAST(t1.x AS Float64)t1.x, y@1 as y]
 ----------------------MemoryExec: partitions=1, partition_sizes=[1]
 
+# create an unbounded table that contains ordered timestamp.
+statement ok
+CREATE UNBOUNDED EXTERNAL TABLE csv_with_timestamps (
+  name VARCHAR,
+  ts TIMESTAMP
+)
+STORED AS CSV
+WITH ORDER (ts DESC)
+LOCATION '../core/tests/data/timestamps.csv'
+
+# below query should work in streaming mode.
+query TT
+EXPLAIN SELECT date_bin('15 minutes', ts) as time_chunks
+  FROM csv_with_timestamps
+  GROUP BY (date_bin('15 minutes', ts))
+  ORDER BY time_chunks DESC
+  LIMIT 5
+----
+logical_plan
+Limit: skip=0, fetch=5
+--Sort: time_chunks DESC NULLS FIRST, fetch=5
+----Projection: date_bin(Utf8("15 minutes"),csv_with_timestamps.ts) AS time_chunks
+------Aggregate: groupBy=[[date_bin(IntervalMonthDayNano("900000000000"), csv_with_timestamps.ts) AS date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)]], aggr=[[]]
+--------TableScan: csv_with_timestamps projection=[ts]
+physical_plan
+GlobalLimitExec: skip=0, fetch=5
+--SortPreservingMergeExec: [time_chunks@0 DESC], fetch=5
+----ProjectionExec: expr=[date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0 as time_chunks]
+------AggregateExec: mode=FinalPartitioned, gby=[date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0 as date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
+--------CoalesceBatchesExec: target_batch_size=2
+----------SortPreservingRepartitionExec: partitioning=Hash([date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0], 8), input_partitions=8, sort_exprs=date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0 DESC
+------------AggregateExec: mode=Partial, gby=[date_bin(900000000000, ts@0) as date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
+--------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
+----------------StreamingTableExec: partition_sizes=1, projection=[ts], infinite_source=true, output_ordering=[ts@0 DESC]
+
+query P
+SELECT date_bin('15 minutes', ts) as time_chunks
+  FROM csv_with_timestamps
+  GROUP BY (date_bin('15 minutes', ts))
+  ORDER BY time_chunks DESC
+  LIMIT 5
+----
+2018-12-13T12:00:00
+2018-11-13T17:00:00
+
 statement ok
 drop table t1
 

From b35d343d0a7f59d7c1d7b67be324e2f32fb20820 Mon Sep 17 00:00:00 2001
From: Mustafa Akur <mustafa.akur@synnada.ai>
Date: Fri, 17 Nov 2023 14:13:14 +0300
Subject: [PATCH 2/7] Minor changes

---
 datafusion/physical-expr/src/equivalence.rs | 29 ++++++++++++---------
 1 file changed, 16 insertions(+), 13 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence.rs b/datafusion/physical-expr/src/equivalence.rs
index 87031b12a949..a5ddb93a78f2 100644
--- a/datafusion/physical-expr/src/equivalence.rs
+++ b/datafusion/physical-expr/src/equivalence.rs
@@ -691,22 +691,25 @@ impl OrderingEquivalenceClass {
     /// For instance, If we already have the ordering [a ASC, b ASC, c DESC],
     /// then there is no need to keep ordering [a ASC, b ASC] in the state.
     fn remove_redundant_entries(&mut self) {
+        // Get leading orderings (e.g first sort expr in each lexicographical ordering) among orderings
         let leading_ordering_exprs = self
             .orderings
             .iter()
             .flat_map(|ordering| ordering.first().map(|sort_expr| sort_expr.expr.clone()))
             .collect::<Vec<_>>();
 
-        // Remove leading orderings that are beyond index 0, to simplify ordering.
+        // Remove leading orderings that are at the end of the lexicographical ordering.
         self.orderings.iter_mut().for_each(|ordering| {
-            let mut counter = 0;
-            ordering.retain(|sort_expr| {
-                // Either first entry or is not leading ordering
-                let should_retain = counter == 0
-                    || !physical_exprs_contains(&leading_ordering_exprs, &sort_expr.expr);
-                counter += 1;
-                should_retain
-            });
+            while ordering.len() > 1 {
+                let last_sort_expr = &ordering[ordering.len() - 1];
+                if physical_exprs_contains(&leading_ordering_exprs, &last_sort_expr.expr)
+                {
+                    ordering.pop();
+                } else {
+                    // last ordering expr is not leading. Stop removing from the end.
+                    break;
+                }
+            }
         });
 
         let mut idx = 0;
@@ -2395,7 +2398,7 @@ mod tests {
             let table_data_with_properties =
                 generate_table_for_eq_properties(&eq_properties, N_ELEMENTS, N_DISTINCT)?;
 
-            let exp_fn = create_physical_expr(
+            let floor_a = create_physical_expr(
                 &BuiltinScalarFunction::Floor,
                 &[col("a", &test_schema)?],
                 &test_schema,
@@ -2413,7 +2416,7 @@ mod tests {
                 col("d", &test_schema)?,
                 col("e", &test_schema)?,
                 col("f", &test_schema)?,
-                exp_fn,
+                floor_a,
                 a_plus_b,
             ];
 
@@ -3294,7 +3297,7 @@ mod tests {
             let table_data_with_properties =
                 generate_table_for_eq_properties(&eq_properties, N_ELEMENTS, N_DISTINCT)?;
 
-            let exp_fn = create_physical_expr(
+            let floor_a = create_physical_expr(
                 &BuiltinScalarFunction::Floor,
                 &[col("a", &test_schema)?],
                 &test_schema,
@@ -3312,7 +3315,7 @@ mod tests {
                 col("d", &test_schema)?,
                 col("e", &test_schema)?,
                 col("f", &test_schema)?,
-                exp_fn,
+                floor_a,
                 a_plus_b,
             ];
 

From bb77516304b7862372a6e0f76bad7f6876705cad Mon Sep 17 00:00:00 2001
From: Mehmet Ozan Kabak <ozankabak@gmail.com>
Date: Mon, 20 Nov 2023 10:34:24 +0300
Subject: [PATCH 3/7] Review Part 1

---
 .../replace_with_order_preserving_variants.rs |  93 ++++---
 datafusion/physical-expr/src/equivalence.rs   | 229 +++++++++---------
 2 files changed, 181 insertions(+), 141 deletions(-)

diff --git a/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs b/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs
index 7f8c9b852cb1..5f130848de11 100644
--- a/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs
+++ b/datafusion/core/src/physical_optimizer/replace_with_order_preserving_variants.rs
@@ -357,15 +357,19 @@ mod tests {
         let physical_plan =
             sort_preserving_merge_exec(vec![sort_expr("a", &schema)], sort);
 
-        let expected_input = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+        let expected_input = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortExec: expr=[a@0 ASC NULLS LAST]",
             "    RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
-        let expected_optimized = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
+        let expected_optimized = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortPreservingRepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8, sort_exprs=a@0 ASC NULLS LAST",
             "    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "      CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "      CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         assert_optimized!(expected_input, expected_optimized, physical_plan);
         Ok(())
     }
@@ -434,19 +438,20 @@ mod tests {
         let physical_plan =
             sort_preserving_merge_exec(vec![sort_expr("a", &schema)], sort);
 
-        let expected_input = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+        let expected_input = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortExec: expr=[a@0 ASC NULLS LAST]",
             "    RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "      FilterExec: c@1 > 3",
             "        RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         let expected_optimized = [
             "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortPreservingRepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8, sort_exprs=a@0 ASC NULLS LAST",
             "    FilterExec: c@1 > 3",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
             "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
-
         ];
         assert_optimized!(expected_input, expected_optimized, physical_plan);
         Ok(())
@@ -466,19 +471,23 @@ mod tests {
         let physical_plan =
             sort_preserving_merge_exec(vec![sort_expr("a", &schema)], sort);
 
-        let expected_input = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+        let expected_input = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortExec: expr=[a@0 ASC NULLS LAST]",
             "    CoalesceBatchesExec: target_batch_size=8192",
             "      FilterExec: c@1 > 3",
             "        RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "          RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "            CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
-        let expected_optimized = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+            "            CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
+        let expected_optimized = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  CoalesceBatchesExec: target_batch_size=8192",
             "    FilterExec: c@1 > 3",
             "      SortPreservingRepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8, sort_exprs=a@0 ASC NULLS LAST",
             "        RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         assert_optimized!(expected_input, expected_optimized, physical_plan);
         Ok(())
     }
@@ -499,21 +508,25 @@ mod tests {
         let physical_plan =
             sort_preserving_merge_exec(vec![sort_expr("a", &schema)], sort);
 
-        let expected_input = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+        let expected_input = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortExec: expr=[a@0 ASC NULLS LAST]",
             "    CoalesceBatchesExec: target_batch_size=8192",
             "      FilterExec: c@1 > 3",
             "        RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "          CoalesceBatchesExec: target_batch_size=8192",
             "            RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "              CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
-        let expected_optimized = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+            "              CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
+        let expected_optimized = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  CoalesceBatchesExec: target_batch_size=8192",
             "    FilterExec: c@1 > 3",
             "      SortPreservingRepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8, sort_exprs=a@0 ASC NULLS LAST",
             "        CoalesceBatchesExec: target_batch_size=8192",
             "          RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "            CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "            CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         assert_optimized!(expected_input, expected_optimized, physical_plan);
         Ok(())
     }
@@ -531,18 +544,22 @@ mod tests {
         let physical_plan: Arc<dyn ExecutionPlan> =
             coalesce_partitions_exec(coalesce_batches_exec);
 
-        let expected_input = ["CoalescePartitionsExec",
+        let expected_input = [
+            "CoalescePartitionsExec",
             "  CoalesceBatchesExec: target_batch_size=8192",
             "    FilterExec: c@1 > 3",
             "      RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "        RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
-        let expected_optimized = ["CoalescePartitionsExec",
+            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
+        let expected_optimized = [
+            "CoalescePartitionsExec",
             "  CoalesceBatchesExec: target_batch_size=8192",
             "    FilterExec: c@1 > 3",
             "      RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "        RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "          CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         assert_optimized!(expected_input, expected_optimized, physical_plan);
         Ok(())
     }
@@ -570,7 +587,7 @@ mod tests {
             "        FilterExec: c@1 > 3",
             "          RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "            RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "              CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"
+            "              CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
         ];
         let expected_optimized = [
             "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
@@ -603,16 +620,20 @@ mod tests {
             sort,
         );
 
-        let expected_input = ["SortPreservingMergeExec: [c@1 ASC]",
+        let expected_input = [
+            "SortPreservingMergeExec: [c@1 ASC]",
             "  SortExec: expr=[c@1 ASC]",
             "    RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
-        let expected_optimized = ["SortPreservingMergeExec: [c@1 ASC]",
+            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
+        let expected_optimized = [
+            "SortPreservingMergeExec: [c@1 ASC]",
             "  SortExec: expr=[c@1 ASC]",
             "    RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         assert_optimized!(expected_input, expected_optimized, physical_plan);
         Ok(())
     }
@@ -628,15 +649,19 @@ mod tests {
         let physical_plan =
             sort_exec(vec![sort_expr("a", &schema)], coalesce_partitions, false);
 
-        let expected_input = ["SortExec: expr=[a@0 ASC NULLS LAST]",
+        let expected_input = [
+            "SortExec: expr=[a@0 ASC NULLS LAST]",
             "  CoalescePartitionsExec",
             "    RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
-        let expected_optimized = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
+        let expected_optimized = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortPreservingRepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8, sort_exprs=a@0 ASC NULLS LAST",
             "    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "      CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "      CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], infinite_source=true, output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         assert_optimized!(expected_input, expected_optimized, physical_plan);
         Ok(())
     }
@@ -766,15 +791,19 @@ mod tests {
         let physical_plan =
             sort_preserving_merge_exec(vec![sort_expr("a", &schema)], sort);
 
-        let expected_input = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+        let expected_input = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortExec: expr=[a@0 ASC NULLS LAST]",
             "    RepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8",
             "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
-        let expected_optimized = ["SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
+            "        CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
+        let expected_optimized = [
+            "SortPreservingMergeExec: [a@0 ASC NULLS LAST]",
             "  SortPreservingRepartitionExec: partitioning=Hash([c@1], 8), input_partitions=8, sort_exprs=a@0 ASC NULLS LAST",
             "    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
-            "      CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true"];
+            "      CsvExec: file_groups={1 group: [[file_path]]}, projection=[a, c, d], output_ordering=[a@0 ASC NULLS LAST], has_header=true",
+        ];
         assert_optimized!(expected_input, expected_optimized, physical_plan, true);
         Ok(())
     }
diff --git a/datafusion/physical-expr/src/equivalence.rs b/datafusion/physical-expr/src/equivalence.rs
index a5ddb93a78f2..11c814b7e2a2 100644
--- a/datafusion/physical-expr/src/equivalence.rs
+++ b/datafusion/physical-expr/src/equivalence.rs
@@ -138,86 +138,81 @@ impl EquivalenceClass {
 /// projection.
 #[derive(Debug, Clone)]
 pub struct ProjectionMapping {
-    /// `(source expression)` --> `(target expression)`
-    /// Indices in the vector corresponds to the indices after projection.
-    inner: Vec<(Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)>,
+    /// Mapping between source expressions and target expressions.
+    /// Vector indices correspond to the indices after projection.
+    map: Vec<(Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)>,
 }
 
 impl ProjectionMapping {
     /// Constructs the mapping between a projection's input and output
     /// expressions.
     ///
-    /// For example, given the input projection expressions (`a+b`, `c+d`)
-    /// and an output schema with two columns `"c+d"` and `"a+b"`
-    /// the projection mapping would be
+    /// For example, given the input projection expressions (`a + b`, `c + d`)
+    /// and an output schema with two columns `"c + d"` and `"a + b"`, the
+    /// projection mapping would be:
+    ///
     /// ```text
-    ///  [0]: (c+d, col("c+d"))
-    ///  [1]: (a+b, col("a+b"))
+    ///  [0]: (c + d, col("c + d"))
+    ///  [1]: (a + b, col("a + b"))
     /// ```
-    /// where `col("c+d")` means the column named "c+d".
+    ///
+    /// where `col("c + d")` means the column named `"c + d"`.
     pub fn try_new(
         expr: &[(Arc<dyn PhysicalExpr>, String)],
         input_schema: &SchemaRef,
     ) -> Result<Self> {
         // Construct a map from the input expressions to the output expression of the projection:
-        let mut inner = vec![];
-        for (expr_idx, (expression, name)) in expr.iter().enumerate() {
-            let target_expr = Arc::new(Column::new(name, expr_idx)) as _;
-
-            let source_expr = expression.clone().transform_down(&|e| match e
-                .as_any()
-                .downcast_ref::<Column>(
-            ) {
-                Some(col) => {
-                    // Sometimes, expression and its name in the input_schema doesn't match.
-                    // This can cause problems. Hence in here we make sure that expression name
-                    // matches with the name in the inout_schema.
-                    // Conceptually, source_expr and expression should be same.
-                    let idx = col.index();
-                    let matching_input_field = input_schema.field(idx);
-                    let matching_input_column =
-                        Column::new(matching_input_field.name(), idx);
-                    Ok(Transformed::Yes(Arc::new(matching_input_column)))
-                }
-                None => Ok(Transformed::No(e)),
-            })?;
-
-            inner.push((source_expr, target_expr));
-        }
-        Ok(Self { inner })
+        expr.iter()
+            .enumerate()
+            .map(|(expr_idx, (expression, name))| {
+                let target_expr = Arc::new(Column::new(name, expr_idx)) as _;
+                expression
+                    .clone()
+                    .transform_down(&|e| match e.as_any().downcast_ref::<Column>() {
+                        Some(col) => {
+                            // Sometimes, an expression and its name in the input_schema
+                            // doesn't match. This can cause problems, so we make sure
+                            // that the expression name matches with the name in `input_schema`.
+                            // Conceptually, `source_expr` and `expression` should be the same.
+                            let idx = col.index();
+                            let matching_input_field = input_schema.field(idx);
+                            let matching_input_column =
+                                Column::new(matching_input_field.name(), idx);
+                            Ok(Transformed::Yes(Arc::new(matching_input_column)))
+                        }
+                        None => Ok(Transformed::No(e)),
+                    })
+                    .map(|source_expr| (source_expr, target_expr))
+            })
+            .collect::<Result<Vec<_>>>()
+            .map(|map| Self { map })
     }
 
     /// Iterate over pairs of (source, target) expressions
     pub fn iter(
         &self,
     ) -> impl Iterator<Item = &(Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)> + '_ {
-        self.inner.iter()
+        self.map.iter()
     }
 
-    /// This function returns target value for given expression
+    /// This function returns the target expression for a given source expression.
     ///
     /// # Arguments
     ///
-    /// * `expr` - Source (e.g key) physical expression
+    /// * `expr` - Source physical expression.
     ///
     /// # Returns
     ///
-    /// An `Option` containing a the target (e.g value) for the source expression.
-    /// `None` means that source is not found inside the mapping.
+    /// An `Option` containing the target for the given source expression,
+    /// where a `None` value means that `expr` is not inside the mapping.
     pub fn target_expr(
         &self,
         expr: &Arc<dyn PhysicalExpr>,
     ) -> Option<Arc<dyn PhysicalExpr>> {
-        if let Some(idx) = self
-            .inner
+        self.map
             .iter()
-            .position(|(source, _target)| source.eq(expr))
-        {
-            let (_source, target) = &self.inner[idx];
-            Some(target.clone())
-        } else {
-            None
-        }
+            .find(|(source, _)| source.eq(expr))
+            .map(|(_, target)| target.clone())
     }
 }
 
@@ -236,7 +231,7 @@ impl EquivalenceGroup {
 
     /// Creates an equivalence group from the given equivalence classes.
     fn new(classes: Vec<EquivalenceClass>) -> Self {
-        let mut result = EquivalenceGroup { classes };
+        let mut result = Self { classes };
         result.remove_redundant_entries();
         result
     }
@@ -279,12 +274,13 @@ impl EquivalenceGroup {
                 // If the given left and right sides belong to different classes,
                 // we should unify/bridge these classes.
                 if first_idx != second_idx {
-                    // By convention make sure second_idx is larger than first_idx.
+                    // By convention, make sure `second_idx` is larger than `first_idx`.
                     if first_idx > second_idx {
                         (first_idx, second_idx) = (second_idx, first_idx);
                     }
-                    // Remove second_idx from self.classes then merge its values with class at first_idx.
-                    // Convention above makes sure that first_idx is still valid after second_idx removal.
+                    // Remove the class at `second_idx` and merge its values with
+                    // the class at `first_idx`. The convention above makes sure
+                    // that `first_idx` is still valid after removing `second_idx`.
                     let other_class = self.classes.swap_remove(second_idx);
                     self.classes[first_idx].extend(other_class);
                 }
@@ -445,9 +441,9 @@ impl EquivalenceGroup {
             // If the given expression is not inside the mapping, try to project
             // expressions considering the equivalence classes.
             for (source, target) in mapping.iter() {
-                // If we match an equivalent expression to source,
-                // then we can project. For example, if we have the mapping
-                // (a as a1, a + c) and the equivalence class (a, b), expression `b` projects to `a1`.
+                // If we match an equivalent expression to `source`, then we can
+                // project. For example, if we have the mapping `(a as a1, a + c)`
+                // and the equivalence class `(a, b)`, expression `b` projects to `a1`.
                 if self
                     .get_equivalence_class(source)
                     .map_or(false, |group| group.contains(expr))
@@ -461,15 +457,12 @@ impl EquivalenceGroup {
         if children.is_empty() {
             // Leaf expression should be inside mapping.
             return None;
-        } else if let Some(children) = children
+        }
+        children
             .into_iter()
             .map(|child| self.project_expr(mapping, &child))
             .collect::<Option<Vec<_>>>()
-        {
-            return Some(expr.clone().with_new_children(children).unwrap());
-        }
-        // Arriving here implies the expression was invalid after projection.
-        None
+            .map(|children| expr.clone().with_new_children(children).unwrap())
     }
 
     /// Projects `ordering` according to the given projection mapping.
@@ -533,8 +526,8 @@ impl EquivalenceGroup {
         Self::new(classes)
     }
 
-    /// Returns the equivalence class that contains `expr`.
-    /// If none of the equivalence classes contains `expr`, returns `None`.
+    /// Returns the equivalence class containing `expr`. If no equivalence class
+    /// contains `expr`, returns `None`.
     fn get_equivalence_class(
         &self,
         expr: &Arc<dyn PhysicalExpr>,
@@ -687,47 +680,35 @@ impl OrderingEquivalenceClass {
         self.remove_redundant_entries();
     }
 
-    /// Removes redundant orderings from this equivalence class.
-    /// For instance, If we already have the ordering [a ASC, b ASC, c DESC],
-    /// then there is no need to keep ordering [a ASC, b ASC] in the state.
+    /// Removes redundant orderings from this equivalence class. For instance,
+    /// if we already have the ordering `[a ASC, b ASC, c DESC]`, then there is
+    /// no need to keep ordering `[a ASC, b ASC]` in the state.
     fn remove_redundant_entries(&mut self) {
-        // Get leading orderings (e.g first sort expr in each lexicographical ordering) among orderings
-        let leading_ordering_exprs = self
-            .orderings
-            .iter()
-            .flat_map(|ordering| ordering.first().map(|sort_expr| sort_expr.expr.clone()))
-            .collect::<Vec<_>>();
-
-        // Remove leading orderings that are at the end of the lexicographical ordering.
-        self.orderings.iter_mut().for_each(|ordering| {
-            while ordering.len() > 1 {
-                let last_sort_expr = &ordering[ordering.len() - 1];
-                if physical_exprs_contains(&leading_ordering_exprs, &last_sort_expr.expr)
-                {
-                    ordering.pop();
-                } else {
-                    // last ordering expr is not leading. Stop removing from the end.
-                    break;
-                }
-            }
-        });
-
-        let mut idx = 0;
-        while idx < self.orderings.len() {
-            let mut removal = self.orderings[idx].is_empty();
-            for (ordering_idx, ordering) in self.orderings[0..idx].iter().enumerate() {
-                if let Some(right_finer) = finer_side(ordering, &self.orderings[idx]) {
-                    if right_finer {
-                        self.orderings.swap(ordering_idx, idx);
+        let mut work = true;
+        while work {
+            work = false;
+            let mut idx = 0;
+            while idx < self.orderings.len() {
+                let mut ordering_idx = idx + 1;
+                let mut removal = self.orderings[idx].is_empty();
+                while ordering_idx < self.orderings.len() {
+                    work |= resolve_overlap(&mut self.orderings, idx, ordering_idx);
+                    if self.orderings[idx].is_empty() {
+                        removal = true;
+                        break;
+                    }
+                    work |= resolve_overlap(&mut self.orderings, ordering_idx, idx);
+                    if self.orderings[ordering_idx].is_empty() {
+                        self.orderings.swap_remove(ordering_idx);
+                    } else {
+                        ordering_idx += 1;
                     }
-                    removal = true;
-                    break;
                 }
-            }
-            if removal {
-                self.orderings.swap_remove(idx);
-            } else {
-                idx += 1;
+                if removal {
+                    self.orderings.swap_remove(idx);
+                } else {
+                    idx += 1;
+                }
             }
         }
     }
@@ -735,8 +716,7 @@ impl OrderingEquivalenceClass {
     /// Returns the concatenation of all the orderings. This enables merge
     /// operations to preserve all equivalent orderings simultaneously.
     pub fn output_ordering(&self) -> Option<LexOrdering> {
-        let output_ordering =
-            self.orderings.iter().flatten().cloned().collect::<Vec<_>>();
+        let output_ordering = self.orderings.iter().flatten().cloned().collect();
         let output_ordering = collapse_lex_ordering(output_ordering);
         (!output_ordering.is_empty()).then_some(output_ordering)
     }
@@ -793,12 +773,18 @@ pub fn add_offset_to_expr(
     // an `Ok` value.
 }
 
-/// Returns `true` if the ordering `rhs` is strictly finer than the ordering `rhs`,
-/// `false` if the ordering `lhs` is at least as fine as the ordering `lhs`, and
-/// `None` otherwise (i.e. when given orderings are incomparable).
-fn finer_side(lhs: LexOrderingRef, rhs: LexOrderingRef) -> Option<bool> {
-    let all_equal = lhs.iter().zip(rhs.iter()).all(|(lhs, rhs)| lhs.eq(rhs));
-    all_equal.then_some(lhs.len() < rhs.len())
+/// Trims `orderings[idx]` if some suffix of it overlaps with a prefix of
+/// `orderings[pre_idx]`. Returns `true` if there is any overlap, `false` otherwise.
+fn resolve_overlap(orderings: &mut [LexOrdering], idx: usize, pre_idx: usize) -> bool {
+    let length = orderings[idx].len();
+    let other_length = orderings[pre_idx].len();
+    for overlap in 1..=length.min(other_length) {
+        if orderings[idx][length - overlap..] == orderings[pre_idx][..overlap] {
+            orderings[idx].truncate(length - overlap);
+            return true;
+        }
+    }
+    false
 }
 
 /// A `EquivalenceProperties` object stores useful information related to a schema.
@@ -2588,6 +2574,7 @@ mod tests {
         let col_a = &col("a", &schema)?;
         let col_b = &col("b", &schema)?;
         let col_c = &col("c", &schema)?;
+        let col_d = &col("d", &schema)?;
 
         let option_asc = SortOptions {
             descending: false,
@@ -2705,6 +2692,30 @@ mod tests {
                     vec![(col_b, option_asc)],
                 ],
             ),
+            // ------- TEST CASE 7 ---------
+            // b, a
+            // c, a
+            // d, b, c
+            (
+                // ORDERINGS GIVEN
+                vec![
+                    // [b ASC, a ASC]
+                    vec![(col_b, option_asc), (col_a, option_asc)],
+                    // [c ASC, a ASC]
+                    vec![(col_c, option_asc), (col_a, option_asc)],
+                    // [d ASC, b ASC, c ASC]
+                    vec![(col_d, option_asc), (col_b, option_asc), (col_c, option_asc)],
+                ],
+                // EXPECTED orderings that is succinct.
+                vec![
+                    // [b ASC, a ASC]
+                    vec![(col_b, option_asc), (col_a, option_asc)],
+                    // [c ASC, a ASC]
+                    vec![(col_c, option_asc), (col_a, option_asc)],
+                    // [d ASC]
+                    vec![(col_d, option_asc)],
+                ],
+            ),            
         ];
         for (orderings, expected) in test_cases {
             let orderings = convert_to_orderings(&orderings);

From b5f0ef87074f9bf42c5334e80fbd3fca6f794555 Mon Sep 17 00:00:00 2001
From: Mustafa Akur <mustafa.akur@synnada.ai>
Date: Mon, 20 Nov 2023 13:21:48 +0300
Subject: [PATCH 4/7] Add new tests

---
 datafusion/physical-expr/src/equivalence.rs | 75 ++++++++++++++++++++-
 1 file changed, 73 insertions(+), 2 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence.rs b/datafusion/physical-expr/src/equivalence.rs
index 11c814b7e2a2..61319fef8513 100644
--- a/datafusion/physical-expr/src/equivalence.rs
+++ b/datafusion/physical-expr/src/equivalence.rs
@@ -2575,6 +2575,7 @@ mod tests {
         let col_b = &col("b", &schema)?;
         let col_c = &col("c", &schema)?;
         let col_d = &col("d", &schema)?;
+        let col_e = &col("e", &schema)?;
 
         let option_asc = SortOptions {
             descending: false,
@@ -2704,7 +2705,11 @@ mod tests {
                     // [c ASC, a ASC]
                     vec![(col_c, option_asc), (col_a, option_asc)],
                     // [d ASC, b ASC, c ASC]
-                    vec![(col_d, option_asc), (col_b, option_asc), (col_c, option_asc)],
+                    vec![
+                        (col_d, option_asc),
+                        (col_b, option_asc),
+                        (col_c, option_asc),
+                    ],
                 ],
                 // EXPECTED orderings that is succinct.
                 vec![
@@ -2715,7 +2720,73 @@ mod tests {
                     // [d ASC]
                     vec![(col_d, option_asc)],
                 ],
-            ),            
+            ),
+            // ------- TEST CASE 8 ---------
+            // b, e
+            // c, a
+            // d, b, e, c, a
+            (
+                // ORDERINGS GIVEN
+                vec![
+                    // [b ASC, e ASC]
+                    vec![(col_b, option_asc), (col_e, option_asc)],
+                    // [c ASC, a ASC]
+                    vec![(col_c, option_asc), (col_a, option_asc)],
+                    // [d ASC, b ASC, e ASC, c ASC, a ASC]
+                    vec![
+                        (col_d, option_asc),
+                        (col_b, option_asc),
+                        (col_e, option_asc),
+                        (col_c, option_asc),
+                        (col_a, option_asc),
+                    ],
+                ],
+                // EXPECTED orderings that is succinct.
+                vec![
+                    // [b ASC, e ASC]
+                    vec![(col_b, option_asc), (col_e, option_asc)],
+                    // [c ASC, a ASC]
+                    vec![(col_c, option_asc), (col_a, option_asc)],
+                    // [d ASC]
+                    vec![(col_d, option_asc)],
+                ],
+            ),
+            // ------- TEST CASE 9 ---------
+            // b
+            // a, b, c
+            // d, a, b
+            (
+                // ORDERINGS GIVEN
+                vec![
+                    // [b ASC]
+                    vec![(col_b, option_asc)],
+                    // [a ASC, b ASC, c ASC]
+                    vec![
+                        (col_a, option_asc),
+                        (col_b, option_asc),
+                        (col_c, option_asc),
+                    ],
+                    // [d ASC, a ASC, b ASC]
+                    vec![
+                        (col_d, option_asc),
+                        (col_a, option_asc),
+                        (col_b, option_asc),
+                    ],
+                ],
+                // EXPECTED orderings that is succinct.
+                vec![
+                    // [b ASC]
+                    vec![(col_b, option_asc)],
+                    // [a ASC, b ASC, c ASC]
+                    vec![
+                        (col_a, option_asc),
+                        (col_b, option_asc),
+                        (col_c, option_asc),
+                    ],
+                    // [d ASC]
+                    vec![(col_d, option_asc)],
+                ],
+            ),
         ];
         for (orderings, expected) in test_cases {
             let orderings = convert_to_orderings(&orderings);

From 8ee9f1e0ba623b0119e34569a07e17386652d8b1 Mon Sep 17 00:00:00 2001
From: Mehmet Ozan Kabak <ozankabak@gmail.com>
Date: Mon, 20 Nov 2023 17:09:57 +0300
Subject: [PATCH 5/7] Review Part 2

---
 datafusion/physical-expr/src/equivalence.rs | 189 +++++++++-----------
 1 file changed, 89 insertions(+), 100 deletions(-)

diff --git a/datafusion/physical-expr/src/equivalence.rs b/datafusion/physical-expr/src/equivalence.rs
index 61319fef8513..f9f03300f5e9 100644
--- a/datafusion/physical-expr/src/equivalence.rs
+++ b/datafusion/physical-expr/src/equivalence.rs
@@ -1028,40 +1028,40 @@ impl EquivalenceProperties {
         let normalized_reqs = eq_properties.normalize_sort_requirements(reqs);
         for normalized_req in normalized_reqs {
             // Check whether given ordering is satisfied
-            if !eq_properties.ordering_satisfy_single_req(&normalized_req) {
+            if !eq_properties.ordering_satisfy_single(&normalized_req) {
                 return false;
             }
-            // - Treat satisfied ordering as constant in the next iterations. Since in lexicographical ordering
-            //   next orderings are only considered as long as their left side have same values
-            //   (e.g for them their left side is constant).
+            // Treat satisfied keys as constants in subsequent iterations. We
+            // can do this because the "next" key only matters in a lexicographical
+            // ordering when the keys to its left have the same values.
             //
-            // Please note that, these expressions are not properly constant. This is just implementation
-            // and this interpretation doesn't effect outside users in anyway.
+            // Note that these expressions are not properly "constants". This is just
+            // an implementation strategy confined to this function.
             //
-            // As an example:
-            // If the requirement is `[a ASC, b + c ASC]`.
-            // and existing orderings are `[a ASC, b ASC], [c ASC]`.
-            // From the analysis above, we know that `[a ASC]` is satisfied.
-            // Then here, we add column `a` as constant to the state.
-            // This enables us to deduce that `b + c ASC` is satisfied, given `a` is constant.
+            // For example, assume that the requirement is `[a ASC, (b + c) ASC]`,
+            // and existing equivalent orderings are `[a ASC, b ASC]` and `[c ASC]`.
+            // From the analysis above, we know that `[a ASC]` is satisfied. Then,
+            // we add column `a` as constant to the algorithm state. This enables us
+            // to deduce that `(b + c) ASC` is satisfied, given `a` is constant.
             eq_properties =
                 eq_properties.add_constants(std::iter::once(normalized_req.expr));
         }
         true
     }
 
-    /// Determines whether the ordering specified by a given `PhysicalSortRequirement` is satisfied
-    /// based on the internal orderings, equivalent classes, and constant expressions.
+    /// Determines whether the ordering specified by the given sort requirement
+    /// is satisfied based on the orderings within, equivalence classes, and
+    /// constant expressions.
     ///
     /// # Arguments
     ///
-    /// - `req`: A reference to a `PhysicalSortRequirement` for which the ordering satisfaction
-    ///   needs to be determined.
+    /// - `req`: A reference to a `PhysicalSortRequirement` for which the ordering
+    ///   satisfaction check will be done.
     ///
     /// # Returns
     ///
-    /// Returns `true` if the specified ordering is satisfied; otherwise, returns `false`.
-    fn ordering_satisfy_single_req(&self, req: &PhysicalSortRequirement) -> bool {
+    /// Returns `true` if the specified ordering is satisfied, `false` otherwise.
+    fn ordering_satisfy_single(&self, req: &PhysicalSortRequirement) -> bool {
         let expr_ordering = self.get_expr_ordering(req.expr.clone());
         let ExprOrdering { expr, state, .. } = expr_ordering;
         match state {
@@ -1198,31 +1198,31 @@ impl EquivalenceProperties {
 
     /// Projects constants based on the provided `ProjectionMapping`.
     ///
-    /// This function takes a `ProjectionMapping` and identifies and projects constants based on
-    /// the existing constants and the mapping. It ensures that constants are appropriately
-    /// propagated through the projection expressions.
+    /// This function takes a `ProjectionMapping` and identifies/projects
+    /// constants based on the existing constants and the mapping. It ensures
+    /// that constants are appropriately propagated through the projection.
     ///
     /// # Arguments
     ///
-    /// - `mapping`: A reference to a `ProjectionMapping` representing the mapping of source
-    ///   expressions to target expressions in the projection.
+    /// - `mapping`: A reference to a `ProjectionMapping` representing the
+    ///   mapping of source expressions to target expressions in the projection.
     ///
     /// # Returns
     ///
-    /// Returns a vector of `Arc<dyn PhysicalExpr>` containing the projected constants.
+    /// Returns a `Vec<Arc<dyn PhysicalExpr>>` containing the projected constants.
     fn projected_constants(
         &self,
         mapping: &ProjectionMapping,
     ) -> Vec<Arc<dyn PhysicalExpr>> {
-        // Project existing constants
-        // As an example assume that a+b is known to be constant. If projection were:
-        // `a as a_new`, `b as b_new`; we would project constant `a+b` as `a_new+b_new`
+        // First, project existing constants. For example, assume that `a + b`
+        // is known to be constant. If the projection were `a as a_new`, `b as b_new`,
+        // then we would project constant `a + b` as `a_new + b_new`.
         let mut projected_constants = self
             .constants
             .iter()
             .flat_map(|expr| self.eq_group.project_expr(mapping, expr))
             .collect::<Vec<_>>();
-        // Add projection expressions that are known to be constant.
+        // Add projection expressions that are known to be constant:
         for (source, target) in mapping.iter() {
             if self.is_expr_constant(source)
                 && !physical_exprs_contains(&projected_constants, target)
@@ -1240,9 +1240,6 @@ impl EquivalenceProperties {
         projection_mapping: &ProjectionMapping,
         output_schema: SchemaRef,
     ) -> Self {
-        let projected_constants = self.projected_constants(projection_mapping);
-        let projected_eq_group = self.eq_group.project(projection_mapping);
-
         let mut projected_orderings = self
             .oeq_class
             .iter()
@@ -1251,7 +1248,7 @@ impl EquivalenceProperties {
         for (source, target) in projection_mapping.iter() {
             let expr_ordering = ExprOrdering::new(source.clone())
                 .transform_up(&|expr| Ok(update_ordering(expr, self)))
-                // It is guaranteed to always return Ok.
+                // Guaranteed to always return `Ok`.
                 .unwrap();
             if let SortProperties::Ordered(options) = expr_ordering.state {
                 // Push new ordering to the state.
@@ -1262,9 +1259,9 @@ impl EquivalenceProperties {
             }
         }
         Self {
-            eq_group: projected_eq_group,
+            eq_group: self.eq_group.project(projection_mapping),
             oeq_class: OrderingEquivalenceClass::new(projected_orderings),
-            constants: projected_constants,
+            constants: self.projected_constants(projection_mapping),
             schema: output_schema,
         }
     }
@@ -1285,110 +1282,111 @@ impl EquivalenceProperties {
     ) -> (LexOrdering, Vec<usize>) {
         let mut eq_properties = self.clone();
         let mut result = vec![];
-        // Algorithm is as follows:
-        // - 1. Iterate over all expressions and insert the expressions that are known to be ordered
-        //      into result_ordering.
-        // - 2. Treat inserted expressions as constants (add them as constant to the state)
-        // - 3. Go back to step 1.
-        // - Continue the above iteration as long as no new expression is inserted (Algorithm reached a fixed point).
-
+        // The algorithm is as follows:
+        // - Iterate over all the expressions and insert ordered expressions
+        //   into the result.
+        // - Treat inserted expressions as constants (i.e. add them as constants
+        //   to the state).
+        // - Continue the above procedure until no expression is inserted; i.e.
+        //   the algorithm reaches a fixed point.
+        // This algorithm should reach a fixed point in at most `exprs.len()`
+        // iterations.
         let mut search_indices = (0..exprs.len()).collect::<IndexSet<_>>();
-        // Algorithm above should reach a fixed point at-most `exprs.len()` number of iterations.
-        // We could have use loop{}, However, to guarantee we exit anyway (in-case of bugs).
-        // We use upper limit number of iterations (worst-case).
         for _idx in 0..exprs.len() {
             // Get ordered expressions with their indices.
             let ordered_exprs = search_indices
                 .iter()
                 .flat_map(|&idx| {
-                    let expr = exprs[idx].clone();
                     let ExprOrdering { expr, state, .. } =
-                        eq_properties.get_expr_ordering(expr);
+                        eq_properties.get_expr_ordering(exprs[idx].clone());
                     if let SortProperties::Ordered(options) = state {
-                        let sort_expr = PhysicalSortExpr { expr, options };
-                        Some((sort_expr, idx))
+                        Some((PhysicalSortExpr { expr, options }, idx))
                     } else {
                         None
                     }
                 })
                 .collect::<Vec<_>>();
-
             // We reached a fixed point, exit.
             if ordered_exprs.is_empty() {
                 break;
             }
-
-            // - Remove indices that have an ordering from search_indices
-            // - Treat ordered expressions as constant in the next iterations. Since in lexicographical ordering
-            //   next orderings are only considered as long as their left side have same values
-            //   (e.g for them their left side is constant).
+            // Remove indices that have an ordering from `search_indices`, and
+            // treat ordered expressions as constants in subsequent iterations.
+            // We can do this because the "next" key only matters in a lexicographical
+            // ordering when the keys to its left have the same values.
             //
-            // Please note that, these expressions are not properly constant. This is just implementation
-            // and this interpretation doesn't effect outside users in anyway.
-            for (sort_expr, idx) in &ordered_exprs {
+            // Note that these expressions are not properly "constants". This is just
+            // an implementation strategy confined to this function.
+            for (PhysicalSortExpr { expr, .. }, idx) in &ordered_exprs {
                 eq_properties =
-                    eq_properties.add_constants(std::iter::once(sort_expr.expr.clone()));
+                    eq_properties.add_constants(std::iter::once(expr.clone()));
                 search_indices.remove(idx);
             }
-
             // Add new ordered section to the state.
             result.extend(ordered_exprs);
         }
         result.into_iter().unzip()
     }
 
-    /// Checks whether a given expression is constant.
-    ///
-    /// This function determines whether the provided expression is constant based on the known constants.
+    /// This function determines whether the provided expression is constant
+    /// based on the known constants.
     ///
     /// # Arguments
     ///
-    /// - `expr`: A reference to a `Arc<dyn PhysicalExpr>` representing the expression to be checked.
+    /// - `expr`: A reference to a `Arc<dyn PhysicalExpr>` representing the
+    ///   expression to be checked.
     ///
     /// # Returns
     ///
-    /// Returns `true` if the expression is constant within the equivalence group; otherwise, returns `false`.
+    /// Returns `true` if the expression is constant according to equivalence
+    /// group, `false` otherwise.
     fn is_expr_constant(&self, expr: &Arc<dyn PhysicalExpr>) -> bool {
-        // As an example, assume that we know columns [a, b] are constant.
-        // `a`, `b`, `a+b` will all return `true`, whereas `c` will return `false`.
+        // As an example, assume that we know columns `a` and `b` are constant.
+        // Then, `a`, `b` and `a + b` will all return `true` whereas `c` will
+        // return `false`.
         let normalized_constants = self.eq_group.normalize_exprs(self.constants.to_vec());
         let normalized_expr = self.eq_group.normalize_expr(expr.clone());
-        is_expr_constant_util(&normalized_constants, &normalized_expr)
+        is_constant_recurse(&normalized_constants, &normalized_expr)
     }
 
     /// Retrieves the ordering information for a given physical expression.
     ///
-    /// This function constructs an `ExprOrdering` object for the provided expression, which encapsulates
-    /// information about the expression's ordering, including its state and associated options.
+    /// This function constructs an `ExprOrdering` object for the provided
+    /// expression, which encapsulates information about the expression's
+    /// ordering, including its [`SortProperties`].
     ///
     /// # Arguments
     ///
-    /// - `expr`: An `Arc<dyn PhysicalExpr>` representing the physical expression for which ordering information is sought.
+    /// - `expr`: An `Arc<dyn PhysicalExpr>` representing the physical expression
+    ///   for which ordering information is sought.
     ///
     /// # Returns
     ///
-    /// Returns an `ExprOrdering` object containing the ordering information for the given expression.
+    /// Returns an `ExprOrdering` object containing the ordering information for
+    /// the given expression.
     pub fn get_expr_ordering(&self, expr: Arc<dyn PhysicalExpr>) -> ExprOrdering {
-        let expr_ordering = ExprOrdering::new(expr.clone());
-        expr_ordering
+        ExprOrdering::new(expr.clone())
             .transform_up(&|expr| Ok(update_ordering(expr, self)))
-            // It is guaranteed to always return Ok.
+            // Guaranteed to always return `Ok`.
             .unwrap()
     }
 }
 
-/// Checks whether a given expression is constant. This function determines whether the
-/// provided expression is constant based on the known constants.
+/// This function determines whether the provided expression is constant
+/// based on the known constants.
 ///
 /// # Arguments
 ///
-/// - `constants`: A reference to a `[Arc<dyn PhysicalExpr>]` containing expressions known to be a constant.
-/// - `expr`: A reference to a `Arc<dyn PhysicalExpr>` representing the expression to be checked.
+/// - `constants`: A `&[Arc<dyn PhysicalExpr>]` containing expressions known to
+///   be a constant.
+/// - `expr`: A reference to a `Arc<dyn PhysicalExpr>` representing the expression
+///   to check.
 ///
 /// # Returns
 ///
-/// Returns `true` if the expression is constant within the equivalence group; otherwise, returns `false`.
-fn is_expr_constant_util(
+/// Returns `true` if the expression is constant according to equivalence
+/// group, `false` otherwise.
+fn is_constant_recurse(
     constants: &[Arc<dyn PhysicalExpr>],
     expr: &Arc<dyn PhysicalExpr>,
 ) -> bool {
@@ -1396,13 +1394,7 @@ fn is_expr_constant_util(
         return true;
     }
     let children = expr.children();
-    if children.is_empty() {
-        false
-    } else {
-        children
-            .iter()
-            .all(|child| is_expr_constant_util(constants, child))
-    }
+    !children.is_empty() && children.iter().all(|c| is_constant_recurse(constants, c))
 }
 
 /// Calculate ordering equivalence properties for the given join operation.
@@ -1518,27 +1510,24 @@ fn update_ordering(
     eq_properties: &EquivalenceProperties,
 ) -> Transformed<ExprOrdering> {
     // We have a Column, which is one of the two possible leaf node types:
-    let eq_group = &eq_properties.eq_group;
-    let normalized_expr = eq_group.normalize_expr(node.expr.clone());
-    let oeq_class = &eq_properties.normalized_oeq_class();
+    let normalized_expr = eq_properties.eq_group.normalize_expr(node.expr.clone());
     if eq_properties.is_expr_constant(&normalized_expr) {
         node.state = SortProperties::Singleton;
-        return Transformed::Yes(node);
-    } else if let Some(options) = oeq_class.get_options(&normalized_expr) {
+    } else if let Some(options) = eq_properties
+        .normalized_oeq_class()
+        .get_options(&normalized_expr)
+    {
         node.state = SortProperties::Ordered(options);
-        return Transformed::Yes(node);
-    }
-    if !node.expr.children().is_empty() {
+    } else if !node.expr.children().is_empty() {
         // We have an intermediate (non-leaf) node, account for its children:
         node.state = node.expr.get_ordering(&node.children_states);
-        Transformed::Yes(node)
     } else if node.expr.as_any().is::<Literal>() {
         // We have a Literal, which is the other possible leaf node type:
         node.state = node.expr.get_ordering(&[]);
-        Transformed::Yes(node)
     } else {
-        Transformed::No(node)
+        return Transformed::No(node);
     }
+    Transformed::Yes(node)
 }
 
 #[cfg(test)]
@@ -3716,11 +3705,11 @@ mod tests {
 
         let constants = vec![col_a.clone(), col_b.clone()];
         let expr = b_plus_d.clone();
-        assert!(!is_expr_constant_util(&constants, &expr));
+        assert!(!is_constant_recurse(&constants, &expr));
 
         let constants = vec![col_a.clone(), col_b.clone(), col_d.clone()];
         let expr = b_plus_d.clone();
-        assert!(is_expr_constant_util(&constants, &expr));
+        assert!(is_constant_recurse(&constants, &expr));
         Ok(())
     }
 }

From 3319c8d4684f370da9d82a0d8ea228c1dbda2c84 Mon Sep 17 00:00:00 2001
From: Mehmet Ozan Kabak <ozankabak@gmail.com>
Date: Wed, 22 Nov 2023 23:15:56 +0300
Subject: [PATCH 6/7] Address review feedback

---
 .../sqllogictest/test_files/groupby.slt       | 87 ++++++++++++++++---
 1 file changed, 73 insertions(+), 14 deletions(-)

diff --git a/datafusion/sqllogictest/test_files/groupby.slt b/datafusion/sqllogictest/test_files/groupby.slt
index 192f87cc7007..cc15e795396e 100644
--- a/datafusion/sqllogictest/test_files/groupby.slt
+++ b/datafusion/sqllogictest/test_files/groupby.slt
@@ -3844,7 +3844,7 @@ ProjectionExec: expr=[SUM(alias1)@1 as SUM(DISTINCT t1.x), MAX(alias1)@2 as MAX(
 
 # create an unbounded table that contains ordered timestamp.
 statement ok
-CREATE UNBOUNDED EXTERNAL TABLE csv_with_timestamps (
+CREATE UNBOUNDED EXTERNAL TABLE unbounded_csv_with_timestamps (
   name VARCHAR,
   ts TIMESTAMP
 )
@@ -3855,38 +3855,97 @@ LOCATION '../core/tests/data/timestamps.csv'
 # below query should work in streaming mode.
 query TT
 EXPLAIN SELECT date_bin('15 minutes', ts) as time_chunks
-  FROM csv_with_timestamps
-  GROUP BY (date_bin('15 minutes', ts))
+  FROM unbounded_csv_with_timestamps
+  GROUP BY date_bin('15 minutes', ts)
   ORDER BY time_chunks DESC
-  LIMIT 5
+  LIMIT 5;
 ----
 logical_plan
 Limit: skip=0, fetch=5
 --Sort: time_chunks DESC NULLS FIRST, fetch=5
-----Projection: date_bin(Utf8("15 minutes"),csv_with_timestamps.ts) AS time_chunks
-------Aggregate: groupBy=[[date_bin(IntervalMonthDayNano("900000000000"), csv_with_timestamps.ts) AS date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)]], aggr=[[]]
---------TableScan: csv_with_timestamps projection=[ts]
+----Projection: date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts) AS time_chunks
+------Aggregate: groupBy=[[date_bin(IntervalMonthDayNano("900000000000"), unbounded_csv_with_timestamps.ts) AS date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)]], aggr=[[]]
+--------TableScan: unbounded_csv_with_timestamps projection=[ts]
 physical_plan
 GlobalLimitExec: skip=0, fetch=5
 --SortPreservingMergeExec: [time_chunks@0 DESC], fetch=5
-----ProjectionExec: expr=[date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0 as time_chunks]
-------AggregateExec: mode=FinalPartitioned, gby=[date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0 as date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
+----ProjectionExec: expr=[date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0 as time_chunks]
+------AggregateExec: mode=FinalPartitioned, gby=[date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0 as date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
 --------CoalesceBatchesExec: target_batch_size=2
-----------SortPreservingRepartitionExec: partitioning=Hash([date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0], 8), input_partitions=8, sort_exprs=date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)@0 DESC
-------------AggregateExec: mode=Partial, gby=[date_bin(900000000000, ts@0) as date_bin(Utf8("15 minutes"),csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
+----------SortPreservingRepartitionExec: partitioning=Hash([date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0], 8), input_partitions=8, sort_exprs=date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)@0 DESC
+------------AggregateExec: mode=Partial, gby=[date_bin(900000000000, ts@0) as date_bin(Utf8("15 minutes"),unbounded_csv_with_timestamps.ts)], aggr=[], ordering_mode=Sorted
 --------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
 ----------------StreamingTableExec: partition_sizes=1, projection=[ts], infinite_source=true, output_ordering=[ts@0 DESC]
 
 query P
 SELECT date_bin('15 minutes', ts) as time_chunks
-  FROM csv_with_timestamps
-  GROUP BY (date_bin('15 minutes', ts))
+  FROM unbounded_csv_with_timestamps
+  GROUP BY date_bin('15 minutes', ts)
   ORDER BY time_chunks DESC
-  LIMIT 5
+  LIMIT 5;
 ----
 2018-12-13T12:00:00
 2018-11-13T17:00:00
 
+# Since extract is not a monotonic function, below query should not run.
+query error
+SELECT extract(month from ts) as months
+  FROM unbounded_csv_with_timestamps
+  GROUP BY extract(month from ts)
+  ORDER BY months DESC
+  LIMIT 5;
+----
+DataFusion error: PipelineChecker
+caused by
+Error during planning: Aggregate Error: `GROUP BY` clauses with columns without ordering and GROUPING SETS are not supported for unbounded inputs.
+
+
+statement ok
+CREATE EXTERNAL TABLE csv_with_timestamps (
+  name VARCHAR,
+  ts TIMESTAMP
+)
+STORED AS CSV
+WITH ORDER (ts DESC)
+LOCATION '../core/tests/data/timestamps.csv';
+
+# below query should run since it operates on a bounded source and have a sort
+# at the top of its plan.
+query TT
+EXPLAIN SELECT extract(month from ts) as months
+  FROM csv_with_timestamps
+  GROUP BY extract(month from ts)
+  ORDER BY months DESC
+  LIMIT 5;
+----
+logical_plan
+Limit: skip=0, fetch=5
+--Sort: months DESC NULLS FIRST, fetch=5
+----Projection: date_part(Utf8("MONTH"),csv_with_timestamps.ts) AS months
+------Aggregate: groupBy=[[date_part(Utf8("MONTH"), csv_with_timestamps.ts)]], aggr=[[]]
+--------TableScan: csv_with_timestamps projection=[ts]
+physical_plan
+GlobalLimitExec: skip=0, fetch=5
+--SortPreservingMergeExec: [months@0 DESC], fetch=5
+----SortExec: TopK(fetch=5), expr=[months@0 DESC]
+------ProjectionExec: expr=[date_part(Utf8("MONTH"),csv_with_timestamps.ts)@0 as months]
+--------AggregateExec: mode=FinalPartitioned, gby=[date_part(Utf8("MONTH"),csv_with_timestamps.ts)@0 as date_part(Utf8("MONTH"),csv_with_timestamps.ts)], aggr=[]
+----------CoalesceBatchesExec: target_batch_size=2
+------------RepartitionExec: partitioning=Hash([date_part(Utf8("MONTH"),csv_with_timestamps.ts)@0], 8), input_partitions=8
+--------------AggregateExec: mode=Partial, gby=[date_part(MONTH, ts@0) as date_part(Utf8("MONTH"),csv_with_timestamps.ts)], aggr=[]
+----------------RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1
+------------------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/timestamps.csv]]}, projection=[ts], output_ordering=[ts@0 DESC], has_header=false
+
+query R
+SELECT extract(month from ts) as months
+  FROM csv_with_timestamps
+  GROUP BY extract(month from ts)
+  ORDER BY months DESC
+  LIMIT 5;
+----
+12
+11
+
 statement ok
 drop table t1
 

From 13dcd6c863a0a74b8041fa26f4b4346952753e65 Mon Sep 17 00:00:00 2001
From: Mustafa Akur <mustafa.akur@synnada.ai>
Date: Thu, 23 Nov 2023 09:11:06 +0300
Subject: [PATCH 7/7] Remove error message check in the test

---
 datafusion/sqllogictest/test_files/groupby.slt | 7 ++-----
 1 file changed, 2 insertions(+), 5 deletions(-)

diff --git a/datafusion/sqllogictest/test_files/groupby.slt b/datafusion/sqllogictest/test_files/groupby.slt
index cc15e795396e..756d3f737439 100644
--- a/datafusion/sqllogictest/test_files/groupby.slt
+++ b/datafusion/sqllogictest/test_files/groupby.slt
@@ -3888,18 +3888,15 @@ SELECT date_bin('15 minutes', ts) as time_chunks
 2018-11-13T17:00:00
 
 # Since extract is not a monotonic function, below query should not run.
+# when source is unbounded.
 query error
 SELECT extract(month from ts) as months
   FROM unbounded_csv_with_timestamps
   GROUP BY extract(month from ts)
   ORDER BY months DESC
   LIMIT 5;
-----
-DataFusion error: PipelineChecker
-caused by
-Error during planning: Aggregate Error: `GROUP BY` clauses with columns without ordering and GROUPING SETS are not supported for unbounded inputs.
-
 
+# Create a table where timestamp is ordered
 statement ok
 CREATE EXTERNAL TABLE csv_with_timestamps (
   name VARCHAR,
