PostgreSQL Sharding in 2026: Horizontal Scaling Without Citus or Managed Cloud Services

Introduction: When Vertical Scaling Hits Its Limits

PostgreSQL is one of the most reliable and feature-rich relational database engines available. But even it reaches a ceiling under growing load. Vertical scaling — adding more CPU, RAM, or NVMe storage — works up to a point: roughly 32–64 cores and a few terabytes of data. Beyond that, problems emerge: rising write latency, WAL pressure, unacceptable VACUUM times, and index degradation.

In 2026, a typical high-load system is a microservice architecture handling tens of millions of events per day, a multi-tenant SaaS platform, or an IoT data collector. Horizontal scaling through sharding is no longer optional — it's a necessity. Yet Citus requires licensing and a specific operational model, while managed solutions (Aurora, AlloyDB, Neon) introduce vendor lock-in and drive up costs. This article focuses on self-hosted PostgreSQL sharding using your own infrastructure.

Core Sharding Strategies

Hash Sharding

Data is distributed across shards based on a hash of the shard key. For example, shard_id = hash(user_id) % N. This ensures even record distribution and eliminates hotspot shards when keys are randomly distributed.

Pros: balanced load, simple routing logic, scales well when adding shards with resharding.

Cons: range queries on the shard key are inefficient — you must scan all shards. Resharding when the shard count changes requires data migration.

Use case: user-oriented workloads where queries always include user_id and range queries on that field are not needed.

Range Sharding

Each shard is responsible for a range of values: for example, user_id 1 to 1,000,000 goes to shard 1, 1,000,001 to 2,000,000 to shard 2, and so on. Alternatively, sharding by date: January events go to shard 1, February to shard 2.

Pros: efficient range queries, precise routing, well-suited for time-series data.

Cons: hotspot risk — new records always land on the last shard. Requires rebalancing when growth is uneven.

Use case: analytical event tables with queries over time ranges; archiving old data by detaching shards.

Directory-Based Sharding

A separate metadata table stores the mapping: which tenant or entity lives on which shard. The router consults this table before every request.

Pros: maximum flexibility, ability to manually move tenants between shards, data isolation for large clients.

Cons: an extra lookup on every request (mitigated by caching); single point of failure for the shard map (mitigated by replication).

Use case: multi-tenant SaaS platforms that require data isolation and the ability to migrate a customer to a dedicated shard.

Application-Level Sharding Implementation

Application-level sharding is the most common approach in production systems that don't use Citus. Routing logic lives in the application or a dedicated proxy layer. Key components:

• Shard Map — a data structure (hash table, config file, or separate database) that maps a shard key to the connection string of a specific PostgreSQL node.
• Router — a component that accepts a request and returns the appropriate connection pool.
• Metadata Store — storage for the sharding schema, migration versions, and shard states.

Example shard map in Go (pseudocode):

type ShardMap struct {\n    Shards []ShardNode\n    TotalShards int\n}\n\ntype ShardNode struct {\n    ID   int\n    DSN  string\n    Pool *pgxpool.Pool\n}\n\nfunc (sm *ShardMap) GetShard(userID int64) *ShardNode {\n    idx := int(userID) % sm.TotalShards\n    return &sm.Shards[idx]\n}\n

For directory-based sharding, the shard map is stored in a dedicated PostgreSQL instance with a replica. The mapping cache is updated via Redis pub/sub whenever the schema changes.

It's important to version your sharding schema: if you change the number of shards or the hashing algorithm, existing data must be migrated, or routing will break. Use a dedicated shard_schema_version table with a full change history.

Foreign Data Wrappers and postgres_fdw for Federated Queries

When you need to run queries spanning multiple shards, you can use postgres_fdw — the standard PostgreSQL extension for accessing remote PostgreSQL servers. This lets you set up a "coordinator" — a single PostgreSQL node through which cross-shard queries are executed.

