Real Postgres

Neki is built as a sharding and scaling solution for real Postgres. It's not a fork, nor a wire-compatible reimplementation, nor a MySQL sharding idea wearing a Postgres label. Neki uses ordinary PostgreSQL instances that store rows in Postgres data pages using MVCC, carry out transactions, and work as you would expect with psql and other Postgres drivers. Neki builds around those instances to let you shard them, scale them, and manage them as one database.

Let's take a look:

PostgresManager

Using vanilla Postgres means Neki needs a way to run and manage each instance. That includes starting and stopping Postgres, owning its data directory, and configuring replication so a new instance can join a shard. PostgresManager handles this coordination, running as the first process in the Postgres container and managing the postgres process directly.

Sidecar

Postgres uses a separate backend process for each connection and limits how many can be open at once. Neki’s Sidecar sits in front of each instance and pools connections, letting many client connections share fewer Postgres backends.

The Router, which is the component that accepts external client connections, communicates with the Postgres nodes via these Sidecars.

It also reports each Postgres instance's health and whether it is a primary or replica, so the rest of the cluster knows whether it can receive write queries.

The pool doesn't treat every connection the same way. The length of time a connection is checked out for use varies depending on what it's being used for. A multi-statement transaction holds on to its connection until commit or rollback. A session-scoped advisory lock needs a connection of its own, because the lock has to outlive whatever transaction is open at the time and can't share that connection. Everything else checks a connection out and hands it back the moment the statement finishes.

The Sidecar knows which of the three to use because the Router sends the necessary information with the query: autocommit, an open transaction, or a session that has to stay on one backend.

Shards

Each Postgres instance gets its own Sidecar and PostgresManager pair. Real deployments need more than one instance: a primary and its replicas. Neki calls that group a shard, the unit it splits data across. It's always advised to run a shard with a primary and 2+ replicas for high availability, as well as for additional read query capacity.

A shard is considered one Postgres cluster. Its replicas are physical copies of the primary, so they share a catalog and the same object identifiers.

Object Identifiers (OIDs) are how Postgres tracks objects internally, rather than by name. A client reads a column’s type OID off the wire to interpret its bytes and may cache that OID for later re-use. A custom type therefore needs to carry the same OID no matter which shard answers the query. Independent shards can assign that type different OIDs, so Neki designates one shard in the entire Neki cluster as the authoritative shard. This shard is the source of truth for translating custom type OIDs in responses from other shards to match. It ensures OIDs are consistent across the many shards of the Neki cluster.

The authoritative shard's Sidecar also watches for schema changes and reports them to the Routers. This keeps the Routers' view of the schema current when a table is renamed or a column is dropped.

Admin

In a distributed system, instances can fail independently while the rest of the system lives on. Neki is no different. A primary or replica can go down at any moment while its fellow instances on the shard are healthy. The Admin's job is to detect failures, promote a replica, and maintain each shard’s durability policy.

It health-checks every Sidecar, tracks replication lag for each replica, and decides when a shard needs a new primary. When a primary goes down, it coordinates an emergency failover, promoting a replica to take its place. It can also coordinate a planned switchover, which are needed for intentional node resizes and version upgrades. In both situations, Admin uses pg_rewind to bring diverged instances onto the new primary’s timeline, copying only the data that changed since the timelines diverged.

Each shard has a durability policy that determines when a commit is acknowledged:

• Async: The primary acknowledges the commit without waiting for a replica.
• Sync: The primary waits for a replica to confirm the commit, protecting against the loss of a single node.
• Cross-zone sync: The primary waits for confirmation from a replica in another availability zone, protecting against the loss of the primary’s zone.

Postgres enforces whichever one is configured, using its own synchronous replication machinery. The Admin keeps that configuration correct as replicas join or leave shards, or a failover moves the primary to a different zone.

Much of Admin’s work, however, doesn’t involve changing the primary. It repoints replicas to the correct replication source and corrects roles when Postgres and the topology disagree.

Operator

Neki’s components need to be deployed, updated, and replaced when their machines fail. Neki is built Kubernetes-first, and the Operator manages this full lifecycle.

The Operator models a cluster as a hierarchy. A cluster owns routers and shards, and each shard owns the pods running its Postgres instances and Sidecars. When the Neki cluster configuration changes, the Operator works out which pods need to be created, updated, or removed.

How it replaces an instance depends on whether that instance is still running. For a live instance, the Operator builds a replacement and confirms it has caught up before deleting the old one. If a node fails and loses its ephemeral storage, the Operator rebuilds the lost instance from scratch once its safety checks pass.

Admin and the Router handle the database side of those disruptions. Admin coordinates a switchover for planned primary replacements or a failover when a primary goes down. The Router can buffer queries that are safe to retry while a healthy primary becomes available.

Router

We've talked a lot about how the Neki cluster operates and handles failure internally. What we've yet to dive into is how applications use the thing!

The Router is the entry point for clients connecting to a Neki cluster, presenting a single Postgres wire-protocol endpoint to connect to a (potentially) massive sharded database. Applications use Postgres drivers to send SQL and open transactions without managing connections to individual shards.

Authentication and role checks are done as if it were the Postgres instance itself, and the protocol's own extended-query flow and prepared-statement lifecycle are all built into the Router.

Once a query arrives, the Router runs a Postgres-compatible parser against the authoritative shard's catalog, plans it against the current sharding layout, and sends it to whichever Sidecar needs to run it over gRPC.

Not every query can run on a single shard. A join may need data from several shards or an aggregate may need to read from all of them. The Router coordinates that work as a distributed query.

Whenever possible, it leaves the work to the Postgres instances. If both sides of a join are on the same shard, the Router sends the join to that shard. When a join needs to run across shards, the Router executes it itself, choosing between nested-loop, hash, and merge joins based on cost estimations.

Earlier, we covered how Admin promotes a new primary during a switchover or failover. If that happens, the Router can buffer queries, giving the Admin time to complete the handover. For queries that can safely be retried after failing against a primary, the Router buffers the query and waits, for a fixed time, for a healthy primary. Once a healthy primary is available, the Router releases queued queries gradually.

Data Topology

Router, Sidecars, and Admin all need a consistent picture of which shards exist, what key ranges they own, and which tables are sharded at all. If the Router's copy is wrong, a query can land on the wrong shard. This is all specified with a Data Topology, and etcd holds the single, authoritative copy of it. When the Data Topology changes, the Router, Sidecars, and Admin pick up the updated configuration without a restart or manual synchronization.

The Data Topology defines shard groups, named sets of physical shards, each owning a range of routing keys. Each table belongs to a shard group. Shard indexes specify the columns or expressions and the strategy used to turn row values into routing keys. Those keys determine which shard receives each row.

Replicator

As a database grows, its layout may need to change. Tables need to be imported, shards need to be split, and schemas need to change all while applications keep using the database.

Neki's Replicator handles the data movement behind all such operations. It runs as a separate process colocated with a shard's Sidecar and Postgres. It is responsible for copying existing rows to new destinations, and also keeping the data current by decoding changes from a Postgres logical replication stream and applying them as SQL.

Three workflows use the Replicator:

• MoveTables relocates a set of tables, including imports from an external Postgres instance
• Reshard redistributes data across shard key ranges, allowing a shard to be split when it outgrows its capacity
• OnlineDDL changes a table's schema by building a shadow table alongside the original and keeping it current through the same change-data-capture pipeline MoveTables and Reshard use to relocate rows. A final rename swaps the new table into place. This supports changes such as repartitioning a table, alongside changes that would otherwise require a blocking operation.

Once the data has been copied and the destination is caught up, the workflow switches from the original tables or shards to their replacements. This is the cutover. The Router uses the same buffering mechanism that handles primary changes for this step. It buffers queries during that switch and releases them afterward.

Together, these components let Neki scale Postgres horizontally while presenting a single database to applications.

Get started

Neki is in Platform Preview right now.

Start a Neki cluster today: build on it from scratch, or import an existing Postgres database.

Neki is built from lessons we’ve learned over eight years of running some of the largest sharded MySQL clusters in the world. Thousands of production workloads with millions of queries per second for companies where even a few seconds of downtime is a very public event. We know what it means to power the world’s biggest tier 0 workloads.

