diff --git a/datafusion/physical-expr/src/equivalence/properties.rs b/datafusion/physical-expr/src/equivalence/properties.rs index 2471d9249e163..386e74b4a6dcb 100644 --- a/datafusion/physical-expr/src/equivalence/properties.rs +++ b/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::>(); @@ -839,7 +845,7 @@ fn is_constant_recurse( constants: &[Arc], expr: &Arc, ) -> bool { - if physical_exprs_contains(constants, expr) { + if physical_exprs_contains(constants, expr) || expr.as_any().is::() { 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::>(); + 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()?; diff --git a/datafusion/sqllogictest/test_files/order.slt b/datafusion/sqllogictest/test_files/order.slt index 61888eb8006d7..2ea78448b940a 100644 --- a/datafusion/sqllogictest/test_files/order.slt +++ b/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 diff --git a/datafusion/sqllogictest/test_files/subquery.slt b/datafusion/sqllogictest/test_files/subquery.slt index 1ca9045f1bd00..1f2870a8c92fa 100644 --- a/datafusion/sqllogictest/test_files/subquery.slt +++ b/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 diff --git a/datafusion/sqllogictest/test_files/union.slt b/datafusion/sqllogictest/test_files/union.slt index b4e338875e247..5550fa68a6fca 100644 --- a/datafusion/sqllogictest/test_files/union.slt +++ b/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] diff --git a/datafusion/sqllogictest/test_files/window.slt b/datafusion/sqllogictest/test_files/window.slt index afdd4a9b487b2..fa4445d4cd4c9 100644 --- a/datafusion/sqllogictest/test_files/window.slt +++ b/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]