Scaling MySQL: replicas, sharding and Vitess

A single MySQL primary takes every write your application makes, and one server has hard limits: the writes it can commit, the disk it holds, the memory its busy rows need. When one server is no longer enough, four moves come before sharding, each costing more to run than the last: tuning queries, read replicas, a cache and partitioning.

Scaling MySQL well means climbing those moves in order, not jumping to the hardest. The ladder below ends at sharding, where Vitess spreads one database across many MySQL servers. You will see what each move helps, why sharding is the last resort, and how Vitess routes a query to the right shard.

Scale in cost order, not straight to sharding

Reach for the cheapest fix first: tuning queries and adding a missing index costs nothing and can turn a slow page fast. When reads are the limit, a read replica is a second server that copies every change from the primary and answers read queries, so reads scale with the number of replicas. A replica adds no write capacity, though: every write still lands on the one primary, and a replica can lag behind it, a replication lagreplication lag: The time between a transaction's commit on the source and its commit on a replica. SHOW REPLICA STATUS estimates it in whole seconds as Seconds_Behind_Source; performance_schema keeps the timestamps of each transaction a worker applies. More in “Replication lag and parallel apply” readers must tolerate. A cache in front of MySQL answers hot reads from memory, so a repeated query need not reach the database at all.

When one table grows too large, partitioningRANGE partitioning: Partitioning in which each partition holds the values from the previous partition's bound up to its own VALUES LESS THAN bound; RANGE COLUMNS compares column values such as dates directly, and a last partition bounded by MAXVALUE takes everything above. More in “Partitioning in MySQL: pruning and retention” splits it into ranges on a single server, so a query scans only the partitions it needs and old data drops in one statement. Only shardingsharding: Splitting a table's rows across many database servers so that reads and writes spread out and each server holds only a slice of the data. splits the rows across many servers, so reads and writes both spread out and each server holds a slice of the data. Sharding is the last resort because it moves joins and transactions across servers: a query that does not carry the shard keyshard key: The column whose value decides which shard a row lives on; a query that filters on it with an equality can be routed to a single shard. must reach every shard. The table below weighs them.

Vitess puts a MySQL proxy in front of a fleet of servers

VitessAn open-source system that spreads one logical MySQL database across many MySQL servers and routes each query to the servers holding its rows; a graduated project of the Cloud Native Computing Foundation. lets a fleet of MySQL servers answer as one database: each statement goes to the servers that hold its rows, and the application never needs to know which they are. The application connects to a proxy, VTGateVTGate: The Vitess proxy that speaks the MySQL protocol to applications and routes each query to the shards that hold its rows., with the ordinary MySQL protocol, so a driver or ORM needs no change, and by default VTGate reports its version as MySQL 8.4.

Beside each MySQL server runs a VTTabletVTTablet: The Vitess process that runs beside each MySQL server, pooling connections, rewriting queries and managing that one server; a mysqld with its VTTablet is a tablet, and a shard has one or more., which pools connections to it, rewrites queries and guards it. A logical database is a keyspacekeyspace: In Vitess, a logical database the application sees as one, which can be split into shards spread across many MySQL servers., which can be split into shards, and the cluster’s layout lives in a topology server such as etcd. Step through one query below.

A vindex decides which shard a row lives on

A vindexvindex: A Vitess index that maps a column value to a shard, so the query planner can route a query to the shard or shards that hold its rows. maps a column value to a shard. The primary vindex sits on the shard key and, like a primary key, must be unique, so an equality on the shard key routes a query to exactly one shard. A query whose filter gives VTGate no vindex to route on becomes a scatter queryscatter query: A query whose filter gives a sharded database no shard key or other index to route on, so it is sent to every shard and the partial answers are merged.: VTGate sends it to every shard and merges the answers. The command-line tool vtexplain shows the difference without a running cluster: give it a schema, a VSchema and queries and it prints, per statement, which shards the planner picks. The run below shards customer and corder by a hash vindex on customer_id, over two shards and then four.