When we released PlanetScale Postgres a year and a half ago, we knew we needed to do more. In that time we’ve onboarded several thousands of customers on PlanetScale, some of them rivaling the size of our largest MySQL customers. Time and time again, we watched teams approach the ceiling of a single machine with Postgres. Metal bought them time, but with customers hitting the upper limit of what a single machine is capable of, we found there was no good option to hand them. Enter Neki.

What is Neki?

Neki is sharded Postgres from PlanetScale. It lets you scale a Postgres database across many machines while keeping real Postgres on every shard.

Your application connects to a Neki router over the standard Postgres wire protocol, so your existing drivers, ORMs, and connection string keep working. Each shard is a full Postgres cluster with one primary and at least two replicas across 3 availability zones. There is no custom storage engine, so extensions, SQL support, and performance behave the way Postgres does.

You choose the shard key and control how tables are grouped and distributed through a JSON data topology. Schema changes, version upgrades, failovers, imports, and resharding all run as built-in fully online workflows. You also get the PlanetScale features you already rely on, including Insights, schema recommendations, branching, and MCP.

You don't have to shard on day one. Run Neki as a single primary with replicas, and when you outgrow one machine, resharding is a workflow you run against the cluster you already have.

Why Neki?

You already know the problems that come with fast-growing Postgres databases: tables too large to vacuum or index without affecting traffic, backups taking hours, connection limits, maintenance windows for schema changes, transaction wraparound and so much more.

You can move to a bigger instance, but eventually you run out of big enough machines, and the problems don’t scale linearly as you add more cores and IOPS.

The existing answers each ask you to give something up. Application-level sharding pushes routing into your code. Postgres-”compatible” distributed databases hide the shard key from you, take away your extensions, and add complexity and latency which becomes difficult to handle and debug.

So we built Neki with a few principles, the biggest one being: stick to Postgres, don’t work around it, fake it, or turn away from it.

How does Neki work?

We architected Neki from first principles for Postgres, with real Postgres on every shard. There are four moving parts.

Neki routers

Your application first connects to a Neki router. The router speaks the Postgres wire protocol so your existing drivers and ORMs keep working with a single connection string. A router has a full Postgres query parser, a distributed query planner, query buffering and more. It parses your query, builds a plan that decides which shards should run it, sends the work out, and combines the results back into one stream. Routers can scale vertically and horizontally, so no single router becomes the bottleneck.

Sharding and shard groups

Every shard in Neki is real Postgres with 1 primary and at least 2 replicas, spread across availability zones. There is no modified storage engine. Extensions, SQL support, and performance behave the way Postgres behaves, because it is Postgres.

Shards are organized into shard groups, so different tables or workloads can live on different sets of shards. Each shard uses a configuration profile that defines its instance size, replica count, storage, Postgres parameters, and extensions, so you can size each group for its own traffic.

Connection pooling

Sidecars run alongside every Postgres instance. This is the piece that makes Neki's connection handling meaningfully better than just sticking PgBouncer in front of a database. Because Neki controls both ends of the connection, the router side and the Postgres side, it can size pools to what each instance can actually serve instead of estimating from outside the process.

Control plane

The control plane tracks the health of every node, runs planned switchovers and unplanned failovers, and coordinates the workflows that reshard data, apply schema changes, and perform version upgrades.

Data topology

Tying it together is the data topology, a JSON configuration that maps your logical tables onto physical shards. You define shard indexes, which specify the column Neki routes on and how that value gets hashed, and shard groups, which control how many shards a set of tables spreads across and which shards those are. Routers cache the topology and consult it on every plan.

What you get beyond sharding

Everything you would normally schedule a maintenance window for runs as a built-in workflow in Neki. Workflows provision new target nodes, catch them up with replication, switch traffic with a __neki metafunction, and retire the old nodes. All through the same psql connection your application uses.

This online operations model covers schema changes, version upgrades, planned and unplanned failovers, imports, and resharding.

Neki also includes all of the features you’ve come to rely on with PlanetScale: Insights, schema recommendations, branching, MCP, and more.

You can also run Neki unsharded, as a single primary with replicas. You get the improved connection pooling, online DDL, zero downtime upgrades, and health monitoring before you need to shard. When you do, resharding is a workflow you run against the cluster you already have.

What is a platform preview?

We wanted to get Neki into your hands as soon as possible. You should not run production workloads on Neki during the platform preview. The product is still changing, and some of those changes will be breaking.

If you have any feedback, questions, or face any issues during the platform preview, please let us know. Fill out a support ticket or join our Discord

Try Neki today

Sign in to PlanetScale, opt in to the platform preview, and create a Neki cluster. Read the Neki docs for more information about Neki's architecture, how to shard, and more.

If you have a large Postgres cluster and are curious whether Neki is a good fit, get in touch. We would love to do a private demo for your team, dig into your schema and query patterns, and give you real suggestions on how to shard.

You've just been paged. Every query is failing to execute. Find out why.

Years ago at GitHub, this was a question Sam Lambert would ask in interviews.

The solution involved a downed database that started with a MySQL schema change needing an exclusive lock on a table, but it couldn't get one. Another session had already touched that table and never committed. That open transaction blocked the schema change.

Once the schema change was waiting, new queries on the same table queued up behind it. Nothing was deadlocked. Nothing would time out on its own. The whole pile-up sat there until someone found the connection at the top of the chain and killed it.

