Database Partitioning and Sharding: How They Work, When to Use Each, and the Trade-offs That Matter

Every database starts small enough that a single server handles everything comfortably. Then the data grows, the query load increases, and the single-server model starts showing its limits — slow range scans, lock contention, replication lag, backup windows that eat into write availability.

Partitioning and sharding are the two fundamental strategies for breaking a large dataset into smaller, more manageable pieces. They are related but distinct, and conflating them leads to architecture decisions that seem reasonable until production proves otherwise.

This post covers both precisely: what each strategy is, how every variant works under the hood, when to reach for each, and what you give up when you do.

The Fundamental Distinction

Before any mechanism, the definition:

Partitioning splits a table into smaller physical segments that are managed by the same database server. The database engine is aware of the partition structure and routes queries transparently. The application typically does not know or care that the data is partitioned.

Sharding splits data across multiple independent database servers (or clusters). Each server — called a shard — owns a subset of the data. The application (or a routing layer in front of it) must know which shard holds the data it wants and direct queries accordingly.

Partitioning:                      Sharding:

┌─────────────────────────┐        ┌────────────┐  ┌────────────┐
│     Single DB Server    │        │  Shard 1   │  │  Shard 2   │
│                         │        │  (Server A)│  │  (Server B)│
│  ┌──────┐  ┌──────┐     │        │            │  │            │
│  │Part 1│  │Part 2│     │        │  users     │  │  users     │
│  │Jan   │  │Feb   │     │        │  id 1–50M  │  │  id 50M+   │
│  └──────┘  └──────┘     │        └────────────┘  └────────────┘
│  ┌──────┐  ┌──────┐     │
│  │Part 3│  │Part 4│     │        Application must know:
│  │Mar   │  │Apr   │     │        "User 72M is on Shard 2"
│  └──────┘  └──────┘     │
│                         │        DB server does NOT know
│  DB engine routes       │        about other shards
│  queries transparently  │
└─────────────────────────┘

The consequences of this distinction are significant:

• Partitioning is a database-internal optimization. It improves query performance and maintenance without changing application architecture.
• Sharding is a distributed systems decision. It introduces the full set of distributed systems problems: no cross-shard transactions, no cross-shard joins, eventual consistency concerns, resharding complexity.

Start with partitioning. Graduate to sharding only when a single server — even well-partitioned — can no longer handle your workload.

Part 1: Partitioning

How the Database Engine Uses Partitions

When you query a partitioned table, the query planner uses partition pruning to eliminate partitions that cannot possibly contain rows matching the query’s filter conditions.

-- Table partitioned by order month
SELECT * FROM orders WHERE created_at BETWEEN '2025-01-01' AND '2025-01-31';

-- Without partitioning: full table scan or index scan across all data
-- With partitioning: planner prunes all partitions except January 2025
--   Execution plan reads ONLY orders_2025_01 — O(partition_size) not O(table_size)

This is the core performance benefit: queries with a filter on the partition key access only the relevant partition(s). The larger the table relative to a partition, the greater the speedup.

Secondary benefits:

• Partition-wise maintenance: VACUUM, ANALYZE, REINDEX can run on one partition at a time.
• Fast bulk deletes: dropping old data is DROP TABLE orders_2024_01 — an O(1) metadata operation — instead of DELETE FROM orders WHERE created_at < '2025-01-01' — an O(rows_deleted) write operation that generates WAL and triggers autovacuum.
• Partition-local indexes: each partition has its own index, which is smaller and therefore faster to scan and maintain.

Strategy 1: Range Partitioning

Data is divided based on ranges of a column’s values. Most commonly used with timestamps (partition by month or year) or sequential IDs.

-- PostgreSQL declarative partitioning
CREATE TABLE orders (
    id          BIGSERIAL,
    customer_id BIGINT NOT NULL,
    total       NUMERIC(12,2) NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL,
    status      TEXT NOT NULL
) PARTITION BY RANGE (created_at);

-- Create monthly partitions
CREATE TABLE orders_2025_01
    PARTITION OF orders
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

