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

Flow uses batching mode for aggregation and TQL workloads. Simple non-aggregation Flow queries currently use the deprecated streaming mode and are not recommended for new workloads.

Create a Source Table

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

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:

/* 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:

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 repeated evaluation of the full query. Scheduled SQL flows can use joins, subqueries, and SQL CTEs when the SQL query engine can plan the query. TQL flows require EVAL INTERVAL. Batching time-window aggregate flows can run without EVAL INTERVAL.

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 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 section.
• EVAL INTERVAL is an optional interval for scheduled full-query evaluation. TQL flows require 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 for details.

A simple example to create a flow:

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.

Scheduled full-query SQL and TQL flows execute unfiltered snapshots unless the query contains its own 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 when creating tables.

Setting a reasonable time interval for EXPIRE AFTER is helpful to limit how far back the batching engine needs to recompute results and to avoid excessive resource usage. It serves a similar purpose to bounding lateness in stream processing systems, but new Flow workloads should use batching mode.

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.

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

Experimental incremental source reads

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:

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:

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 a scheduled full-query SQL Flow, planner-valid joins, subqueries, and SQL CTEs are supported; an unsupported plan fails when the Flow is created. For batching time-window aggregates, GROUP BY commonly includes the time-window expression. See Expressions for functions commonly used in Flow queries, and Define time window for fixed windows.

Refer to Continuous Aggregation 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:

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.

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 statements and system tables to inspect Flow definitions and runtime information:

SHOW FLOWS;
SHOW CREATE FLOW my_flow;
SHOW FLOW STATUS LIKE 'my%';
SELECT * FROM information_schema.flows;
SELECT * FROM information_schema.flow_statistics;

SHOW FLOWS lists flows, SHOW CREATE FLOW returns a Flow definition, and SHOW FLOW STATUS returns runtime statistics. The information_schema tables provide definition and statistics details. Runtime fields can initially be NULL, and values can lag behind the latest state in distributed deployments.

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.

ADMIN FLUSH_FLOW('<flow-name>')

Delete a flow

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

DROP FLOW [IF EXISTS] <name>

For example:

DROP FLOW IF EXISTS my_flow;