[RFC] Streaming Queries

# Context

We want to support a streaming queries concept in ClickHouse similar to Apache Flink and RisingWave. This RFC is a continuation of https://github.com/ClickHouse/ClickHouse/issues/42990 with additional details and implementation notes.

# Terminology

## Event Time and Time Attribute

All streaming queries operate on `event time` - a `DateTime`-family column in the source table that represents when an event actually occurred. It must be stored in the table, not generated during query execution (though it can be auto-generated on insert using a default expression with `now()`).

We assume that `event time` should correlate with commit order of rows, which is the most common case. However, it is not required to be strictly increasing - data inserted into the source may arrive slightly out of commit order. These fluctuations should be handled by a properly configured `watermark` applied to the stream.

Each tuple in a stream is associated with a `time attribute` that represents its position on the event-time axis. Time attributes are recalculated by each query operator according to its logic.

## Watermarks

A `watermark` is a monotonically non-decreasing assertion that no future events with event time earlier than the watermark value will arrive. Formally, a watermark `W` guarantees that all events with event time `< W` have been observed.

The main purpose of watermarks is to track progress of data in the stream. In stateful operators like aggregation or joins, watermarks serve as markers indicating that some portion of state can be evicted because it will not be needed by any future data. Data inserted into the stream after watermark `W` with `event time < W` is discarded.

Watermarks are calculated from the `time attributes` of the stream. Each query operator emits watermarks based on its own invariants, with the constraint that it must never emit data older than the previously emitted watermark.

## Windows

A `window` defines bounds over the `time attribute` space within which a computation is performed. Computations include `window join`, `window aggregation`, etc. Windows are used to periodically evict old state once it is outdated according to the latest emitted `watermark`.

This RFC discusses only two types of windows:
1. `TUMBLE INTERVAL N` - non-overlapping windows of N time units.
2. `HOP INTERVAL N EVERY M` - overlapping windows of size N time units with a slide of M time units.

Since `TUMBLE N = HOP N N`, everywhere below we consider only `HOP`.

# Cursor Model

A cursor is a persistent read position in the source. It can be thought of as an offset in the data from which the query starts reading. The cursor advances in commit order.

A cursor is represented as a tree where each edge corresponds to a logical component of the source, for example a partition or shard. In the query pipeline, reading will likely be split by partition.

## Queues

For queue sources like Kafka, RabbitMQ, NATS, etc., the cursor is simply an offset (integer) within each partition of all topics. For example, a Kafka cursor looks like this:

```
{
'partition-1': 10,
'partition-2': 20
}
```

## MergeTree

`MergeTree` can be represented as a queue where the commit sequence of rows is sorted by the pair `(_block_number, _block_offset)`. Commit order in `MergeTree` is meaningful only within a single partition, so the cursor for a table is a combination of per-partition cursors:

```
{
'partition-1': {
'block_number': 10,
'block_offset': 20,
},
'partition-2': {
'block_number': 10,
'block_offset': 20,
}
}
```

This means that `MergeTree` can be used in streaming queries only with the `enable_block_number_column` / `enable_block_offset_column` settings enabled.

# SQL Extension

The streaming model is based on a single primitive: the `STREAM` keyword placed after a table reference in a `FROM` clause. This turns an ordinary `SELECT` into a continuous query that never terminates, processing new rows as they are committed to the source table. Watermark, windowing, and cursor positioning are specified as keyword clauses on this table expression. This approach matches the CQL standard specification.

```
table_expression
STREAM
[FROM BEGINNING | FROM CURSOR '{...}']
[WATERMARK FOR column AS expr]
[HOP INTERVAL window_size EVERY hop_interval]
```

All clauses are optional except `STREAM` itself. `WATERMARK` is required when the table expression uses a window clause (`HOP`/`TUMBLE`) or participates in an interval join - both need watermarks for window closing and state eviction.

# Query Execution

In streaming queries, query plan operators introduce an additional column to their output data stream - `time_attr`. This column is recalculated after each operator so that the next operator can correctly place each row on the event-time axis, enabling chaining of stream processors.

## System Chunks

It is convenient to include metadata in the data flow of the pipeline. Watermarks and checkpoint requests are pushed into the pipeline as empty chunks with extended chunk info.

Using this approach, we can split data in streams into segments and, based on these segments, flush intermediate data or create consistent snapshots.

## MergeTree Source

