[avantgardnerio]

Which issue does this PR close?

• Closes Optimizer should push limits past window functions #17346.

Rationale for this change

We would like plans with window functions to go faster.

What changes are included in this PR?

A proof-of-concept level optimizer rule push limits past some window functions.

Are these changes tested?

So far only by the existing sqllogictest suite.

Are there any user-facing changes?

Queries with window functions may go faster, particularly if the TableProvider supports TopK pushdown.

[akurmustafa]

Thanks @avantgardnerio for this PR

[akurmustafa]

As far as I can see, we update latest_max only for WindowFrameUnits::Rows cases. If I am not mistaken, it is also valid to update latest_max for WindowFrameUnits::Range and WindowFrameUnits::Groups as longs as end_bound is WindowFrameBound::Preceding(_). I think, we can extend checks to include this use case also.
This should change the plan for following kind of queries

SELECT
    c9,
    SUM(c9) OVER(ORDER BY c9 ASC RANGE BETWEEN 5 PRECEDING AND 1 PRECEDING) as sum1
    FROM aggregate_test_100
    LIMIT 5

However, we can do this change in subsequent PRs. I think in current form, this PR is correct and we can merge as is. Thanks @avantgardnerio for this PR.

[avantgardnerio]

Good point! I will file a subsequent PR 😄

[avantgardnerio]

I tried this and I think it's unsafe:

1. query result mismatch:
[SQL] SELECT
 SUM(c1) OVER (ORDER BY c2 DESC) as summation1
 FROM null_cases
 LIMIT 5;
[Diff] (-expected|+actual)
-   962
-   962
-   962
-   962
-   962
+   263
+   263
+   263
+   263
+   263

and

[Diff] (-expected|+actual)
    logical_plan
    01)Projection: sum(null_cases.c1) ORDER BY [null_cases.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS summation1
    02)--Limit: skip=0, fetch=5
    03)----WindowAggr: windowExpr=[[sum(null_cases.c1) ORDER BY [null_cases.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
    04)------TableScan: null_cases projection=[c1, c2]
    physical_plan
    01)ProjectionExec: expr=[sum(null_cases.c1) ORDER BY [null_cases.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as summation1]
    02)--GlobalLimitExec: skip=0, fetch=5
    03)----BoundedWindowAggExec: wdw=[sum(null_cases.c1) ORDER BY [null_cases.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Field { name: "sum(null_cases.c1) ORDER BY [null_cases.c2 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, frame: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
-   04)------SortExec: expr=[c2@1 DESC], preserve_partitioning=[false]
+   04)------SortExec: TopK(fetch=5), expr=[c2@1 DESC], preserve_partitioning=[false]
    05)--------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/core/tests/data/null_cases.csv]]}, projection=[c1, c2], file_type=csv, has_header=true

I think this is failing because there are multiple entries with the same value, so it needs more than the limit.

[akurmustafa]

However, for this case end bound is Current Row. I think, when end bound is WindowFrameBound::Preceding(_) we can still do this optimization for WindowFrameUnits::Range and WindowFrameUnits::Groups