Frame window function

benchmarks/src/Window/MovingAverageScenario.php

@@ -0,0 +1,40 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Window;
+
+use Flow\Benchmarks\BenchmarkConfig;
+use Flow\Benchmarks\Datasets\Datasets;
+
+use function Flow\ETL\DSL\average;
+use function Flow\ETL\DSL\current_row;
+use function Flow\ETL\DSL\data_frame;
+use function Flow\ETL\DSL\preceding;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\window;
+use function Flow\Floe\DSL\from_floe;
+
+final readonly class MovingAverageScenario
+{
+    public function __construct(
+        private int $rows,
+    ) {}
+
+    public function run(): void
+    {
+        data_frame(BenchmarkConfig::builder())
+            ->read(from_floe(Datasets::orders($this->rows)->floe()))
+            ->withEntry(
+                'moving_average',
+                average(ref('discount'))
+                    ->over(
+                        window()
+                            ->partitionBy(ref('seller_id'))
+                            ->orderBy(ref('created_at'))
+                            ->rowsBetween(preceding(10), current_row()),
+                    ),
+            )
+            ->run();
+    }
+}

benchmarks/src/Window/RankScenario.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Window;
+
+use Flow\Benchmarks\BenchmarkConfig;
+use Flow\Benchmarks\Datasets\Datasets;
+
+use function Flow\ETL\DSL\data_frame;
+use function Flow\ETL\DSL\rank;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\window;
+use function Flow\Floe\DSL\from_floe;
+
+final readonly class RankScenario
+{
+    public function __construct(
+        private int $rows,
+    ) {}
+
+    public function run(): void
+    {
+        data_frame(BenchmarkConfig::builder())
+            ->read(from_floe(Datasets::orders($this->rows)->floe()))
+            ->withEntry('rank', rank()->over(window()->partitionBy(ref('seller_id'))->orderBy(ref('created_at'))))
+            ->run();
+    }
+}

benchmarks/src/Window/RowNumberScenario.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Window;
+
+use Flow\Benchmarks\BenchmarkConfig;
+use Flow\Benchmarks\Datasets\Datasets;
+
+use function Flow\ETL\DSL\data_frame;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\row_number;
+use function Flow\ETL\DSL\window;
+use function Flow\Floe\DSL\from_floe;
+
+final readonly class RowNumberScenario
+{
+    public function __construct(
+        private int $rows,
+    ) {}
+
+    public function run(): void
+    {
+        data_frame(BenchmarkConfig::builder())
+            ->read(from_floe(Datasets::orders($this->rows)->floe()))
+            ->withEntry(
+                'row_number',
+                row_number()->over(window()->partitionBy(ref('seller_id'))->orderBy(ref('created_at'))),
+            )
+            ->run();
+    }
+}

benchmarks/src/Window/RunningTotalScenario.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Window;
+
+use Flow\Benchmarks\BenchmarkConfig;
+use Flow\Benchmarks\Datasets\Datasets;
+
+use function Flow\ETL\DSL\data_frame;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\sum;
+use function Flow\ETL\DSL\window;
+use function Flow\Floe\DSL\from_floe;
+
+final readonly class RunningTotalScenario
+{
+    public function __construct(
+        private int $rows,
+    ) {}
+
+    public function run(): void
+    {
+        data_frame(BenchmarkConfig::builder())
+            ->read(from_floe(Datasets::orders($this->rows)->floe()))
+            ->withEntry(
+                'running_total',
+                sum(ref('discount'))->over(window()->partitionBy(ref('seller_id'))->orderBy(ref('created_at'))),
+            )
+            ->run();
+    }
+}

benchmarks/src/Window/WholePartitionScenario.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Window;
+
+use Flow\Benchmarks\BenchmarkConfig;
+use Flow\Benchmarks\Datasets\Datasets;
+
+use function Flow\ETL\DSL\data_frame;
+use function Flow\ETL\DSL\ref;
+use function Flow\ETL\DSL\sum;
+use function Flow\ETL\DSL\window;
+use function Flow\Floe\DSL\from_floe;
+
+final readonly class WholePartitionScenario
+{
+    public function __construct(
+        private int $rows,
+    ) {}
+
+    public function run(): void
+    {
+        data_frame(BenchmarkConfig::builder())
+            ->read(from_floe(Datasets::orders($this->rows)->floe()))
+            ->withEntry('seller_total', sum(ref('discount'))->over(window()->partitionBy(ref('seller_id'))))
+            ->run();
+    }
+}