CREATE TABLE orders_2025_02
    PARTITION OF orders
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

CREATE TABLE orders_2025_03
    PARTITION OF orders
    FOR VALUES FROM ('2025-03-01') TO ('2025-04-01');

-- Default partition catches rows that don't match any defined range
-- (important: without this, inserts outside defined ranges fail)
CREATE TABLE orders_default
    PARTITION OF orders DEFAULT;

Automating partition creation with a scheduled job:

-- Function to create the next month's partition
CREATE OR REPLACE FUNCTION create_monthly_partition(
    base_table TEXT,
    partition_date DATE
) RETURNS VOID AS $$
DECLARE
    partition_name TEXT;
    start_date     DATE;
    end_date       DATE;
BEGIN
    start_date     := DATE_TRUNC('month', partition_date);
    end_date       := start_date + INTERVAL '1 month';
    partition_name := base_table || '_' || TO_CHAR(start_date, 'YYYY_MM');

    EXECUTE FORMAT(
        'CREATE TABLE IF NOT EXISTS %I
         PARTITION OF %I
         FOR VALUES FROM (%L) TO (%L)',
        partition_name, base_table, start_date, end_date
    );
END;
$$ LANGUAGE plpgsql;

-- Call this monthly, e.g., from pg_cron:
SELECT create_monthly_partition('orders', NOW() + INTERVAL '1 month');

When range partitioning excels:

• Time-series data (logs, events, metrics, financial records) where queries almost always filter by date range.
• Data with natural retention policies — drop the oldest partition to reclaim space instantly.
• Reporting workloads that query one or two months at a time.

Hotspot risk: All current writes go to the latest partition. If write throughput is very high, the current partition becomes a contention point. Mitigate by using sub-partitioning (partition by month, then by hash within each month) or by spreading writes across multiple shards.

Strategy 2: Hash Partitioning

Data is assigned to partitions by computing a hash of the partition key modulo the number of partitions. This distributes rows uniformly across partitions regardless of the key’s value distribution.

CREATE TABLE user_events (
    id          BIGSERIAL,
    user_id     BIGINT NOT NULL,
    event_type  TEXT NOT NULL,
    payload     JSONB,
    occurred_at TIMESTAMPTZ NOT NULL
) PARTITION BY HASH (user_id);

-- 8 partitions — use powers of 2 for easy resharding later
CREATE TABLE user_events_p0 PARTITION OF user_events
    FOR VALUES WITH (MODULUS 8, REMAINDER 0);
CREATE TABLE user_events_p1 PARTITION OF user_events
    FOR VALUES WITH (MODULUS 8, REMAINDER 1);
-- ... through p7
CREATE TABLE user_events_p7 PARTITION OF user_events
    FOR VALUES WITH (MODULUS 8, REMAINDER 7);

When hash partitioning excels:

• High-cardinality keys (user IDs, order IDs) with uniform access patterns.
• Write-heavy workloads where even write distribution across partitions eliminates hotspots.
• When you need parallelism — the query planner can scan partitions in parallel.

The trade-off: Range queries on the partition key are inefficient. WHERE user_id BETWEEN 1000 AND 2000 must check all partitions because the hash function scrambles ordering. Hash partitioning optimizes point lookups and uniform write throughput at the cost of range scan efficiency.

Strategy 3: List Partitioning

Data is divided by explicit enumerated values of the partition key. Useful for categorical data with a small, well-known set of values.

CREATE TABLE orders (
    id          BIGSERIAL,
    region      TEXT NOT NULL,
    customer_id BIGINT NOT NULL,
    total       NUMERIC(12,2) NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL
) PARTITION BY LIST (region);

CREATE TABLE orders_us
    PARTITION OF orders
    FOR VALUES IN ('us-east', 'us-west', 'us-central');

CREATE TABLE orders_eu
    PARTITION OF orders
    FOR VALUES IN ('eu-west', 'eu-central', 'eu-north');

