Stateful queries

Ingest Engine supports two semantics for streaming SQL, which we call Dataflow semantics and updating semantics. There’s a core problem of trying to execute a SQL query on an unbounded, streaming source: how do we know when to compute aggregates and joins, given that we will always see more data in the future? In Dataflow semantics, which are introduced with the use of time-oriented windows, like HOP and TUMBLE, we compute aggregates for a window once the watermark passes. This is a powerful model, as allows the streaming system to signal completeness to its consumers, so they don’t need to reason about when they are able to trust the results. However, the requirement that all computations are windowed can be limiting. Updating semantics, on the other hand, allow for more flexibility in the types of queries that can be expressed, including most of normal batch SQL. It works by treating the input stream as a table that is constantly being updated, and the output as updates on a materialized view of the query. When writing to a sink, the output is a stream of updates (in Debezium format), representing additions, updates, and deletes to the materialized view.

Reading from updating sources

Source connectors such as Kafka can specify the format as 'debezium_json' to read Debezium formatted messages. Updating sources need at least one primary key, which tells Ingest Engine which rows are logically the same. The primary key is specified in the DDL, like this:

CREATE TABLE debezium_source (
    id INT PRIMARY KEY,
    customer_name TEXT,
    price FLOAT,
    order_date TIMESTAMP,
    status TEXT
) WITH (
    connector = 'kafka',
    format = 'debezium_json',
    type = 'source',
    bootstrap_servers = 'localhost:9092',
    topic = 'orders_cdc'
);

Writing to updating sinks

Updating queries can be written to sinks with the debezium_json format. This output can then be consumed by the Debezium sink connector to write to a RDBMS like MySQL or Postgres. For a complete example of this, see the PostgreSQL CDC to Iceberg example.

TTLs

The base semantics of updating tables require that, for any event that comes in, we must be able to update the state the output. However doing this with complete correctness would require storing data for all time. This is generally intractable in a streaming system without blowing up our state. Therefore, updating states have a time-to-live (TTL) associated with them. This TTL is the maximum amount of time we will store a key after we last saw an event for it. (A key might be something like a user id or a transaction id; generally this is the thing being grouped by in an aggregation or joined on.) By default, the TTL is 1 day, but it can be configured with the SET updating_ttl command, which takes a SQL interval. For example, to set the TTL to 1 hour:

SET updating_ttl = '1 hour';

Currently all queries in a pipeline share the same TTL. In the future, we may allow different TTLs for different queries.

TTL eviction example

If TTL is set to 1 hour and a customer hasn’t placed an order in over 1 hour, their aggregation state is dropped. The next order from that customer starts a fresh aggregation. Consider a SUM(amount) grouped by customer_id with updating_ttl = '1 hour':

Customer c1 orders at 10:00 (amount=50), 10:15 (amount=60), 10:30 (amount=40) - running total = 150.00
No orders from c1 for 1.5 hours (exceeds the 1 hour TTL)
State for c1 is evicted
Customer c1 orders at 12:00 (amount=40.00) - treated as a new group, total = 40.00 (not 190.00)

This trades off correctness for bounded state. Choose a TTL that balances accuracy with your memory constraints.

Updating aggregates

Aggregating data without a window will result in an updating output. This will emit an insert the first time data is processed for a group and subsequent data will retract the prior value and then insert the new value. Aggregates are buffered in the operator, occasionally flushing. By default flushing happens every 1 second, but can be overridden with the pipeline.update-aggregate-flush-interval config. For instance, the following query

CREATE TABLE impulse WITH (
    'connector' = 'impulse',
    event_rate = '100'
);

SELECT count(*) as rows
FROM impulse
HAVING rows < 500;

will produce output data like the following:

before	after	op
null	`{ "rows": 100}`	`"c"`
`{ "rows": 100 }`	`{ "rows": 200}`	`"u"`
`{ "rows": 200 }`	`{ "rows": 300}`	`"u"`
`{ "rows": 300 }`	`{ "rows": 400}`	`"u"`
`{ "rows": 400 }`	null	`"d"`

For examples of manipulating updating data, see the Debezium documentation.

Worked example: aggregating orders by customer

Given an input stream orders:

order_id	customer_id	amount	event_time
1	c1	50.00	2024-01-01 10:00:15
2	c2	30.00	2024-01-01 10:00:30
3	c1	25.00	2024-01-01 10:00:45
4	c1	75.00	2024-01-01 10:01:10
5	c2	40.00	2024-01-01 10:01:40

And this query:

SELECT customer_id, SUM(amount) as total_spent, COUNT(*) as order_count
FROM orders
GROUP BY customer_id

The updating output (in Debezium format) is:

op	customer_id	total_spent	order_count
c	c1	50.00	1
c	c2	30.00	1
u	c1	75.00	2
u	c1	150.00	3
u	c2	70.00	2

"c" = create - emitted the first time a group key is seen (first order for that customer).
"u" = update - emitted when a group’s aggregate changes. For each update, the old value is retracted and the new value is inserted.

Updating joins

See the join documentation for more details on how to use Joins in updating queries.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

Reading from updating sources

Writing to updating sinks

TTLs

TTL eviction example

Updating aggregates

Worked example: aggregating orders by customer

Updating joins

​Reading from updating sources

​Writing to updating sinks

​TTLs

​TTL eviction example

​Updating aggregates

​Worked example: aggregating orders by customer

​Updating joins

Reading from updating sources

Writing to updating sinks

TTLs

TTL eviction example

Updating aggregates

Worked example: aggregating orders by customer

Updating joins