Data Partitioning and Sharding Strategies for .NET

Sharding distributes data across independent databases. It can increase write and storage capacity when load spreads across shards, but queries and transactions that cross shard boundaries become more expensive.

The shard key determines those boundaries. Choose it against the queries your application actually runs, including reporting and administrative work.

Partitioning vs Sharding: Same Word Family, Different Contracts

Partitioning splits a table into pieces within one database server. PostgreSQL declarative partitioning, for example, splits a huge table by month or by hash, the planner prunes partitions automatically, and your application keeps issuing the same SQL against the same connection string. It solves big-table problems: index bloat, slow bulk deletes, cold data management. I covered it in PostgreSQL table partitioning.

Sharding splits data across database servers. Now there are N connection strings, and someone must answer "which server has this row?" on every query. That someone is your application (or a proxy layer doing the same job). Sharding solves a different problem: the write path or working set no longer fits one machine, no matter how big.

Native table partitioning usually keeps routing inside the database; application-managed sharding exposes it to your code.

Do You Need Sharding Yet?

Benchmark a single node with your transaction size, indexes, durability settings, and hardware before deciding that it has reached its limit. Before sharding, consider:

1. Index and query tuning, then vertical scaling.
2. Read replicas for read-heavy load.
3. Caching frequently repeated reads.
4. Table partitioning for big-table pain.
5. Archiving cold data out of the hot path.

Shard when writes or data size exceed what steps 1 through 5 can carry. Migrating an existing database also needs capacity headroom, so plan the move before the current server is saturated.

Choosing a Shard Key

Every row is stored on the shard its key maps to. Changing the key usually means migrating data and coordinating routing changes while requests continue. The cost depends on data volume and how much downtime the application can accept.

A good shard key scores on three axes:

• Query alignment: the key appears in (nearly) all hot queries, so they route to exactly one shard.
• Distribution: rows and load spread evenly, no whale shard.
• Locality of transactions: rows modified together are stored on the same shard, so transactions stay single-shard.

Walk the classic candidates for an e-commerce system:

• TenantId (B2B/SaaS): useful when most queries and transactions stay inside one tenant. Several tenants can share a shard, so tenant authorization and filtering still apply. Large tenants can skew distribution; see multi-tenant applications.
• CustomerId (B2C): great for customer-facing flows ("my orders", "my cart"). But every merchant-side query ("orders for product X today") is now cross-shard.
• OrderId: distributes evenly but matches almost no query pattern. Every non-trivial query fans out.
• CreatedAt: fits time-series reads, but concentrates all writes on the newest shard. Usually the worst key for an OLTP write path.

The test that catches bad keys early: write down your top 20 queries, and for each one mark whether the candidate key appears in the WHERE clause. Every unmarked row is a query that pays the cross-shard query tax.

Hash vs Range: How Keys Map to Shards

Hash sharding: shard = Hash(key) % N (in spirit; in practice you want consistent hashing or a slot table so N can change).

• A stable hash distributes many keys approximately evenly. A few large or busy tenants can still dominate a shard.
• Queries without a tenant key, such as "all orders from last week", fan out. Date-range queries for one tenant remain local when tenant ID is the shard key.

Install System.IO.Hashing for this fixed-shard-count example:

dotnet add package System.IO.Hashing

using System.IO.Hashing;

public sealed class HashShardRouter
{
    private readonly string[] _connectionStrings;

    public HashShardRouter(IEnumerable<string> connectionStrings)
    {
        _connectionStrings = connectionStrings.ToArray();
        if (_connectionStrings.Length == 0 ||
            _connectionStrings.Any(string.IsNullOrWhiteSpace))
        {
            throw new ArgumentException("Configure at least one valid shard.");
        }
    }

    public string GetConnectionString(Guid tenantId)
    {
        // Keep the hash algorithm and GUID byte encoding stable across clients.
        var hash = XxHash32.HashToUInt32(tenantId.ToByteArray());
        var shard = (int)(hash % (uint)_connectionStrings.Length);

        return _connectionStrings[shard];
    }
}

Do not use GetHashCode() as a persistent routing contract; string.GetHashCode() can vary across processes. Keep the shard list order, hash algorithm, and key byte encoding identical on every instance. Changing the count or order moves keys without moving their rows. Use a migration protocol and a versioned directory or slot table before adding shards.

Range sharding: contiguous key ranges per shard (tenants A-F here, G-M there; or time windows).

• Range predicates can target the shards whose ranges overlap the query. A range spanning several shards still requires fan-out.
• Hotspots are the default. With time-based ranges, all current writes land on the newest shard, so your cluster turns back into one hot machine plus N-1 archives.
• A range directory maps ranges to shards. Moving a large tenant needs a targeted data migration plus a coordinated directory update.

Use hash routing when hot queries include the shard key, and range routing when queries can prune key ranges and the write distribution is acceptable. Directory-based routing (explicit key-to-shard mapping) is worth its operational cost the moment you have tenants big enough to need their own shard.

The Cross-Shard Query Tax