-- On the coordinator: attach shards as foreign servers\nCREATE EXTENSION postgres_fdw;\n\nCREATE SERVER shard1\n    FOREIGN DATA WRAPPER postgres_fdw\n    OPTIONS (host 'shard1.internal', port '5432', dbname 'events_db');\n\nCREATE SERVER shard2\n    FOREIGN DATA WRAPPER postgres_fdw\n    OPTIONS (host 'shard2.internal', port '5432', dbname 'events_db');\n\nCREATE USER MAPPING FOR app_user\n    SERVER shard1\n    OPTIONS (user 'app_user', password 'secret');\n\nCREATE USER MAPPING FOR app_user\n    SERVER shard2\n    OPTIONS (user 'app_user', password 'secret');\n\n-- Create foreign tables\nCREATE FOREIGN TABLE events_shard1 (\n    id          BIGINT,\n    user_id     BIGINT,\n    event_type  TEXT,\n    payload     JSONB,\n    created_at  TIMESTAMPTZ\n) SERVER shard1\nOPTIONS (table_name 'events');\n\nCREATE FOREIGN TABLE events_shard2 (\n    id          BIGINT,\n    user_id     BIGINT,\n    event_type  TEXT,\n    payload     JSONB,\n    created_at  TIMESTAMPTZ\n) SERVER shard2\nOPTIONS (table_name 'events');\n\n-- Combine via a VIEW\nCREATE VIEW events_all AS\n    SELECT * FROM events_shard1\n    UNION ALL\n    SELECT * FROM events_shard2;\n

Now a query like SELECT * FROM events_all WHERE user_id = 42 will be executed on the coordinator with predicate pushdown to both shards. PostgreSQL is smart enough to push the filter condition to the remote servers, minimizing network traffic.

Limitations of postgres_fdw: aggregations and sorts with LIMIT are executed locally on the coordinator after fetching data from the shards. With large data volumes, this creates significant load on the coordinator. For heavy analytical queries, it's better to run queries on each shard in parallel from the application and perform merge-sort on the results.

Managing Transactions and Data Consistency

The biggest pain point of sharding is distributed transactions. PostgreSQL supports two-phase commit (2PC) via PREPARE TRANSACTION / COMMIT PREPARED, but this is operationally complex: stalled prepared transactions block VACUUM and consume slots.

-- The coordinator begins a distributed transaction\nBEGIN;\n-- On shard 1\nINSERT INTO events (user_id, event_type, created_at)\n    VALUES (42, 'purchase', now());\nPREPARE TRANSACTION 'txn_20260101_001';\n\n-- On shard 2\nINSERT INTO user_stats (user_id, total_purchases)\n    VALUES (42, 1)\n    ON CONFLICT (user_id)\n    DO UPDATE SET total_purchases = user_stats.total_purchases + 1;\nPREPARE TRANSACTION 'txn_20260101_001';\n\n-- If both shards responded OK:\nCOMMIT PREPARED 'txn_20260101_001'; -- on both shards\n

In practice, microservice systems rarely use 2PC. Instead, they use the Saga pattern: a sequence of local transactions, each of which publishes an event triggering the next step. On failure, compensating transactions are executed.

Example Saga for a funds transfer between shards:

1. Shard A: debit funds, set operation status to PENDING, publish debit_completed event.
2. Shard B: credit funds, set status to COMPLETED, publish credit_completed event.
3. Shard A: update operation status to SUCCESS.
4. On failure at step 2: publish credit_failed; Shard A compensates — refunds the amount and sets status to ROLLED_BACK.

For eventual consistency across shards, use the outbox pattern: write transactionally to a local outbox_events table and use a separate process to deliver events. This guarantees at-least-once delivery without a centralized transaction manager.

Practical Example: Sharding an Events Table by user_id

Let's look at a real-world scenario: an events table with 500 million rows growing by 5 million per day. We'll shard by user_id using hash sharding across 4 shards.

Schema on each shard:

-- Run on each of the 4 PostgreSQL servers\nCREATE TABLE events (\n    id          BIGSERIAL PRIMARY KEY,\n    user_id     BIGINT       NOT NULL,\n    event_type  VARCHAR(64)  NOT NULL,\n    payload     JSONB        DEFAULT '{}',\n    created_at  TIMESTAMPTZ  NOT NULL DEFAULT now()\n);\n\nCREATE INDEX idx_events_user_id     ON events (user_id);\nCREATE INDEX idx_events_created_at  ON events (created_at DESC);\nCREATE INDEX idx_events_type_user   ON events (event_type, user_id);\n\n-- Intra-shard partitioning by month (range on created_at)\nCREATE TABLE events_2026_01\n    PARTITION OF events\n    FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');\n\nCREATE TABLE events_2026_02\n    PARTITION OF events\n    FOR VALUES FROM ('2026-02-01') TO ('2026-03-01');\n

Routing logic in the application:

-- Determine shard for user_id = 123456\n-- shard_index = 123456 % 4 = 0 → shard_0\n\n-- Query on the target shard:\nSELECT id, event_type, payload, created_at\nFROM events\nWHERE user_id = 123456\n  AND created_at >= now() - INTERVAL '30 days'\nORDER BY created_at DESC\nLIMIT 50;\n

Cross-shard aggregation via FDW coordinator:

-- Event type statistics for the last 7 days\n-- Executed on the coordinator with predicate pushdown to all shards\nSELECT\n    event_type,\n    count(*)           AS total,\n    count(DISTINCT user_id) AS unique_users\nFROM events_all\nWHERE created_at >= now() - INTERVAL '7 days'\nGROUP BY event_type\nORDER BY total DESC;\n

To fetch data for user 123456, the application goes directly to shard 0 — bypassing the coordinator, with minimal latency. Analytical reports use the FDW coordinator, and the load it generates is isolated from OLTP traffic.

Monitoring and Observability for a Sharded Cluster

Monitoring a sharded PostgreSQL cluster is more complex than monitoring a monolith: you need to aggregate metrics from all nodes and correlate them. Key metrics and tools:

• pg_stat_statements on each shard — top slow queries, normalized by parameters. Collect via Prometheus + postgres_exporter.
• pg_stat_replication — replication lag on each shard. Critical for read replicas: if lag exceeds 5 seconds, the read replica is removed from rotation.
• Shard balance metrics — table size and row count on each shard. A deviation of more than 20% from the average is a signal to reshard.
• Cross-shard query latency — execution time of queries through the FDW coordinator. Trace via OpenTelemetry with shard_id annotations.
• Connection pool utilization — PgBouncer on each shard. Monitor via SHOW POOLS and SHOW STATS.

-- Useful query for monitoring shard balance\n-- Run on the coordinator via FDW\nSELECT\n    'shard1' AS shard,\n    count(*) AS row_count,\n    pg_size_pretty(pg_total_relation_size('events')) AS table_size\nFROM events_shard1\nUNION ALL\nSELECT\n    'shard2',\n    count(*),\n    pg_size_pretty(pg_total_relation_size('events'))\nFROM events_shard2;\n

For distributed tracing, set application_name in the connection string to include the shard_id and request_id. This lets you correlate PostgreSQL logs with application traces in Jaeger or Tempo.

Pitfalls and Common Mistakes

• Choosing the wrong shard key. The key must appear in the majority of queries. Sharding events by created_at seems logical, but if queries always filter by user_id, every request will fan out to all shards.
• Not designing for resharding. Hardcoding the shard count in your code leads to a painful migration down the road. Plan for resharding from day one: use consistent hashing or virtual shards (e.g., 1,024 virtual → 4 physical).
• Global sequences. BIGSERIAL on each shard generates independent sequences — ID collisions are possible when doing cross-shard JOINs. Use UUIDv7 or Snowflake IDs with an embedded shard_id.
• Cross-shard JOINs in the hot path. Joining tables across shards means a network round-trip plus a merge on the coordinator. Denormalize your data or use co-location: keep related data for the same user on the same shard.
• Ignoring VACUUM on shards. Under heavy write load, each shard needs aggressive autovacuum. Set autovacuum_vacuum_cost_delay = 2ms and autovacuum_max_workers = 6 independently on each node.
• No circuit breaker for shards. If one shard degrades, without a circuit breaker you risk a cascading failure. Implement the circuit breaker pattern in the router layer with fallback to a read replica.

Conclusion: Choosing the Right Strategy

Sharding PostgreSQL without Citus or managed cloud services is entirely feasible and justified when done with proper planning. Strategy selection depends on your data access patterns:

• Hash sharding — when all queries include a single key (user_id, tenant_id) and range queries on that key are not required. Suitable for most OLTP systems.
• Range sharding — when data is time-oriented and regularly archived. Great for time-series and log storage.
• Directory-based sharding — for multi-tenant SaaS platforms with data isolation requirements and the need to migrate customers between shards.

Start with application-level sharding and a small number of shards (4–8). Use postgres_fdw for analytical cross-shard queries, keeping them isolated from OLTP traffic. Design for resharding from the very beginning. Use the Saga pattern instead of 2PC for distributed operations. And invest seriously in observability — without metrics from every shard, you'll be flying blind.

Horizontal PostgreSQL scaling is neither magic nor a silver bullet. It's an engineering trade-off between system complexity and scalability. Make that trade-off deliberately.