# Manage Flows

> Describes how to manage flows in GreptimeDB, including creating, updating, and deleting flows. It explains the syntax for creating flows, the importance of sink tables, and how to use the EXPIRE AFTER clause. Examples of SQL queries for managing flows are provided.

# Manage Flows

Each `flow` is a continuous aggregation query in GreptimeDB.
It continuously updates the aggregated data based on the incoming data.
This document describes how to create, and delete a flow.

:::note
`EVAL INTERVAL` schedules batching evaluations but does not select batching mode. TQL workloads require it. See [Create a flow](#create-a-flow) for execution routing and instant-TTL restrictions.
:::

## Create a Source Table

Before creating a flow, you need to create a source table to store the raw data. Like this:

```sql
CREATE TABLE temp_sensor_data (
  sensor_id INT,
  loc STRING,
  temperature DOUBLE,
  ts TIMESTAMP TIME INDEX,
  PRIMARY KEY(sensor_id, loc)
);
```
For new workloads, avoid `WITH ('ttl' = 'instant')` on Flow source tables. This is a legacy pattern and is not recommended for new aggregation or TQL workloads. Keep source data with an appropriate retention policy instead.

## Create a Sink Table

A flow stores its aggregated data in a sink table. When the sink table does not exist, `CREATE FLOW`
automatically creates it when the query result is sufficient to infer its schema. Pre-create the sink when you
need control over its schema or layout, or when inference is complex. An existing sink table is validated against
the flow's query result. The source and sink tables must be different tables.

The sink table has to be compatible with the flow's query result:

- **Column order and type**: For a pre-created SQL sink, match the query output columns in order and type.
- **Time index**: Specify the `TIME INDEX` for the sink table, typically using the time window column generated by the time window function.
- **Update time**: For an auto-created batching SQL sink, Flow adds an `update_at` column for the update time. TQL sinks follow the query output and do not automatically add `update_at`. A pre-created SQL sink can either match the query output width or include one extra trailing timestamp column for update time.
- **Tags**: Use `PRIMARY KEY` to specify Tags, which together with the time index serve as a unique identifier for row data and optimize query performance.

For example:

```sql
/* Create sink table */
CREATE TABLE temp_alerts (
  sensor_id INT,
  loc STRING,
  max_temp DOUBLE,
  time_window TIMESTAMP TIME INDEX,
  update_at TIMESTAMP,
  PRIMARY KEY(sensor_id, loc)
);

CREATE FLOW temp_monitoring
SINK TO temp_alerts
AS
SELECT
  sensor_id,
  loc,
  max(temperature) AS max_temp,
  date_bin('10 seconds'::INTERVAL, ts) AS time_window
FROM temp_sensor_data
GROUP BY
  sensor_id,
  loc,
  time_window
HAVING max_temp > 100;
```

The sink table has the columns `sensor_id`, `loc`, `max_temp`, `time_window`, and `update_at`.

- The first four columns correspond to the query result columns of flow: `sensor_id`, `loc`, `max(temperature)` and `date_bin('10 seconds'::INTERVAL, ts)` respectively.
- The `time_window` column is specified as the `TIME INDEX` for the sink table.
- The `update_at` column is the last one in the schema to store the update time of the data.
- The `PRIMARY KEY` at the end of the schema definition specifies `sensor_id` and `loc` as the tag columns.
  This means the flow will insert or update data based on the tags `sensor_id` and `loc` along with the time index `time_window`.

## Create a flow

The grammar to create a flow is:

```sql
CREATE [ OR REPLACE ] FLOW [ IF NOT EXISTS ] <flow-name>
SINK TO <sink-table-name>
[ EXPIRE AFTER <expr> ]
[ EVAL INTERVAL <interval> ]
[ COMMENT '<string>' ]
[ WITH (<flow-option> = <value> [, ...]) ]
AS
<SQL>;
```

The clauses must appear in the order shown: `EXPIRE AFTER` comes before `EVAL INTERVAL`.
`EVAL INTERVAL` schedules batching evaluations. TQL flows require it. SQL plans containing `Aggregate` or `Distinct` use
batching, unless an instant-TTL source selects legacy streaming; ordinary projections and non-aggregate joins also use
legacy streaming. Streaming ignores `EVAL INTERVAL`. For batching time-window SQL, an evaluation can be incremental
rather than a full-query evaluation. Batching SQL without a usable time window requires `EVAL INTERVAL` and executes the full query; time-window aggregates can run without it.

When `OR REPLACE` is specified, any existing flow with the same name will be updated to the new version. It's important to note that this only affects the flow task itself; the source and sink tables will remain unchanged.

Conversely, when `IF NOT EXISTS` is specified, the command will have no effect if the flow already exists, rather than reporting an error. Additionally, please note that `OR REPLACE` cannot be used in conjunction with `IF NOT EXISTS`.

- `flow-name` is a unique identifier at the catalog level.
- `sink-table-name` is the table name where the materialized aggregated data is stored.
  It can be an existing table or a new one; see [Create a Sink Table](#create-a-sink-table) for creation and validation behavior.
- `EXPIRE AFTER` is an optional interval to expire data from the Flow engine. For more details, please refer to the [`EXPIRE AFTER`](#expire-after) section.
- `EVAL INTERVAL` is an optional interval for batching evaluations; streaming ignores it.
- `COMMENT` is the description of the flow.
- `WITH` specifies flow options.
  The user-facing options documented below are `defer_on_missing_source` and the experimental `experimental_enable_incremental_read`.
- `SQL` part defines the continuous aggregation query.
  It defines the source tables that provide data for the flow.
  Each flow can have multiple source tables.
  Please refer to [Write a SQL query](#write-a-sql-query) for details.

A simple example to create a flow:

```sql
CREATE FLOW IF NOT EXISTS my_flow
SINK TO my_sink_table
EXPIRE AFTER '1 hour'::INTERVAL
COMMENT 'My first flow in GreptimeDB'
AS
SELECT
    max(temperature) as max_temp,
    date_bin('10 seconds'::INTERVAL, ts) as time_window
FROM temp_sensor_data
GROUP BY time_window;
```

The created flow groups `max(temperature)` into 10-second windows and stores the result in `my_sink_table`. Data within the last hour is used in the flow.

### EXPIRE AFTER

The `EXPIRE AFTER` clause specifies the interval after which data will expire from the flow engine.

For a Flow with a usable time-window expression, data in the source table older than the specified interval is excluded from calculations, and older sink rows are not updated. This limits the state and recomputation range for time-window flows, including stateful queries such as those involving `GROUP BY`.

Batching plans without a usable time-window expression require `EVAL INTERVAL` and execute unfiltered snapshots unless their query has a time predicate; `EXPIRE AFTER` does not add a time filter. It does not delete data from either table. If you want to delete data from the source or sink table, please [set the `TTL` option](/user-guide/manage-data/overview.md#manage-data-retention-with-ttl-policies) when creating tables.

For example, if the flow engine processes the aggregation at 10:00:00 and the `'1 hour'::INTERVAL` is set,
any input data that arrive now with a time index older than 1 hour (before 09:00:00) will expire and be ignored.
Only data timestamped from 09:00:00 onwards will be used in the aggregation and to update the sink table.

### Defer creation when a source is missing

By default, creating a Flow fails if one of its source tables does not exist. Set
`defer_on_missing_source` to `true` to persist a pending Flow instead of failing. The Flow is not scheduled while its
sources remain unresolved, and it is not activated automatically when those tables are created. `CREATE OR REPLACE`
cannot activate a pending Flow.

```sql
CREATE FLOW pending_flow
SINK TO pending_sink
WITH (defer_on_missing_source = 'true')
AS
SELECT * FROM source_created_later;
```

After all source tables have been created, drop and recreate the Flow to activate it.

```sql
DROP FLOW pending_flow;
CREATE FLOW pending_flow
SINK TO pending_sink
AS
SELECT * FROM source_created_later;
```

### Experimental incremental source reads

:::warning Experimental feature
The `experimental_enable_incremental_read` option is experimental.
Its behavior and limitations may change in future releases.
:::

For batching SQL flows whose source tables are append-only, you can enable incremental source reads:

```sql
CREATE TABLE temp_sensor_data (
  sensor_id INT,
  loc STRING,
  temperature DOUBLE,
  ts TIMESTAMP TIME INDEX,
  PRIMARY KEY(sensor_id, loc)
) WITH ('append_mode' = 'true');

CREATE FLOW temp_monitoring
SINK TO temp_alerts
WITH (experimental_enable_incremental_read = 'true')
AS
SELECT
  sensor_id,
  loc,
  max(temperature) AS max_temp,
  date_bin('10 seconds'::INTERVAL, ts) AS time_window
FROM temp_sensor_data
GROUP BY
  sensor_id,
  loc,
  time_window;
```

When enabled, Flow attempts to read only newly appended source rows after the initial full snapshot.
This is an execution optimization and does not change the query result. The optimization is not a persistence
contract: the first run, and a run after a restart or when incremental reading is not safe, may use a full snapshot.

The current limitations are:

- All source tables must be append-only tables created with `append_mode = 'true'`.
  Flow creation fails if any source table is not append-only.
- The optimization applies only to eligible batching SQL flows. TQL flows and plans that do not support incremental
  reads use the normal full-snapshot behavior.

### Write a SQL query

The SQL after `AS` is planned as a standard SQL query. A typical batching time-window aggregate has this shape:

```sql
SELECT AGGR_FUNCTION(column1, column2,..) [, TIME_WINDOW_FUNCTION() as time_window]
FROM <source_table>
GROUP BY {time_window | column1, column2,.. };
```

The query engine and Flow plan determine which SQL expressions and clauses are supported. For batching SQL flows
with `Aggregate` or `Distinct` plans, `EVAL INTERVAL` schedules evaluations; TQL flows require it. Planner-valid
joins, subqueries, and SQL CTEs are supported for batching SQL flows with `EVAL INTERVAL`. The query planner must
still produce a valid plan; unsupported queries fail when the Flow is created. For batching time-window aggregates,
`GROUP BY` commonly includes the time-window expression. See [Expressions](expressions.md)
for functions commonly used in Flow queries, and [Define time window](#define-time-window) for fixed windows.

Refer to [Continuous Aggregation](continuous-aggregation.md) for more examples of how to use continuous aggregation in real-time analytics, monitoring, and dashboards.

### Define time window

A time window is a crucial attribute of your continuous aggregation query.
It determines how data is aggregated within the flow.
These time windows are left-closed and right-open intervals.

A time window represents a specific range of time.
Data from the source table is mapped to the corresponding window based on the time index column.
The time window also defines the scope for each calculation of an aggregation expression,
resulting in one row per time window in the result table.

You can use `date_bin()` after the `SELECT` keyword to define fixed time windows.
For example:

```sql
SELECT
    max(temperature) as max_temp,
    date_bin('10 seconds'::INTERVAL, ts) as time_window
FROM temp_sensor_data
GROUP BY time_window;
```

In this example, the `date_bin('10 seconds'::INTERVAL, ts)` function creates 10-second time windows starting from UTC 00:00:00.
The `max(temperature)` function calculates the maximum temperature value within each time window.

For more details on the behavior of the function,
please refer to [`date_bin`](/reference/sql/functions/df-functions.md#date_bin).

:::tip NOTE
The time-window expression helps Flow determine how to update results incrementally. The appropriate window size
depends on the workload and query semantics.
:::

## Inspect flows

Use the following commands and system table to inspect Flow definitions and runtime information:

| Command | Purpose |
| --- | --- |
| `SHOW FLOWS;` | List flows. |
| `SHOW CREATE FLOW my_flow;` | Return a Flow definition. |
| `SELECT * FROM information_schema.flows;` | View Flow definitions and the nullable runtime fields `state_size` and `last_execution_time`. |

## Flush a flow

The flow engine automatically processes aggregation operations within a short period(i.e. few seconds) when new data arrives in the source table.
However, you can manually trigger the flow engine to process the aggregation operation immediately using the `ADMIN FLUSH_FLOW` command.

```sql
ADMIN FLUSH_FLOW('<flow-name>')
```

## Delete a flow

To delete a flow, use the following `DROP FLOW` clause:

```sql
DROP FLOW [IF EXISTS] <name>
```

For example:

```sql
DROP FLOW IF EXISTS my_flow;
```