Reading from `MergeTree` is done in commit order, starting from the provided cursor position. Every part in `MergeTree` is already ordered by block numbers since merges are only possible over contiguous ranges of inserted blocks. This means reading parts in order of their left block number is enough to guarantee commit order across parts. However, data inside each part is sorted by the table's `ORDER BY` key, not by the `(_block_number, _block_offset)` pair. To restore commit order, we either re-sort the part's data during the query or use an explicitly specified projection:

```
PROJECTION _commit_order (SELECT * ORDER BY _block_number, _block_offset)
```

During execution, the query subscribes to new parts that are inserted or fetched by background operations. Polling and repopulation with new data are handled by a background process that can be explicitly triggered in `MergeTreeTransaction` during commit into the parts index. Most of the repopulation logic is already implemented in this PR: https://github.com/ClickHouse/ClickHouse/pull/63312

## Time Attributes and Watermarks

Initial calculation of the `time_attr` column is done right after the source read step in the query plan. The time attribute is an alias to the column specified in the `WATERMARK FOR` clause.

Watermarks are calculated based on the expression in the `WATERMARK` clause. The calculation can be done in the same step as `time_attr` or in the next one.

An important detail for watermark calculation: it must filter rows within each chunk based on the current watermark and the processed chunk prefix. If a row in the middle of a chunk advances the watermark, the new value must be applied immediately to the remaining rows in that chunk.

## Time Attributes Recalculation

The `time_attr` column is recalculated after each query plan operator and included in its output header:

| Operator | `time_attr` value |
|-----------------|--------------------|
| Source | The event time column declared in `WATERMARK FOR` |
| Filter/Expression/Limit | Passthrough from input |
| Window aggregate | `window_end` |
| Window join | `window_end` |
| Global aggregate | Max `time_attr` in the chunk |
| Interval join | `max(left.time_attr, right.time_attr)` of the matching pair |
| Temporal join | Stream-side event time |

## Watermarks Recalculation

Each operator type recalculates and emits watermarks differently. The key invariant is: an operator must never emit
a watermark `W_out` unless it guarantees that all output rows with `time_attr`
`< W_out` have already been emitted.

| Operator | Output watermark | When emitted |
|----------|------------------|--------------|
| Source | `WATERMARK_EXPR(max(event_time))` over all rows seen | After each chunk |
| Filter/Expression/Limit | Same as input | After each chunk |
| Global Aggregate | Same as input | After emitting prefix aggregation results |
| Window Aggregate | Advances to `window_end` of the latest closed window | After emitting window results |
| Interval Join | `min(W_left, W_right) - max_interval` | After processing pending matches |
| Window Join | Advances to `window_end` of the latest closed window | After emitting window results |
| Temporal Join | Same as input | After each chunk |

## Complex Operators

Many window operators maintain dynamically changing state while processing the stream. For overlapping windows, the stream is split into non-overlapping panes, and a combination of panes composes each window.

Panes are calculated from the `time_attr` column for each row in a chunk using the formula:

```
pane_size = GCD(hop_interval, window_size)
pane_id = floor(time_attr / pane_size) * pane_size
```

Window operators add two additional columns to the output stream: `window_start` and `window_end`. These columns are semantically included as implicit grouping keys and in join expressions, but the implementation may handle this internally by partitioning data by windows.

## Streaming Aggregation

### Window Aggregation

To support window aggregation, the operator internally decomposes data into non-overlapping panes based on the `time_attr` column and computes partial aggregates for each pane.

When the input watermark exceeds the end of a window, the panes composing that window will not be updated further, and the window result can be finalized by merging the partial aggregates. After that, panes not covered by any open window can be discarded from the operator's state.

The operator internally maintains a watermark equal to the `window_end` of the first non-closed window. The output watermark is emitted only after a window closes and the aggregation results are fully flushed.

### Global Aggregation

Global aggregation is not recommended for high-cardinality keys because the operator's state is potentially unbounded - it must maintain aggregated values for all distinct keys seen so far and emit cumulative results after each chunk. However, it is useful when the aggregated state is small and bounded, for example `count()` or `median(column)`.

The implementation should be similar to what `AggregatingTransform` does today, but it must emit its output after each chunk. The output `time_attr` of each row should be the maximum `time_attr` of all processed rows that produced this prefix aggregate.

## Streaming Join

### Temporal Join

The most common pattern: join a stream against the current state of a static table. It does not require any stream extensions like watermark generation or windows because every row from the stream is joined against static data.

```sql
SELECT
e.event_time, e.event_type,
u.name, u.email
FROM events STREAM AS e
JOIN users AS u ON e.user_id = u.user_id
```

This should already work if we build a hash table for the static table side rather than the stream side.

### Interval Join

Example:

