Enhance/Refactor Ordering Equivalence Properties

datafusion/common/src/stats.rs

@@ -19,6 +19,8 @@
 
 use std::fmt::Display;
 
+use arrow::datatypes::DataType;
+
 use crate::ScalarValue;
 
 /// Statistics for a relation
@@ -70,3 +72,24 @@ pub struct ColumnStatistics {
     /// Number of distinct values
     pub distinct_count: Option<usize>,
 }
+
+impl ColumnStatistics {
+    pub fn is_singleton(&self) -> bool {
+        match (&self.min_value, &self.max_value) {
+            // Min and max values are the same and not infinity.
+            (Some(min), Some(max)) => !min.is_null() && !max.is_null() && (min == max),
+            (_, _) => false,
+        }
+    }
+
+    /// Returns the [`ColumnStatistics`] corresponding to the given datatype by assigning infinite bounds.
+    pub fn new_with_unbounded_column(dt: &DataType) -> ColumnStatistics {
+        let null = ScalarValue::try_from(dt.clone()).ok();
+        ColumnStatistics {
+            null_count: None,
+            max_value: null.clone(),
+            min_value: null,
+            distinct_count: None,
+        }
+    }
+}

datafusion/core/src/physical_optimizer/enforce_distribution.rs

@@ -53,8 +53,7 @@ use datafusion_physical_expr::utils::{
     map_columns_before_projection, ordering_satisfy_requirement_concrete,
 };
 use datafusion_physical_expr::{
-    expr_list_eq_strict_order, normalize_expr_with_equivalence_properties, PhysicalExpr,
-    PhysicalSortRequirement,
+    expr_list_eq_strict_order, PhysicalExpr, PhysicalSortRequirement,
 };
 
 use datafusion_common::internal_err;
@@ -804,36 +803,21 @@ fn try_reorder(
     } else if !equivalence_properties.classes().is_empty() {
         normalized_expected = expected
             .iter()
-            .map(|e| {
-                normalize_expr_with_equivalence_properties(
-                    e.clone(),
-                    equivalence_properties.classes(),
-                )
-            })
+            .map(|e| equivalence_properties.normalize_expr(e.clone()))
             .collect::<Vec<_>>();
         assert_eq!(normalized_expected.len(), expected.len());
 
         normalized_left_keys = join_keys
             .left_keys
             .iter()
-            .map(|e| {
-                normalize_expr_with_equivalence_properties(
-                    e.clone(),
-                    equivalence_properties.classes(),
-                )
-            })
+            .map(|e| equivalence_properties.normalize_expr(e.clone()))
             .collect::<Vec<_>>();
         assert_eq!(join_keys.left_keys.len(), normalized_left_keys.len());
 
         normalized_right_keys = join_keys
             .right_keys
             .iter()
-            .map(|e| {
-                normalize_expr_with_equivalence_properties(
-                    e.clone(),
-                    equivalence_properties.classes(),
-                )
-            })
+            .map(|e| equivalence_properties.normalize_expr(e.clone()))
             .collect::<Vec<_>>();
         assert_eq!(join_keys.right_keys.len(), normalized_right_keys.len());
 

datafusion/core/src/physical_plan/filter.rs

@@ -41,12 +41,13 @@ use datafusion_common::{plan_err, DataFusionError, Result};
 use datafusion_execution::TaskContext;
 use datafusion_expr::Operator;
 use datafusion_physical_expr::expressions::BinaryExpr;
-use datafusion_physical_expr::intervals::utils::check_support;
 use datafusion_physical_expr::{
     analyze, split_conjunction, AnalysisContext, ExprBoundaries,
     OrderingEquivalenceProperties, PhysicalExpr,
 };
 
+use datafusion_physical_expr::intervals::utils::check_support;
+use datafusion_physical_expr::utils::collect_columns;
 use futures::stream::{Stream, StreamExt};
 use log::trace;
 
@@ -153,7 +154,19 @@ impl ExecutionPlan for FilterExec {
     }
 
     fn ordering_equivalence_properties(&self) -> OrderingEquivalenceProperties {
-        self.input.ordering_equivalence_properties()
+        let stats = self.statistics();
+        // Add the columns that have only one value (singleton) after filtering to constants.
+        if let Some(col_stats) = stats.column_statistics {
+            let constants = collect_columns(self.predicate())
+                .into_iter()
+                .filter(|column| col_stats[column.index()].is_singleton())
+                .map(|column| Arc::new(column) as Arc<dyn PhysicalExpr>)
+                .collect::<Vec<_>>();
+            let filter_oeq = self.input.ordering_equivalence_properties();
+            filter_oeq.with_constants(constants)
+        } else {
+            self.input.ordering_equivalence_properties()
+        }
     }
 
     fn with_new_children(
@@ -197,7 +210,14 @@ impl ExecutionPlan for FilterExec {
         let input_stats = self.input.statistics();
         let input_column_stats = match input_stats.column_statistics {
             Some(stats) => stats,
-            None => return Statistics::default(),
+            None => self
+                .schema()
+                .fields
+                .iter()
+                .map(|field| {
+                    ColumnStatistics::new_with_unbounded_column(field.data_type())
+                })
+                .collect::<Vec<_>>(),
         };
 
         let starter_ctx =