benchmarks/suites/Window/WindowBench.php

@@ -0,0 +1,58 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\Benchmarks\Window;
+
+use Generator;
+use PhpBench\Attributes as Bench;
+
+final class WindowBench
+{
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['window'])]
+    public function bench_moving_average(array $params): void
+    {
+        (new MovingAverageScenario((int) $params['rows']))->run();
+    }
+
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['window'])]
+    public function bench_rank(array $params): void
+    {
+        (new RankScenario((int) $params['rows']))->run();
+    }
+
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['window'])]
+    public function bench_row_number(array $params): void
+    {
+        (new RowNumberScenario((int) $params['rows']))->run();
+    }
+
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['window'])]
+    public function bench_running_total(array $params): void
+    {
+        (new RunningTotalScenario((int) $params['rows']))->run();
+    }
+
+    #[Bench\ParamProviders('rows')]
+    #[Bench\Groups(['window'])]
+    public function bench_whole_partition(array $params): void
+    {
+        (new WholePartitionScenario((int) $params['rows']))->run();
+    }
+
+    /**
+     * The orders dataset spreads rows over 5 sellers, so every row here lands in a partition of
+     * rows/5. Frame-materializing aggregates are quadratic in partition size, hence a lower default
+     * than the other suites - 100_000 rows would put bench_whole_partition into the minutes.
+     */
+    public function rows(): Generator
+    {
+        $rows = (int) (getenv('FLOW_BENCH_WINDOW_ROWS') ?: 10_000);
+
+        yield number_format($rows) => ['rows' => $rows];
+    }
+}

documentation/components/core/window-functions.md

@@ -4,12 +4,13 @@
 
 [TOC]
 
-Window functions are a special type of function that perform calculations on a set of rows that are related to the current row. 
-Unlike regular aggregate functions, use of a window function does not cause rows to become grouped into a single output row — the rows retain their separate identities. 
-Behind the scenes, the window function is able to access more than just the current row of the query result.
+Window functions are a special type of function that perform calculations on a set of rows that are related to the
+current row. Unlike regular aggregate functions, use of a window function does not cause rows to become grouped into a
+single output row — the rows retain their separate identities. Behind the scenes, the window function is able to access
+more than just the current row of the query result.
 
-To narrow window function to a specific set of rows, you need to use `window`.
-Window is used to partition dataset into smaller partitions on which later window function will be applied.
+To narrow window function to a specific set of rows, you need to use `window`. Window is used to partition dataset into
+smaller partitions on which later window function will be applied.
 
 ### Window Functions:
 
@@ -20,8 +21,8 @@ Window is used to partition dataset into smaller partitions on which later windo
 - [`AVERAGE`](/src/core/etl/src/Flow/ETL/Function/Average.php)
 - [`COUNT`](/src/core/etl/src/Flow/ETL/Function/Count.php)
 
-
-All window functions are implementing [`WindowFunction`](/src/core/etl/src/Flow/ETL/Function/WindowFunction.php) interface.
+All window functions are implementing [`WindowFunction`](/src/core/etl/src/Flow/ETL/Function/WindowFunction.php)
+interface.
 
 ### Example
 