CREATE TABLE orders_apac
    PARTITION OF orders
    FOR VALUES IN ('ap-southeast', 'ap-northeast', 'ap-south');

CREATE TABLE orders_other
    PARTITION OF orders DEFAULT;

When list partitioning excels:

• Data residency requirements: EU customer data must stay on EU servers (combine with tablespaces pointing to different storage paths or, better, with sharding).
• Tenant isolation in multi-tenant systems: one partition per tenant or tenant group.
• Categorical data where queries almost always filter by category.

The trade-off: Partition sizes can be unequal if some categories have dramatically more data than others. A “US” partition that is 10× larger than “EU” eliminates most of the benefit.

Strategy 4: Composite (Sub) Partitioning

Partitions are themselves partitioned — combining strategies for finer granularity.

-- Partition by range (year), then by hash (user_id) within each year
CREATE TABLE events (
    id          BIGSERIAL,
    user_id     BIGINT NOT NULL,
    event_type  TEXT NOT NULL,
    occurred_at TIMESTAMPTZ NOT NULL
) PARTITION BY RANGE (occurred_at);

-- Year-level partitions
CREATE TABLE events_2024 PARTITION OF events
    FOR VALUES FROM ('2024-01-01') TO ('2025-01-01')
    PARTITION BY HASH (user_id);

-- Sub-partitions within 2024
CREATE TABLE events_2024_p0 PARTITION OF events_2024
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE events_2024_p1 PARTITION OF events_2024
    FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE events_2024_p2 PARTITION OF events_2024
    FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE events_2024_p3 PARTITION OF events_2024
    FOR VALUES WITH (MODULUS 4, REMAINDER 3);

Result: Queries filtering by occurred_at prune to the year-level partition. Queries filtering by user_id within a year then target one of 4 hash sub-partitions. Point lookups (user_id = X AND occurred_at BETWEEN ...) read 1 of 4 sub-partitions of the relevant year — a very small fraction of the total table.

Partitioning: Indexes and Constraints

In PostgreSQL declarative partitioning, indexes and constraints are defined on the parent table and automatically applied to each partition:

-- Index on parent applies to all partitions
CREATE INDEX ON orders (customer_id, created_at);
-- Creates: orders_2025_01_customer_id_created_at_idx
--          orders_2025_02_customer_id_created_at_idx
--          ... (one per partition)

-- Unique constraints must include the partition key
-- (PostgreSQL limitation: uniqueness cannot be enforced across partitions)
CREATE UNIQUE INDEX ON orders (id, created_at);
-- NOT: CREATE UNIQUE INDEX ON orders (id); ← will fail

This is an important limitation: globally unique constraints cannot be enforced across partitions in most databases without including the partition key in the constraint. For a globally unique id, use a sequence or UUID at the application level and accept that the database does not enforce uniqueness across partitions.

Part 2: Sharding

Sharding moves from a database-internal optimization to an architectural decision that affects your entire application. It is the answer to questions partitioning cannot solve:

• Write throughput exceeds what one server can sustain even with SSDs.
• The dataset exceeds the storage capacity of one server.
• You need geographic data locality (data must be physically hosted in a specific region).
• Read replicas are saturated and you need more read parallelism than one primary can support.

The Sharding Key

The most consequential decision in sharding design is the sharding key — the value used to determine which shard owns a given row.

The ideal sharding key:

1. Has high cardinality — enough distinct values to distribute data evenly across shards.
2. Is present in most queries — queries without the sharding key must fan out to all shards.
3. Does not cause hotspots — access frequency should be roughly uniform across key values.
4. Does not change — changing a row’s sharding key requires moving it to a different shard (expensive).
5. Maps to the application’s access patterns — if most queries are “get everything for user X,” then user_id is the natural key.

Common sharding keys by use case:

Sharding Strategy 1: Range-Based Sharding

Each shard owns a contiguous range of sharding key values.

Shard 1: user_id  1       –  50,000,000
Shard 2: user_id  50,000,001 – 100,000,000
Shard 3: user_id  100,000,001 – 150,000,000
Shard 4: user_id  150,000,001 – MAX

