Window aggregates output order broken due to hash repartitioning

[crepererum]

Describe the bug

It seems that under certain circumstances, window aggregation outputs are not sorted. In my test, this requires set datafusion.execution.batch_size = 1;, but there might be other cases.

To Reproduce

Use this sqllogictest:

The output is NOT deterministic!

statement ok
CREATE TABLE t (
    k VARCHAR,
    v Int,
    time TIMESTAMP WITH TIME ZONE
);

statement ok
INSERT INTO t (k, v, time) VALUES
    ('a', 1, '1970-01-01T00:01:00.00Z'),
    ('a', 1, '1970-01-01T00:02:00.00Z'),
    ('a', 1, '1970-01-01T00:03:00.00Z'),
    ('a', 2, '1970-01-01T00:03:00.00Z'),
    ('a', 1, '1970-01-01T00:04:00.00Z'),
    ('b', 3, '1970-01-01T00:01:00.00Z'),
    ('b', 3, '1970-01-01T00:02:00.00Z'),
    ('b', 4, '1970-01-01T00:03:00.00Z'),
    ('b', 4, '1970-01-01T00:03:00.00Z');

query TPI
SELECT
    k,
    time,
    COUNT(v) OVER (
        PARTITION BY k
        ORDER BY time
        RANGE BETWEEN INTERVAL '2 minutes' PRECEDING AND CURRENT ROW
    ) AS normal_count
FROM t
ODER BY k, time;
----
a 1970-01-01T00:01:00Z 1
a 1970-01-01T00:02:00Z 2
a 1970-01-01T00:03:00Z 4
a 1970-01-01T00:03:00Z 4
a 1970-01-01T00:04:00Z 4
b 1970-01-01T00:01:00Z 1
b 1970-01-01T00:02:00Z 2
b 1970-01-01T00:03:00Z 4
b 1970-01-01T00:03:00Z 4

query TT
EXPLAIN SELECT
    k,
    time,
    COUNT(v) OVER (
        PARTITION BY k
        ORDER BY time
        RANGE BETWEEN INTERVAL '2 minutes' PRECEDING AND CURRENT ROW
    ) AS normal_count
FROM t
ODER BY k, time;
----
logical_plan
01)Projection: oder.k, oder.time, count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW AS normal_count
02)--WindowAggr: windowExpr=[[count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW AS count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW]]
03)----SubqueryAlias: oder
04)------TableScan: t projection=[k, v, time]
physical_plan
01)ProjectionExec: expr=[k@0 as k, time@2 as time, count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
02)--BoundedWindowAggExec: wdw=[count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
03)----SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[false]
04)------DataSourceExec: partitions=1, partition_sizes=[1]


statement ok
set datafusion.execution.batch_size = 1;

query TPI
SELECT
    k,
    time,
    COUNT(v) OVER (
        PARTITION BY k
        ORDER BY time
        RANGE BETWEEN INTERVAL '2 minutes' PRECEDING AND CURRENT ROW
    ) AS normal_count
FROM t
ODER BY k, time;
----
b 1970-01-01T00:01:00Z 1
b 1970-01-01T00:02:00Z 2
a 1970-01-01T00:01:00Z 1
a 1970-01-01T00:02:00Z 2
a 1970-01-01T00:03:00Z 4
a 1970-01-01T00:03:00Z 4
b 1970-01-01T00:03:00Z 4
b 1970-01-01T00:03:00Z 4
a 1970-01-01T00:04:00Z 4

query TT
EXPLAIN SELECT
    k,
    time,
    COUNT(v) OVER (
        PARTITION BY k
        ORDER BY time
        RANGE BETWEEN INTERVAL '2 minutes' PRECEDING AND CURRENT ROW
    ) AS normal_count