"""Shows how Vitess routes queries across shards, with vtexplain and no running cluster.

vtexplain reads a schema and a VSchema, builds the same query planner VTGate uses, and prints, for each statement,
which shard or shards the query goes to and the SQL each shard runs. It needs no mysqld and no etcd: it starts a
fake MySQL server of its own on a Unix socket in TMPDIR. macOS refuses a Unix socket path longer than 103 characters,
so this program sets TMPDIR to a short relative folder, "tmp", and runs vtexplain from its own folder; an absolute
TMPDIR inside a long folder path makes vtexplain panic.

Only what vtexplain writes to standard output, the per-shard plans, is printed. Its standard error, a few hundred
lines of time-stamped logs from the fake tablets, is dropped.

The keyspace "customer" is sharded with one hash vindex on customer_id, mapped onto both the customer table and the
corder table, so rows of a customer and that customer's orders hash to the same shard. The same four queries are
planned first over two shards and then over four, which is what a reshard from two to four shards changes.
"""

import os
import subprocess
import sys
from pathlib import Path


def toolchain(name):
    """The unpacked Vitess release, a folder toolchains/<name>: in a folder above this program or above the folder it
    runs in, or in one of the folders of a tmp folder there. The filesystem root is not searched."""
    seen = set()
    for start in (Path.cwd().resolve(), Path(__file__).resolve().parent):
        for parent in (start, *start.parents[:-1]):
            if parent in seen:
                continue
            seen.add(parent)
            found = parent / "toolchains" / name
            if found.is_dir():
                return found
            scratch = parent / "tmp"
            if scratch.is_dir():
                for folder in sorted(scratch.iterdir()):
                    found = folder / "toolchains" / name
                    if found.is_dir():
                        return found
    sys.exit(f"the toolchain {name} was not found")


VTEXPLAIN = toolchain("vitess-24.0.5") / "bin" / "vtexplain"

SCHEMA = (
    "CREATE TABLE customer (customer_id BIGINT NOT NULL, email VARCHAR(128), PRIMARY KEY (customer_id));\n"
    "CREATE TABLE corder (order_id BIGINT NOT NULL, customer_id BIGINT NOT NULL, total INT, PRIMARY KEY (order_id));\n"
)

VSCHEMA = """{
  "customer": {
    "sharded": true,
    "vindexes": { "hash": { "type": "hash" } },
    "tables": {
      "customer": { "column_vindexes": [ { "column": "customer_id", "name": "hash" } ] },
      "corder":   { "column_vindexes": [ { "column": "customer_id", "name": "hash" } ] }
    }
  }
}
"""

QUERIES = [
    "SELECT * FROM customer WHERE customer_id = 5",
    "SELECT COUNT(*) FROM corder",
    "SELECT c.email, o.total FROM customer c JOIN corder o ON c.customer_id = o.customer_id WHERE c.customer_id = 5",
    "SELECT * FROM corder WHERE total > 100",
]


def write(name, text):
    Path(name).write_text(text)
    print(f"$ cat {name}")
    print(text.rstrip("\n"))
    print()


def plan(shards):
    sql = "; ".join(QUERIES)
    env = dict(os.environ, TMPDIR="tmp")
    print(f"$ vtexplain --shards {shards} --vschema-file vschema.json --schema-file schema.sql \\")
    print('      --sql "$(cat queries.sql)"')
    done = subprocess.run(
        [str(VTEXPLAIN), "--shards", str(shards), "--vschema-file", "vschema.json",
         "--schema-file", "schema.sql", "--sql", sql],
        stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, cwd=".", env=env)
    print(done.stdout.rstrip("\n"))
    print()


def main():
    Path("tmp").mkdir(exist_ok=True)
    write("schema.sql", SCHEMA)
    write("vschema.json", VSCHEMA)
    write("queries.sql", "\n".join(q + ";" for q in QUERIES) + "\n")
    print("# Two shards: customer_id=5 hashes to one shard; COUNT(*) and the non-key filter scatter to both.")
    print()
    plan(2)
    print("# Four shards: the single-shard lookups still touch one shard; the scatters now touch all four.")
    print()
    plan(4)