// Shard routing for range-based sharding
type ShardRange struct {
    Min    int64
    Max    int64
    DSN    string
}

type RangeRouter struct {
    ranges []ShardRange // sorted by Min
}

func (r *RangeRouter) ShardFor(userID int64) (string, error) {
    for _, sr := range r.ranges {
        if userID >= sr.Min && userID <= sr.Max {
            return sr.DSN, nil
        }
    }
    return "", fmt.Errorf("no shard for user_id %d", userID)
}

Pros: Range scans on the sharding key (e.g., “all users with ID between X and Y”) touch only the relevant shard(s). Easy to reason about which data is where.

Cons: Sequential ID generation concentrates all new writes on the last shard (the same hotspot problem as range partitioning). User growth is not uniform — some ranges may grow faster than others, leading to unequal shard sizes over time. Rebalancing requires moving data across shards.

Sharding Strategy 2: Hash-Based Sharding

The sharding key is hashed, and the hash modulo the number of shards determines the target shard.

type HashRouter struct {
    shards []string // DSNs, indexed by shard number
}

func (r *HashRouter) ShardFor(userID int64) string {
    // Use a stable hash function — not Go's map hash (non-deterministic)
    h := fnv64(userID)
    shardIndex := h % uint64(len(r.shards))
    return r.shards[shardIndex]
}

func fnv64(id int64) uint64 {
    h := fnv.New64a()
    binary.Write(h, binary.LittleEndian, id)
    return h.Sum64()
}

The resharding problem: With 4 shards, user_id % 4 determines the shard. When you add a 5th shard, user_id % 5 redistributes ~80% of data to different shards. This requires migrating the majority of data — an expensive, risky operation.

The solution: consistent hashing.

Sharding Strategy 3: Consistent Hashing

Consistent hashing places both shard nodes and data keys on a virtual ring. Each key is owned by the nearest node clockwise on the ring. When a node is added or removed, only the keys between the new node and its predecessor need to move.

type ConsistentHashRouter struct {
    ring     []uint64            // sorted hash positions
    nodeMap  map[uint64]string   // position → DSN
    replicas int                 // virtual nodes per shard (for even distribution)
}

func NewConsistentHashRouter(shards []string, replicas int) *ConsistentHashRouter {
    r := &ConsistentHashRouter{nodeMap: make(map[uint64]string), replicas: replicas}
    for _, shard := range shards {
        r.addShard(shard)
    }
    sort.Slice(r.ring, func(i, j int) bool { return r.ring[i] < r.ring[j] })
    return r
}

func (r *ConsistentHashRouter) addShard(dsn string) {
    for i := 0; i < r.replicas; i++ {
        key := fmt.Sprintf("%s#%d", dsn, i)
        h := hashString(key)
        r.ring = append(r.ring, h)
        r.nodeMap[h] = dsn
    }
}

func (r *ConsistentHashRouter) ShardFor(userID int64) string {
    h := fnv64(userID)
    // Binary search for the first ring position >= h
    idx := sort.Search(len(r.ring), func(i int) bool {
        return r.ring[i] >= h
    })
    if idx == len(r.ring) {
        idx = 0 // wrap around
    }
    return r.nodeMap[r.ring[idx]]
}

Adding a shard: Insert the new node’s virtual positions into the ring. Only keys between the new node and its predecessor on the ring need to be migrated — roughly 1/n of total data, where n is the new total number of shards.

Virtual nodes (replicas): Each physical shard is represented by multiple points on the ring (100–200 is typical). This smooths out the key distribution — without virtual nodes, a small number of physical nodes places them unevenly on the ring, causing some shards to own significantly more keys than others.

Sharding Strategy 4: Directory-Based Sharding (Lookup Table)

A separate metadata store maps sharding key ranges (or individual keys) to shard identifiers. The application consults the directory for every request.

type DirectoryRouter struct {
    // In production: Redis cache in front of a database
    cache  *redis.Client
    db     *sql.DB
}