```sql
SELECT
c.click_time, i.impression_time,
c.user_id, i.ad_id
FROM clicks
STREAM
WATERMARK FOR click_time AS click_time
AS c
JOIN impressions
STREAM
WATERMARK FOR impression_time AS impression_time
AS i
ON c.user_id = i.user_id
AND c.click_time BETWEEN i.impression_time
AND i.impression_time + INTERVAL 1 HOUR
```

Basically, this type of join adds the ability to connect two streams with flexible intervals where each row in the left stream can be joined with the right stream within an interval calculated specifically for that row.

To support this, we need to maintain hash tables for each of the join sides where keys will be equi-join expression operands and values must be a sorted data structure with pairs ``. Sorting by `time_attr` helps to reduce data after watermark advancement by simply removing the prefix of values.

To support removal of stale data we can rewrite expression `l.a between r.b - x and r.b + y` to `l.a < r.b + y` and if we include watermark into this inequality it will be `W(l) < l.a < r.b + y -> r.b > W(l) - y` that basically means that for all new elements from stream l we will need only elements from stream r which have time attribute higher than `W(l) - y`. The same logic can be applied to the right stream watermark.

### Window Join

Example:

```sql
SELECT
pv.user_id, pv.page, p.product, p.amount
FROM page_views
STREAM WATERMARK FOR event_time AS event_time - INTERVAL 10 SECOND
TUMBLE INTERVAL 1 HOUR
AS pv
JOIN purchases
STREAM WATERMARK FOR event_time AS event_time - INTERVAL 5 SECOND
TUMBLE INTERVAL 1 HOUR
AS p
ON pv.user_id = p.user_id
```

Window join is a regular relation join on a segment of a stream (window). The requirement for a window join is to have equal windowing semantics on both sides of the join.

Implementation is very similar to the pane aggregation strategy discussed above. For each side of a join, we will maintain disjoint hash tables for each of the intersecting panes, and each row of each stream will land into one of the panes.

When a window is closing because of watermark advancement, we will need to emit the join result before emitting the watermark to the output stream. The join result is a cross-pane join where each pane must be joined with every other pane using the join semantics provided in the query.

# Query Snapshots

Streaming queries form a DAG of operators. To take a globally consistent snapshot of a query, we use the Asynchronous Barrier Snapshotting (ABS) algorithm.

Each source operator periodically emits a checkpoint barrier into its output stream. When a downstream operator receives a barrier on one input, it blocks that input and waits for barriers from all other inputs. After barrier alignment, the operator dumps its state to persistent storage (local disk, S3, etc.), pushes the barrier downstream, and unblocks all inputs.

Barriers are emitted from sources based on a timeout calculated by each source independently. Sources may be slightly out of sync, but this is acceptable since each downstream operator calibrates itself upon receiving the first barrier.

## What State is Captured

| Operator | Snapshot contents |
|----------|-------------------|
| Source | Cursor position, current watermark |
| Filter/Expression/Limit | Nothing (stateless) |
| Global Aggregate | Aggregate state, current watermark |
| Window Aggregate | Partial aggregate states per pane, current watermark |
| Temporal Join | Nothing (hash table rebuilt from dimension table on restart) |
| Interval Join | Hash tables for both sides, current watermark |
| Window Join | Buffered data per pane for both sides, current watermark |

## Snapshot Commits

Snapshot metadata is persisted in Keeper after the barrier reaches the end of the pipeline.

To provide more flexibility during upgrades, only stateful processors' data will be stored in snapshots. We must not serialize data <-> processor mapping because it would be challenging to change the query plan/pipeline after that. Because of this, the proposed way is to store data for query segments like source/aggregation/join etc. and map these segments into the query plan dynamically during recovery.

```
/clickhouse/query_snapshots/{query_id}/{snapshot_id}
{
"segments": {
"source": {
"table-1": "local path on snapshots disk"
...
}
"aggregation": "local path on snapshots disk",
"join": "local path on snapshots disk"
...
}
}
```

## Recovery From Snapshots

After a failure, the query can be resumed from any completed snapshot. To do this, the query pipeline operators are restored from the snapshot state and the sources are rewound to the snapshotted cursors.

**Recovery correctness** depends on:

1. **Source replayability:** `MergeTree` tables are inherently replayable - data is persistent and can be re-read from any `(_block_number, _block_offset)` cursor. Kafka sources are replayable via offset seeking.

2. **Snapshot consistency:** The ABS algorithm ensures that the snapshot captures a consistent cut across all operators. Source cursors, watermarks, and operator states are all recorded at the same logical point in the data stream. This means that each row is included in all related operators' states exactly once.