Database Sharding with Spring Boot and PostgreSQL: Routing, Resharding, and Operational Boundaries

Database sharding is not a normal first step in scaling PostgreSQL. It converts one database boundary into several independent failure, transaction, backup, migration, and observability boundaries.

That cost can be justified when one primary can no longer meet a measured requirement after the simpler options have been exhausted. It is not justified merely because the application uses microservices or because the orders table is large.

This guide designs application-level sharding for a Spring Boot 4.1 service using JPA and PostgreSQL 18. The focus is the part that simplified tutorials omit: stable routing, transaction scope, global uniqueness, cross-shard behavior, resharding, and recovery.

TL;DR Prove that one PostgreSQL primary is the bottleneck before sharding. Choose a shard key that keeps the most important transactions and queries on one shard. Make routing deterministic, versioned, observable, and independent of mutable profile fields. Treat one request touching two shards as a distributed workflow, not a normal JPA transaction. Use globally unique identifiers and database constraints inside each shard. Design resharding, backup validation, and shard evacuation before the first production split.

First, Confirm That You Need Sharding

Sharding is one option among several.

Declarative partitioning splits one logical table inside one PostgreSQL cluster. Sharding distributes data across independent database servers. Partitioning can improve pruning and maintenance, but it does not distribute the primary's write work to multiple machines.

Before sharding, collect:

• working-set and storage growth;
• write-ahead log volume;
• p50, p95, and p99 query latency;
• CPU, I/O, memory, and lock saturation;
• connection usage;
• largest relations and indexes;
• top queries by total time;
• tenant or key distribution;
• recovery time and backup size.

The decision should name the exhausted resource and the capacity target a shard adds.

Define the Sharding Unit

Assume an order service owns:

customer
  -> order
      -> order_line
      -> payment_attempt
      -> shipment_request
      -> outbox_event

If the primary access pattern is “all orders for one customer,” customer_id is a candidate shard key. All related rows should carry that key so routing does not require a lookup on another database.

CREATE TABLE customer_order (
    customer_id UUID NOT NULL,
    order_id UUID NOT NULL,
    status VARCHAR(40) NOT NULL,
    total_amount NUMERIC(19, 4) NOT NULL,
    version BIGINT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL,

    PRIMARY KEY (customer_id, order_id)
);

CREATE INDEX customer_order_recent_idx
    ON customer_order (
        customer_id,
        created_at DESC
    );

The composite primary key makes the shard key part of the local uniqueness definition and supports the dominant query.

Choose a Shard Key Deliberately

A good shard key is:

• present before the transaction starts;
• immutable for the life of the aggregate;
• distributed enough to avoid hot shards;
• shared by rows that must commit together;
• included in common queries;
• safe to expose only when appropriate.

Weak keys include:

• country, because populations and traffic are uneven and users can move;
• subscription plan, because it changes and has low cardinality;
• creation month, when all writes target the newest shard;
• auto-increment ID modulo shard count, if IDs are allocated only after choosing a shard;
• a random row ID when queries group by customer and must fan out.

Hash routing

slot = hash(customerId) mod 4096
shard = placementTable[slot]

Using many logical slots is more flexible than:

shard = hash(customerId) mod numberOfShards

With direct modulo, changing from four to five shards remaps most keys. A logical slot table lets operators move selected slots while leaving other customers in place.

Range routing

Ranges simplify scans and intentional placement:

[0000, 3fff] -> shard-a
[4000, 7fff] -> shard-b

They can create hot ranges if new identifiers are time ordered. Range boundaries also require monitoring and manual or automated splitting.

Directory routing

A directory stores the exact placement:

customer 8a... -> shard-c

This supports tenant isolation and targeted movement, but every request needs a reliable placement lookup. Cache it carefully, version it, and define behavior when the directory is unavailable.

Make the Routing Contract Explicit

public record ShardId(String value) {

    public ShardId {
        if (value == null || value.isBlank()) {
            throw new IllegalArgumentException(
                    "shard id is required");
        }
    }
}

public record ShardPlacement(
        int routingVersion,
        int slot,
        ShardId shardId,
        PlacementState state) {
}

public interface ShardDirectory {

    ShardPlacement locate(UUID customerId);
}

