Splitting a table into shards

A table can be split by the identities of its records into shards. Each shard is a range of its own: it is logged on its own, it can be led by its own node, and a peer can subscribe to it alone. This is how writes to one table are taken by more than one machine.

Splitting arrived in 0.4.0-beta, and a change feed that follows a split table in 0.7.0-beta. A table is split when it is declared, and from 0.19.0-beta its map can be changed while it serves — a shard split again, or two neighbours merged — by a statement an operator runs. From 0.24.0-beta the cluster can do that itself within bounds the table states, and generated identities can be spread over every shard instead of landing in the last one.

Declaring the shards: SPLIT AT

DEFINE TABLE orders (total int, customer string) IDENTITY uuid SPLIT AT 'g', 'p';
INFO FOR TABLE orders;

The points name where each shard begins. Two points make three shards, numbered from 1 in key order: 1 holds everything before 'g', 2 holds 'g' up to but not including 'p', and 3 holds 'p' and everything after. A point is written exactly as the identity it bounds would be addressed — 100, 'm', uuid '…' — and the order is the order a span read walks: every integer before every text, every text before every uuid.

INFO FOR TABLE reports the map, each bound in the spelling the clause takes back and NONE at the open ends, and the definition it carries includes the SPLIT AT, so a table re-created from it is split the same way.

Four declarations are refused, each by name:

• SplitNeedsGeneratedUuid — SPLIT AT on a table numbered by a counter. The counter is one value the whole store shares, so every insert would route through one place. Declare the table IDENTITY uuid, or name each record yourself.
• SplitPointsOutOfOrder — points that do not ascend strictly, or a point written twice. The list is refused rather than sorted, because a sorted list would accept a boundary written in the wrong place and say nothing.
• SplitOnAKindThatIsNotRecords — SPLIT AT on a DEFINE TABLE that is not a plain table of records: an edge table, or a graph's node table, which a walk reaches from its neighbours and has no span to confine to one shard.
• SplitPointIsNotWritten — a point supplied as a parameter (SPLIT AT $from). A shard map is a declaration somebody reviews and a boundary every later write is routed by, so its points are written in the statement.

The other kinds — a collection, a space, a bucket, a queue, a series, a topic, a vault, a vector or geo store — have no SPLIT AT in their grammar at all, so the clause is refused where it is written, as an unexpected word, before any of the refusals above can apply.

Where generated identities land. IDENTITY uuid mints UUID v7, whose leading bits are a timestamp, so every record the store names lands in the last shard: splitting over generated identities is sharding by time. That keeps old records apart from new ones; it does not spread writes. To spread writes, use SPREAD (next), or name records by a key whose leading part varies — a tenant, a region, an account — and split on that.

Spreading new records over every shard: IDENTITY uuid SPREAD

DEFINE TABLE orders (total int) IDENTITY uuid SPREAD SPLIT AT '40', '80', 'c0';

A table that names its records with SPREAD begins each identity with a bucket — two hex digits taken from the UUID's random part — so a new record is orders:'3f:01926f…' and new records land evenly over the 256 buckets, each bucket keeping creation order. The split points choose how many shards share the buckets: '40', '80', 'c0' is four shards with a quarter each. A read on a node holding part of the table fans out per shard, not per bucket, so a bucket costs nothing at read time. What it gives up: ORDER BY id is no longer creation order across buckets.

• SpreadNeedsGeneratedUuid — SPREAD on a table that does not declare IDENTITY uuid.

With PARTITION BY the bucket follows the region — 'de:3f:01926f…' — so a region spreads within its own span.

Changing the shards: ALTER TABLE … SPLIT AT · MERGE SHARD

A split table's map changes while it serves:

ALTER TABLE orders SPLIT AT 'm';
ALTER TABLE orders MERGE SHARD 4, 5;

SPLIT AT retires the shard holding the point and mints two in its place; MERGE SHARD retires two neighbouring shards, named by the numbers INFO FOR TABLE reports, and mints one. A new shard always takes the next number the table has never used. No record moves: a retired shard's log keeps what was written to it and is still collected, fed and backed up, and every write committed after the change is filed by the new map. INFO FOR TABLE reports the map's version and its retired shards.

