Chapter 27: Working with Large Data Sets

As data volume grows, a single database instance or table can become a bottleneck. Large data sets require strategies that spread data across partitions, nodes, or even multiple engines while still providing a coherent query interface. This chapter introduces partitioning, sharding, and distributed query techniques used by modern systems to manage scale without losing consistency.

Partitioning and sharding basics

Partitioning divides one logical table into multiple physical parts within the same database. Sharding goes a step further, splitting data across multiple database servers. Both approaches improve performance by limiting how much data each query must scan and by parallelizing operations.

Horizontal and vertical partitioning

Partitioning can happen by rows (horizontal) or columns (vertical):

Example: range partitioning by date

Most engines support declarative partitioning syntax. This example shows PostgreSQL style range partitions.

CREATE TABLE sales (
  id          BIGSERIAL PRIMARY KEY,
  region      TEXT NOT NULL,
  sale_date   DATE NOT NULL,
  amount      NUMERIC(10,2)
) PARTITION BY RANGE (sale_date);

CREATE TABLE sales_2025q1 PARTITION OF sales
  FOR VALUES FROM ('2025-01-01') TO ('2025-04-01');

CREATE TABLE sales_2025q2 PARTITION OF sales
  FOR VALUES FROM ('2025-04-01') TO ('2025-07-01');

Queries automatically use the correct partition when filtering by the partition key, and each partition can have its own index and storage parameters.

Hash partitioning

Hash partitioning distributes rows more evenly when ranges are not predictable, such as user ids or session keys.

CREATE TABLE sessions (
  session_id BIGINT,
  user_id    BIGINT,
  started_at TIMESTAMP,
  PRIMARY KEY (session_id)
) PARTITION BY HASH (user_id);

CREATE TABLE sessions_p0 PARTITION OF sessions FOR VALUES WITH (MODULUS 4, REMAINDER 0);
CREATE TABLE sessions_p1 PARTITION OF sessions FOR VALUES WITH (MODULUS 4, REMAINDER 1);
CREATE TABLE sessions_p2 PARTITION OF sessions FOR VALUES WITH (MODULUS 4, REMAINDER 2);
CREATE TABLE sessions_p3 PARTITION OF sessions FOR VALUES WITH (MODULUS 4, REMAINDER 3);

Sharding across servers

Sharding divides data across multiple servers (each with its own schema). Applications or middleware determine which shard to query based on a shard key. For example, users 1–10000 on shard 1, 10001–20000 on shard 2, and so on.

# Pseudo code
def shard_for_user(user_id):
    shard_index = (user_id % NUM_SHARDS)
    return connections[shard_index]

conn = shard_for_user(15234)
cur = conn.cursor()
cur.execute("SELECT * FROM orders WHERE user_id = %s", (15234,))

Each shard runs independently, and joins across shards require application logic or special frameworks to merge results.

Rebalancing and growth

Shards can fill unevenly as data grows. To rebalance, resharding tools copy data to new nodes based on a new key distribution. Use consistent hashing to minimize the amount of data moved when adding shards.

-- Conceptual representation
Hash(user_id) % N_shards   -- determines target shard
-- When N_shards increases, only a fraction of data moves

Federated or external partitions

Some databases (for example MySQL with the FEDERATED engine, or PostgreSQL with postgres_fdw) support tables that point to data in remote servers. These act as partitions hosted elsewhere but queried locally.

CREATE SERVER remote_db
  FOREIGN DATA WRAPPER postgres_fdw
  OPTIONS (host 'reports.example.com', dbname 'analytics');

CREATE FOREIGN TABLE analytics.sales_2025
  (id BIGINT, region TEXT, amount NUMERIC)
  SERVER remote_db
  OPTIONS (table_name 'sales_2025');

Federated queries and distributed SQL

Federated and distributed SQL systems allow a single query to span multiple nodes or databases as if they were one. They handle data location, parallel execution, and result aggregation transparently to the client.

Federated queries

Federation is common when integrating heterogeneous sources such as operational and analytics databases. Tools like postgres_fdw, mysql_fdw, or cross engine layers like Presto or Trino can query across data sets and merge results at runtime.

SELECT o.id, o.amount, c.region
FROM orders o
JOIN foreign_customers c ON o.customer_id = c.id
WHERE c.region = 'EU';

The federated layer manages connections to both data sources and streams results back to the client. This allows unified analysis without duplicating data, though at a performance cost for large joins.

Distributed SQL engines

Distributed SQL platforms (for example CockroachDB, YugabyteDB, Google Spanner) distribute both data and queries automatically. They replicate data across nodes, provide ACID transactions, and scale horizontally without manual sharding.

These systems appear as a single logical database but partition data internally. They use consensus protocols (such as Raft or Paxos) to coordinate replicas and ensure consistency.

Trade offs and considerations

Distributed SQL simplifies scaling but introduces coordination overhead. Latency increases when replicas span regions, and schema changes may propagate slowly. Evaluate whether your workload requires true distributed consistency or can tolerate eventual models (for example in caching or analytics).

Query federation vs data replication

Federation queries remote data on demand. Replication copies data locally. Federation reduces duplication but adds latency; replication speeds queries but needs synchronization jobs. Many systems combine both—federating rarely used sources while replicating high traffic tables.

Practical scaling workflow

1. Start with a single node, strong indexes, and tuned queries.
2. Add read replicas to handle heavy read workloads.
3. Introduce partitioning for large tables (by date, tenant, or id ranges).
4. Move to sharding or distributed SQL only when you outgrow single host capacity.