@Component
public class SlotShardDirectory
        implements ShardDirectory {

    private static final int SLOT_COUNT = 4096;

    private final SlotPlacementRepository placements;

    public SlotShardDirectory(
            SlotPlacementRepository placements) {
        this.placements = placements;
    }

    @Override
    public ShardPlacement locate(UUID customerId) {
        int slot = Math.floorMod(
                stableHash(customerId),
                SLOT_COUNT);
        return placements.findRequired(slot);
    }
}

The hash algorithm is persisted architecture. Changing Java's implementation or switching languages must not silently reroute existing data. Define test vectors:

customer UUID -> expected slot

and run them in every implementation.

Route Before Starting the Transaction

Spring's AbstractRoutingDataSource can select a configured DataSource using a routing key.

public final class ShardContext {

    private static final ThreadLocal<ShardId> CURRENT =
            new ThreadLocal<>();

    private ShardContext() {
    }

    public static void set(ShardId shardId) {
        if (CURRENT.get() != null) {
            throw new IllegalStateException(
                    "shard context already set");
        }
        CURRENT.set(shardId);
    }

    public static ShardId required() {
        ShardId shardId = CURRENT.get();
        if (shardId == null) {
            throw new IllegalStateException(
                    "shard context is missing");
        }
        return shardId;
    }

    public static void clear() {
        CURRENT.remove();
    }
}

public class ShardRoutingDataSource
        extends AbstractRoutingDataSource {

    @Override
    protected Object determineCurrentLookupKey() {
        return ShardContext.required().value();
    }
}

Set and clear the context around the service invocation:

@Component
public class ShardExecutor {

    private final ShardDirectory directory;
    private final TransactionTemplate transactions;

    public ShardExecutor(
            ShardDirectory directory,
            TransactionTemplate transactions) {
        this.directory = directory;
        this.transactions = transactions;
    }

    public <T> T inCustomerShard(
            UUID customerId,
            Supplier<T> work) {

        ShardPlacement placement =
                directory.locate(customerId);

        ShardContext.set(placement.shardId());
        try {
            return transactions.execute(status ->
                    work.get());
        }
        finally {
            ShardContext.clear();
        }
    }
}

The shard is selected before the transaction obtains a connection. An annotation aspect with the wrong ordering can start a transaction first and route too late.

Fail when the context is missing. A default shard is dangerous because a routing bug becomes valid-looking data written to the wrong database.

Thread-local routing is suitable for a synchronous Spring MVC/JPA path when it is strictly cleared. It should not be copied unchanged into Reactor pipelines or manually created asynchronous tasks. Context propagation must match the execution model.

Keep JPA Transactions Inside One Shard

@Service
public class PlaceOrderService {

    private final ShardExecutor shards;
    private final OrderRepository orders;
    private final OutboxRepository outbox;

    public PlaceOrderService(
            ShardExecutor shards,
            OrderRepository orders,
            OutboxRepository outbox) {
        this.shards = shards;
        this.orders = orders;
        this.outbox = outbox;
    }

    public UUID place(PlaceOrderCommand command) {
        return shards.inCustomerShard(
                command.customerId(),
                () -> {
                    CustomerOrder order =
                            CustomerOrder.create(command);
                    orders.save(order);
                    outbox.save(
                            OutboxEvent.orderPlaced(order));
                    return order.getOrderId();
                });
    }
}

The order and its outbox row commit in one local PostgreSQL transaction on one shard.

This does not work as one ordinary JPA transaction:

debit account on shard-a
credit account on shard-b

Options include:

• redesign ownership so the invariant lives on one shard;
• use a ledger service with its own consistency boundary;
• implement a durable workflow with explicit pending and compensated states;
• use a distributed transaction only after accepting its operational cost and availability trade-offs.

Do not hide two local commits behind one method name and call the result atomic.

Global Identifiers and Uniqueness

Database sequences generate unique values only inside their database unless ranges are coordinated.

Prefer an application-generated UUID or another documented globally unique format for public aggregate IDs. The identifier solves global identity, not every uniqueness rule.

An email address that must be unique across all customers cannot be enforced by an independent unique index on each shard. Common designs are:

1. a centralized identity or reservation service;
2. a globally consistent directory keyed by normalized email;
3. assigning the uniqueness domain to one shard;
4. relaxing the requirement and including tenant scope.

A reservation table might be:

CREATE TABLE global_email_reservation (
    normalized_email VARCHAR(320) PRIMARY KEY,
    customer_id UUID NOT NULL UNIQUE,
    reserved_at TIMESTAMPTZ NOT NULL
);

Creating the reservation and customer row still crosses boundaries. Model the workflow, idempotency, cleanup of abandoned reservations, and retry behavior.

Idempotency Must Survive Retries

Clients can retry after a timeout without knowing whether the first attempt committed.

CREATE TABLE command_deduplication (
    customer_id UUID NOT NULL,
    command_id UUID NOT NULL,
    result_type VARCHAR(80) NOT NULL,
    result_payload JSONB NOT NULL,
    completed_at TIMESTAMPTZ NOT NULL,

    PRIMARY KEY (customer_id, command_id)
);

Store the command result in the same shard and transaction as the order. The shard key must be part of the command so a retry reaches the same database.

An in-memory idempotency cache fails after restart and cannot protect concurrent instances reliably.

Cross-Shard Reads Need Their Own Product Contract

“List all orders” becomes a distributed query.

Possible approaches:

• require a shard key and keep the endpoint local;
• query shards in parallel with a strict limit and merge results;
• maintain a search or reporting projection;
• export events to an analytical store;
• use PostgreSQL foreign data wrappers for controlled administrative queries.

A fan-out request needs:

• a total deadline;
• per-shard timeouts;
• a maximum number of shards;
• bounded concurrency;
• deterministic merge ordering;
• a partial-result policy;
• a pagination design that does not use one global offset.

Offset pagination across shards is unstable and expensive. A cursor can carry a position per shard plus a routing version, but it becomes larger as shard count grows. For customer-facing global search, a dedicated projection is often simpler.

postgres_fdw can expose remote PostgreSQL tables through foreign tables, but it does not turn separate servers into one low-latency local database. Remote estimates, network transfer, connection management, transaction behavior, and unsupported operations must be considered.

Schema Changes Across Shards

Every shard must run a compatible schema.

Use expand-and-contract:

1. add a nullable column or new table;
2. deploy code that can read old and new forms;
3. backfill in bounded batches;
4. switch writes and reads;
5. enforce the new constraint;
6. remove the old representation later.

Track migration state:

shard-a -> schema 128
shard-b -> schema 128
shard-c -> schema 127, migration running

Application releases must tolerate the allowed version window. Stop rollout if a shard migration fails; do not let a schema outlier remain invisible.

Plan Resharding Before Production

Suppose logical slots 1200 through 1399 move from shard-a to shard-d.

A safe migration needs states such as:

STABLE
COPYING
DUAL_WRITE
VERIFYING
CUTOVER
SOURCE_READ_ONLY
COMPLETE

One possible flow:

1. Record the migration and source watermark.
2. Copy a consistent snapshot from the source.
3. Capture changes after the watermark through an outbox, logical decoding, or another durable change stream.
4. Apply changes idempotently to the destination.
5. Compare counts, key ranges, and business-level checksums.
6. briefly fence writes or use versioned dual writes for cutover.
7. update the routing directory atomically.
8. keep the source readable for a defined rollback window.
9. monitor destination traffic and discrepancies.
10. delete source data only after backup and rollback policy allow it.

Naive dual writing from the application is unsafe:

write source succeeds
write destination fails

If dual writes are used, each target needs idempotency, durable retry, version conflict handling, and reconciliation. A change log sourced from the committed database is often easier to reason about.

Prevent stale routers

Every placement record should have a routing version. During movement, a stale application instance may still send traffic to the old shard.

Defenses include:

• short-lived cached placements plus invalidation;
• source-side forwarding during a bounded window;
• a shard ownership table checked on writes;
• rejecting a request with MOVED metadata so the caller refreshes placement;
• fencing tokens that prevent an old owner from accepting new versions.

Caching the directory forever makes cutover nondeterministic.

Availability and Recovery

After sharding, “the database is healthy” is no longer one state.

Define:

• whether failure of one shard degrades only its customers;
• how the router reports an unavailable placement;
• failover behavior inside each shard;
• recovery point and recovery time objectives per shard;
• backup encryption and retention;
• restore tests for an individual shard and the routing directory;
• a procedure for rebuilding projections and outbox relays.

Backups taken independently are not a globally consistent snapshot. If a business process spans shards, recovery must understand workflow state and reconciliation rather than assuming all databases represent the same instant.

