Sharding and Partitioning Basics for Go Developers

When one PostgreSQL server stops being enough, two words arrive in the conversation: partitioning and sharding. They are often used as if they meant the same thing, and they solve different problems at different costs. Partitioning keeps your data in one database and divides a big table into smaller physical pieces. Sharding puts different slices of your data on different servers and makes the application responsible for finding them. This article explains both from a Go developer's point of view, shows partitioning working on PostgreSQL 18.6, and measures what a change in the number of shards does to where keys live.

It is part of the database cornerstone. If reads are the bottleneck, read replicas come first, and multi tenant layouts have their own trade-offs in multi tenant databases.

Partitioning: one database, many tables

A partitioned table looks like a single table to queries and is stored as several. PostgreSQL supports three schemes: range (by date, the common one for logs and events), list (by region or status) and hash (by a key, to spread rows evenly). The lab uses hash partitioning on customer_id:

package main

import (
	"context"
	"database/sql"
	"fmt"
	"hash/fnv"
	"log"
	"strings"

	_ "github.com/jackc/pgx/v5/stdlib"
)

const dsn = "postgres://postgres:[email protected]:55432/lab?sslmode=disable"

// modShard places a key with a hash and a remainder.
func modShard(key string, n int) int {
	h := fnv.New64a()
	h.Write([]byte(key))
	return int(h.Sum64() % uint64(n))
}

// jump is Google's jump consistent hash: growing from n to n+1 shards moves about 1/(n+1) of the keys.
func jump(key uint64, n int) int {
	var b, j int64 = -1, 0
	for j < int64(n) {
		b = j
		key = key*2862933555777941757 + 1
		j = int64(float64(b+1) * (float64(int64(1)<<31) / float64((key>>33)+1)))
	}
	return int(b)
}

func jumpShard(key string, n int) int {
	h := fnv.New64a()
	h.Write([]byte(key))
	return jump(h.Sum64(), n)
}

func main() {
	ctx := context.Background()
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()
	for _, q := range []string{
		"drop table if exists orders620",
		"create table orders620 (id bigint not null, customer_id int not null, total int not null, primary key (customer_id, id)) partition by hash (customer_id)",
		"create table orders620_p0 partition of orders620 for values with (modulus 4, remainder 0)",
		"create table orders620_p1 partition of orders620 for values with (modulus 4, remainder 1)",
		"create table orders620_p2 partition of orders620 for values with (modulus 4, remainder 2)",
		"create table orders620_p3 partition of orders620 for values with (modulus 4, remainder 3)",
		"insert into orders620 select i, i % 1000, i from generate_series(1, 200000) i",
		"analyze orders620",
	} {
		if _, err := db.ExecContext(ctx, q); err != nil {
			log.Fatal(err)
		}
	}

	fmt.Println("rows per partition (hash of customer_id into 4):")
	for p := 0; p < 4; p++ {
		var n int
		_ = db.QueryRowContext(ctx, fmt.Sprintf("select count(*) from orders620_p%d", p)).Scan(&n)
		fmt.Printf("  orders620_p%d %d\n", p, n)
	}

	plan := func(q string) {
		rows, err := db.QueryContext(ctx, "explain (costs off) "+q)
		if err != nil {
			log.Fatal(err)
		}
		defer rows.Close()
		var lines []string
		for rows.Next() {
			var l string
			_ = rows.Scan(&l)
			lines = append(lines, l)
		}
		fmt.Println("  " + strings.Join(lines, "\n  "))
	}
	fmt.Println("plan with the partition key in the where clause:")
	plan("select sum(total) from orders620 where customer_id = 42")
	fmt.Println("plan without it (every partition is searched):")
	plan("select sum(total) from orders620 where id = 42")

	// Application level sharding: what a change in the number of shards moves.
	const keys = 100000
	moved := func(place func(string, int) int, from, to int) float64 {
		n := 0
		for i := 0; i < keys; i++ {
			k := fmt.Sprintf("customer-%d", i)
			if place(k, from) != place(k, to) {
				n++
			}
		}
		return 100 * float64(n) / keys
	}
	fmt.Println("share of 100,000 keys that change shard when the shard count grows:")
	for _, step := range [][2]int{{4, 5}, {4, 8}, {8, 9}} {
		fmt.Printf("  %d -> %d shards: hash modulo moves %5.1f%%, jump consistent hash moves %5.1f%%\n",
			step[0], step[1], moved(modShard, step[0], step[1]), moved(jumpShard, step[0], step[1]))
	}
	counts := make([]int, 4)
	for i := 0; i < keys; i++ {
		counts[jumpShard(fmt.Sprintf("customer-%d", i), 4)]++
	}
	fmt.Println("keys per shard with jump hash over 4 shards:", counts)
}

rows per partition (hash of customer_id into 4):
  orders620_p0 51800
  orders620_p1 46800
  orders620_p2 55200
  orders620_p3 46200
plan with the partition key in the where clause:
  Aggregate
    ->  Bitmap Heap Scan on orders620_p2 orders620
          Recheck Cond: (customer_id = 42)
          ->  Bitmap Index Scan on orders620_p2_pkey
                Index Cond: (customer_id = 42)
