Add constant expression support to equivalence properties

datafusion/physical-expr/src/equivalence/properties.rs

@@ -31,7 +31,7 @@ use crate::{
     PhysicalSortRequirement,
 };
 
-use arrow_schema::SchemaRef;
+use arrow_schema::{SchemaRef, SortOptions};
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_common::{JoinSide, JoinType};
 
@@ -710,10 +710,16 @@ impl EquivalenceProperties {
                 .flat_map(|&idx| {
                     let ExprOrdering { expr, data, .. } =
                         eq_properties.get_expr_ordering(exprs[idx].clone());
-                    if let SortProperties::Ordered(options) = data {
-                        Some((PhysicalSortExpr { expr, options }, idx))
-                    } else {
-                        None
+                    match data {
+                        SortProperties::Ordered(options) => {
+                            Some((PhysicalSortExpr { expr, options }, idx))
+                        }
+                        SortProperties::Singleton => {
+                            // Assign default ordering to constant expressions
+                            let options = SortOptions::default();
+                            Some((PhysicalSortExpr { expr, options }, idx))
+                        }
+                        SortProperties::Unordered => None,
                     }
                 })
                 .collect::<Vec<_>>();
@@ -839,7 +845,7 @@ fn is_constant_recurse(
     constants: &[Arc<dyn PhysicalExpr>],
     expr: &Arc<dyn PhysicalExpr>,
 ) -> bool {
-    if physical_exprs_contains(constants, expr) {
+    if physical_exprs_contains(constants, expr) || expr.as_any().is::<Literal>() {
         return true;
     }
     let children = expr.children();
@@ -1881,6 +1887,36 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_find_longest_permutation2() -> Result<()> {
+        // Schema satisfies following orderings:
+        // [a ASC], [d ASC, b ASC], [e DESC, f ASC, g ASC]
+        // and
+        // Column [a=c] (e.g they are aliases).
+        // At below we add [d ASC, h DESC] also, for test purposes
+        let (test_schema, mut eq_properties) = create_test_params()?;
+        let col_h = &col("h", &test_schema)?;
+
+        // Add column h as constant
+        eq_properties = eq_properties.add_constants(vec![col_h.clone()]);
+
+        let test_cases = vec![
+            // TEST CASE 1
+            // ordering of the constants are treated as default ordering.
+            // This is the convention currently used.
+            (vec![col_h], vec![(col_h, SortOptions::default())]),
+        ];
+        for (exprs, expected) in test_cases {
+            let exprs = exprs.into_iter().cloned().collect::<Vec<_>>();
+            let expected = convert_to_sort_exprs(&expected);
+            let (actual, _) = eq_properties.find_longest_permutation(&exprs);
+            assert_eq!(actual, expected);
+        }
+
+        Ok(())
+    }
+
     #[test]
     fn test_get_meet_ordering() -> Result<()> {
         let schema = create_test_schema()?;

datafusion/sqllogictest/test_files/order.slt

@@ -769,18 +769,18 @@ SortPreservingMergeExec: [m@0 ASC NULLS LAST,t@1 ASC NULLS LAST]
 --SortExec: expr=[m@0 ASC NULLS LAST,t@1 ASC NULLS LAST]
 ----InterleaveExec
 ------ProjectionExec: expr=[Int64(0)@0 as m, t@1 as t]
---------AggregateExec: mode=FinalPartitioned, gby=[Int64(0)@0 as Int64(0), t@1 as t], aggr=[]
+--------AggregateExec: mode=FinalPartitioned, gby=[Int64(0)@0 as Int64(0), t@1 as t], aggr=[], ordering_mode=PartiallySorted([0])
 ----------CoalesceBatchesExec: target_batch_size=8192
 ------------RepartitionExec: partitioning=Hash([Int64(0)@0, t@1], 2), input_partitions=2
 --------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------AggregateExec: mode=Partial, gby=[0 as Int64(0), t@0 as t], aggr=[]
+----------------AggregateExec: mode=Partial, gby=[0 as Int64(0), t@0 as t], aggr=[], ordering_mode=PartiallySorted([0])
 ------------------ProjectionExec: expr=[column1@0 as t]
 --------------------ValuesExec
 ------ProjectionExec: expr=[Int64(1)@0 as m, t@1 as t]
---------AggregateExec: mode=FinalPartitioned, gby=[Int64(1)@0 as Int64(1), t@1 as t], aggr=[]
+--------AggregateExec: mode=FinalPartitioned, gby=[Int64(1)@0 as Int64(1), t@1 as t], aggr=[], ordering_mode=PartiallySorted([0])
 ----------CoalesceBatchesExec: target_batch_size=8192
 ------------RepartitionExec: partitioning=Hash([Int64(1)@0, t@1], 2), input_partitions=2
 --------------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-----------------AggregateExec: mode=Partial, gby=[1 as Int64(1), t@0 as t], aggr=[]
+----------------AggregateExec: mode=Partial, gby=[1 as Int64(1), t@0 as t], aggr=[], ordering_mode=PartiallySorted([0])
 ------------------ProjectionExec: expr=[column1@0 as t]
 --------------------ValuesExec

datafusion/sqllogictest/test_files/subquery.slt

@@ -264,10 +264,10 @@ ProjectionExec: expr=[t1_id@0 as t1_id, SUM(t2.t2_int)@1 as t2_sum]
 ------CoalesceBatchesExec: target_batch_size=2
 --------RepartitionExec: partitioning=Hash([t2_id@1], 4), input_partitions=4
 ----------ProjectionExec: expr=[SUM(t2.t2_int)@2 as SUM(t2.t2_int), t2_id@0 as t2_id]
-------------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id, Utf8("a")@1 as Utf8("a")], aggr=[SUM(t2.t2_int)]
+------------AggregateExec: mode=FinalPartitioned, gby=[t2_id@0 as t2_id, Utf8("a")@1 as Utf8("a")], aggr=[SUM(t2.t2_int)], ordering_mode=PartiallySorted([1])
 --------------CoalesceBatchesExec: target_batch_size=2
 ----------------RepartitionExec: partitioning=Hash([t2_id@0, Utf8("a")@1], 4), input_partitions=4
-------------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id, a as Utf8("a")], aggr=[SUM(t2.t2_int)]
+------------------AggregateExec: mode=Partial, gby=[t2_id@0 as t2_id, a as Utf8("a")], aggr=[SUM(t2.t2_int)], ordering_mode=PartiallySorted([1])
 --------------------MemoryExec: partitions=4, partition_sizes=[1, 0, 0, 0]
 
 query II rowsort

datafusion/sqllogictest/test_files/union.slt

@@ -547,10 +547,10 @@ Union
 physical_plan
 UnionExec
 --ProjectionExec: expr=[Int64(1)@0 as a]
-----AggregateExec: mode=FinalPartitioned, gby=[Int64(1)@0 as Int64(1)], aggr=[]
+----AggregateExec: mode=FinalPartitioned, gby=[Int64(1)@0 as Int64(1)], aggr=[], ordering_mode=Sorted
 ------CoalesceBatchesExec: target_batch_size=2
 --------RepartitionExec: partitioning=Hash([Int64(1)@0], 4), input_partitions=1
-----------AggregateExec: mode=Partial, gby=[1 as Int64(1)], aggr=[]
+----------AggregateExec: mode=Partial, gby=[1 as Int64(1)], aggr=[], ordering_mode=Sorted
 ------------PlaceholderRowExec
 --ProjectionExec: expr=[2 as a]
 ----PlaceholderRowExec
@@ -578,10 +578,10 @@ Union
 physical_plan
 UnionExec
 --ProjectionExec: expr=[COUNT(*)@1 as count, n@0 as n]
-----AggregateExec: mode=FinalPartitioned, gby=[n@0 as n], aggr=[COUNT(*)]
+----AggregateExec: mode=FinalPartitioned, gby=[n@0 as n], aggr=[COUNT(*)], ordering_mode=Sorted
 ------CoalesceBatchesExec: target_batch_size=2
 --------RepartitionExec: partitioning=Hash([n@0], 4), input_partitions=1
-----------AggregateExec: mode=Partial, gby=[n@0 as n], aggr=[COUNT(*)]
+----------AggregateExec: mode=Partial, gby=[n@0 as n], aggr=[COUNT(*)], ordering_mode=Sorted
 ------------ProjectionExec: expr=[5 as n]
 --------------PlaceholderRowExec
 --ProjectionExec: expr=[1 as count, MAX(Int64(10))@0 as n]

datafusion/sqllogictest/test_files/window.slt

@@ -4053,3 +4053,52 @@ SELECT c3,
 125 217013822852 8701233618
 123 213388536442 11293564174
 97 210796205886 14767488750
+
+statement ok
+create table a (a bigint) AS VALUES (1);
+
+query I
+select count(*) over (partition by a order by a) from (select * from a where a = 1);
+----
+1
+
+query TT
+EXPLAIN select count(*) over (partition by a order by a) from (select * from a where a = 1);
+----
+logical_plan
+Projection: COUNT(*) PARTITION BY [a.a] ORDER BY [a.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+--WindowAggr: windowExpr=[[COUNT(UInt8(1)) PARTITION BY [a.a] ORDER BY [a.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS COUNT(*) PARTITION BY [a.a] ORDER BY [a.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+----Filter: a.a = Int64(1)
+------TableScan: a projection=[a]
+physical_plan
+ProjectionExec: expr=[COUNT(*) PARTITION BY [a.a] ORDER BY [a.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@1 as COUNT(*) PARTITION BY [a.a] ORDER BY [a.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]
+--BoundedWindowAggExec: wdw=[COUNT(*) PARTITION BY [a.a] ORDER BY [a.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name: "COUNT(*) PARTITION BY [a.a] ORDER BY [a.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range, start_bound: Preceding(Int64(NULL)), end_bound: CurrentRow }], mode=[Sorted]
+----CoalesceBatchesExec: target_batch_size=4096
+------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
+--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+----------CoalesceBatchesExec: target_batch_size=4096
+------------FilterExec: a@0 = 1
+--------------MemoryExec: partitions=1, partition_sizes=[1]
+
+query I
+select ROW_NUMBER() over (partition by a) from (select * from a where a = 1);
+----
+1
+
+query TT
+EXPLAIN select ROW_NUMBER() over (partition by a) from (select * from a where a = 1);
+----
+logical_plan
+Projection: ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+--WindowAggr: windowExpr=[[ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]]
+----Filter: a.a = Int64(1)
+------TableScan: a projection=[a]
+physical_plan
+ProjectionExec: expr=[ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@1 as ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]
+--BoundedWindowAggExec: wdw=[ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)) }], mode=[Sorted]
+----CoalesceBatchesExec: target_batch_size=4096
+------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
+--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+----------CoalesceBatchesExec: target_batch_size=4096
+------------FilterExec: a@0 = 1
+--------------MemoryExec: partitions=1, partition_sizes=[1]