The routing directory can be a more critical dependency than any one shard. Replicate, back up, restore-test, and observe it accordingly.

Observability

Every trace and structured log should include:

• shard ID;
• logical slot or placement identifier;
• routing version;
• operation name;
• database outcome;
• migration state when relevant;
• retry or idempotency outcome.

Use shard ID as a metric label only when the shard count is controlled. Do not use customerId, orderId, or raw SQL as labels.

Track:

• per-shard query latency and error rate;
• connection-pool utilization;
• storage, WAL, CPU, I/O, locks, and replication lag;
• key and traffic skew;
• fan-out shard count and partial-result rate;
• directory lookup latency and cache hit rate;
• schema version drift;
• reshard copy lag and reconciliation mismatches;
• wrong-shard or stale-routing rejections.

Aggregate metrics alone hide a single hot or failing shard.

Test the Routing Invariant

Unit-test fixed hash vectors:

@ParameterizedTest
@CsvSource({
        "00000000-0000-0000-0000-000000000001, 271",
        "00000000-0000-0000-0000-000000000002, 912"
})
void routingIsStable(String customerId, int expectedSlot) {
    assertThat(router.slot(UUID.fromString(customerId)))
            .isEqualTo(expectedSlot);
}

The example slot values must come from the real chosen hash implementation.

Integration tests should prove:

• rows for one customer always reach one shard;
• missing context fails instead of using a default;
• context is cleared after success and failure;
• transaction creation happens after routing;
• duplicate commands return the stored result;
• a stale routing version is rejected during migration;
• cross-shard queries respect total deadlines;
• schema mismatch blocks an unsafe deployment.

A resharding rehearsal should inject:

• copy worker restart;
• duplicate change events;
• destination outage;
• router cache staleness;
• write during cutover;
• verification mismatch;
• rollback after directory update.

If the movement process has never been rehearsed at realistic scale, the system does not yet have an operational resharding capability.

Common Failure Modes

One shard is much hotter than the others

Measure both stored rows and request cost per key. A few large tenants, a time-correlated key, or one popular endpoint may defeat an apparently even hash distribution.

Records appear on the wrong shard

Remove fallback routing, log the routing version, check transaction/aspect ordering, verify context cleanup, and compare hash test vectors across service versions.

A cross-shard page repeats or skips rows

Global offset pagination is not stable under concurrent writes. Use shard-aware cursors with a deterministic tie-breaker or query a dedicated global projection.

Resharding never catches up

The copy rate may be lower than the write rate. Increase destination capacity, split the movement into smaller slots, throttle eligible source writes, or use a change-capture mechanism with a measurable lag.

A global unique value is duplicated

Local shard indexes cannot enforce global uniqueness. Route that uniqueness domain through a single authoritative reservation boundary and reconcile abandoned or partially completed workflows.

Sharding Readiness Checklist

• a measured single-primary resource is exhausted;
• partitioning, indexing, pooling, replicas, and archival have been evaluated;
• the shard key matches transaction and query locality;
• routing is deterministic and versioned;
• no missing-key path falls back to a default shard;
• IDs are globally unique;
• global uniqueness rules have an explicit owner;
• cross-shard writes are modeled as workflows;
• cross-shard read and pagination behavior is documented;
• every shard receives compatible migrations;
• idempotency survives process restart;
• resharding is implemented and rehearsed;
• the routing directory is backed up and restore-tested;
• per-shard recovery and evacuation procedures exist;
• dashboards reveal skew and individual shard health.

Conclusion

Sharding is successful when most business work stays local to one shard and the exceptional cross-shard paths are explicit. The key decisions are not the number of DataSource beans or the hash function alone. They are ownership, transaction locality, routing stability, uniqueness, recovery, and movement.

Spring Boot and JPA can route a transaction to PostgreSQL cleanly, but they cannot make several independent databases behave like one local transaction. Treat each shard as a real operational boundary, and design resharding before capacity pressure forces an emergency split.

Recommended Articles for Further Reading

• Eventual Consistency with Spring Boot, Kafka, and PostgreSQL
• Zero-Downtime Data Migration and Schema Evolution

Official References

• PostgreSQL 18: Table Partitioning
• PostgreSQL 18: postgres_fdw
• Spring Framework: AbstractRoutingDataSource
• Spring Framework: Transaction Management
• Hibernate ORM: Multitenancy