FROM t
ODER BY k, time;
----
logical_plan
01)Projection: oder.k, oder.time, count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW AS normal_count
02)--WindowAggr: windowExpr=[[count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW AS count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW]]
03)----SubqueryAlias: oder
04)------TableScan: t projection=[k, v, time]
physical_plan
01)ProjectionExec: expr=[k@0 as k, time@2 as time, count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
02)--BoundedWindowAggExec: wdw=[count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(oder.v) PARTITION BY [oder.k] ORDER BY [oder.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
03)----SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[true]
04)------CoalesceBatchesExec: target_batch_size=1
05)--------RepartitionExec: partitioning=Hash([k@0], 4), input_partitions=4
06)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
07)------------DataSourceExec: partitions=1, partition_sizes=[1]

Expected behavior

Both queries in the test file above should always result in an ordered output.

Additional context

Reproduced with 3d4fdf2 .

This could be related to #15833.

[zhuqi-lucas]

Thank you @crepererum I can't reproduce this, it may cause "ODER BY k, time;" this is a typo, need to change to ORDER BY k, time;

Here is the result, it's right.

statement ok
set datafusion.execution.batch_size = 1;

query TPI
SELECT
    k,
    time,
    COUNT(v) OVER (
        PARTITION BY k
        ORDER BY time
        RANGE BETWEEN INTERVAL '2 minutes' PRECEDING AND CURRENT ROW
    ) AS normal_count
FROM t
ORDER BY k, time;
----
a 1970-01-01T00:01:00Z 1
a 1970-01-01T00:02:00Z 2
a 1970-01-01T00:03:00Z 4
a 1970-01-01T00:03:00Z 4
a 1970-01-01T00:04:00Z 4
b 1970-01-01T00:01:00Z 1
b 1970-01-01T00:02:00Z 2
b 1970-01-01T00:03:00Z 4
b 1970-01-01T00:03:00Z 4

query TT
EXPLAIN VERBOSE SELECT
    k,
    time,
    COUNT(v) OVER (
        PARTITION BY k
        ORDER BY time
        RANGE BETWEEN INTERVAL '2 minutes' PRECEDING AND CURRENT ROW
    ) AS normal_count