func (r *DirectoryRouter) ShardFor(ctx context.Context, tenantID string) (string, error) {
    // Check cache first
    dsn, err := r.cache.Get(ctx, "shard:"+tenantID).Result()
    if err == nil {
        return dsn, nil
    }

    // Cache miss: look up in metadata database
    err = r.db.QueryRowContext(ctx,
        "SELECT shard_dsn FROM shard_directory WHERE tenant_id = $1",
        tenantID,
    ).Scan(&dsn)
    if err != nil {
        return "", fmt.Errorf("shard lookup for tenant %s: %w", tenantID, err)
    }

    // Populate cache with a TTL matching your resharding frequency
    r.cache.Set(ctx, "shard:"+tenantID, dsn, 5*time.Minute)
    return dsn, nil
}

Pros: Maximum flexibility. A tenant can be moved to a different shard by updating one row in the directory — no rehashing, no ring recalculation. Hot tenants can be isolated on their own shards without restructuring the overall sharding scheme.

Cons: The directory is a critical dependency. If it is unavailable, no queries can be routed. Requires careful caching to avoid making the directory a latency bottleneck on every request. The directory itself must be highly available — often replicated across regions.

Where to Put the Routing Logic

The shard router can live in three places, each with different trade-offs:

Option A: Application-level routing

The application itself contains the routing logic and maintains connections to all shards.

type ShardedDB struct {
    router  Router
    conns   map[string]*sql.DB // DSN → connection pool
}

func (db *ShardedDB) QueryUser(ctx context.Context, userID int64) (*User, error) {
    dsn, err := db.router.ShardFor(userID)
    if err != nil {
        return nil, err
    }
    conn := db.conns[dsn]
    // execute query on the correct shard
    var u User
    err = conn.QueryRowContext(ctx,
        "SELECT id, email, name FROM users WHERE id = $1", userID,
    ).Scan(&u.ID, &u.Email, &u.Name)
    return &u, err
}

Pros: Low latency (routing decision is local). Full control over routing logic. Cons: Every service must implement routing. Schema changes and resharding require coordinated deployments across all services.

Option B: Proxy layer (Vitess, ProxySQL, PgBouncer + custom routing)

A proxy sits between the application and the database shards. The application connects to the proxy using standard SQL; the proxy routes queries to the correct shard.

Application  →  Vitess VTGate (proxy)  →  Shard 1 (MySQL)
                                       →  Shard 2 (MySQL)
                                       →  Shard 3 (MySQL)

Vitess (originally built at YouTube for MySQL) is the most mature solution here. It parses SQL, inspects the sharding key in the query, and routes to the appropriate shard — transparently to the application. It also handles cross-shard scatter-gather queries, connection pooling across shards, and online resharding.

Pros: Application sees a single database endpoint. Routing is centralised. Online resharding without application changes. Cons: Proxy is an additional network hop and failure domain. SQL compatibility is not always complete (complex queries may fail routing analysis).

Option C: Built-in distributed SQL (CockroachDB, YugabyteDB, Spanner)

Distributed SQL databases present a standard SQL interface but internally shard and replicate data automatically. The application uses them exactly like a single-server database.

