[MAJOR] Equivalence System Overhaul (#16217)

* introduce Soft & Hard RequiredInputOrderings
remove usage of prefer_existing_sort
as default set requirements Hard
set soft on AggregateExec and BoundedWindowExec since they have InputOrderMode functionalities

* add documentation to replace_with_partial_sort
simplify indentation

* add documentation to analyze_immediate_sort_removal
simplify indentation

* remove prefer_existing_sort effects
remove prefer_existing_sort based test cases

* remove prefer_existing_sort configuration

* remove prefer_existing_sort configuration

* add documentation

* add documentation

* add documentation

* fix imports and test cases

* fix imports and test cases

* implement RequiredInputOrdering as vectors

* implement RequiredInputOrdering as vectors
return alternative on BoundedWindowAggExec

* fix test cases

* change doc

* revert prefer_existing_sort flag

* fix changes

* fix test case

* make LexRequirement private

* ensure RequiredInputOrdering inner requirement can not be empty
simplify sort_pushdown.rs

* add default test cases
add requirements compatible test cases

* doc fixes

* fix clippy and docs

* format code

* format code

* doc fix

* add TODO test cases with test_soft_hard_requirements prefix

* Review Part 1

* Review Part 2

* Review Part 3

* Review Part 4

* Review Part 5

* Review Part 6

* Enforce non-degeneracy for LexRequirement

* Enforce non-degeneracy for LexOrdering (Part 1)

* Enforce non-degeneracy for LexOrdering (Part 2)

* fix first phase of merge conflicts and other bugs

* Fix sqllogictests except the schema mismatch

* Cleanup Part 1

* Cleanup Part 2

* Cleanup Part 3

* do not initialize Trivial accumulators if ordering is set

* initialize TrivialFirstPrimitiveGroupsAccumulator struct and return

* fix clippy

* fix merge conflicts

* fix typos
remove TrivialFirstPrimitiveGroupsAccumulator
make groups accumulator available only when order requirement is set

* format code

* Add requirement_satisfied back in

* Replace AsRef with ordinary & for LexOrdering

* Further cleanup

* add OutputRequirementExec fetches to sort adding

* Simplify remove_redundant_entries

* Work with iterators in ordering_satisfy_requirement

* Fix doctests

* Cleanup LexOrdering APIs

* Cleanup LexOrdering APIs 2

* Add reverse_each to LexOrdering

* Use LexOrdering instead of Arc<[PhysicalSortExpr]>

* Use PhysicalSortExpr slices in contexts where we simply list sort expressions

* Generalize add_new_ordering APIs

* Simplifications

* More cleanups

* API Simplifications

* Improve comments

* Use vector in Expr structs

* Fix doctests

* Simplify sort

* Simplify the get_finer_aggregate_exprs_requirement function

* Avoid hidden clones

* bugfix

* Simplify the get_finer_aggregate_exprs_requirement function

* Simplify the function with_reorder

* Fix with_reorder bug

* Simplify the function with_reorder (Part 2)

* Simplify

* DRY

* Simplifications

* Improve add_equal_condition

* Improve docs

* Simplifications

* Simplifications

* RequiredInputOrdering -> OrderingAlternatives

* Simplify new_with_orderings

* Transition to fallible LexOrdering constructor

* Transition to fallible LexOrdering constructor - 2

* Transition to fallible LexOrdering constructor - 3

* Transition to fallible LexOrdering constructor - 4

* Transition to fallible LexOrdering constructor - 5

* Transition to fallible LexOrdering constructor - 6

* Transition to fallible LexOrdering constructor - 7

* Transition to fallible LexOrdering constructor - 8

* Transition to fallible LexOrdering constructor - 9

* Transition to fallible LexOrdering constructor - 10

* Transition to fallible LexOrdering constructor - 11

* Simplify constant expressions

* Simplify constant expressions - 2

* Simplify constant expressions - 3

* Simplify constant expressions - 4

* Simplify constant expressions - 5

* Simplify constant expressions - 6

* Simplify constant expressions - 7

* Simplify constant expressions - 8

* Simplify constant expressions - 9

* Fix imports

* Remove explicit constant tracking from equivalences

* Resolve logical conflict

* Remove the unusual take API, instead use the from trait

* Simplify projection mapping - 1

* Use a map instead of a vector in ProjectionMapping

* Simplify DependencyMap

* Simplify DependencyMap - 2

* Simplify DependencyMap - 3

* Incorporate Jay's suggestions

* Simplifications

* Fix doctest

* Improve docstrings

* Update/cast the constant value accordingly when schema changes

* Improve ProjectionMapping

* Remove DerefMut from ProjectionTargets to preserve non-emptiness

* Docstring

* Optimize project_expr by fetching equivalence classes only once

* Project multiple expressions more efficiently at once

* Project multiple expressions more efficiently at once - 2

* Project multiple expressions more efficiently at once - 3

* Project multiple expressions more efficiently at once - 4

* Move normalization of sort expressions to equivalence group

* Improve comments

* Improve display for EquivalenceProperties

* More idiomatic code

* More succinct code

* Remove extend_orderings from EquivalenceProperties

* Simplify with_reorder

* Store normalized orderings - 1

* Reduce time complexity of normalization w.r.t. number of equivalence classes

* Simplify bridge_classes logic

* Remove TODOs

* Simplify generate_dependency_orderings

* normalized orderings - 2

* normalized orderings - 3

* undo normalized orderings

* Fix logical conflicts

* Fix imports

* Remove noop code

* Move add_offset_to_expr

* Remove mutation from LexOrdering

* Remove unwraps

* Remove unwraps - 2

* Remove unwraps - 3

* Remove unwraps - 4

* Remove collapse from LexOrdering

* Remove unwraps - 5

* Remove unwraps - 6

* Remove unwraps - 7

* Remove unwraps - 8

* Remove unwraps - 9

* Remove unwraps - 10

* Remove collapse from LexRequirement

* Simplify ordering_satisfy

* Enforce uniqueness in LexOrdering

* Fix with_reorder

* Use tee

* Fix reorder api

* Comment grammar

* Remove unwraps

* Cache normalized orderings

* Minor: remove an unecessary clone in common_sort_prefix_length

* Address reviews

---------

Co-authored-by: mertak-synnada <mertak67+synaada@gmail.com>
Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>
Co-authored-by: Berkay Şahin <124376117+berkaysynnada@users.noreply.github.com>

Author: 4 people

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)