plan without it (every partition is searched):
  Aggregate
    ->  Append
          ->  Index Scan using orders620_p0_pkey on orders620_p0 orders620_1
                Index Cond: (id = 42)
          ->  Index Scan using orders620_p1_pkey on orders620_p1 orders620_2
                Index Cond: (id = 42)
          ->  Index Scan using orders620_p2_pkey on orders620_p2 orders620_3
                Index Cond: (id = 42)
          ->  Index Scan using orders620_p3_pkey on orders620_p3 orders620_4
                Index Cond: (id = 42)
share of 100,000 keys that change shard when the shard count grows:
  4 -> 5 shards: hash modulo moves  80.0%, jump consistent hash moves  20.0%
  4 -> 8 shards: hash modulo moves  50.0%, jump consistent hash moves  50.1%
  8 -> 9 shards: hash modulo moves  88.9%, jump consistent hash moves  11.2%
keys per shard with jump hash over 4 shards: [25219 24754 24858 25169]

create table orders620 (...) partition by hash (customer_id) declares the parent, and four partition of ... for values with (modulus 4, remainder N) statements create the pieces. 200,000 rows were inserted through the parent, and PostgreSQL placed each row in a piece itself. The counts per partition were 51,800, 46,800, 55,200 and 46,200: roughly even, as a hash should give, with the spread of a thousand customers that each have 200 orders.

The primary key is (customer_id, id), not id alone. PostgreSQL requires the partition key to be part of every unique constraint on a partitioned table, because each partition's index can only guarantee uniqueness inside itself. That is the first sign of what partitioning costs: global uniqueness on any other column is gone, unless the key includes it.

Pruning, the point of it

The benefit shows in the plans. A query with the partition key in its where clause:

• select sum(total) from orders620 where customer_id = 42 was planned against one partition, orders620_p2. The other three were never touched. This is partition pruning.
• select sum(total) from orders620 where id = 42, without the key, searched all four, an Append over four index scans. Each of them is fast here, and at 200 partitions it is 200 index lookups for a query that wanted one row.

So partitioning pays when most queries carry the partition key, and it costs something when they do not. Choose the key from the access pattern, not from the data's shape.

What partitioning is good for:

• Dropping old data instantly. Range partitions by month turn "delete last year's rows" (a huge, slow delete that bloats the table) into drop table orders_2025_10, which is immediate and leaves nothing to vacuum.
• Keeping the hot part small. Indexes on the newest partition fit in memory while old partitions sit cold.
• Maintenance in pieces. Vacuum, reindex and backups work on smaller objects.
• Parallelism in some plans, with a modest effect.

What it does not do: it does not add servers, so write throughput and CPU are the same one machine's. If the bottleneck is the disk, memory or CPU of a single server, partitioning moves it around but does not remove it. It also adds planning cost per query and a migration to get existing data in, which for a live table is a project of its own; see zero downtime migrations.

Sharding: many databases

Sharding divides the data across servers, so each holds a fraction and takes a fraction of the traffic. Writes scale, storage scales, and the application becomes part of the database: it must know which shard holds a key. The simplest router hashes a shard key, here the customer, and picks a server:

• shard = hash(customer_id) % n is the common first version.
• Every query for one customer goes to one shard. Queries that span customers fan out to all of them and merge in Go.

The costs are large, and they are why sharding comes last:

Before sharding, exhaust the options that keep one primary: indexes, query fixes, caching, read replicas, a bigger machine and partitioning. A single modern server with fast disks handles terabytes and tens of thousands of queries a second. Teams reach for sharding years before they need it and pay the cost every day.

Adding a shard

The decision that gets made once and is hard to change is how keys map to shards, because the number of shards will grow. The lab measures that. It takes 100,000 customer keys and counts how many land on a different shard when the count changes:

With hash % n, adding one shard to four sends 80 percent of all data to a different server: nearly everything must be copied. A consistent scheme moves only what it has to: 1 in 5 keys to make the fifth shard its fair share. The lab uses jump consistent hash, a short algorithm from Google that needs no table and spreads keys evenly (25,219, 24,754, 24,858 and 25,169 per shard over four). The catch is that it numbers shards from 0 to n and only supports growing at the end, which suits most services.

Doubling, 4 to 8, moves half the keys under either scheme, because modulo with a power of two is already consistent for doubling. That is why teams often start with many logical shards (say 256) mapped onto a few servers, and later move whole logical shards between servers without rehashing any key. If you shard, do it that way: fix a large number of logical shards on day one, and keep a small table that says which server holds which.

Choosing a shard key

A good key is in almost every query, spreads load evenly and keeps related data together. A customer or tenant id is the usual answer for business software: the customer's orders, invoices and users live together, so most queries touch one shard. Bad keys are low cardinality (a country, with one giant shard), time (every new write hits the newest shard) and anything users do not supply on most requests. A key that creates a hot shard, one customer with a thousand times the traffic of the rest, needs special handling, like moving that customer to a shard of its own.

What to do in Go

Keep the routing in one place. A Shards type that holds the pools and a For(key) method is all the application needs, and every repository takes a pool from it, as in the repository pattern. Never let the shard count leak into handlers. Include the shard key in every table and every query by design, generate globally unique ids so rows can move, and write the code so that a cross shard query is visible in a review, because it is expensive. Also keep the option of not sharding: if the code goes through a router that returns the same pool for every key, you can test with one database and deploy with many.

The syntax and the limits are in the PostgreSQL manual's chapter on table partitioning, which covers the unique key rule and pruning, and the jump hash algorithm is described in its authors' paper, A Fast, Minimal Memory, Consistent Hash Algorithm.