-- CockroachDB: looks like PostgreSQL to the application
-- Internally: data is automatically split into ranges and distributed
CREATE TABLE orders (
    id          UUID DEFAULT gen_random_uuid() PRIMARY KEY,
    customer_id UUID NOT NULL,
    total       NUMERIC(12,2) NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- No partitioning DDL needed — CockroachDB manages range splits automatically
-- The application just queries normally
SELECT * FROM orders WHERE customer_id = $1;

Pros: Dramatically simpler application code. ACID transactions across shards. Automatic rebalancing. Cons: Higher latency per query (consensus protocol overhead — typically 2–5ms per write vs sub-millisecond for local writes). Vendor lock-in. More expensive at scale. Less control over data placement.

The Hard Problems Sharding Introduces

Cross-Shard Queries

A query that needs data from multiple shards must either:

1. Fan out — execute the query on all shards in parallel and merge the results in the application.
2. Denormalize — duplicate data that would otherwise require a cross-shard join.
3. Redesign — restructure the query to be shard-local.

// Scatter-gather: fan out a query to all shards
func (db *ShardedDB) SearchUsers(ctx context.Context, query string) ([]*User, error) {
    var (
        mu      sync.Mutex
        results []*User
        wg      sync.WaitGroup
        errs    []error
    )

    for dsn, conn := range db.conns {
        dsn, conn := dsn, conn
        wg.Add(1)
        go func() {
            defer wg.Done()
            rows, err := conn.QueryContext(ctx,
                "SELECT id, email, name FROM users WHERE name ILIKE $1", "%"+query+"%",
            )
            if err != nil {
                mu.Lock()
                errs = append(errs, fmt.Errorf("shard %s: %w", dsn, err))
                mu.Unlock()
                return
            }
            defer rows.Close()
            for rows.Next() {
                var u User
                rows.Scan(&u.ID, &u.Email, &u.Name)
                mu.Lock()
                results = append(results, &u)
                mu.Unlock()
            }
        }()
    }

    wg.Wait()
    if len(errs) > 0 {
        return nil, errors.Join(errs...)
    }
    return results, nil
}

Scatter-gather works but has costs: latency is the max across all shards (not the average), and the application must merge, sort, and paginate results manually. Pagination across shards is particularly difficult — “page 3 of results sorted by name” requires fetching and sorting results from all shards.

Cross-Shard Transactions

ACID transactions across shards require a distributed transaction protocol — typically two-phase commit (2PC) or a saga pattern.

2PC across shards:

Coordinator
  │
  ├──► Shard 1: PREPARE (lock rows, write to WAL, hold)
  ├──► Shard 2: PREPARE (lock rows, write to WAL, hold)
  │
  │    (all prepared successfully)
  │
  ├──► Shard 1: COMMIT
  └──► Shard 2: COMMIT

If either shard fails to prepare, all are told to ROLLBACK. If a shard fails after PREPARE but before COMMIT, the transaction hangs in a prepared state until the coordinator recovers or a human intervenes.

2PC adds latency (multiple network round-trips), reduces availability (a shard failure blocks the transaction), and requires the coordinator to be highly available. Most sharding architectures avoid cross-shard transactions by design — through careful sharding key selection that keeps related data co-located.

Hotspots and Uneven Distribution

Even with a good hash function, real-world data is not perfectly uniform:

• A small number of “celebrity” users generate dramatically more activity than average users.
• Newly-created records are accessed more frequently than old ones.
• Certain geographic regions generate disproportionate traffic.

Mitigation strategies:

// For celebrity/whale users: dedicated shards
func (r *DirectoryRouter) ShardFor(ctx context.Context, userID int64) (string, error) {
    // Check for explicitly assigned shard (hot users or VIP tenants)
    if dsn, ok := r.hotUserShards[userID]; ok {
        return dsn, nil
    }
    // Fall through to normal hash routing
    return r.hashRouter.ShardFor(userID), nil
}

-- For hot rows: use application-level caching (Redis)
-- before hitting the database shard at all.
-- A celebrity user's profile should be in Redis, not generating
-- 100k/s reads on one shard.

Resharding

Moving data between shards without downtime is one of the hardest operational problems in sharding. The general approach:

Phase 1: Dual-write
  All writes go to both old shard and new shard.
  New shard is not yet serving reads.

Phase 2: Backfill
  Copy historical data from old shard to new shard.
  Dual-write continues, ensuring no writes are missed.

Phase 3: Verify
  Compare checksums or row counts between old and new shard.
  Validate data consistency.

Phase 4: Cutover
  Update routing table to direct reads to new shard.
  Old shard becomes a fallback for a brief window.

Phase 5: Cleanup
  After confidence period, stop dual-writes.
  Archive or delete data from old shard.

This process takes days to weeks for large datasets and requires careful coordination between the application team, DBA, and operations. Vitess automates much of this for MySQL; Citus does so for PostgreSQL.

Partitioning vs Sharding: Decision Framework

Start here:
  ↓
Is your bottleneck query performance or maintenance on a large table?
  YES → Partitioning. Choose the strategy based on your dominant query pattern:
          │
          ├── Time-series data, range queries on date → Range partitioning
          ├── Uniform access, no natural range → Hash partitioning
          ├── Categorical data (region, tenant) → List partitioning
          └── Complex access patterns → Composite partitioning
  NO  ↓
Does a single server handle all writes?
  YES → You don't need sharding. Optimize queries, add read replicas.
  NO  ↓
Does data size exceed one server's storage?
  YES → Sharding required.
  NO  ↓
Is write throughput the bottleneck?
  YES → Sharding required (or evaluate distributed SQL).
  NO  ↓
Is geographic data locality required?
  YES → Sharding (or global distributed SQL like Spanner).
  NO  → Add read replicas. You probably don't need sharding.

Practical Example: PostgreSQL + Citus

Citus extends PostgreSQL with transparent sharding. It is the least disruptive path to sharding for PostgreSQL users.

-- Install Citus extension
CREATE EXTENSION citus;

-- Add worker nodes (each is a PostgreSQL server)
SELECT citus_add_node('worker-1', 5432);
SELECT citus_add_node('worker-2', 5432);
SELECT citus_add_node('worker-3', 5432);

-- Create your table normally
CREATE TABLE orders (
    id          BIGSERIAL,
    customer_id BIGINT NOT NULL,
    total       NUMERIC(12,2) NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (customer_id, id)  -- sharding key must be in PK
);

-- Distribute the table (this shards it across workers)
SELECT create_distributed_table('orders', 'customer_id');
-- Citus creates 32 shards by default, distributes across workers

-- From this point: the application queries normally
-- Citus coordinator routes to the correct worker shard
SELECT * FROM orders WHERE customer_id = 12345;
-- Routed to the single shard that owns customer_id 12345

-- Cross-shard query: Citus executes on all shards and merges
SELECT COUNT(*), SUM(total) FROM orders WHERE created_at > NOW() - INTERVAL '1 day';
-- Pushed down to all shards in parallel, results aggregated on coordinator

Key Takeaways

Partitioning and sharding address different problems at different scales:

1. Partition first, shard later. Partitioning is low-risk, transparent to the application, and solves the majority of large-table performance problems. Sharding introduces distributed systems complexity that cannot be undone without significant rework.

2. The partition key (or sharding key) is the most important decision. A poorly chosen key creates hotspots that make the technique worse than no partitioning at all. Choose based on your dominant query pattern, not on what’s easy to implement.

3. Range partitioning is for time-series; hash partitioning is for uniform distribution. If you find yourself wanting both, use composite (sub) partitioning.

4. Sharding changes your application architecture permanently. Cross-shard queries, cross-shard transactions, and resharding are genuinely hard. Design your data model to minimize cross-shard operations — the sharding key should be the entity that most queries are scoped to.

5. The directory pattern is the most flexible sharding approach. It allows tenant isolation, hot-shard mitigation, and online resharding without rehashing. Its cost is an additional dependency (the directory) that must be highly available and fast.

6. Consistent hashing solves the resharding problem for hash-based approaches. hash % n_shards is simple but brittle. Consistent hashing with virtual nodes moves only 1/n of data when adding a shard.

7. Distributed SQL (CockroachDB, Spanner) hides sharding complexity but adds latency. For workloads that require cross-shard ACID transactions and can tolerate 2–5ms write latency, distributed SQL is often simpler than hand-rolled sharding. For workloads where sub-millisecond write latency is critical, explicit sharding with application-level routing wins.

8. Hotspots are a runtime property, not a design-time guarantee. Monitor shard sizes and query distributions continuously. A sharding scheme that is balanced today may become imbalanced as your user base evolves. Build the operational tooling to detect and respond to hotspots before they become outages.