Shard the block table: Notion
Notion split its block table into 480 logical shards, then streamed changes through Kafka to a data lake.
Notion kept every block, the unit a page is made of, in one Postgres database for five years. By 2021 the block table held more than 20 billion rows, and VACUUM had started to stall. The team split it into 480 logical shards across 32 physical databases, then grew to 96 physical databases in 2023 while keeping 480 logical shards.
The system#
A logical shard is a Postgres schema, and a physical database holds several of them. The application knows the shard, so a query goes to one database when it has the key. The analytics path moved off the primary: Debezium reads the Postgres write-ahead log, Kafka carries the changes, and Hudi writes them to S3. By 2024 the lake handled tens of megabytes a second of row changes, and the block table had passed 200 billion rows.
The board#
Drag to pan. Hold Ctrl or Cmd and scroll to zoom.
| Where | Setting | Value | Why |
|---|---|---|---|
| App Server | Instance size | large | Large, because the app serves blocks from many shards. |
| App Server | Copies | 3 | Three copies stand for the app tier over the shards. |
| Database | Shards | 4 | Notion splits its block table across logical shards. Four is the board’s widest, and it stands for 480. |
| Database | Shard key | hash | A hash spreads blocks and pages, so no shard takes one workspace’s whole document tree. |
| Database | Instance size | medium | Medium for each shard, so a busy shard shows. |
| Event Stream | Partitions | 12 | Many partitions let the Warehouse read every shard’s changes at once. |
| Event Stream | Acks | all | A change is not acknowledged until every replica has it. |
| Event Stream | Keyed by entity | On | One block’s changes stay in order. |
| Event Stream | Retention | 3600 | An hour of retention lets a failed Warehouse replay its work. |
Research this Design needs: Change data capture, Event streaming, Data warehousing, Sharding, Change data capture, Event streaming.
The Database has four shards on a hash key, standing for the 480 logical shards. The CDC Connector is Debezium, the Event Stream is Kafka, and the Data Warehouse is the lake that the changes land in. The app never writes twice: it writes the block, and the connector reads the change.
Why it works#
The shard keeps the online table small enough for VACUUM and for one machine's IOPS. The log keeps analytics off the primary, and it gives the lake an ordered stream of changes. Change data capture and object storage and lifecycle rules are the two paths.
Where the game stops#
The board has four shards, not 480, and no schemas inside a physical database. It does not show Hudi, the 480 Fivetran connectors the team replaced, the S3 layer the lake writes to, or the dark reads that made the cutover safe. The caption is Quorum's result, not Notion's change rate.
Sources#
Herding elephants: Lessons learned from sharding Postgres at Notion, Notion: the 480 logical shards, the 32 physical databases, and the migration.
The Great Re-shard: adding Postgres capacity (again) with zero downtime, Notion: the move to 96 physical databases and the shard map.
Building and scaling Notion's data lake, Notion: Debezium, Kafka, Hudi, S3, and the 200 billion block rows.