From eb238ad79393178a80cb5d7ef0de0e50724b8af5 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 23 Sep 2026 14:20:15 -0400 Subject: [PATCH 1/5] wip Persist 101 - data layout in blob storage, writers, TODOs --- doc/developer/101-persist.md | 114 +++++++++++++++++++++++++++++++++++ 1 file changed, 114 insertions(+) create mode 100644 doc/developer/101-persist.md diff --git a/doc/developer/101-persist.md b/doc/developer/101-persist.md new file mode 100644 index 0000000000000..58e12021f03b0 --- /dev/null +++ b/doc/developer/101-persist.md @@ -0,0 +1,114 @@ +# Persist 101 +Persist is a data store backed by immutable blobs. +## Semantics +**First, _what_ is Persist?** + +It's a durable store of time-varying collections. + +Data is organized into collections called "shards". +Individual records within a collection are _(key, value, multiplicity)_. + +For the time window between a shard's **"since"** (lower bound) and its **"frontier"** (upper bound), we know the history of the collection and can query its contents. +- Before the lower bound, we no longer remember the history of the collection. +- At or beyond the frontier, the collection's state is still being decided. + +We can only write (append) to a collection _at or beyond_ the frontier. We cannot alter history before that point. + +## Data Layout in Blob Storage +**What does it actually look like?** +### Layout: Shard > Batch > Run > Part +Within a shard, records are organized into: +- Batches. Together, they cover the entire `[Since, Frontier)` time range without overlapping. Each timestamp within the range belongs to a single batch. +- Runs. Each batch is made up of one or more runs. Runs are an artifact of how batches are written, not a semantic separation of records. +- Parts. Each run is made up of one or more parts. Each _part_ of a _run_ is responsible for a range of keys. + Within a part, records are typically sorted by key, and within a run, parts are sorted by their key ranges (which do not overlap). + - _Parts are internally sorted by key only if they've been through compaction. For large shards, most parts have been through compaction._ +#### Data Layout in a Part File +Part files are Parquet, where the columns are: +- `t` - timestamp +- `d` - diff +- `k_s` - Arrow-encoded Parquet "key", which is either: + - `ok` - struct with one field per relation column + - `err` - binary-encoded `DataflowError` +- `v_s` - unused +Our Parquet writer is simple: +- only one row group +- no statistics (Predicate push-down chooses which part files to read based on stats we write into shard state--outside the part files themselves.) + +_Note: There is also a legacy format with columns `k, v, t, d` and a migration format with `k, v, t, d, k_s, v_s`._ +#### Interlude on Future Work: Predicate Push-Down _within_ a Part +Today, we always read and decode an entire part file. +If we stop doing that, maybe we can support point (key + time) lookups and improve predicate push-down. + +Claude's diagram of Parquet chunks and pages within our one row group: +``` +File +└── Row group (a horizontal slice of rows, all columns) + ├── Column chunk "t" (all of t's values for those rows) + │ ├── Page (~1 MB of t values) + │ ├── Page + │ └── Page + ├── Column chunk "d" + │ └── Page ... + └── Column chunk "k_s" + └── Page .. +``` + +Steps to doing better: +- Split the file into multiple row groups. +- Enable `Chunk` or `Page` statistics, which gives us a page index (in the parquet footer) with stats (e.g. column min/max/nulls) for those segmentations of the part file. +- Add "Range Read" to our Blob Store interface to read from specific offsets within a file. Then, only read the row groups or pages whose stats fit the predicate. +## Writers +**What uses Persist?** +### Materialized View sink +Multiple replicas can write to the same shard, including replicas running different versions of the code. +- We can't assume all writers agree on the collection's contents. + - e.g. After a bugfix, newer replicas will disagree with older replicas. +- A writer can't assume the previous batches in Persist were written by writers it agrees with. + +Therefore, if a writer wants the shard to match its own view of the collection, it must _read the shard_, calculate the diff, and commit the diff. +We call this behavior _self-correction_ because the winning writer erases any accumulated mistakes from the collection. + +_On each replica:_ + +1. One worker chooses the time range for a batch. + - It also _selects which worker_ is responsible for `compare_and_append`ing the batch. +2. All workers write their own parts for the batch. +3. The _selected worker_ `compare_and_append`s all the workers' runs. + +### Source exports' Persist sinks +For each of a _source's_ (e.g. database) _exports_ (e.g. table), we run a `persist_sink` dataflow to record that export's collection of records. + +_On each replica:_ +1. One worker chooses the time range for a batch. +2. All workers write their own parts for the batch. + - _N.B._ Because sources don't guarantee key ordering, each part can span the entire key range. Therefore, each part is its own run. + - _Fun fact:_ Workers write their parts as single-timestamp batches, which the leader worker consolidates into one Persist batch, typically spanning multiple timestamps. +3. One worker `compare_and_append`s the batch with each worker's run(s). + - _Multiple Replicas:_ Kafka and load-generator sources run on multiple replicas, which compete for a successful `compare_and_append`. + Postgres, MySQL, and SQL Server only run on a single replica. + +_Fun fact x2:_ The source exports' Persist sink implementation is derived from the materialized view sink, with the self-correction step removed. +### Tables, `txn-wal` +### More Writers +TODO +These other components write to Persist but are not drivers of its design: +- Sinks: only record progress +- Builtin tables during 0dt upgrade: because the new generation can't modify `txn-wal`. (doc says it's a temporary hack?) +- COPY FROM: uses the table path +- Webhook sources +- Storage-controller collections +- Catalog +- Expression cache +- Builtin schema migration shard +- Dropped-shard cleanup +## Readers +TODO +### Reading a Snapshot +- Get all _batches_ for the shard, up to the _time_ we want to read. +- In parallel, read all the _parts_ for all the _runs_ in those batches. +TODO +### Interlude on Future Work: Consolidate on Read +TODO +## Consensus +TODO \ No newline at end of file From eb66703665b3c32616cb9119896d5b81f2980f9a Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Thu, 24 Sep 2026 10:49:13 -0400 Subject: [PATCH 2/5] txn-wal section --- doc/developer/101-persist.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/doc/developer/101-persist.md b/doc/developer/101-persist.md index 58e12021f03b0..761a74a016d35 100644 --- a/doc/developer/101-persist.md +++ b/doc/developer/101-persist.md @@ -90,6 +90,25 @@ _On each replica:_ _Fun fact x2:_ The source exports' Persist sink implementation is derived from the materialized view sink, with the self-correction step removed. ### Tables, `txn-wal` +In each environment, the storage controller batches updates from all tables (user tables and system tables) into a group commit, timestamped by the timestamp oracle, and appends it to a `txn-wal` shard. + +For every table with data changes, the storage controller then copies from the group commit in `txn-wal` into the table's shard (and deletes that table's commit from `txn-wal` to mark it as "done"). + +If we didn't group-commit all tables to `txn-wal`: +- We couldn't support multi-table transactions. +- We'd still tick each table separately. i.e. Even if a table didn't change, we'd still need to append to its shard every tick. Now, only `txn-wal` needs to tick. + +#### Migrating system tables during 0dt upgrades +The new generation starts out read-only. The new storage controller follows `txn-wal` but cannot write to it. + +For any system tables with schema changes that cannot be migrated in place, the new coordinator creates new table shards and writes their IDs to the migration shard. +The new storage controller writes the catalog state to the new table shards directly, without passing through `txn-wal` first. + +On promotion, the new generation restarts, no longer in read-only mode: +- The new catalog performs its migration, with the new table shard IDs from the migration shard. +- The new storage controller takes over `txn-wal` and registers the new table shards there. +- The coordinator reads system tables from their shards and, via `txn-wal`, replaces their rows with the current state from the catalog. + ### More Writers TODO These other components write to Persist but are not drivers of its design: From f57e9fa89c33f7f50906aed92e853569cffdbdd1 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Tue, 29 Sep 2026 17:12:48 -0400 Subject: [PATCH 3/5] placeholders and formatting --- doc/developer/101-persist.md | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/doc/developer/101-persist.md b/doc/developer/101-persist.md index 761a74a016d35..6a7dfb80a4236 100644 --- a/doc/developer/101-persist.md +++ b/doc/developer/101-persist.md @@ -111,6 +111,7 @@ On promotion, the new generation restarts, no longer in read-only mode: ### More Writers TODO + These other components write to Persist but are not drivers of its design: - Sinks: only record progress - Builtin tables during 0dt upgrade: because the new generation can't modify `txn-wal`. (doc says it's a temporary hack?) @@ -121,13 +122,32 @@ These other components write to Persist but are not drivers of its design: - Expression cache - Builtin schema migration shard - Dropped-shard cleanup + ## Readers TODO + +_Claude says: leased vs. critical readers, since holds, heartbeats, and what happens when a reader loses its lease._ + ### Reading a Snapshot - Get all _batches_ for the shard, up to the _time_ we want to read. - In parallel, read all the _parts_ for all the _runs_ in those batches. + TODO + ### Interlude on Future Work: Consolidate on Read TODO + ## Consensus -TODO \ No newline at end of file +TODO + +_Claude says: Shard state is stored as a log of diffs plus periodic rollups, updated by compare-and-set. Every operation on a shard (appends, heartbeats, compaction, GC) goes through that one compare-and-set, per PER-38._ + +## Compaction +TODO + +_Claude says: writers trigger it after writing. Include physical vs. logical compaction, incremental compaction, and how one slow reader holds back compaction for everyone (PER-42)._ + +## Garbage Collection +TODO + +_Claude says: state GC and blob GC. GC is turned off in cloud, and dropped shards leave orphaned blobs (about 200 TB at one customer)._ From 67d6ec27526d92681fa91941ad4db73f7c6c1b50 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 30 Sep 2026 13:51:01 -0400 Subject: [PATCH 4/5] short summary of architecture --- doc/developer/101-persist.md | 57 ++++++++++++++++++++++++++++-------- 1 file changed, 44 insertions(+), 13 deletions(-) diff --git a/doc/developer/101-persist.md b/doc/developer/101-persist.md index 6a7dfb80a4236..d682cb3add636 100644 --- a/doc/developer/101-persist.md +++ b/doc/developer/101-persist.md @@ -1,21 +1,30 @@ # Persist 101 -Persist is a data store backed by immutable blobs. -## Semantics -**First, _what_ is Persist?** -It's a durable store of time-varying collections. +## Semantics +Persist is a durable store of time-varying collections. -Data is organized into collections called "shards". -Individual records within a collection are _(key, value, multiplicity)_. +Each time-varying collection is called a "shard". -For the time window between a shard's **"since"** (lower bound) and its **"frontier"** (upper bound), we know the history of the collection and can query its contents. +For the time window between a shard's **"since"** (lower bound) and its **"upper"** (upper bound), we know the history of the collection and can query its contents. - Before the lower bound, we no longer remember the history of the collection. -- At or beyond the frontier, the collection's state is still being decided. +- At or beyond the upper, the collection's state is still being decided. + +We can only write (append) to a collection _at or beyond_ the upper. We cannot alter history before that point. + +Individual records within a shard (as recorded at a given point in history) are _(key, value, multiplicity)_. + +## Architecture + +**Blob Storage:** Each shard's contents are written to (immutable) blobs. + +**Consensus:** All coordination and bookkeeping happens in the consensus database. For each shard: +- _Trace:_ The since, the upper, and all the blobs with our data. +- _Readers:_ What each reader is looking at, so we don't delete it before they're done. +- _Writers:_ Each writer's last write, for idempotent retries. -We can only write (append) to a collection _at or beyond_ the frontier. We cannot alter history before that point. +**Clients:** There is no Persist "server". All clients interact directly with blob storage and consensus. ## Data Layout in Blob Storage -**What does it actually look like?** ### Layout: Shard > Batch > Run > Part Within a shard, records are organized into: - Batches. Together, they cover the entire `[Since, Frontier)` time range without overlapping. Each timestamp within the range belongs to a single batch. @@ -59,7 +68,6 @@ Steps to doing better: - Enable `Chunk` or `Page` statistics, which gives us a page index (in the parquet footer) with stats (e.g. column min/max/nulls) for those segmentations of the part file. - Add "Range Read" to our Blob Store interface to read from specific offsets within a file. Then, only read the row groups or pages whose stats fit the predicate. ## Writers -**What uses Persist?** ### Materialized View sink Multiple replicas can write to the same shard, including replicas running different versions of the code. - We can't assume all writers agree on the collection's contents. @@ -109,12 +117,12 @@ On promotion, the new generation restarts, no longer in read-only mode: - The new storage controller takes over `txn-wal` and registers the new table shards there. - The coordinator reads system tables from their shards and, via `txn-wal`, replaces their rows with the current state from the catalog. +_Note:_ Docs say the new generation's behavior is a hack, and it should be unified with the `txn-wal` approach. + ### More Writers -TODO These other components write to Persist but are not drivers of its design: - Sinks: only record progress -- Builtin tables during 0dt upgrade: because the new generation can't modify `txn-wal`. (doc says it's a temporary hack?) - COPY FROM: uses the table path - Webhook sources - Storage-controller collections @@ -142,6 +150,29 @@ TODO _Claude says: Shard state is stored as a log of diffs plus periodic rollups, updated by compare-and-set. Every operation on a shard (appends, heartbeats, compaction, GC) goes through that one compare-and-set, per PER-38._ +### Shard State + +Claude's diagram of shard state: +``` +State +├── shard_id, seqno, walltime_ms, hostname +└── collections: StateCollections + ├── version state format version (0dt compatibility) + ├── trace: Trace the batch list: since, upper, spine + │ └── HollowBatch desc (lower, upper, since), len, run_splits, run_meta + │ └── RunPart + │ ├── Single(BatchPart) + │ │ ├── Hollow blob key, size, key_lower, stats, schema_id + │ │ └── Inline the updates themselves, stored in state + │ └── Many(HollowRunRef) pointer to a blob that lists more parts + ├── leased_readers id → since, seqno, last heartbeat, lease duration + ├── critical_readers id → since, opaque token + ├── writers id → last heartbeat, last write token, last write upper + ├── schemas schema id → encoded key/val schemas + ├── rollups seqno → blob key of that rollup + ├── active_rollup, active_gc, last_gc_req +``` + ## Compaction TODO From 0550983e80a77bec8d7d037a0fd90a66554d70be Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Thu, 1 Oct 2026 13:58:17 -0400 Subject: [PATCH 5/5] continuity, formatting, short intro to consensus --- doc/developer/101-persist.md | 31 +++++++++++++++++++++++++++---- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/doc/developer/101-persist.md b/doc/developer/101-persist.md index d682cb3add636..a928acb8374a8 100644 --- a/doc/developer/101-persist.md +++ b/doc/developer/101-persist.md @@ -25,13 +25,19 @@ Individual records within a shard (as recorded at a given point in history) are **Clients:** There is no Persist "server". All clients interact directly with blob storage and consensus. ## Data Layout in Blob Storage +At its core, a Persist shard is just an append-only log, where writers append batches of blobs +and readers sift through the blobs to find what they need. + +Here is what those blobs look like: + ### Layout: Shard > Batch > Run > Part Within a shard, records are organized into: -- Batches. Together, they cover the entire `[Since, Frontier)` time range without overlapping. Each timestamp within the range belongs to a single batch. -- Runs. Each batch is made up of one or more runs. Runs are an artifact of how batches are written, not a semantic separation of records. -- Parts. Each run is made up of one or more parts. Each _part_ of a _run_ is responsible for a range of keys. +- _Batches._ Together, they cover the entire `[Since, Frontier)` time range without overlapping. Each timestamp within the range belongs to a single batch. +- _Runs._ Each batch is made up of one or more runs. Runs are an artifact of how batches are written, not a semantic separation of records. i.e. Two runs can cover exactly the same time and key ranges. +- _Parts._ Each run is made up of one or more parts. Each _part_ of a _run_ is responsible for a range of keys. Within a part, records are typically sorted by key, and within a run, parts are sorted by their key ranges (which do not overlap). - _Parts are internally sorted by key only if they've been through compaction. For large shards, most parts have been through compaction._ + #### Data Layout in a Part File Part files are Parquet, where the columns are: - `t` - timestamp @@ -40,14 +46,16 @@ Part files are Parquet, where the columns are: - `ok` - struct with one field per relation column - `err` - binary-encoded `DataflowError` - `v_s` - unused + Our Parquet writer is simple: - only one row group - no statistics (Predicate push-down chooses which part files to read based on stats we write into shard state--outside the part files themselves.) _Note: There is also a legacy format with columns `k, v, t, d` and a migration format with `k, v, t, d, k_s, v_s`._ + #### Interlude on Future Work: Predicate Push-Down _within_ a Part Today, we always read and decode an entire part file. -If we stop doing that, maybe we can support point (key + time) lookups and improve predicate push-down. +If we stop doing that, maybe we can improve predicate push-down and even support point (key + time) lookups. Claude's diagram of Parquet chunks and pages within our one row group: ``` @@ -67,7 +75,22 @@ Steps to doing better: - Split the file into multiple row groups. - Enable `Chunk` or `Page` statistics, which gives us a page index (in the parquet footer) with stats (e.g. column min/max/nulls) for those segmentations of the part file. - Add "Range Read" to our Blob Store interface to read from specific offsets within a file. Then, only read the row groups or pages whose stats fit the predicate. + ## Writers +### First, a bare-minimum introduction to the consensus database + +In order to append to a shard, a writer must perform an atomic `compare_and_append` operation: +1. By comparing sequence numbers, does the actual shard state match the latest shard state the writer has seen? +2. Durably append the writer's change to the shard's state (with the next sequence number). + - Example state change: Append a batch of blobs for the latest timestamp range. + +Any client (writers, readers, others) that needs to catch up on the latest shard state performs these operations: +1. `head`, to get an outline of the latest shard state, including a pointer to the latest state "rollup" blob. +2. `scan`, to read from the most recent state rollup through all the subsequent state changes. + +_Shard state is recorded as a sequence of incremental updates (stored inline in the consensus database) interspersed with periodic full-state rollups (stored in blobs), +and each client builds its own view of shard state by applying the updates to a rollup._ + ### Materialized View sink Multiple replicas can write to the same shard, including replicas running different versions of the code. - We can't assume all writers agree on the collection's contents.