Queries without the shard key need a different execution plan:

Fan-out reads. "Top 100 orders above a given total" without the shard key means querying every shard and merging the results. This example uses a separate EF Core context for each query and bounds concurrency at four shards:

using System.Collections.Concurrent;
using Microsoft.EntityFrameworkCore;

public sealed class CrossShardQueries(IReadOnlyList<string> shards)
{
    public async Task<IReadOnlyList<Order>> GetTopOrdersAsync(
        decimal minTotal, CancellationToken ct)
    {
        var results = new ConcurrentBag<Order>();
        await Parallel.ForEachAsync(shards, new ParallelOptions
        {
            MaxDegreeOfParallelism = 4,
            CancellationToken = ct
        }, async (connectionString, token) =>
        {
            var options = new DbContextOptionsBuilder<AppDbContext>()
                .UseNpgsql(connectionString).Options;
            await using var db = new AppDbContext(options);
            var rows = await db.Orders.AsNoTracking()
                .Where(o => o.Total >= minTotal)
                .OrderByDescending(o => o.Total).ThenBy(o => o.Id)
                .Take(100).ToListAsync(token);

            foreach (var row in rows) results.Add(row);
        });

        return results.OrderByDescending(o => o.Total).ThenBy(o => o.Id)
            .Take(100).ToList();
    }
}

One unavailable shard fails the query rather than returning a misleading partial top 100. The model and provider setup appear below; this assumes globally unique IDs and PostgreSQL's native uuid column mapping, whose ordering matches the application comparison.

A local .NET 10 check populated two PostgreSQL databases with 120 orders each and ran this merge:

EF Core shards: two 120-row shards merge to global top 100, ordered by total then ID.

For pagination across shards, page 50 needs each shard to produce enough rows to guarantee global ordering. Offset pagination degenerates; cursor-based pagination with per-shard cursors is the workable version.

Cross-shard transactions need explicit coordination. The router below does not provide it: one DbContext transaction covers one shard. Keep transactions local where possible, or design eventual consistency with an outbox and sagas; distributed transaction support depends on your database and coordinator.

Joins across shards move into application code or into a second system.

The standard escape hatches each add architecture of their own:

• Duplicate reference data (small lookup tables) onto every shard.
• Global secondary index: maintain an "email to tenant" lookup in a shared store for the login-by-email problem, then route.
• Analytics elsewhere: replicate shards into a warehouse (or search index) and send every "across everything" question there instead of fanning out OLTP.

If the taxed queries turn out to be your core product, that is the signal the key (or sharding itself) was wrong for this system.

.NET Mechanics: EF Core Across Shards

EF Core has no concept of shards; a DbContext wraps one connection. Route when creating the context, after authenticating and authorizing the tenant:

Install the EF Core provider:

dotnet add package Npgsql.EntityFrameworkCore.PostgreSQL

Add this registration to Program.cs in an ASP.NET Core application:

using Microsoft.EntityFrameworkCore;

builder.Services.AddSingleton(new HashShardRouter(
    builder.Configuration.GetSection("Shards").Get<string[]>()
        ?? throw new InvalidOperationException("Configure Shards.")));

builder.Services.AddDbContext<AppDbContext>((sp, options) =>
{
    var tenant = sp.GetRequiredService<ITenantProvider>().CurrentTenant;
    var router = sp.GetRequiredService<HashShardRouter>();
    options.UseNpgsql(router.GetConnectionString(tenant.Id));
});

public sealed record Tenant(Guid Id);

// Implement and register this with your authenticated tenant context.
public interface ITenantProvider
{
    Tenant CurrentTenant { get; }
}

public sealed class AppDbContext(DbContextOptions<AppDbContext> options)
    : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();
}

public sealed class Order
{
    public Guid Id { get; set; }
    public Guid TenantId { get; set; }
    public decimal Total { get; set; }
}

Configure Shards as an ordered array of connection strings, and register your ITenantProvider as a scoped service. Do not route from an unverified request header. Apply tenant filters to queries because multiple tenants can share a shard; routing does not replace authorization.

Run schema migrations against every shard, and generate IDs that do not require cross-shard coordination, such as GUID v7. Avoid sharing a context across concurrent shard queries.

For PostgreSQL, Citus provides distributed tables and query routing. Its distribution key still determines which operations can execute locally.

Summary

• Partitioning is inside one server and transparent; sharding is across servers and restructures your application. Exhaust the single-node ladder first.
• The shard key is costly to change. Evaluate candidates against your main queries, measured workload distribution, and transaction locality.
• Hash routing spreads keys, but busy tenants can still create uneven load. Ranges without a shard key can require fan-out; range routing allows pruning but can concentrate writes. Use a stable hash, never GetHashCode().
• Queries without the key need fan-out, a secondary index, or a separate read model. Transactions spanning shards need a coordination strategy.

Before migrating, test routing consistency across instances, an unavailable shard, and a shard move under concurrent traffic. Those cases determine whether the application can operate the layout after its first deployment.