> For the complete documentation index, see [llms.txt](https://docs.e6data.com/ingestion-engine/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.e6data.com/ingestion-engine/sql/streaming/stateful.md).

# Stateful queries

e6 Ingestion 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](/ingestion-engine/sql/streaming/windows.md), 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](https://debezium.io/) 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](https://debezium.io/) formatted messages.

Updating sources need at least one *primary key*, which tells e6 Ingestion Engine which rows are logically the same. The primary key is specified in the DDL, like this:

```sql
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](https://debezium.io/documentation/reference/stable/connectors/jdbc.html) to write to a RDBMS like MySQL or Postgres.

For a complete example of this, see this tutorial.

## 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:

```sql
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'`:

1. Customer `c1` orders at 10:00 (amount=50), 10:15 (amount=60), 10:30 (amount=40) - running total = **150.00**
2. No orders from `c1` for 1.5 hours (exceeds the 1 hour TTL)
3. State for `c1` is evicted
4. 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

```sql
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:

```sql
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](/ingestion-engine/sql/streaming/joins.md#updating-joins) for more details on how to use Joins in updating queries.


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.e6data.com/ingestion-engine/sql/streaming/stateful.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.