<!-- .. include:: ../../includes/batch-streaming-labels.rst
:start-after: batch-streaming-labels_start
:end-before: batch-streaming-labels_end -->

<a id="flink-sql-over-aggregation"></a>

# OVER Aggregation Queries in Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® enables computing an aggregated value for every row over a range
of ordered rows.

## Syntax

<!-- .. include:: ../../includes/batch-streaming-labels.rst
:start-after: execution-mode-batch-and-streaming_start
:end-before: execution-mode-batch-and-streaming_end -->
```sql
SELECT
  agg_func(agg_col) OVER (
    [PARTITION BY column1[, column2, ...]]
    ORDER BY time_column
    range_definition),
  ...
FROM ...
```

## Description

Compute an aggregated value for every row over a range of ordered rows.

`OVER` aggregates compute an aggregated value for every input row over
a range of ordered rows. In contrast to a `GROUP BY` aggregate,
an `OVER` aggregate doesn’t reduce the number of result rows to a single
row for every group. Instead, an `OVER` aggregate produces an aggregated
value for every input row.

You can define multiple `OVER` window aggregates in a `SELECT`
clause. However, for streaming queries, the `OVER` windows for all
aggregates must be identical due to a current limitation.

**Table types.** An `OVER` aggregation requires an
[append-only table](../../concepts/dynamic-tables.md#flink-sql-dynamic-tables-append-only-table) as
input, and it produces an append-only table.

## ORDER BY

`OVER` windows are defined on an ordered sequence of rows. Because
tables do not have an inherent order, the `ORDER BY` clause is
mandatory. For streaming queries, Flink currently only supports `OVER`
windows that are defined with an ascending [time attribute](../../concepts/timely-stream-processing.md#flink-sql-time-attributes)
order. Additional orderings are not supported.

## PARTITION BY

`OVER` windows can be defined on a partitioned table. In the presence
of a `PARTITION BY` clause, the aggregate is computed for each input
row only over the rows of its partition.

## Range definitions

The range definition specifies how many rows are included in the
aggregate. The range is defined with a `BETWEEN` clause that defines a
lower and an upper boundary. All rows between these boundaries are
included in the aggregate. Flink only supports `CURRENT ROW` as the
upper boundary.

There are two options to define the range, `ROWS` intervals and
`RANGE` intervals.

### RANGE intervals

A `RANGE` interval is defined on the values of the `ORDER BY` column,
which in Flink is always a time attribute. The following `RANGE`
interval defines that all rows with a time attribute of at most 30
minutes less than the current row are included in the aggregate.

```sql
RANGE BETWEEN INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW
```

### ROW intervals

A `ROWS` interval is a count-based interval. It defines exactly how
many rows are included in the aggregate. The following `ROWS` interval
defines that the 10 rows preceding the current row and the current row
(so 11 rows in total) are included in the aggregate.

```sql
ROWS BETWEEN 10 PRECEDING AND CURRENT ROW
```

Use the `WINDOW` clause to define an `OVER` window outside of the
`SELECT` clause. The `WINDOW` clause can make queries more readable
and also allows you to reuse the window definition for multiple aggregates.

```sql
SELECT order_id, order_time, amount,
  SUM(amount) OVER w AS sum_amount,
  AVG(amount) OVER w AS avg_amount
FROM orders
WINDOW w AS (
  PARTITION BY product
  ORDER BY order_time
  RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW)
```

## Example

The following query computes for every order the sum of amounts of all
orders for the same product that were received within one hour before
the current order.

```sql
SELECT order_id, order_time, amount,
  SUM(amount) OVER (
    PARTITION BY product
    ORDER BY order_time
    RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
  ) AS one_hour_prod_amount_sum
FROM orders
```

## Related content

- Confluent Developer: [OVER aggregations](https://developer.confluent.io/courses/flink-sql/over-windows/)
- [Time Attributes](../../concepts/timely-stream-processing.md#flink-sql-time-attributes)
- [Flink SQL Queries](overview.md#flink-sql-queries)
- [Flink SQL Functions](../functions/overview.md#flink-sql-functions-overview)
- [Statements](../statements/overview.md#flink-sql-statements-overview)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).