Refused by name: NotASplitTable, SplitPointOnABoundary, ShardNotLive, ShardsNotAdjacent, and ShardNamedByAReplica — a row naming the shard by REPLICATES SHARD or LEADS SHARD; change the row first. A node gathering a shard from a node whose map differs is refused ShardMapMoved, which is retriable once the change has reached it.

A table that splits and merges itself

ALTER TABLE events SPLIT AUTOMATICALLY ABOVE 100000 RECORDS
    OR 500 WRITES PER SECOND
    MERGE BELOW 20000 RECORDS;
ALTER TABLE events SPLIT MANUALLY;

A split table can ask the cluster to keep its shards within bounds. The node leading the store line looks every few seconds: a shard holding more records than ABOVE, or taking more writes a second than OR … WRITES PER SECOND, is split — at the record half the bound into it, so a much larger shard splits again on later passes — and two neighbouring shards holding fewer than MERGE BELOW together are merged. At most one change per table per pass, and each is exactly the SPLIT AT or MERGE SHARD above: it is in the log, every node applies it, and the node logs it. SPLIT MANUALLY stops it; it is off unless asked.

• AutoSplitOnAnUnsplitTable — the table has no SPLIT AT.
• AutoSplitWouldOscillate — the merge bound is at least half the split bound, so the halves of a split would merge straight back.

Counting a shard walks its records up to the bound, so a pass costs each shard at most its bound. INFO FOR TABLE reports the bounds as auto_split, and on the store's leader — the node that counts — also sampled: each shard's records as the last pass counted them (complete: false means at least that many), its writes_per_second, and the last_act taken on the table. /metrics carries the counts as tessari_shard_records{table, shard}.

A new shard is led by the store line until a row places it — the cluster splits tables, and placing a shard on a node stays a statement you run (Who leads what).

A shard on its own node: REPLICATES SHARD

A subscription can name one shard, fully qualified and never relative to the session's USE:

DEFINE REPLICA part AT 'db-4.internal:9000'
    NODE '5d0c2e7a91f84b36a2e1c7d9f0a3b648'
    REPLICATES SHARD prod.shop.orders 2;

A shard the table does not have — or a table that is not split — is refused rather than stored, because a subscription to nothing is the quietest failure a cluster has. A peer subscribed to one shard is given the definitions above it — its database's, and every table's in it — so its readers can name what it holds. Definitions travel down; records never travel sideways.

A leader per shard: LEADS

A member row can name one shard its node stands to lead:

ALTER REPLICA b LEADS SHARD prod.shop.orders 2;

The shard becomes a line of its own, with its own leader, epochs and lease, so writes to two shards led by two nodes are taken by both at once. Making, moving, preferring and giving back a placement, and letting the store's leader even them out, is Who leads what. A transaction writing shards two nodes lead is refused SpansLeaderships unless it says COMMIT ACROSS LEADERS (Transactions across leaders).

One region per shard: PARTITION BY

A table can name the field that decides where its records live:

DEFINE TABLE customers (region string, name string)
    IDENTITY uuid
    PARTITION BY region
    SPLIT AT 'de', 'fr';

The store names each new record '<region>:<uuid v7>', so a region's records sort together and SPLIT AT over region names gives each region a shard. Every write is checked, on the node taking it and on every node applying it:

• PartitionNeedsGeneratedUuid — PARTITION BY without IDENTITY uuid.
• PartitionMismatch — the record's identity does not begin with its region and a :, or the region is absent, is not text, or holds a :. An UPDATE that changes the region is refused too: delete the record and create it again.

A read whose WHERE fixes the region — WHERE region = 'de', alone or beside other conditions joined by AND — reads only that region's records, and EXPLAIN names the one shard it touches. Place a region on a node with LEADS SHARD prod.shop.customers 2; a node that holds only its own region (REPLICATES SHARD …) answers a read of the whole table by gathering the rest.

Reading the whole table on a node that holds part of it

A node holding some shards answers a SELECT over the whole table, a span, one record or a WHERE by fetching the shards it lacks from their leaders and running the statement over its own records and the fetched ones together:

SELECT count(*) AS n FROM orders;

on a node holding shard 3 counts shards 1 and 2 as well. A point read or a span asks only for the shards it touches, and a span inside what the node holds asks nothing.

Some of the work goes to the leaders, under your visibility. The fields you may read travel with the question, and a leader takes every other field away before it evaluates anything — so a grant hiding a field hides it from work done on the leader exactly as from a local read. Only expressions that read nothing but the record travel (no subquery, other record, clock, generator or full-text match), and every value in them travels as a value, never as statement text:

• a WHERE — the leader sends only the records it keeps, and this node tests each one again;
• a LIMIT, behind a WHERE only when the WHERE travelled too — with no ORDER BY a shard past the bound is not asked at all, and under one each leader sends only its own first records by the statement's keys and this node merges them;
• a grouping read, whatever its folds — each leader sends what each group needs instead of the records, and the answer is the one a walk of the whole table gives, floats included: they are summed exactly, so the order the shards answer in changes nothing. count, sum, mean, min, max, variance and stddev send one state per group; median sends its values as runs of exact values and counts; collect sends its values in key order; and increase, rate and delta send a summary — how many samples, the first, the last and the rises — which this node joins end to end. When two shards' samples interleave in time the summaries cannot be joined, and every leader is asked once more, for the samples (from 0.32.0-beta). A value a fold refuses fetches the records as before;
• a join side or a FETCH into a shard this node lacks — the far side is narrowed to the keys the near side holds, and its records are fetched under your visibility, so a field you cannot read matches nothing.

The answer is complete and it is not one snapshot. Each fetched shard is its leader's state when it was asked. The answer says so with the note gathered, naming the shards fetched. For the same reason some reads are never gathered — a read inside a transaction or under VERSION, and the read an UPDATE or DELETE makes — and they are refused instead:

• NotHeldHere — a read that needs shards this node lacks, where gathering does not apply. It names the shards held elsewhere.
• NotGathered — a shard whose leader is not known, cannot be reached or declines. The whole read is refused, naming the shard and what refused. Nothing is answered in part.
• GatheredTooMuch — more than 100 000 records — or groups, for a read the leaders fold — would have to be held. A WHERE or LIMIT that travelled counts only what it kept, so SELECT count(*) answers over any number of records. A fold that holds its group counts what it holds beside it: a median its distinct values, a collect its values, a counter its samples only when it was asked for them — so a median over a few distinct values answers over any number of records, and a collect, whose answer is its values, still stops at 100 000. Otherwise read a span inside the shards this node holds, or read on a node holding the whole table.

Who may be asked. A shard's leader hands its records only to a member whose subscription holds part of the same table — any of its shards, or its database. Placing one shard of a table on a node therefore lets that node read the whole table; a follower confined to another tenancy fetches nothing.

Following a split table

A change feed over a split table — or over a database that holds one — reads the database's log and each shard's log, and delivers their changes in the order the node committed them. Nothing is missed and nothing is repeated, however the writes were spread over the shards.

What changes is how a subscriber says where it was. Each log counts its own positions, so two changes on one feed can carry the same sequence from two different logs, and no single number says how far the feed got. Every change on such a feed therefore carries a cursor — one position per log, opaque — and a subscriber resumes by sending the last one back. The node resumes after that change; unlike from, the cursor needs no +1. All five clients carry it from 0.2.0, and so does the WebSocket feed at /watch; a feed over a table that is not split is unchanged and never sends one. See the protocol.

A feed over a split table on a node that does not write all of it is refused by name, as is a cursor the node cannot read — a feed that silently skipped a shard's changes would look exactly like a quiet shard.

Upgrading followers to 0.4.0-beta

Two fixes in this release change what a follower holds, and neither can repair what an older follower already applied. Give every follower bootstrapped before 0.4.0-beta a fresh copy of the state:

• a user's grants now travel wherever the user goes — a namespace, database or shard follower used to hold a restricted user unrestricted;
• a follower now applies one writer's logs in the order the writer committed them — it used to be able to end at an older value than its leader.