if __name__ == "__main__":
    main()

$ cat schema.sql
CREATE TABLE customer (customer_id BIGINT NOT NULL, email VARCHAR(128), PRIMARY KEY (customer_id));
CREATE TABLE corder (order_id BIGINT NOT NULL, customer_id BIGINT NOT NULL, total INT, PRIMARY KEY (order_id));

$ cat vschema.json
{
  "customer": {
    "sharded": true,
    "vindexes": { "hash": { "type": "hash" } },
    "tables": {
      "customer": { "column_vindexes": [ { "column": "customer_id", "name": "hash" } ] },
      "corder":   { "column_vindexes": [ { "column": "customer_id", "name": "hash" } ] }
    }
  }
}

$ cat queries.sql
SELECT * FROM customer WHERE customer_id = 5;
SELECT COUNT(*) FROM corder;
SELECT c.email, o.total FROM customer c JOIN corder o ON c.customer_id = o.customer_id WHERE c.customer_id = 5;
SELECT * FROM corder WHERE total > 100;

# Two shards: customer_id=5 hashes to one shard; COUNT(*) and the non-key filter scatter to both.

$ vtexplain --shards 2 --vschema-file vschema.json --schema-file schema.sql \
      --sql "$(cat queries.sql)"
----------------------------------------------------------------------
SELECT * FROM customer WHERE customer_id = 5

1 customer/-80: select * from customer where customer_id = 5 limit 10001

----------------------------------------------------------------------
SELECT COUNT(*) FROM corder

1 customer/-80: select count(*) from corder limit 10001
1 customer/80-: select count(*) from corder limit 10001

----------------------------------------------------------------------
SELECT c.email, o.total FROM customer c JOIN corder o ON c.customer_id = o.customer_id WHERE c.customer_id = 5

1 customer/-80: select c.email, o.total from customer as c, corder as o where c.customer_id = 5 and c.customer_id = o.customer_id limit 10001

----------------------------------------------------------------------
SELECT * FROM corder WHERE total > 100

1 customer/-80: select * from corder where total > 100 limit 10001
1 customer/80-: select * from corder where total > 100 limit 10001

----------------------------------------------------------------------
<nil>

# Four shards: the single-shard lookups still touch one shard; the scatters now touch all four.

$ vtexplain --shards 4 --vschema-file vschema.json --schema-file schema.sql \
      --sql "$(cat queries.sql)"
----------------------------------------------------------------------
SELECT * FROM customer WHERE customer_id = 5

1 customer/40-80: select * from customer where customer_id = 5 limit 10001

----------------------------------------------------------------------
SELECT COUNT(*) FROM corder

1 customer/-40: select count(*) from corder limit 10001
1 customer/40-80: select count(*) from corder limit 10001
1 customer/80-c0: select count(*) from corder limit 10001
1 customer/c0-: select count(*) from corder limit 10001

----------------------------------------------------------------------
SELECT c.email, o.total FROM customer c JOIN corder o ON c.customer_id = o.customer_id WHERE c.customer_id = 5

1 customer/40-80: select c.email, o.total from customer as c, corder as o where c.customer_id = 5 and c.customer_id = o.customer_id limit 10001

----------------------------------------------------------------------
SELECT * FROM corder WHERE total > 100

1 customer/-40: select * from corder where total > 100 limit 10001
1 customer/40-80: select * from corder where total > 100 limit 10001
1 customer/80-c0: select * from corder where total > 100 limit 10001
1 customer/c0-: select * from corder where total > 100 limit 10001

----------------------------------------------------------------------
<nil>

A lookup of customer_id = 5 hit one shard, and the join stayed there because it joins on the shard key. The COUNT(*) and the filter on total, which is not the shard key, scattered to every shard. Planned over four shards, as after a reshard, the lookup stayed on one shard and each scatter widened from two shards to four.