@@ -63,3 +64,118 @@ Output:
 +----+--------+------------+--------+------+
 2 rows
 ```
+
+## Frames
+
+A frame narrows a window function down to a subset of the partition, relative to the current row. This is what makes
+moving averages, running totals and trailing counts expressible.
+
+```php
+// every row in the department gets the same number
+average(ref('salary'))->over(window()->partitionBy(ref('department')));
+
+// 3-row moving average
+average(ref('salary'))->over(
+    window()
+        ->partitionBy(ref('department'))
+        ->orderBy(ref('date'))
+        ->rowsBetween(preceding(2), current_row())
+);
+```
+
+### Default frame
+
+When no frame is defined explicitly, Flow follows the SQL default:
+
+| Window              | Default frame                                                                            |
+|---------------------|------------------------------------------------------------------------------------------|
+| `orderBy()` present | all rows from the start of the partition up to and including the current row's **peers** |
+| `orderBy()` absent  | the whole partition                                                                      |
+
+Peers are rows that compare equal to the current row on **every** `ORDER BY` reference. This is why
+`partitionBy()` without `orderBy()` still means "the whole partition", and why tied rows all see the same value:
+
+```php
+// date:   2024-01-01, 2024-01-01, 2024-01-02, 2024-01-03
+// salary:        100,        200,        300,        400
+sum(ref('salary'))->over(window()->orderBy(ref('date')));
+// result:        300,        300,        600,       1000
+//                 ^^^^^^^^^^^^^^ both rows tie on date, so both see both
+```
+
+To restore the pre-0.43 behaviour of aggregating over the entire partition, ask for it explicitly:
+
+```php
+sum(ref('salary'))->over(
+    window()->orderBy(ref('date'))->rowsBetween(unbounded_preceding(), unbounded_following())
+);
+```
+
+### Frame bounds
+
+`rowsBetween()` takes a start and an end bound, built with these DSL functions:
+
+| Function                 | Meaning                               |
+|--------------------------|---------------------------------------|
+| `unbounded_preceding()`  | the first row of the partition        |
+| `preceding(int $offset)` | `$offset` rows before the current row |
+| `current_row()`          | the current row                       |
+| `following(int $offset)` | `$offset` rows after the current row  |
+| `unbounded_following()`  | the last row of the partition         |
+
+Bounds are clamped to the partition, so `preceding(2)` on the first row simply starts at the first row. A frame that
+falls entirely outside the partition is empty — `sum()` and `average()` then return `null`
+and `count()` returns `0`, matching SQL.
+
+`rowsBetween()` counts physical rows (SQL `ROWS` mode). `RANGE` with offsets, `GROUPS` and `EXCLUDE`
+are not supported yet.
+
+A frame on a window without `orderBy()` is permitted, matching PostgreSQL and Spark.
+
+### Ranking functions ignore frames
+
+`row_number()`, `rank()` and `dense_rank()` always see the whole ordered partition. A frame defined alongside them is
+accepted and ignored, exactly as in PostgreSQL. They do require an explicit
+`orderBy()`.
+
+`rank()` and `dense_rank()` accept multiple `orderBy()` columns - rows are peers when they match on all of them - and
+follow the ordering direction, so ascending and descending windows rank in opposite directions as in PostgreSQL.
+
+### Example
+
+```php
+data_frame()
+    ->read(
+        from_array([
+            ['id' => 1, 'department' => 'IT', 'date' => '2024-01-01', 'salary' => 100],
+            ['id' => 2, 'department' => 'IT', 'date' => '2024-01-02', 'salary' => 200],
+            ['id' => 3, 'department' => 'IT', 'date' => '2024-01-03', 'salary' => 300],
+            ['id' => 4, 'department' => 'IT', 'date' => '2024-01-04', 'salary' => 400],
+        ])
+    )
+    ->withEntry(
+        'moving_avg',
+        average(ref('salary'))->over(
+            window()
+                ->partitionBy(ref('department'))
+                ->orderBy(ref('date'))
+                ->rowsBetween(preceding(2), current_row())
+        )
+    )
+    ->write(to_output(false))
+    ->run();
+```
+
+Output:
+
+```console
++----+------------+------------+--------+------------+
+| id | department |       date | salary | moving_avg |
++----+------------+------------+--------+------------+
+|  1 |         IT | 2024-01-01 |    100 |        100 |
+|  2 |         IT | 2024-01-02 |    200 |        150 |
+|  3 |         IT | 2024-01-03 |    300 |        200 |
+|  4 |         IT | 2024-01-04 |    400 |        300 |
++----+------------+------------+--------+------------+
+4 rows
+```