[MAJOR] Equivalence System Overhaul

benchmarks/src/sort.rs

@@ -70,28 +70,31 @@ impl RunOpt {
         let sort_cases = vec![
             (
                 "sort utf8",
-                LexOrdering::new(vec![PhysicalSortExpr {
+                [PhysicalSortExpr {
                     expr: col("request_method", &schema)?,
                     options: Default::default(),
-                }]),
+                }]
+                .into(),
             ),
             (
                 "sort int",
-                LexOrdering::new(vec![PhysicalSortExpr {
+                [PhysicalSortExpr {
                     expr: col("response_bytes", &schema)?,
                     options: Default::default(),
-                }]),
+                }]
+                .into(),
             ),
             (
                 "sort decimal",
-                LexOrdering::new(vec![PhysicalSortExpr {
+                [PhysicalSortExpr {
                     expr: col("decimal_price", &schema)?,
                     options: Default::default(),
-                }]),
+                }]
+                .into(),
             ),
             (
                 "sort integer tuple",
-                LexOrdering::new(vec![
+                [
                     PhysicalSortExpr {
                         expr: col("request_bytes", &schema)?,
                         options: Default::default(),
@@ -100,11 +103,12 @@ impl RunOpt {
                         expr: col("response_bytes", &schema)?,
                         options: Default::default(),
                     },
-                ]),
+                ]
+                .into(),
             ),
             (
                 "sort utf8 tuple",
-                LexOrdering::new(vec![
+                [
                     // sort utf8 tuple
                     PhysicalSortExpr {
                         expr: col("service", &schema)?,
@@ -122,11 +126,12 @@ impl RunOpt {
                         expr: col("image", &schema)?,
                         options: Default::default(),
                     },
-                ]),
+                ]
+                .into(),
             ),
             (
                 "sort mixed tuple",
-                LexOrdering::new(vec![
+                [
                     PhysicalSortExpr {
                         expr: col("service", &schema)?,
                         options: Default::default(),
@@ -139,7 +144,8 @@ impl RunOpt {
                         expr: col("decimal_price", &schema)?,
                         options: Default::default(),
                     },
-                ]),
+                ]
+                .into(),
             ),
         ];
         for (title, expr) in sort_cases {

datafusion-examples/examples/custom_file_format.rs

@@ -21,27 +21,24 @@ use arrow::{
     array::{AsArray, RecordBatch, StringArray, UInt8Array},
     datatypes::{DataType, Field, Schema, SchemaRef, UInt64Type},
 };
-use datafusion::physical_expr::LexRequirement;
 use datafusion::{
     catalog::Session,
     common::{GetExt, Statistics},
-};
-use datafusion::{
-    datasource::physical_plan::FileSource, execution::session_state::SessionStateBuilder,
-};
-use datafusion::{
     datasource::{
         file_format::{
             csv::CsvFormatFactory, file_compression_type::FileCompressionType,
             FileFormat, FileFormatFactory,
         },
-        physical_plan::{FileScanConfig, FileSinkConfig},
+        physical_plan::{FileScanConfig, FileSinkConfig, FileSource},
         MemTable,
     },
     error::Result,
+    execution::session_state::SessionStateBuilder,
+    physical_expr_common::sort_expr::LexRequirement,
     physical_plan::ExecutionPlan,
     prelude::SessionContext,
 };
+
 use object_store::{ObjectMeta, ObjectStore};
 use tempfile::tempdir;
 

datafusion-examples/examples/function_factory.rs

@@ -150,10 +150,6 @@ impl ScalarUDFImpl for ScalarFunctionWrapper {
         Ok(ExprSimplifyResult::Simplified(replacement))
     }
 
-    fn aliases(&self) -> &[String] {
-        &[]
-    }
-
     fn output_ordering(&self, _input: &[ExprProperties]) -> Result<SortProperties> {
         Ok(SortProperties::Unordered)
     }

datafusion/catalog/src/listing_schema.rs

@@ -25,9 +25,7 @@ use std::sync::{Arc, Mutex};
 use crate::{SchemaProvider, TableProvider, TableProviderFactory};
 
 use crate::Session;
-use datafusion_common::{
-    Constraints, DFSchema, DataFusionError, HashMap, TableReference,
-};
+use datafusion_common::{DFSchema, DataFusionError, HashMap, TableReference};
 use datafusion_expr::CreateExternalTable;
 
 use async_trait::async_trait;
@@ -143,7 +141,7 @@ impl ListingSchemaProvider {
                             order_exprs: vec![],
                             unbounded: false,
                             options: Default::default(),
-                            constraints: Constraints::empty(),
+                            constraints: Default::default(),
                             column_defaults: Default::default(),
                         },
                     )

datafusion/catalog/src/memory/table.rs

@@ -23,25 +23,22 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use crate::TableProvider;
-use datafusion_common::error::Result;
-use datafusion_expr::Expr;
-use datafusion_expr::TableType;
-use datafusion_physical_expr::create_physical_sort_exprs;
-use datafusion_physical_plan::repartition::RepartitionExec;
-use datafusion_physical_plan::{
-    common, ExecutionPlan, ExecutionPlanProperties, Partitioning,
-};
 
 use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
+use datafusion_common::error::Result;
 use datafusion_common::{not_impl_err, plan_err, Constraints, DFSchema, SchemaExt};
 use datafusion_common_runtime::JoinSet;
-use datafusion_datasource::memory::MemSink;
-use datafusion_datasource::memory::MemorySourceConfig;
+use datafusion_datasource::memory::{MemSink, MemorySourceConfig};
 use datafusion_datasource::sink::DataSinkExec;
 use datafusion_datasource::source::DataSourceExec;
 use datafusion_expr::dml::InsertOp;
-use datafusion_expr::SortExpr;
+use datafusion_expr::{Expr, SortExpr, TableType};
+use datafusion_physical_expr::{create_physical_sort_exprs, LexOrdering};
+use datafusion_physical_plan::repartition::RepartitionExec;
+use datafusion_physical_plan::{
+    common, ExecutionPlan, ExecutionPlanProperties, Partitioning,
+};
 use datafusion_session::Session;
 
 use async_trait::async_trait;
@@ -89,7 +86,7 @@ impl MemTable {
                 .into_iter()
                 .map(|e| Arc::new(RwLock::new(e)))
                 .collect::<Vec<_>>(),
-            constraints: Constraints::empty(),
+            constraints: Constraints::default(),
             column_defaults: HashMap::new(),
             sort_order: Arc::new(Mutex::new(vec![])),
         })
@@ -239,16 +236,13 @@ impl TableProvider for MemTable {
         if !sort_order.is_empty() {
             let df_schema = DFSchema::try_from(self.schema.as_ref().clone())?;
 
-            let file_sort_order = sort_order
-                .iter()
-                .map(|sort_exprs| {
-                    create_physical_sort_exprs(
-                        sort_exprs,
-                        &df_schema,
-                        state.execution_props(),
-                    )
-                })
-                .collect::<Result<Vec<_>>>()?;
+            let eqp = state.execution_props();
+            let mut file_sort_order = vec![];
+            for sort_exprs in sort_order.iter() {
+                let physical_exprs =
+                    create_physical_sort_exprs(sort_exprs, &df_schema, eqp)?;
+                file_sort_order.extend(LexOrdering::new(physical_exprs));
+            }
             source = source.try_with_sort_information(file_sort_order)?;
         }
 

datafusion/catalog/src/stream.rs

@@ -256,7 +256,7 @@ impl StreamConfig {
         Self {
             source,
             order: vec![],
-            constraints: Constraints::empty(),
+            constraints: Constraints::default(),
         }
     }
 
@@ -350,15 +350,10 @@ impl TableProvider for StreamTable {
         input: Arc<dyn ExecutionPlan>,
         _insert_op: InsertOp,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        let ordering = match self.0.order.first() {
-            Some(x) => {
-                let schema = self.0.source.schema();
-                let orders = create_ordering(schema, std::slice::from_ref(x))?;
-                let ordering = orders.into_iter().next().unwrap();
-                Some(ordering.into_iter().map(Into::into).collect())
-            }
-            None => None,
-        };
+        let schema = self.0.source.schema();
+        let orders = create_ordering(schema, &self.0.order)?;
+        // It is sufficient to pass only one of the equivalent orderings:
+        let ordering = orders.into_iter().next().map(Into::into);
 
         Ok(Arc::new(DataSinkExec::new(
             input,

datafusion/common/src/functional_dependencies.rs

@@ -36,35 +36,31 @@ pub enum Constraint {
 }
 
 /// This object encapsulates a list of functional constraints:
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+#[derive(Clone, Debug, Default, Eq, Hash, PartialEq, PartialOrd)]
 pub struct Constraints {
     inner: Vec<Constraint>,
 }
 
 impl Constraints {
-    /// Create empty constraints
-    pub fn empty() -> Self {
-        Constraints::new_unverified(vec![])
-    }
-
     /// Create a new [`Constraints`] object from the given `constraints`.
-    /// Users should use the [`Constraints::empty`] or [`SqlToRel::new_constraint_from_table_constraints`] functions
-    /// for constructing [`Constraints`]. This constructor is for internal
-    /// purposes only and does not check whether the argument is valid. The user
-    /// is responsible for supplying a valid vector of [`Constraint`] objects.
+    /// Users should use the [`Constraints::default`] or [`SqlToRel::new_constraint_from_table_constraints`]
+    /// functions for constructing [`Constraints`] instances. This constructor
+    /// is for internal purposes only and does not check whether the argument
+    /// is valid. The user is responsible for supplying a valid vector of
+    /// [`Constraint`] objects.
     ///
     /// [`SqlToRel::new_constraint_from_table_constraints`]: https://docs.rs/datafusion/latest/datafusion/sql/planner/struct.SqlToRel.html#method.new_constraint_from_table_constraints
     pub fn new_unverified(constraints: Vec<Constraint>) -> Self {
         Self { inner: constraints }
     }
 
-    /// Check whether constraints is empty
-    pub fn is_empty(&self) -> bool {
-        self.inner.is_empty()
+    /// Extends the current constraints with the given `other` constraints.
+    pub fn extend(&mut self, other: Constraints) {
+        self.inner.extend(other.inner);
     }
 
-    /// Projects constraints using the given projection indices.
-    /// Returns None if any of the constraint columns are not included in the projection.
+    /// Projects constraints using the given projection indices. Returns `None`
+    /// if any of the constraint columns are not included in the projection.
     pub fn project(&self, proj_indices: &[usize]) -> Option<Self> {
         let projected = self
             .inner
@@ -74,14 +70,14 @@ impl Constraints {
                     Constraint::PrimaryKey(indices) => {
                         let new_indices =
                             update_elements_with_matching_indices(indices, proj_indices);
-                        // Only keep constraint if all columns are preserved
+                        // Only keep the constraint if all columns are preserved:
                         (new_indices.len() == indices.len())
                             .then_some(Constraint::PrimaryKey(new_indices))
                     }
                     Constraint::Unique(indices) => {
                         let new_indices =
                             update_elements_with_matching_indices(indices, proj_indices);
-                        // Only keep constraint if all columns are preserved
+                        // Only keep the constraint if all columns are preserved:
                         (new_indices.len() == indices.len())
                             .then_some(Constraint::Unique(new_indices))
                     }
@@ -93,15 +89,9 @@ impl Constraints {
     }
 }
 
-impl Default for Constraints {
-    fn default() -> Self {
-        Constraints::empty()
-    }
-}
-
 impl IntoIterator for Constraints {
     type Item = Constraint;
-    type IntoIter = IntoIter<Constraint>;
+    type IntoIter = IntoIter<Self::Item>;
 
     fn into_iter(self) -> Self::IntoIter {
         self.inner.into_iter()

datafusion/core/benches/physical_plan.rs

@@ -50,11 +50,8 @@ fn sort_preserving_merge_operator(
 
     let sort = sort
         .iter()
-        .map(|name| PhysicalSortExpr {
-            expr: col(name, &schema).unwrap(),
-            options: Default::default(),
-        })
-        .collect::<LexOrdering>();
+        .map(|name| PhysicalSortExpr::new_default(col(name, &schema).unwrap()));
+    let sort = LexOrdering::new(sort).unwrap();
 
     let exec = MemorySourceConfig::try_new_exec(
         &batches.into_iter().map(|rb| vec![rb]).collect::<Vec<_>>(),

datafusion/core/benches/sort.rs

@@ -71,7 +71,6 @@ use std::sync::Arc;
 use arrow::array::StringViewArray;
 use arrow::{
     array::{DictionaryArray, Float64Array, Int64Array, StringArray},
-    compute::SortOptions,
     datatypes::{Int32Type, Schema},
     record_batch::RecordBatch,
 };
@@ -272,14 +271,11 @@ impl BenchCase {
 
 /// Make sort exprs for each column in `schema`
 fn make_sort_exprs(schema: &Schema) -> LexOrdering {
-    schema
+    let sort_exprs = schema
         .fields()
         .iter()
-        .map(|f| PhysicalSortExpr {
-            expr: col(f.name(), schema).unwrap(),
-            options: SortOptions::default(),
-        })
-        .collect()
+        .map(|f| PhysicalSortExpr::new_default(col(f.name(), schema).unwrap()));
+    LexOrdering::new(sort_exprs).unwrap()
 }
 
 /// Create streams of int64 (where approximately 1/3 values is repeated)

datafusion/core/benches/spm.rs

@@ -21,7 +21,6 @@ use arrow::array::{ArrayRef, Int32Array, Int64Array, RecordBatch, StringArray};
 use datafusion_execution::TaskContext;
 use datafusion_physical_expr::expressions::col;
 use datafusion_physical_expr::PhysicalSortExpr;
-use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
 use datafusion_physical_plan::{collect, ExecutionPlan};
 
@@ -70,7 +69,7 @@ fn generate_spm_for_round_robin_tie_breaker(
     let partitiones = vec![rbs.clone(); partition_count];
 
     let schema = rb.schema();
-    let sort = LexOrdering::new(vec![
+    let sort = [
         PhysicalSortExpr {
             expr: col("b", &schema).unwrap(),
             options: Default::default(),
@@ -79,7 +78,8 @@ fn generate_spm_for_round_robin_tie_breaker(
             expr: col("c", &schema).unwrap(),
             options: Default::default(),
         },
-    ]);
+    ]
+    .into();
 
     let exec = MemorySourceConfig::try_new_exec(&partitiones, schema, None).unwrap();
     SortPreservingMergeExec::new(sort, exec)