(By the way, you'd avoid this MySQL lock on PlanetScale Vitess since ALTER would have been an online schema change, copying in the background and only taking a brief lock at cutover.)

This isn't rare, and it isn't exclusive to MySQL. You can reproduce the same pile-up on Postgres just as easily.

Want to see a stuck database?

You won't need elevated permissions. A normal application connection is enough.

All it takes is an unhandled exception and a poorly timed migration.

Say you have a Postgres database with a hot table named orders.

On one connection, a transaction begins a SELECT query, but the app throws an exception.

-- Connection A: Ordinary application user
BEGIN;

SELECT * FROM orders WHERE id = 123;

-- ...then the app throws an exception 💀

The SELECT query finished, but Postgres keeps holding the connection, waiting for the transaction to end with COMMIT or ROLLBACK, which it never does because of the app's thrown exception.

While Postgres waits on that connection, another connection attempts to run a migration.

-- Connection B: Migration role waiting on lock
ALTER TABLE orders ADD COLUMN foo integer;

The SELECT query only needed a read lock (ACCESS SHARE). The migration needs a lock that blocks all other access (ACCESS EXCLUSIVE), so it waits.

Postgres will not let later queries jump the queue. Every new query on orders waits behind the already-stuck migration, which is itself stuck behind the initial SELECT connection that never closed.

-- Connections C, D, E, etc: Waiting on the lock
SELECT *
FROM orders
WHERE customer_id = ?;

If the app doesn't handle errors gracefully and depends on quick responses from orders, you've got downtime.

Not because of a hack, but because one unhandled exception prevented a COMMIT, kept a connection open, blocked a migration, and that migration blocked everything else.

By the way, this is preventable. It happens because the default for idle_in_transaction_session_timeout is disabled. Give it a value and Postgres times out transactions that haven't closed.

Freeing a stuck database

Debugging the state of connections in MySQL and Postgres can be clunky.

As covered in See what your database is doing right now with Connections it is possible to run a query on a loop in your terminal to see active connections. But it's not an ideal interface.

It also depends on your database having spare connection slots. The worst-case scenario is your database is overwhelmed with connections that haven't been closed, meaning even you can't connect to debug and kill them.

Because this experience is so terrible, we've added the ability to view and kill connections in Postgres and MySQL databases on PlanetScale.

Connections in the dashboard

PlanetScale's connections tooling uses a reserved administrative connection so it can still show you a list of active connections and processes even when connections are exhausted and your application can't connect.

In the PlanetScale dashboard, you can now click the Connections tab to view a live list of connections. You'll see a bunch of useful columns like Process ID, State, and Duration, but most importantly, you will see a column for Blocked Queries.

From here, it's easy to see which connections are currently blocking others from proceeding. You can click into any connection to see more details, such as the query currently being performed by that connection.

Kill process gives you three options, from least to most disruptive:

• Cancel query stops the running statement and leaves the connection open, so the app can send another query
• Terminate transaction rolls back the open transaction and closes the connection, but only if that same transaction is still running
• Terminate connection drops the backend entirely, whether it is idle or mid-query

Thinking back to the downtime example, the idle SELECT had already finished, so canceling the query will not release the lock. In that scenario you would need to terminate the transaction or the connection.

Over on Vitess, each process has two options: cancel the query or terminate the connection. In a sharded database, you can view the process list per keyspace and shard.

Connections via CLI

If you'd rather work with a terminal UI, launch:

pscale branch connections top <database> <branch>

This opens an interactive live view that refreshes about once every second. The most important connections are sorted toward the top. You can navigate through the view using keyboard shortcuts and open any connection to see more details.

If your agents prefer the current state of connections in JSON, use this one-shot command instead of the live view:

pscale branch connections show <database> <branch> --format json

That output includes the IDs needed to cancel a query or terminate a connection, which you can still require confirmation for. See the CLI reference for the kill commands that go with it.

Migrate today, stress less later

The correct answer to land your dream job today is "Use PlanetScale."

Next time you're paged because everything is erroring, I hope you've already migrated over. The interview answer hasn't changed, but connection management on PlanetScale has made it easier than ever before.

Product success leads to lots of data, lots of data leads to big tables, but big tables lead to predictable problems.

A table can be large (many rows), wide (many columns), or "fat" (oversized values). Any of these can cause you problems.

How a large table can cause an outage

First, let me tell you a story.

A customer had two tables, A and B. Table B is large and unpartitioned. It has a foreign key to A with cascade delete turned on. A single delete on A has to delete every child row in B. That can be 100s or 1000s of rows in B for every one in A.

In this customer's case, deletions from A ran in a background job and attempted to remove around 100,000 rows from A. The deletes took so long that they timed out. The job retried. Each attempt generated a pile of WAL (the write-ahead log Postgres uses to record every change), which saturated the network and CPU between the primary and its replicas. This caused the replicas to lag.

The application checked replica LSN (a replica's position in that WAL stream) before sending a read to a secondary. The secondaries were not caught up, so every query went to the primary.

The primary was already busy writing WAL. It also had to respond to all of the reads. And so a simple cascade delete on a large table quickly became an outage.

Possible fixes for a large table

The problems covered in this post are common to all Postgres databases but are made worse by those containing large tables. You might reach for one of a few solutions, which we'll examine throughout.

Partitioning splits a large table based on a key value. For example, if your large table stores rows with a specific date, such as when an event occurred, cutting that table into partitions of months or years creates more, smaller, and more manageable tables.

This can be useful for some large-table problems.

Vertical scaling refers to adding more resources to make more space or brute-forcing your way out of performance bottlenecks such as increasing CPU, RAM, or buying larger and faster disks.

It's almost always a band-aid solution that only hides the real problem.

Sharding (horizontal scaling) splits the large table, or the entire database, into separate database clusters connected by a router to act as one. Just as Vitess is sharded MySQL, Neki is sharded Postgres.

With more, smaller, and isolated database clusters, almost all of your expected large table problems are resolved. More on that later in the post.

Slow, late vacuum

You may already know that deleted rows in Postgres aren't automatically removed; they are marked for deletion. These marked rows hold space until the vacuum process reclaims it. Read Every UPDATE leaves a ghost to learn more.

What you might not realize is that vacuum runs per table.

Vacuuming a single large table can tie up a worker for a long time. Vacuum uses a ring buffer so it does not evict the rest of shared_buffers (Postgres's shared page cache), but on a table larger than RAM it still competes with user queries for the same disks.

Large tables take longer to vacuum, and they also wait longer to start.

On Postgres 17, the default trigger is 20% of the table. A table with ~500 million rows needs ~100 million dead tuples before autovacuum begins. You can modify this value cluster-wide or per table by changing autovacuum_vacuum_scale_factor.

Postgres 18 added autovacuum_vacuum_max_threshold with a default value of 100,000,000. So a billion-row table triggered autovacuum at 100 million (still a high number) instead of 200 million. Below 500 million rows, the old scale factor was still in play.

Postgres 19 adds parallel autovacuum, off by default. Extra workers can finish one large table faster, but that still doesn't change when vacuum starts.

Autovacuum also runs ANALYZE, which has the same default-doesn't-scale problem. It samples a fixed number of rows no matter how large the table is. So the planner's estimates potentially get worse as the table grows. Raise the target with ALTER TABLE ... SET STATISTICS on the columns the planner actually needs to get right.

Vacuum on large tables is late to start, slow to finish, with more dead rows while it runs. Each cycle is worse than the last, even when vacuum reports success.

Partitioning can be a good first move to split your large table into more reasonable-size chunks, if it has a suitable key.

Partitioning also helps avoid autovacuum's late start. Each partition is its own smaller heap, so the 20% trigger fires after far fewer dead tuples. Workers vacuum partitions in parallel and ANALYZE samples a smaller relation.

Disks and shared_buffers are still shared with the rest of the cluster, so it's not a perfect solution.

Vertical scaling can help more workers finish sooner with faster disks. That does not change the trigger on one huge heap, and the table may still not fit in RAM.

Sharding (horizontal scaling) gives you the best of both worlds. Smaller tables on smaller databases, each with their own isolated resources.

Sharding also solves foundational parts of Postgres that can't be resolved by more resources.

Every write transaction gets a 32-bit ID, and vacuum uses these IDs (xmin) to decide which row versions are still visible. These IDs are cluster-wide, not per-table. Vacuum freezes old tuples so XIDs can be reused; wraparound is the age of the oldest unfrozen XID (relfrozenxid in pg_class), not how many XIDs the database has ever issued. If that age reaches around 2.1 billion, Postgres goes into read-only mode. Postgres backups under the hood covers transaction wraparound in more detail.

Since each shard is its own cluster, a large table separated across shards is less likely to put a cluster into read-only mode.

Additionally with isolated I/O, any work vacuum does on one shard does not contend with queries on another. ANALYZE becomes a per-shard operation.

Wasted time on incomplete repacking

Vacuum can be prevented from reclaiming space even if it finishes successfully.

Vacuum can't remove a dead tuple while any transaction in the cluster still needs it.

But a slow vacuum on a large table may have wasted its time to completion. Many of the rows it found to remove may still be required by transactions that started before vacuum finished and haven't completed yet. Vacuum will need to walk the entire table again later.

This can be an issue on tables of any size, but is more annoying on large tables as they take longer to complete.

Free space that was successfully reclaimed can be reused for new rows, but the table file usually does not shrink. That leftover file size is bloat.

Should your table get bloated, PlanetScale Insights scans once a day and opens a recommendation when estimated bloat is over 25% and 100MB.

To compact a bloated table while reads and writes continue, enable pg_squeeze on the database's Clusters page (that requires a restart), then run a one-time squeeze or register the table for regular cleanup.

Be aware that pg_squeeze has no throttle. Squeezing a large table can saturate I/O and CPU and cause an outage. Only run it when the cluster has spare capacity.

Partitioning creates more, smaller heaps. So a wasted pass is cheaper and can be retried sooner. Squeeze is per partition, so the spare disk you need to run squeeze is the size of a partition, not the whole table.

Splitting the table does not split the snapshot horizon. xmin is still cluster-wide, so a long query anywhere on the cluster still pins every partition.

You can drop old or bloated partitions instead of compacting them.

Vertical scaling adds more spare disk space, which may be the only reason squeeze can run at all and/or finish sooner. That lets you compact the same large file faster.

Sharding creates distinct Postgres database clusters, so a long query or dump on one shard doesn't pin vacuum on any other. Squeeze runs only on the copy in its shard, not the entire large table. Each cluster tracks its own xmin.

In a sharded database, wasted vacuum passes are cheaper and less likely to happen.

Slow queries hold connections longer

It is impossible to write a Postgres article about performance and not mention that Postgres has a connection-per-process architecture. Connection hygiene matters even more when your database has a large table.

Sequential scans (reading large segments of the table from start to finish), big sorts, and heavy joins against a large table take longer, so they hold connections longer.

Raising max_connections feels like a logical solution to avoid "too many clients already," but this only lets more concurrent connections perform these slow queries. At best, this starves other workloads of new connections. At worst, a traffic spike on your large table is more likely to trigger an out-of-memory (OOM) event.

A large table will not stay in shared_buffers. More processes scanning it means more cache eviction, more disk usage, slower queries, and longer-held connections.

Partitioning splits your large table, which can help queries finish faster (and release connections sooner) so long as they access only one or a few partitions. But cross-partition queries can get worse.

Buy all the RAM and CPU you like with vertical scaling, but you're still not putting a hundreds-of-GB table into cache. You cannot spend your way out of inefficient connection handling.

Sharding spreads your backends across multiple clusters, where each shard-distinct query is shorter and more likely cached.

A connection held by a large table is one the rest of the product cannot use. With workloads spread across shards there may be less overlap.

A scatter-gather query with no shard key can hold more connections, not fewer. Selecting the right key to shard on is critical to seeing performance benefits from sharding.

Slower backups and recovery

Postgres gives you three ways to take a backup. A large table complicates each one in a unique way.

A logical dump with pg_dump writes live rows, so dead tuples stay out of the output. The dump still walks the bloated heap, and it holds a transaction snapshot the whole time. On a small table, that pin is brief. On a large table, the backup is a long-running transaction against the cluster.

A file system backup copies the data directory, bloat and all. Faster than dumping rows. A consistent copy usually means shutting the database down, or relying on atomic snapshots from the file system. You might be okay with a maintenance window for a database of small tables, but a single large table could add hours just to copy one relation.

Continuous archiving copies the data files while Postgres is still running, then keeps the WAL so you can replay through the modified bits and land on a consistent copy. No dump pin, no shutdown. You still copy the large table, bloat, and its indexes. What grows with the table is restore time. How long an incident waits on this heap plus the WAL written while the copy ran.

That third option is how PlanetScale backs up Postgres. We copy the data files and continuously archive WAL on a throwaway node rather than on your primary, so the backup doesn't compete with production queries for disk and doesn't pin the primary.

We can't skip the size of those files. A large table that doesn't fit in memory, plus its indexes, plus its bloat, is what has to land in object storage.

Partitioning rearranges files but doesn't shrink the copy. A pin held during dump is still cluster-wide. A physical backup includes the same bloat even if it is split across smaller tables.

Vertical scaling means faster copies through more resources. But the files aren't any smaller, so restore is only quicker because the disks are faster.

But neither partitioning nor vertical scaling changes the size of the data.

Sharding breaks down backups and restores into smaller, distinct units of work, allowing them to complete much faster. Recovery time depends on the slowest shard instead of the largest table.

Too many indexes

A large table isn't viable to scan from start to finish and is likely queried in many different ways.

To keep queries fast, you keep adding indexes to the table. While indexes make queries fast, they aren't free. They take disk, get vacuumed, get backed up, and make every write touch more files.

There are no good options here. Prune indexes and your queries get slower. Accumulate indexes and pay for it with an enlarged heap, while also vacuuming, packing, and copying many access paths.

The indexes that make a large table usable are part of what makes it too large to keep indexing.

Indexes on a partitioned table are created on the logical table but stored and updated on each partition. Those smaller indexes are faster to build and cheaper to REINDEX.

Vertical scaling buys you more space to store more indexes ... but that's like fixing traffic congestion by adding another lane to the freeway.

Sharding creates more, smaller databases, so having more indexes is less of a problem. You can afford to create many more indexes than if your large table was on a single cluster since a write only updates indexes on the shard that owns that row. Sharding is the good option for balancing a lot of data and many indexes.

Wide tables split across pages and files

A table is not only large because it has many rows; it might be that the rows are wide with many columns, or each row contains a large amount of data.

If a row exceeds ~2KB, oversized values go to TOAST, a separate table and indexes next to the heap you are already struggling to vacuum.

Each toasted value gets a 32-bit OID, about 4 billion per table. Updates of toasted values take a new OID. High-churn wide tables fill that space, and inserts slow down as Postgres hunts for a free one before the hard stop.

Whatever is left still has to fit into a single 8KB page, as Postgres will not split a row across pages. If a tuple cannot fit in the free space on any existing page, it gets a new page.

For a read-heavy table, you want pages packed tight (a high fillfactor), so scans walk as few pages as possible. For an update-heavy table, you want the opposite. Lower the fillfactor so an update can often stay on the same page instead of allocating a new one.

On a small table, the wrong fillfactor is a few wasted pages. On a large, wide table it could be terabytes of sparse pages, or a constant stream of new ones that vacuum, WAL, and backups all have to follow.

A large table is a heap, plus indexes, plus TOAST, and some operations still walk those files one at a time. pg_database_size() is one example. It stats every file in the database serially. On a large, wide table, that can pin a core.

Partitioning is a genuinely good solution for this class of large-table problems. Fillfactor can differ per-partition. Each partition gets its own TOAST table, so OID space is per slice.

However, with more files in one cluster, pg_database_size() can get worse.

Vertical scaling can hide a bad fillfactor for a while with extra disk space. Faster disks make TOAST vacuum and file stat()s cheaper. But you cannot buy greater limits than 8KB pages, the 2KB TOAST threshold, or the 4 billion OIDs.

Sharding creates individual Postgres instances, so TOAST OID space resets per cluster. A bad fillfactor hurts less because the heap is smaller. File walks are per shard.

Sharding does not, by itself, let hot and cold rows use different fillfactors. Partitioning still wins there.

Sharding is the solution to large tables

Just as partitioning breaks up a single table into many small pieces, sharding distributes the workload of your reads and writes across many database clusters.

Let's revisit the cascade delete story.

When replicas lagged, every read went to the primary. On a sharded database, each shard is its own cluster, with its own WAL and replicas. If the affected rows of A and B lived on one shard, that shard could still lag, and its reads pile onto its primary, but that could not fail the LSN check for the rest of the product. If those 100,000 deletes were spread across shards, the WAL burst would be split. No replica stream would have to replay the whole thing.

Your single-cluster large table likely processes a high throughput of traffic. Vertical scaling can help, but sharding does it better. With more clusters receiving and processing writes, a single table doesn't strain the entire cluster.

Additionally, cluster limits that cannot be solved by partitioning or vertical scaling are resolved. Postgres' hard-coded ceilings of 32-bit XIDs, TOAST OID and more become per-shard limits instead.

Sharded database backups are individually smaller and can be massively parallelized, making database restoration faster as well. Operations finish at the time of the slowest shard, not all at once waiting on large, bloated tables.

Cache eviction is a problem with large tables and their working sets. While more resources won't fit a large table into RAM, you often can per shard. Shard-local queries become shorter and faster.

Neki, sharded Postgres, lets your application write queries as if it were a single database, while splitting a large table into smaller, automatically distributed parts.

It contains the logic for where to send reads and writes. You still have to determine how that data is spread. That fan-out is covered in our post on data topologies.

Sharding shares the large table load across multiple clusters. If that's a problem you need solved, sign up for a PlanetScale account to use Neki today.

The errors started as a trickle, a handful per minute, then fed on themselves:

minute   errors/min   queries/s
  0          5          15,000   <- a burst of work arrives
  2         60           3,900
  4        300           2,500
  6        550           2,000
  8        900           1,900
 10      1,400           1,500   <- error peak = throughput trough
 12        700           1,700
 14        250           2,000
 16          0           2,500   <- locks released, backlog drained
 18          0           8,500   <- full recovery, throughput jumps 5x

The trigger was fairly mundane: A batch job opened a transaction against a hot table, took row locks, and then held them for fifteen minutes without committing. A common application bug, the kind that eventually sneaks into many large codebases.

What happened around this long transaction is the interesting bit! The queries piling up behind that transaction were mostly not blocked on its locks at all. They were simple reads, where InnoDB didn't need to wait for row locks; it reads a consistent snapshot instead. Building that snapshot means walking back through the version history of every row the open transaction had touched, and that history grew for fifteen straight minutes.

This caused reads that normally took milliseconds to start blowing through their 90-second execution ceilings. The application retried them in a tight loop. Within minutes, more than ten thousand requests were piled up inside the storage engine. Processing each required reconstructing ever-longer version chains, and the resulting page reads outpaced the buffer pool's ability to free memory.

Now requests that had nothing to do with the locked rows, that touched entirely different tables, began failing too. The otherwise correctly sized buffer pool suddenly became too small to serve the crowd of queries.

At PlanetScale, we run our MySQL databases with Vitess, whose configured transaction timeout eventually killed the long-running transaction. Its locks were released, and the sixteen-minute backlog drained in about thirty seconds.

One slow transaction should not take down a database, and much of Vitess's plumbing is built around avoiding exactly this situation. So what went wrong? This was caused by the combination of the single long-running query and the ten thousand requests allowed in after.

Migration from Cloud SQL to Vitess

Some context on how a seemingly healthy database could end up like this in the first place.

This workload had recently been migrated onto PlanetScale from Cloud SQL, which ran Managed Connection Pooling, a thread-pool-style layer, in front of the database. A thread pool caps how many statements execute at once, commonly around a thousand, and queues the rest. A client arriving when the pool is full waits for a slot, usually for milliseconds, occasionally for a few hundred milliseconds. This cap can protect InnoDB's internals even under extreme load.

On PlanetScale, every MySQL shard sits behind a Vitess proxy called vttablet. These have a transaction pool, which caps how many transactions can be open against MySQL at once. Unlike a thread pool, when this pool fills, requests wait a set amount of time and then fail with an error.

For this workload, that small difference made the issue way worse. Errors propagated up an application stack that had never needed to handle them (since the Cloud SQL thread pool queued requests, pool-full errors effectively didn't exist there).

The solution that was initially tried was to raise the transaction pool cap and increase the timeout.

The pool was increased to ten thousand, a number chosen not by sizing but because it made the errors stop. However, this now limited Vitess's ability to control the backpressure between the application and the database storage engine.

Why high concurrency reduces MySQL throughput

Picture database transactions as items flowing down a conveyor belt (Factorio, anyone?). If they never interacted, throughput would scale in a straight line: twice the items in flight, twice the work done. That is the dream of a perfectly shared-nothing system, but rarely is that achieved. Somewhere on the belt there is a junction where multiple conveyor belts meet. In MySQL, this is the equivalent of a hot row, a latch, a CPU run queue, etc.

Below is a simple playground to visualize how request rate and queuing impact latency. Adjust the ARRIVALS / S slider to see the impact.

Little's Law describes this well. In steady state, N = X * W. Work-in-flight equals throughput times the time each request spends inside the database. Rearranged, X = N / W.

Adding in-flight requests raises throughput only as long as query execution time holds steady. If each new arrival stretches the execution/wait time of all the other queries within the database, the denominator now grows along with the numerator.

Just how badly can that go? Gunther's Universal Scalability Law splits the cost of concurrency into three terms:

                 γN
X(N) = ───────────────────────────
        1 + α(N−1) + βN(N−1)

• N is how much work you allow to run at once.
• γ is the ideal single-request throughput: the linear scaling coefficient if requests never interacted.
• α (contention) is the cost of taking turns for a shared resource.
• β (coherency) is the cost of requests making each other slower. Put differently, the work a database must do to keep shared state consistent across everything in flight.

Note the multiplier N(N−1), which grows with the number of pairs of requests. When N doubles, this coherency cost roughly quadruples.

Past a critical point, N_max = √((1−α)/β), β dominates, and total throughput actually starts to reverse. Gunther calls it retrograde scaling: beyond the peak, every request you admit requires coordination overhead and makes everything slower.

Let's look at how the problem from earlier maps to this equation. The row locks were α. The transactions that wanted those rows had to take turns, and no amount of concurrency changed the queue's drain rate. β, the real issue, was InnoDB trying to process so many queries at once: ten thousand concurrent snapshot reads, each made more expensive by the version history every other in-flight transaction was generating. The per-request cost rose with the increase in in-flight requests.

This is why focusing on "lock contention" does not explain the issue. The lock is just one of many potential junctions. It is the damage around the junction, the mutual slowdown of everything admitted past it, that scales with the square of concurrency and turns a slow batch job into a wider outage.

Reducing pool size and implementing queuing

To solve this, we implemented the reverse of the change that set this up, in both dimensions at once:

1. Reduce the Vitess transaction pool size from ten thousand to roughly the thousand the old thread pool used. We now did this consciously and with evidence-backed proof: that workload had run successfully for years with a pool of that size.
2. Instead of erroring after waiting, a transaction arriving at a full pool now queues for a slot with a longer timeout.

In other words, we configured Vitess's vttablet layer to mimic the old thread pool's behavior.

For this to be safe, two things must be true. The clients have to tolerate periodic waiting during bursts and the wait itself has to be bounded, so that a truly undersized system announces itself with timeouts at the queue rather than accumulating latency forever. Both are true for this workload.

The morning after rolling out this new configuration, it met its first test. A traffic burst arrived on a different Vitess shard of the same database, one that handles an order of magnitude more steady traffic. At peak, the pool handled around twenty-five thousand transaction pool slot requests per second against its thousand-odd available slots. Here are the results, in the same view we opened with:

minute   slot requests/s   errors/min   queries/s
  0             3,000           0         58,000
 10             9,000           0         61,000
 20            26,000           0         60,000   <- peak pressure, throughput flat
 30            17,000           0         59,000
 40             8,000           0         62,000
 50             4,000           0         60,000
 60             2,000           0         57,000   <- burst drained, queue empty

During the earlier incident, errors climbed to 1,400 a minute and throughput fell to a tenth of what the workload requested.

With the new and improved configuration, the application experienced no disruption and QPS didn't drop.

Over the entire day, database health was much better:

• One rejected transaction. Despite slot requests/s at times jumping as high as 40,000, the queue absorbed effectively everything. Waits sat comfortably within the timeout.
• Fewer than two hundred statements executing inside MySQL at any instant. InnoDB managed the bursts well. Sixty thousand queries a second through fewer than two hundred concurrent slots is Little's Law again: about three milliseconds apiece.
• Lock contention didn't change. The hot-row pattern was just a minor contributor here. We were able to solve the problem without changing the lock contention dynamics.

This allowed the database to do less work at once, allowing it to do more work in total.

Below is another playground to visualize the difference between an open request flow and a gated one. Send bursts of traffic, simulate a locking event, and see the difference between a gated and ungated solution with the buttons below.

When to limit database concurrency

This is not a universal recommendation to tightly limit concurrency. Its applicability must be understood in the context of your workload.

• Workloads built on pessimistic locking and contended shared state.
• Hot rows updated by many workers.
• SELECT ... FOR UPDATE on popular keys.
• Long transactions holding locks while they do unrelated work.
• Counters, balances, job queues.

If that describes your workload, added concurrency might just decrease your throughput!

The principles described here are not MySQL-specific. Postgres can have similar problems, but also a similar solution: PgBouncer's transaction pooling exists to allow for more client concurrency than server concurrency. This is why our Postgres offering comes with a local PgBouncer, and can be configured with dedicated primary and replica PgBouncer instances too.

Building systems that handle backpressure

Backpressure can't always be avoided, so make conscious choices about the expected behavior. The answer is almost never to simply raise a limit out of reach, but to fix what happens when that limit is reached.

Sometimes the fastest thing you can do for a busy system is to simply let less happen at once.

Backups should be taken often, and recovery needs to stay fast. With backups being such a vital part of keeping your data safe, you should know how they work.

The types of Postgres backups

There are three different ways to back up a Postgres database: Logical backups (pg_dump), file system backups, and continuous archiving.

Logical backups

pg_dump is a purely logical backup that exports a consistent snapshot of the database, and is the simplest way to back up a Postgres database. At its core, pg_dump reads the current state of the database and writes the data to a given output.

pg_dump can create dumps in either plain text as SQL statements (with SQL queries such as COPY or INSERT), or custom pg_dump formats designed for use with pg_restore.

Regardless of the format used, the data pg_dump outputs is instructions on how to rebuild the data in the database.

-- An abbreviated snippet of output from pg_dump in plain format
CREATE TABLE public.customers (
    id bigint NOT NULL,
    name text NOT NULL,
    email text NOT NULL,
    country text NOT NULL,
    created_at timestamp with time zone NOT NULL
);

ALTER TABLE public.customers OWNER TO postgres;

COPY public.customers (id, name, email, country, created_at) FROM stdin;
1	Customer 1	user1@example.com	UK	2020-01-02 00:00:00+00
2	Customer 2	user2@example.com	DE	2020-01-03 00:00:00+00
...

Reading, formatting, and compressing a few gigabytes is trivial, but in the terabytes+ this becomes increasingly unfeasible. Even if your system can take a successful pg_dump backup at 30TB, it consumes a lot of resources. User queries must contend with pg_dump for CPU and IO, and in worst case scenarios this brings the database to a crawl.

pg_dump connects to the database through a standard SQL connection, only saving the rows and schema inside the database. It does not save physical page layouts, dead tuples, stored index bytes, or cluster-wide roles/tablespaces. This generally keeps outputs from pg_dump smaller compared to other backup methods.

Logical backups and transaction wraparound

Logical backups leverage Postgres's MVCC support in order to take a consistent backup while the database is online. If you're not familiar with MVCC, read our blog on Postgres MVCC. The section on transaction horizons is particularly important here. When pg_dump starts, it opens a transaction in one of two ways:

-- pg_dump with the default settings
BEGIN;
SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY;
...

-- pg_dump --serializable-deferrable
-- Ensures transactions are serializable while the backup is being taken.
-- Generally not recommended for backups intended for disaster recovery.
-- see the --serializable-deferrable flag for pg_dump
-- https://www.postgresql.org/docs/current/app-pgdump.html
BEGIN;
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE;
...

Every transaction that modifies data gets assigned an internal id, referred to as its xid. Rows in the database contain a hidden variable named xmin. xmin informs Postgres of the xid for the transaction that modified that row, while xmax is which transaction deleted, updated or locked that row.

Although pg_dump may not get its own xid, it pins a transaction snapshot of the active xids at that moment, giving the dump a consistent view of the database for its entire duration.

Remember that xids are 32-bit integers used to track the historical version of rows (or tuples, technically) across transactions. Tuples are the underlying representation of a row on disk. 32 bits can only represent a maximum of ~4 billion numbers, so what happens after we have more than 4 billion transactions?

Postgres treats xids as a circular range. Quoting from depesz's article on transaction wraparound:

Let's explain this bit on smaller numbers. Let's assume that we have the whole range of 0 to 10. Only 11 xids. If my current xid is 5, then xids 0..4 are in the past, and 6..10 are in the future. And if current xid is 3? Future is simple: 4..8. But past now contains two ranges: 0..2, and 9..10. Similarly, if current xid is 9, then past xids will be 4..8, and future would be 10, 0..3.

depesz

To keep the range circular and xids reusable, Postgres uses a freeze bit. This bit marks a tuple as older than any running transaction or snapshot, and should always be read as an "old" tuple.

When autovacuum runs, it internally calls an internal Postgres function GetOldestNonRemovableTransactionId(). This finds the oldest xid that is assigned to an open transaction, otherwise known as the transaction horizon, and autovacuum will go and freeze tuples "older" than that given xid.

A unique situation arises when the following criteria are met:

• A long running open transaction pins the transaction horizon
• More than ~2 billion new transactions have started after the long running transaction

If both of these points are met, new transactions start seeing old unfrozen tuples as if they were written by future transactions. This is called a transaction wraparound. At this point, Postgres will refuse new commands that would create xids, triggering read-only mode to ensure no data loss actually occurs.

Since pg_dump holds its snapshot open for the entire duration of the backup, it prevents the transaction horizon from advancing until the backup finishes. On a large database with sufficiently high throughput, pg_dump may inadvertently trigger this phenomenon bringing the cluster into read-only mode.

Above, we can see this in action. Each transaction has its own "future" and "past" based off its own xid. Every time autovacuum runs, it advances the oldest unfrozen xid. The blue section represents active xids that have not yet been frozen.

When the cluster goes into read-only mode in our visual, new transactions would potentially cause data loss. A new transaction would see unfrozen tuples created by the first transaction after pg_dump ran as if they were created by a future transaction that has not yet run.

On very large clusters (upwards of 30TB), this is a valid concern depending on the write throughput to the database. This isn't to say pg_dump is bad by any means. pg_dump is an extremely useful tool. The point here is demonstrating that for taking scheduled backups, pg_dump is most likely not the correct tool.

File system backup

Another approach is to perform a full file system backup. This is substantially faster than pg_dump for saving and restoring a backup.

A file system backup bypasses the database and directly copies the underlying file system.

In order for file system backups to be usable however, the database must be completely shut down. If the database is online during a backup, rows could be changed after the backup process has started, but before they have been copied, leaving the database in an inconsistent state.

This is explained in more detail in the next section, as this is what continuous archiving solves.

The other drawback to this type of backup (and logical backups) is that changes to data after a backup has been taken are not included in the backup. If data from between "now" and the most recent backup needs to be recovered, it isn't there.

The solution to this is to save the WAL, along with the file system, which is exactly what continuous archiving does.

Offline file system backups are generally a non-starter for production workloads, as no one wants to take down their database for hours to perform a daily backup.

Continuous archiving

The most powerful (and complicated) backup method is continuous archiving. Continuous archiving combines a file system backup with the write-ahead log (WAL).

This process can be done without downtime and generally lower impact on the CPU of the current system. This is what PlanetScale uses to facilitate Postgres and Neki backups.

The first step of taking a continuous archive backup has already started long before you take your first backup: archiving the WAL.

The WAL

The WAL is like your database's logbook. Every change, addition, or deletion of the data in your database is written down in the WAL.

Before Postgres makes any changes to data on disk, it first records the changes in the WAL. This ensures that upon a system crash, Postgres can restore to a consistent state by replaying the data mutations saved in its WAL. See the ARIES Wikipedia for more information on the algorithm and reasons behind WAL.

The WAL is by default saved in 16MB segments as binary files, however the size is configurable per cluster. These can be fed to tools such as pg_waldump yielding human readable output:

rmgr: Heap2       len (rec/tot):     60/  7136, tx:          0, lsn: 0/01C12FF8, prev 0/01C12F80, desc: PRUNE snapshotConflictHorizon: 771, nredirected: 0, ndead: 15
	blkref #0: rel 1663/16384/16387 fork main blk 0 (FPW); hole: offset: 452, length: 1116
rmgr: Heap        len (rec/tot):    119/  6443, tx:        778, lsn: 0/01C14BF0, prev 0/01C12FF8, desc: HOT_UPDATE old_xmax: 778, old_off: 44, old_infobits: [], flags: 0x10, new_xmax: 0, new_off: 85
	blkref #0: rel 1663/16384/16387 fork main blk 1 (FPW); hole: offset: 364, length: 1868
rmgr: Heap        len (rec/tot):    114/   114, tx:        778, lsn: 0/01C16538, prev 0/01C14BF0, desc: HOT_UPDATE old_xmax: 778, old_off: 45, old_infobits: [], flags: 0x10, new_xmax: 0, new_off: 86
	blkref #0: rel 1663/16384/16387 fork main blk 1
...
...
...
rmgr: Heap        len (rec/tot):    106/   106, tx:        778, lsn: 0/01C16F78, prev 0/01C16F08, desc: HOT_UPDATE old_xmax: 778, old_off: 82, old_infobits: [], flags: 0x10, new_xmax: 0, new_off: 108
	blkref #0: rel 1663/16384/16387 fork main blk 1
rmgr: Transaction len (rec/tot):     46/    46, tx:        778, lsn: 0/01C16FE8, prev 0/01C16F78, desc: COMMIT 2026-07-21 02:22:15.866163 UTC
rmgr: Heap        len (rec/tot):    107/   807, tx:        779, lsn: 0/01C17018, prev 0/01C16FE8, desc: HOT_UPDATE old_xmax: 779, old_off: 1, old_infobits: [], flags: 0x10, new_xmax: 0, new_off: 2
	blkref #0: rel 1663/16384/16399 fork main blk 0 (FPW); hole: offset: 508, length: 7492
rmgr: Heap        len (rec/tot):    107/   911, tx:        779, lsn: 0/01C17340, prev 0/01C17018, desc: HOT_UPDATE old_xmax: 779, old_off: 2, old_infobits: [], flags: 0x10, new_xmax: 0, new_off: 3
	blkref #0: rel 1663/16384/16399 fork main blk 8 (FPW); hole: offset: 500, length: 7388

There is a lot of information here, and remember this is only what pg_waldump is showing us. For understanding how continuous archiving works, there are a few parts of WAL we need to understand.

• len (rec/tot): rec is the size in bytes of the specific record data, while tot is the total bytes saved in WAL for this entry.
• lsn: The log sequence number (LSN) is the byte address of the given WAL line.
• FPW: Full page write (FPW) marks where a full page was written to the WAL. This is what causes rec != tot in the len field.
• rel: Tablespace / db / filenode, or more simply put, the location of a table file in a specific database. Tables are represented as filenodes, which are made up of pages, and saved as files on disk. See our MVCC blog for more detail.
• blk: The specific page of the filenode that is being updated or saved.
• desc: What action is happening. HOT_UPDATE, for example, is when an updated tuple with an associated index (in this case from an UPDATE SQL query) is saved to the same page it was originally on.

By default, Postgres enables a setting called full_page_writes. This setting is necessary for continuous archiving, and is forced on during a file system backup (also known as a base backup). This instructs Postgres to save an entire page to WAL the first time the page is updated after a checkpoint.

If the base backup is taken from a replica or standby node, then the operator must ensure full_page_writes is turned on for the primary. When the backup is taken directly on the primary, pg_basebackup (the common CLI tool for taking continuous archives) will force it on.

Every WAL line above containing FPW marks a point where Postgres saved the entire page (blk) within a given filenode (rel) to the WAL.

If you look closely at the rel, blk, and FPW on each line, you can see each first update to a given page (rel/blk) after a checkpoint triggers a new full page write. This is the mechanism that makes continuous archiving possible, as we will see later on.

Archiving data

Saving WAL segments often means streaming them to external storage systems such as S3. This can be done via tools such as wal-g, which are built for this exact purpose.

For any Postgres cluster (including Neki shards) running on PlanetScale, every primary has a wal-g sidecar streaming its Postgres WAL to S3.

When we start a backup, we begin by taking an online file system backup. Interestingly, we don't care if writes or mutations cause the file system to be in an inconsistent state, or even corrupted.

The data files we copy over can be thought of as "smeared" data on disk since they have potentially been "smeared" by queries that happened during our backup. We will take a look at exactly why we don't need to worry about smeared data with an example.

The start time is based on a LSN in the log, known as the backup's start_lsn, while the end_lsn is the point in the WAL at which the backup completed.

Say we start taking a backup at some time stamp. As the backup is copying over files, the database receives a SQL statement:

UPDATE users SET email_confirmed=true WHERE id = 314159

where email_confirmed was previously false. At this point, any one of three things could happen after the query executes.

1. The backup process has already copied the file containing the page for that row where email_confirmed was set to false. This is the best case; our backup is unaffected, and still in a consistent state. We have already copied the file, so we do not care if it is modified after the fact.
2. The backup has not yet read the page where that row lives, and our backup is now out of sync. The backup started at a time where email_confirmed was false, but it will save that data page while it is true!
3. The page is backed up at the same instant it was written to, and is now corrupted. The page now contains mangled bits, and later on we will be unable to read its contents properly.

How does Postgres know how to un-smear these data files?

This is where the full_page_writes capability of the WAL comes into play. After we have copied over the smeared file system backup from S3 to a new container for restoring, Postgres looks over the WAL reading every change between start_lsn and end_lsn.

Since full_page_writes was on when Postgres received the UPDATE users ... query mentioned above, it created a new heap tuple on disk as usual, and additionally logged a full image of the entire 8KB page where that row lives to the WAL.

Regardless of the state of that page from the file system backup, whether it is corrupted, in an inconsistent state, or perfectly healthy, we have enough information within the WAL to reconstruct it. The full page image in the WAL can replace the same page that was originally backed up if the ordinary WAL records are not enough to repair it.

Notice that the actual restored backup is not a snapshot of the database when the backup was started (like how pg_dump works), but a consistent version of the database when it finished backing up.

If we tried to restore the database to a point in-between start_lsn and end_lsn we would potentially be left with smeared data. The restore process must read through the full WAL from start to end before the restore can be marked as healthy.

Point in time recovery

The backups system described above is also what facilitates Postgres' ability to do point-in-time restores.

This is where the "Continuous" part of "Continuous Archiving" comes into play. Since the WAL is continuously saved to S3, we can restore any backup and continuously replay the WAL past our end_lsn up to any arbitrary point in time.

Replaying the WAL is generally more expensive than copying over the file system, however. This is why backups need to happen often, and is why at PlanetScale we back up every cluster every 12 hours by default. This keeps the WAL replay within a reasonable bound (half of a day) and keeps restores fast.

Continuously saving WAL is what also powers cluster resizes. When you resize your cluster on PlanetScale, the following is happening under the hood:

1. We bring up a new node with the desired specs
2. Copy over the file system backup (or mount an EBS volume if we are on an EBS backed node)
3. Replay WAL as close to the current time as possible
4. Stream remaining changes from the primary
5. Cutover to new node

In fact, this is the same process we use to take backups in the first place! We use the latest backup taken (usually around 12 hours ago), bring it online, catch it up to the primary, and finally save its file system as a new backup. This is all done on a throwaway node only used to take a new backup, meaning the entire process never contends with the primary for CPU, IO, or network bandwidth.

Backups at scale

Even with these techniques in place, extremely large databases can end up with backups that take 12+ hours to complete, resulting in ever growing windows between restore points.

When restore is needed, it could take days to get the new node back online. You can attempt a PITR from your most recent backup, but that could mean waiting substantially longer for WAL to not only stream over, but replay onto a new node.

We wrote previously on how Vitess handles backups at this scale, but what about Postgres?

Neki distributes Postgres data across shards so each shard can be backed up and restored independently.

This is 768 servers.

To some, that looks like a lot of computers. To those managing the infrastructure for apps with millions of customers, executing millions of queries per second, pretty normal. Products at this scale frequently require thousands of servers working in unison.

The most difficult infrastructure component to scale is almost always the database. A single database server cannot handle such demand, so we must spread the queries and data out across many servers with database sharding.

Database sharding is the best way to scale a Postgres or MySQL database for anything beyond a few terabytes of data. Let's look at how we go from a small single-node database, to one with a few terabytes spread across four shards, all the way up to one that is sharded across 768 servers and storing a petabyte of data.

Growing pains

To understand why sharding is a necessary part of scaling relational databases, we must understand the bottlenecks of less scalable approaches.

Consider first a simple application architecture.

Most applications you've ever used function in this way, or at least did early in their existence. The software running on a client device connects to an app server over the internet. This app server lives in a data center and handles authentication, page loads, and all the server-side logic for how your application behaves. All the persisted data like user accounts, posts, settings, and messages get stored in and retrieved from the database server (where "database server" is typically Postgres or MySQL, though the focus of this article is Postgres).

Even with a large database servers (10s of CPU cores, 100s of gigabytes of RAM) bottlenecks arise pretty quickly. Typically, it is either CPU constraints due to high query volume, or I/O constraints (IOPS) due to a high volume of reads and writes.

This is summed up nicely by the Universal Scalability Law:

In short, the USL states that resource contention causes scalability to grow sub-linearly with increasing resources, and at a certain point, incoherence causes performance degradation. This is true for Postgres, as with any software system attempting to scale out across many threads or processes on a larger server.

One way to solve this, at least in the short term, is leveraging read-replicas.

In this configuration, you maintain the original server as a primary and add additional replicas as shown above.

The primary sends a continuous stream of messages to every replica to ensure they stay up-to-date with the data changes on the primary. Writes (INSERT, UPDATE, DELETE) can only go to the primary. If writes were allowed to any server, we could end up with conflicting data. Solving this requires complex and slow consensus algorithms, which is possible, but in most cases not ideal for optimal performance.

However, app servers can send read (SELECT) queries to the replicas. Since most apps have a much higher percent of reads compared to writes, this provides a lot more scalability. (Replicas are also necessary for high availability and data durability, even if query traffic does not require them).

The database can scale to handle more traffic by adding replicas. An extreme example of this is OpenAI's use of 50 replicas on a single Primary.

It turns out, scaling servers vertically (increasing CPU / RAM) and adding replicas can only take you so far. There are several bottlenecks that cannot be solved in this way

1) Writes limited to one server

With high enough write volume, no amount of additional read-only replicas will alleviate an issue. Before Postgres can acknowledge a committed write, it must record the change in its write-ahead log (WAL) and flush that log to durable storage. The WAL is a shared resource amongst all connections on the primary. This is essentially a single write bottleneck across your entire database, even if you have tens of replicas.

2) Replicas do not increase data capacity

A replica is a full copy of the primary's data, including all indexes. Adding replicas gives us more places to run reads, but it does not distribute the data.

3) Backups

Backups are an important part of data durability and RPO / RTO guarantees. Taking a backup of a large, monolithic database to object storage can take hours or even days due to the bandwidth limitations of node-to-storage communication. This is unacceptably long for many organizations that rely on frequent and validated backups.

The most proven way to handle this is sharding.

Sharding, with a "d"

Sharding solves these three bottlenecks by distributing the data and queries across many distinct primaries. For data, it is useful because a single node can only store so much and is limited on write throughput. For queries, this is useful because the network interconnects and CPUs can only process so many queries at a time.

Sharding is useful at all scales past a few terabytes of data. For example, with 2 terabytes of data, we may choose a setup with four shards, each storing 500 gigabytes and handling 1/4th of the total query traffic. When we needed to store a petabyte of data (one million gigabytes), we'd need many more shards. In this case, we can use 256 shards, each with a primary + 2 replicas, and each responsible for storing ~4 terabytes. This requires 256 * 3 = 768 servers!

Without a good system in place, this adds significant complexity to our app's backend. With so much going on, how does the system...

• Decide which data goes to which server?
• Decide which queries go to which server?
• Handle queries that need to talk to multiple shards simultaneously?
• Take backups across this spread-out database?
• Monitor system-wide health?
• Respond to a failing server?

There's a lot that could be said in addressing each one of those concerns. But the question to address here in this article is the following:

How can these 768 servers look like 1 cohesive database to our apps?

We want to allow the application servers to go from interacting with a complex system, like this:

To instead interacting with it over a single connection string, making it appear as if it's interfacing with one large, scalable database:

While in reality, utilizing tens or hundreds of shards. Neki for Postgres and Vitess for MySQL solve this. Let's see how.

The proxy layer

The most important amongst several critical pieces here is the proxy layer.

Proxies are middleware servers that sit between two services. In our case, these two services are the application servers and database servers.

Proxies are frequently used with Postgres databases. Even when there's no sharding, they are useful for connection pooling and request queuing. For regular (unsharded) Postgres, PgBouncer is a popular proxy that people use to multiplex 1000s of app connections across fewer direct Postgres connections.

PgBouncer has a simple goal. It's built to accept a large number of connections from many clients and route them through a smaller pool of connections that it continually maintains with Postgres. The query queuing is useful for traffic surges and during database failover, so requests can resume when the new primary comes online. We have a whole blog on PgBouncer if you want to learn more.

Sharding Postgres requires an even more sophisticated proxy. The biggest difference is that, in addition to multiplexing and buffering, the proxy must understand how data is distributed across servers and route SQL queries to the correct shards. Because of this, we refer to it as a router.

When inserting data, the router must be aware of how data is to be distributed. This is known as the sharding strategy.

A common approach is to shard incoming rows based on a hash of an id column. When inserting row like this into the database:

 INSERT INTO users (id, username, email) VALUES
    (1, 'ada', 'ada@example.com'),
    (2, 'grace', 'grace@example.com'),
    (3, 'linus', 'linus@example.com'),
    (4, 'margaret', 'margaret@example.com'),
    (5, 'dennis', 'dennis@example.com'),
    (6, 'barbara', 'barbara@example.com'),
    (7, 'donald', 'donald@example.com'),
    (8, 'james', 'james@example.com');

Each of the four shards is assigned a range of IDs that it's responsible for storing, and the router sends the inserts to the correct shard. The insertions first get sent to the router, where it computes a hash of each ID, then forwards it along to the correct shard.

When it comes to reads, some queries are simple enough such that the router passes them along to a single shard.

SELECT email from user where id = 4;

In this case, all the router needs to do is have an internal mapping of which user IDs live in which servers, and forward that query on. Based on the example above, this would be the first (top) shard.

Some cases are more complex.

SELECT email FROM user
  WHERE id BETWEEN 3 AND 5;

Users with this range of IDs are spread out across several shards. The router must understand the data topology, create a plan for distributing the query to all shards that may contain matching results, aggregate the results back at the router, and send the full result set to the client.

Ultimately, this means the router itself must have a full query parser and routing planner built in.

The router must be able to perform query parsing, planning, connection pooling, and buffering, all within a single system. Complex software is hard to get right.

How does it know?

Every database is unique, with its own schema, tables, and query patterns. How then can a router generically know which data, and which queries, go where?

In both Neki and Vitess, these are specified via JSON files representing the data topology of the system. Vitess' VSchema and Neki's data topology give engineers a ton of flexibility to describe precisely how tables and queries should be distributed. Below is a simplified example of how we would specify a sharding scheme for a user table:

{
  "shard_indexes": {
    "user_hash": {
      "type": "hash"
    }
  },
  "tables": {
    "user": {
      "shard_by": "user_hash",
      "column": "id"
    }
  }
}

This metadata is stored in the router, and tells it that the user table is sharded on its id column using the user_hash shard index. This user_hash shard index uses the router's built-in value hashing. For each incoming row, it hashed the ID, and uses this to send it to the correct shard to be stored.

Since this is all communicated to the router via text and JSON, AI agents are great for configuration and optimization here.

Many proxies, one database

At a scale of 256 shards spanning 768 servers and millions of queries per second, we cannot route all of this traffic through a single proxy. We need many! Perhaps 10, perhaps 100, depending on the shape of the traffic.

We'd still like our apps to think of this as a single server. This is where a Network Load Balancer (NLB) helps.

NLBs have a simple job: Allow connections via a single host/IP, and assign each connection to one of many destinations. This is how traffic is distributed across the routers. Once assigned, a connection remains with the same proxy for its lifetime.

In some cases, an NLB is not necessary. Eliminating an NLB adds slightly more complexity to the app server's connection logic, as it will have to be aware of each router's host, but eliminates a network hop, keeping round-trip latency to a minimum.

The full picture

Now all the pieces are in place to make 768 servers storing 1,000 terabytes of data appear as a single, monolithic database to our apps.

1. An app server is told "connect to the database at mydb.pscale.com"
2. A DNS lookup is performed, returning the NLB's IP address: 123.152.100.4
3. The app requests to connect to the database at 123.152.100.4
4. This routes the connection first through the NLB, then to one of the N proxies
5. The app begins sending database queries, which go app -> NLB (optional) -> proxy -> shards. The complex routing logic is hidden from the application. (NLB not pictured below, for simplicity)

This example shows scaling up to 1 petabyte, but sharding should begin long before this scale. The precise recommendations depend on each database's size, schema, and QPS, but we recommend sharding Postgres and MySQL for anything beyond a few terabytes of data. That's the point where you typically begin hitting the bottlenecks described earlier: long backups, write bottlenecks, etc. If you are facing challenges scaling relational databases, Neki and Vitess are the solutions.

Vitess for MySQL has been used for over a decade to scale the world's biggest relational databases. We have years of experience operating large, sharded databases for our customers, and are the core maintainers of the Vitess project. Neki was developed by the same expert maintainers of Vitess, bringing an even more powerful sharding system to Postgres.

What about everything else?

We've only scratched the surface of everything sharding systems like Neki and Vitess provide. There are so many other interesting details. What's the best way to shard data? How do sharded databases handle failures? How do you change the number of shards? How do you take backups across 256 shards at the same time?

Stay tuned for more here. Follow our RSS feed or on X to stay in the loop.

Happy sharding.