Choosing a shard key, adding a lookup vindex, and resharding from two shards to four Pro: opens with a Pro, Premium or Ultimate pass

Moving a table into a sharded keyspace, and the SQL Vitess and PlanetScale restrict Pro: opens with a Pro, Premium or Ultimate pass

Practise

Work the exercise, then the self-check, weakest idea first.

def route(vschema, query):
    """Return how VTGate routes this query over the sharded keyspace:
    'single-shard', 'scatter' or 'unsharded'. See prompt.md."""
    # Replace this with your router.
    return "scatter"

"""Sample tests: two VSchemas, a shop and a multi-tenant product, and queries like those of the lesson's vtexplain run."""

from route import route

# A shop sharded by customer, with an unsharded product catalogue.
SHOP = {"customer": "customer_id", "corder": "customer_id", "product": None}
# A multi-tenant product sharded by tenant.
SAAS = {"account": "tenant_id", "invoice": "tenant_id"}


def test_lookup_on_shard_key_is_single_shard():
    q = {"tables": ["customer"], "equalities": {"customer": ["customer_id"]}}
    assert route(SHOP, q) == "single-shard"


def test_join_on_shard_key_is_single_shard():
    q = {"tables": ["customer", "corder"],
         "equalities": {"customer": ["customer_id"], "corder": ["customer_id"]}}
    assert route(SHOP, q) == "single-shard"


def test_count_with_no_filter_scatters():
    q = {"tables": ["corder"], "equalities": {}}
    assert route(SHOP, q) == "scatter"


def test_range_on_non_key_scatters():
    q = {"tables": ["corder"], "equalities": {}}
    assert route(SHOP, q) == "scatter"


def test_equality_on_non_key_scatters():
    q = {"tables": ["customer"], "equalities": {"customer": ["email"]}}
    assert route(SHOP, q) == "scatter"


def test_one_pinned_one_not_scatters():
    q = {"tables": ["customer", "corder"],
         "equalities": {"customer": ["customer_id"]}}
    assert route(SHOP, q) == "scatter"


def test_only_unsharded_table_is_unsharded():
    q = {"tables": ["product"], "equalities": {"product": ["sku"]}}
    assert route(SHOP, q) == "unsharded"


def test_sharded_plus_unsharded_uses_only_sharded():
    q = {"tables": ["customer", "product"],
         "equalities": {"customer": ["customer_id"]}}
    assert route(SHOP, q) == "single-shard"


def test_tenant_key_is_single_shard():
    q = {"tables": ["account"], "equalities": {"account": ["tenant_id"]}}
    assert route(SAAS, q) == "single-shard"


def test_tenant_without_key_scatters():
    q = {"tables": ["invoice"], "equalities": {"invoice": ["invoice_no"]}}
    assert route(SAAS, q) == "scatter"

Interview questions

Answer the warm-up aloud before reading its answer, then plan the other three.

Warm-up: what is a scatter query in a sharded database, and why is it usually slower than a single-shard query?

A scatter query is one that a sharded database cannot narrow to particular shards, so the proxy sends it to every shard and merges the answers. It happens when the query’s filter gives the proxy no shard key or other index to route on, so any shard might hold a match. It is slower for two reasons: the work is the sum across all shards, so a count costs about as much as on the unsharded table; and the query is only as fast as the slowest shard. Resharding widens the fan-out, so keep the busy queries on the shard key or add a lookup vindex.

Key takeaways

• Scale in cost order: tune queries and indexes, add read replicas and a cache for reads, partition to shrink the live set, and shard only when writes or data outgrow one server.
• Read replicas and a cache add read capacity but no write capacity, and both leave every write on the single primary.
• Vitess puts VTGate, a MySQL-protocol proxy, in front of a fleet of MySQL servers, and a vindex maps the shard key to a shard: an equality on the shard key reaches one shard, a query with nothing to route on scatters to all, and more shards widen that scatter.