How does ksqlDB materialize state for aggregations and pull queries, and what role do RocksDB state stores and changelog topics play in fault tolerance?

## What 'materialized' means here A **materialized view** is a precomputed, continuously-updated result you can read instantly — as opposed to recomputing it on every request. In ksqlDB, when you create an aggregating table (`CREATE TABLE … AS SELECT … GROUP BY …`), the persistent query keeps the **current aggregate per key** so that **pull queries** can do fast lookups. ### Where the state lives: RocksDB ksqlDB runs on **Kafka Streams**, which stores stateful operator data in **state stores**. By default these are backed by **RocksDB** — an embedded, on-disk **key-value** store (a log-structured merge-tree). It is **local** to each ksqlDB node, written to local disk under the state directory. RocksDB keeps recent writes in in-memory **memtables**, flushes them to immutable **SST files**, and serves reads through an off-heap **block cache**. Because it spills to disk, state can far exceed RAM. RocksDB memory and behavior are tunable via a `RocksDBConfigSetter` (block cache size, write buffers, etc.) — important because many partitions each get their own RocksDB instance and memory can add up. ### Partitioning of state State is sharded by the **partitions** of the input/repartition topic. In a multi-node ksqlDB cluster, partitions (and thus state) are **distributed** across nodes via the consumer-group assignment. Each key lives on exactly one node. A **pull query** for a key is **routed** (using interactive-query metadata) to the node that owns that key's partition — locally read or forwarded. ### Fault tolerance: the changelog topic Local disk is not durable on its own — a node can die and take its RocksDB with it. So every state store is continuously mirrored to a **changelog topic** in Kafka (named internally, e.g. `…-changelog`). This topic is **log-compacted**, so it retains the **latest value per key** (plus tombstones for deletes). On failure, rebalance, or restart: 1. A node is assigned the orphaned partition. 2. It creates a fresh RocksDB store and **replays the changelog** topic for that partition to rebuild exact state. 3. Processing resumes from the committed offset. This is how exactly-the-right state survives crashes without recomputing from the source. ### Speeding up recovery: standby replicas Replaying a large changelog is slow. **Standby replicas** (`ksql.streams.num.standby.replicas` > 0) keep **warm, continuously-updated copies** of state stores on other nodes. On failover, a standby is promoted with little or no replay, cutting recovery time — at the cost of extra disk/network. ksqlDB can even serve **stale reads** from standbys during recovery if configured. ### How pull queries use all this A pull query (`SELECT … WHERE key = …`) hits the **materialized** RocksDB store directly — no recomputation. For windowed tables, the key includes window bounds, and you can range over `WINDOWSTART`/`WINDOWEND`. ### Edge cases / operational notes - **Restore lag**: a pull query during restore may fail or (with config) return stale data from a standby. - **Disk pressure**: large keyspaces or many overlapping windows balloon RocksDB disk; size nodes accordingly. - **Repartition topics**: GROUP BY on a non-key column forces a repartition before the store is built. - **Compaction**: the changelog must be compacted (not deleted by retention) or state could be lost on restore; ksqlDB configures this automatically. - **Memory blowups**: each partition's RocksDB has its own memtables/cache; without a bounded config setter, many partitions can exhaust off-heap memory.