FROM t
ORDER BY k, time;
----
initial_logical_plan
01)Sort: t.k ASC NULLS LAST, t.time ASC NULLS LAST
02)--Projection: t.k, t.time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW AS normal_count
03)----WindowAggr: windowExpr=[[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW]]
04)------TableScan: t
logical_plan after resolve_grouping_function SAME TEXT AS ABOVE
logical_plan after type_coercion
01)Sort: t.k ASC NULLS LAST, t.time ASC NULLS LAST
02)--Projection: t.k, t.time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW AS normal_count
03)----WindowAggr: windowExpr=[[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW AS count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW]]
04)------TableScan: t
analyzed_logical_plan SAME TEXT AS ABOVE
logical_plan after eliminate_nested_union SAME TEXT AS ABOVE
logical_plan after simplify_expressions SAME TEXT AS ABOVE
logical_plan after replace_distinct_aggregate SAME TEXT AS ABOVE
logical_plan after eliminate_join SAME TEXT AS ABOVE
logical_plan after decorrelate_predicate_subquery SAME TEXT AS ABOVE
logical_plan after scalar_subquery_to_join SAME TEXT AS ABOVE
logical_plan after decorrelate_lateral_join SAME TEXT AS ABOVE
logical_plan after extract_equijoin_predicate SAME TEXT AS ABOVE
logical_plan after eliminate_duplicated_expr SAME TEXT AS ABOVE
logical_plan after eliminate_filter SAME TEXT AS ABOVE
logical_plan after eliminate_cross_join SAME TEXT AS ABOVE
logical_plan after eliminate_limit SAME TEXT AS ABOVE
logical_plan after propagate_empty_relation SAME TEXT AS ABOVE
logical_plan after eliminate_one_union SAME TEXT AS ABOVE
logical_plan after filter_null_join_keys SAME TEXT AS ABOVE
logical_plan after eliminate_outer_join SAME TEXT AS ABOVE
logical_plan after push_down_limit SAME TEXT AS ABOVE
logical_plan after push_down_filter SAME TEXT AS ABOVE
logical_plan after single_distinct_aggregation_to_group_by SAME TEXT AS ABOVE
logical_plan after eliminate_group_by_constant SAME TEXT AS ABOVE
logical_plan after common_sub_expression_eliminate SAME TEXT AS ABOVE
logical_plan after optimize_projections
01)Sort: t.k ASC NULLS LAST, t.time ASC NULLS LAST
02)--Projection: t.k, t.time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW AS normal_count
03)----WindowAggr: windowExpr=[[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW AS count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW]]
04)------TableScan: t projection=[k, v, time]
logical_plan after eliminate_nested_union SAME TEXT AS ABOVE
logical_plan after simplify_expressions SAME TEXT AS ABOVE
logical_plan after replace_distinct_aggregate SAME TEXT AS ABOVE
logical_plan after eliminate_join SAME TEXT AS ABOVE
logical_plan after decorrelate_predicate_subquery SAME TEXT AS ABOVE
logical_plan after scalar_subquery_to_join SAME TEXT AS ABOVE
logical_plan after decorrelate_lateral_join SAME TEXT AS ABOVE
logical_plan after extract_equijoin_predicate SAME TEXT AS ABOVE
logical_plan after eliminate_duplicated_expr SAME TEXT AS ABOVE
logical_plan after eliminate_filter SAME TEXT AS ABOVE
logical_plan after eliminate_cross_join SAME TEXT AS ABOVE
logical_plan after eliminate_limit SAME TEXT AS ABOVE
logical_plan after propagate_empty_relation SAME TEXT AS ABOVE
logical_plan after eliminate_one_union SAME TEXT AS ABOVE
logical_plan after filter_null_join_keys SAME TEXT AS ABOVE
logical_plan after eliminate_outer_join SAME TEXT AS ABOVE
logical_plan after push_down_limit SAME TEXT AS ABOVE
logical_plan after push_down_filter SAME TEXT AS ABOVE
logical_plan after single_distinct_aggregation_to_group_by SAME TEXT AS ABOVE
logical_plan after eliminate_group_by_constant SAME TEXT AS ABOVE
logical_plan after common_sub_expression_eliminate SAME TEXT AS ABOVE
logical_plan after optimize_projections SAME TEXT AS ABOVE
logical_plan
01)Sort: t.k ASC NULLS LAST, t.time ASC NULLS LAST
02)--Projection: t.k, t.time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW AS normal_count
03)----WindowAggr: windowExpr=[[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW AS count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW]]
04)------TableScan: t projection=[k, v, time]
initial_physical_plan
01)SortExec: expr=[k@0 ASC NULLS LAST, time@1 ASC NULLS LAST], preserve_partitioning=[false]
02)--ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
03)----BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
04)------DataSourceExec: partitions=1, partition_sizes=[9]
initial_physical_plan_with_stats
01)SortExec: expr=[k@0 ASC NULLS LAST, time@1 ASC NULLS LAST], preserve_partitioning=[false], statistics=[Rows=Exact(9), Bytes=Absent, [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]:)]]
02)--ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count], statistics=[Rows=Exact(9), Bytes=Absent, [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]:)]]
03)----BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted], statistics=[Rows=Exact(9), Bytes=Absent, [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]: Null=Exact(0)),(Col[3]:)]]
04)------DataSourceExec: partitions=1, partition_sizes=[9], statistics=[Rows=Exact(9), Bytes=Exact(6840), [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]: Null=Exact(0))]]
initial_physical_plan_with_schema
01)SortExec: expr=[k@0 ASC NULLS LAST, time@1 ASC NULLS LAST], preserve_partitioning=[false], schema=[k:Utf8View;N, time:Timestamp(Nanosecond, Some("+00:00"));N, normal_count:Int64]
02)--ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count], schema=[k:Utf8View;N, time:Timestamp(Nanosecond, Some("+00:00"));N, normal_count:Int64]
03)----BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted], schema=[k:Utf8View;N, v:Int32;N, time:Timestamp(Nanosecond, Some("+00:00"));N, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW:Int64]
04)------DataSourceExec: partitions=1, partition_sizes=[9], schema=[k:Utf8View;N, v:Int32;N, time:Timestamp(Nanosecond, Some("+00:00"));N]
physical_plan after OutputRequirements
01)OutputRequirementExec: order_by=[(k@0, asc), (time@1, asc)], dist_by=SinglePartition
02)--SortExec: expr=[k@0 ASC NULLS LAST, time@1 ASC NULLS LAST], preserve_partitioning=[false]
03)----ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
04)------BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
05)--------DataSourceExec: partitions=1, partition_sizes=[9]
physical_plan after aggregate_statistics SAME TEXT AS ABOVE
physical_plan after join_selection SAME TEXT AS ABOVE
physical_plan after LimitedDistinctAggregation SAME TEXT AS ABOVE
physical_plan after FilterPushdown SAME TEXT AS ABOVE
physical_plan after EnforceDistribution
01)OutputRequirementExec: order_by=[(k@0, asc), (time@1, asc)], dist_by=SinglePartition
02)--SortExec: expr=[k@0 ASC NULLS LAST, time@1 ASC NULLS LAST], preserve_partitioning=[false]
03)----CoalescePartitionsExec
04)------ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
05)--------BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
06)----------RepartitionExec: partitioning=Hash([k@0], 2), input_partitions=2
07)------------DataSourceExec: partitions=2, partition_sizes=[5, 4]
physical_plan after CombinePartialFinalAggregate SAME TEXT AS ABOVE
physical_plan after EnforceSorting
01)OutputRequirementExec: order_by=[(k@0, asc), (time@1, asc)], dist_by=SinglePartition
02)--SortPreservingMergeExec: [k@0 ASC NULLS LAST, time@1 ASC NULLS LAST]
03)----ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
04)------BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
05)--------SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[true]
06)----------RepartitionExec: partitioning=Hash([k@0], 2), input_partitions=2
07)------------DataSourceExec: partitions=2, partition_sizes=[5, 4]
physical_plan after OptimizeAggregateOrder SAME TEXT AS ABOVE
physical_plan after ProjectionPushdown SAME TEXT AS ABOVE
physical_plan after coalesce_batches
01)OutputRequirementExec: order_by=[(k@0, asc), (time@1, asc)], dist_by=SinglePartition
02)--SortPreservingMergeExec: [k@0 ASC NULLS LAST, time@1 ASC NULLS LAST]
03)----ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
04)------BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
05)--------SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[true]
06)----------CoalesceBatchesExec: target_batch_size=1
07)------------RepartitionExec: partitioning=Hash([k@0], 2), input_partitions=2
08)--------------DataSourceExec: partitions=2, partition_sizes=[5, 4]
physical_plan after coalesce_async_exec_input SAME TEXT AS ABOVE
physical_plan after OutputRequirements
01)SortPreservingMergeExec: [k@0 ASC NULLS LAST, time@1 ASC NULLS LAST]
02)--ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
03)----BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
04)------SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[true]
05)--------CoalesceBatchesExec: target_batch_size=1
06)----------RepartitionExec: partitioning=Hash([k@0], 2), input_partitions=2
07)------------DataSourceExec: partitions=2, partition_sizes=[5, 4]
physical_plan after LimitAggregation SAME TEXT AS ABOVE
physical_plan after LimitPushdown SAME TEXT AS ABOVE
physical_plan after ProjectionPushdown SAME TEXT AS ABOVE
physical_plan after EnsureCooperative SAME TEXT AS ABOVE
physical_plan after FilterPushdown(Post) SAME TEXT AS ABOVE
physical_plan after SanityCheckPlan SAME TEXT AS ABOVE
physical_plan
01)SortPreservingMergeExec: [k@0 ASC NULLS LAST, time@1 ASC NULLS LAST]
02)--ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count]
03)----BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted]
04)------SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[true]
05)--------CoalesceBatchesExec: target_batch_size=1
06)----------RepartitionExec: partitioning=Hash([k@0], 2), input_partitions=2
07)------------DataSourceExec: partitions=2, partition_sizes=[5, 4]
physical_plan_with_stats
01)SortPreservingMergeExec: [k@0 ASC NULLS LAST, time@1 ASC NULLS LAST], statistics=[Rows=Exact(9), Bytes=Absent, [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]:)]]
02)--ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count], statistics=[Rows=Exact(9), Bytes=Absent, [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]:)]]
03)----BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted], statistics=[Rows=Exact(9), Bytes=Absent, [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]: Null=Exact(0)),(Col[3]:)]]
04)------SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[true], statistics=[Rows=Exact(9), Bytes=Exact(6840), [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]: Null=Exact(0))]]
05)--------CoalesceBatchesExec: target_batch_size=1, statistics=[Rows=Exact(9), Bytes=Exact(6840), [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]: Null=Exact(0))]]
06)----------RepartitionExec: partitioning=Hash([k@0], 2), input_partitions=2, statistics=[Rows=Exact(9), Bytes=Exact(6840), [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]: Null=Exact(0))]]
07)------------DataSourceExec: partitions=2, partition_sizes=[5, 4], statistics=[Rows=Exact(9), Bytes=Exact(6840), [(Col[0]: Null=Exact(0)),(Col[1]: Null=Exact(0)),(Col[2]: Null=Exact(0))]]
physical_plan_with_schema
01)SortPreservingMergeExec: [k@0 ASC NULLS LAST, time@1 ASC NULLS LAST], schema=[k:Utf8View;N, time:Timestamp(Nanosecond, Some("+00:00"));N, normal_count:Int64]
02)--ProjectionExec: expr=[k@0 as k, time@2 as time, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW@3 as normal_count], schema=[k:Utf8View;N, time:Timestamp(Nanosecond, Some("+00:00"));N, normal_count:Int64]
03)----BoundedWindowAggExec: wdw=[count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW: Field { name: "count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN IntervalMonthDayNano { months: 0, days: 0, nanoseconds: 120000000000 } PRECEDING AND CURRENT ROW], mode=[Sorted], schema=[k:Utf8View;N, v:Int32;N, time:Timestamp(Nanosecond, Some("+00:00"));N, count(t.v) PARTITION BY [t.k] ORDER BY [t.time ASC NULLS LAST] RANGE BETWEEN 2 minutes PRECEDING AND CURRENT ROW:Int64]
04)------SortExec: expr=[k@0 ASC NULLS LAST, time@2 ASC NULLS LAST], preserve_partitioning=[true], schema=[k:Utf8View;N, v:Int32;N, time:Timestamp(Nanosecond, Some("+00:00"));N]
05)--------CoalesceBatchesExec: target_batch_size=1, schema=[k:Utf8View;N, v:Int32;N, time:Timestamp(Nanosecond, Some("+00:00"));N]
06)----------RepartitionExec: partitioning=Hash([k@0], 2), input_partitions=2, schema=[k:Utf8View;N, v:Int32;N, time:Timestamp(Nanosecond, Some("+00:00"));N]
07)------------DataSourceExec: partitions=2, partition_sizes=[5, 4], schema=[k:Utf8View;N, v:Int32;N, time:Timestamp(Nanosecond, Some("+00:00"));N]

[crepererum]

Yes, that's a typo. I think my follow up question would be: why is this a valid query?

SELECT *
FROM t whatever_alias BY k;

Or more precisely: what does BY k in the first FROM table mean?

[zhuqi-lucas]

Yes, that's a typo. I think my follow up question would be: why is this a valid query?

SELECT *
FROM t whatever_alias BY k;
Or more precisely: what does BY k in the first FROM table mean?

Very good question @crepererum , it may due to:

In DataFusion’s SQL grammar, any identifier immediately following a table reference can be taken as an alias, and if it’s followed by BY col1, col2… it’s treated as a distribution hint (i.e. “hash‑partition on these columns”) rather than a typo of ORDER BY?

That syntax

(FROM <table> <alias> BY <cols>) 

is perfectly valid, there’s no parse or validation error...

I need to investigate more, if we can have a better solution to solve this, thanks !

[crepererum]

I need to investigate more, if we can have a better solution to solve this, thanks !

I'm also OK with accepting the current behavior, even though it's slightly suboptimal UX.