Database Sharding

Overview

Database sharding partitions data horizontally across multiple database instances, with each instance (shard) containing a subset of the total data. Unlike vertical partitioning, which splits tables by columns, or traditional replication, which duplicates entire datasets, sharding distributes rows across independent databases based on a sharding key.

The concept emerged from the need to scale beyond single-server database limitations. When a database grows too large for vertical scaling (adding more resources to one server), sharding distributes the load across multiple servers, each handling a portion of the data and queries.

# Without sharding: single database handles all users
User.where(country: 'US').count
# All 10 million users queried on one database

# With sharding: users distributed across multiple databases
shard_1.users.where(country: 'US').count  # 2 million users
shard_2.users.where(country: 'US').count  # 2 million users
shard_3.users.where(country: 'US').count  # 1 million users
# Total: 5 million US users distributed across 3 shards

Sharding introduces complexity through distributed data management but enables near-linear horizontal scalability. Each shard operates as an independent database with its own storage, memory, and processing resources. Applications must implement routing logic to direct queries to the correct shard based on the sharding key.

Key Principles

Sharding operates on several fundamental principles that distinguish it from other database scaling approaches. The sharding key determines data distribution and must be chosen carefully to balance load and minimize cross-shard queries.

Shared-Nothing Architecture

Each shard operates independently without sharing memory or storage with other shards. This isolation provides fault tolerance—a failed shard affects only its portion of data—and eliminates resource contention between shards. However, transactions spanning multiple shards require distributed coordination.

Shard Key Selection

The shard key determines which shard stores each record. This key must appear in most queries to enable efficient routing. Common patterns include:

• Hash-based: Apply hash function to key, modulo by shard count
• Range-based: Assign key ranges to specific shards
• Geographic: Route by location or region
• Directory-based: Maintain lookup table mapping keys to shards

# Hash-based sharding
def get_shard(user_id)
  shard_index = user_id.hash % SHARD_COUNT
  shards[shard_index]
end

# Range-based sharding
def get_shard(user_id)
  case user_id
  when 1..1_000_000
    shards[0]
  when 1_000_001..2_000_000
    shards[1]
  else
    shards[2]
  end
end

Data Distribution

Records distribute across shards according to the sharding key. Uniform distribution prevents hotspots where one shard handles disproportionate load. Skewed distribution causes performance bottlenecks as some shards become overloaded while others remain underutilized.

Query Routing

Applications must implement routing logic to direct queries to appropriate shards. Single-shard queries execute on one database instance. Cross-shard queries require scatter-gather operations: sending queries to multiple shards, collecting results, and merging them. These operations increase latency and complexity.

Rebalancing

Adding or removing shards requires redistributing data. Hash-based sharding with modulo operations requires moving most data when shard count changes. Consistent hashing minimizes data movement by remapping only a portion of keys. Range-based sharding can split ranges without moving data from unchanged shards.

Design Considerations

Sharding introduces architectural complexity that must be weighed against scalability benefits. The decision to shard depends on data volume, query patterns, growth projections, and team capabilities.

When to Shard

Shard when vertical scaling reaches practical limits. A single database server maxes out at several terabytes of storage and hundreds of thousands of queries per second. Before sharding, exhaust simpler options: read replicas for read-heavy workloads, caching layers, database query optimization, and table partitioning within a single database.

Indicators for sharding include:

• Database size exceeding single-server storage capacity
• Write throughput saturating single-server I/O
• Query latency increasing despite optimization
• Dataset growing faster than vertical scaling can accommodate
• Geographic distribution requiring local data access

Trade-offs

Sharding trades operational complexity for horizontal scalability. Benefits include unlimited scaling potential, fault isolation per shard, and independent resource allocation. Costs include increased application complexity, difficult cross-shard queries, distributed transaction coordination, and operational overhead managing multiple databases.

# Simple non-sharded query
User.joins(:orders)
    .where('orders.created_at > ?', 1.month.ago)
    .sum(:total)

# Equivalent sharded query requires scatter-gather
total = shards.map do |shard|
  shard.users
       .joins(:orders)
       .where('orders.created_at > ?', 1.month.ago)
       .sum(:total)
end.sum

Cross-shard joins become application-level operations. Foreign keys spanning shards cannot be enforced at the database level. Transactions across shards require two-phase commit protocols, which increase latency and reduce availability.

Alternatives

Consider alternatives before implementing sharding:

• Read replicas: Distribute read queries across multiple database copies. Handles read-heavy workloads without sharding complexity.
• Vertical partitioning: Split large tables by columns, storing frequently accessed columns separately. Reduces table size without distributing rows.
• Caching layers: Add Redis or Memcached to reduce database load. Often eliminates need for sharding in read-heavy applications.
• Database clustering: Some databases offer built-in sharding (Citus for PostgreSQL, MongoDB). Reduces application-level complexity.

Shard Key Selection Criteria

The shard key determines distribution effectiveness. Good shard keys exhibit:

• High cardinality: Many distinct values prevent hotspots
• Even distribution: Values spread uniformly across shards
• Query alignment: Key appears in most query WHERE clauses
• Immutability: Changing shard keys requires moving records
• Predictability: Applications can determine shard without lookups

Poor shard key choices include low-cardinality fields (status, category), sequential IDs that create hotspots on recent data, and fields rarely used in queries.

Implementation Approaches

Sharding can be implemented at different architectural layers, each with distinct characteristics and trade-offs. The implementation strategy affects complexity, flexibility, and operational requirements.

Application-Level Sharding

The application contains all sharding logic: determining shards, routing queries, and aggregating results. This approach provides maximum flexibility and works with any database but requires significant application changes.

class ShardedUserRepository
  def initialize
    @shards = [
      ActiveRecord::Base.establish_connection(shard_1_config),
      ActiveRecord::Base.establish_connection(shard_2_config),
      ActiveRecord::Base.establish_connection(shard_3_config)
    ]
  end

  def find_user(user_id)
    shard = get_shard(user_id)
    shard.connection.execute(
      "SELECT * FROM users WHERE id = #{user_id}"
    ).first
  end

  def find_users_by_country(country)
    # Cross-shard query: scatter-gather
    @shards.flat_map do |shard|
      shard.connection.execute(
        "SELECT * FROM users WHERE country = '#{country}'"
      ).to_a
    end
  end

  private

  def get_shard(user_id)
    @shards[user_id.hash % @shards.length]
  end
end

Application-level sharding requires managing multiple database connections, implementing connection pooling per shard, and handling connection failures. Each query must specify the target shard explicitly.

Middleware-Level Sharding

A proxy layer sits between the application and databases, handling shard routing transparently. Applications issue standard SQL queries; the proxy determines target shards and routes queries accordingly. This approach centralizes sharding logic but introduces another infrastructure component.

Libraries like Octopus (Ruby) or Vitess (Go) provide middleware sharding. The application configuration specifies sharding rules; the middleware handles routing:

# Using Octopus gem
class User < ActiveRecord::Base
  octopus_establish_connection
end

# Query routing happens automatically
Octopus.using(:shard_1) do
  User.create(name: 'Alice', user_id: 12345)
end

# Or routing by shard key
user_id = 12345
shard = determine_shard(user_id)
Octopus.using(shard) do
  User.find_by(user_id: user_id)
end

Database-Level Sharding

Some databases provide native sharding support. PostgreSQL with Citus extension, MySQL Cluster, and MongoDB automatically handle shard management, query routing, and data distribution. Applications interact with a single connection; the database manages shards internally.

Database-level sharding reduces application complexity but limits flexibility. Shard key changes and resharding depend on database capabilities. Some databases restrict cross-shard operations or require specific query patterns.

Hybrid Approaches

Production systems often combine approaches. Critical paths use application-level sharding for maximum control, while administrative queries use database-level tools. Read queries might use different routing than writes to optimize for different access patterns.

Shard Mapping Strategies

The mapping between shard keys and physical shards affects rebalancing complexity:

• Static mapping: Fixed assignment of key ranges to shards. Simple but requires data movement when adding shards.
• Dynamic mapping: Maintain a directory service mapping keys to shards. Flexible but introduces lookup overhead and a potential single point of failure.
• Consistent hashing: Hash keys to a ring; shards claim ring sections. Adding shards requires moving only adjacent data.

class ConsistentHashRing
  def initialize(shards)
    @shards = shards
    @ring = {}
    
    # Create virtual nodes for each shard
    shards.each do |shard|
      100.times do |i|
        hash_key = Digest::MD5.hexdigest("#{shard.name}-#{i}").to_i(16)
        @ring[hash_key] = shard
      end
    end
    
    @sorted_keys = @ring.keys.sort
  end
  
  def get_shard(key)
    hash = Digest::MD5.hexdigest(key.to_s).to_i(16)
    
    # Find first ring position >= hash
    idx = @sorted_keys.bsearch_index { |k| k >= hash } || 0
    @ring[@sorted_keys[idx]]
  end
end

Ruby Implementation

Ruby applications typically implement sharding through Active Record extensions or custom repository patterns. Several gems provide sharding support, though many projects implement custom solutions tailored to specific requirements.

Octopus Gem

Octopus extends Active Record to support multiple database connections with transparent routing:

# config/shards.yml
octopus:
  shards:
    shard_1:
      adapter: postgresql
      host: shard1.example.com
      database: app_shard_1
    shard_2:
      adapter: postgresql
      host: shard2.example.com
      database: app_shard_2
    shard_3:
      adapter: postgresql
      host: shard3.example.com
      database: app_shard_3

# app/models/user.rb
class User < ActiveRecord::Base
  octopus_establish_connection

  def self.find_by_user_id(user_id)
    shard = "shard_#{(user_id % 3) + 1}".to_sym
    using(shard).find_by(user_id: user_id)
  end
end

# Usage
Octopus.using(:shard_2) do
  User.create(name: 'Bob', email: 'bob@example.com')
end

Octopus maintains separate connection pools per shard and switches context dynamically. Nested using blocks support multi-level routing. However, cross-shard queries require explicit iteration:

# Cross-shard aggregation
def total_users
  [:shard_1, :shard_2, :shard_3].sum do |shard|
    Octopus.using(shard) { User.count }
  end
end

Custom Repository Pattern

For complex sharding requirements, a custom repository pattern provides more control:

class ShardedRepository
  attr_reader :shards

  def initialize
    @shards = load_shard_connections
    @default_shard = @shards.first
  end

  def find(id)
    shard_for(id).find(id)
  end

  def find_by_shard_key(key, value)
    shard = determine_shard(key)
    shard.where(key => value)
  end

  def create(attributes)
    shard_key = attributes[:user_id]
    raise ArgumentError, 'Shard key required' unless shard_key
    
    shard = determine_shard(shard_key)
    shard.create(attributes)
  end

  def cross_shard_query(&block)
    results = @shards.map do |shard|
      Thread.new { yield(shard) }
    end
    
    results.map(&:value).flatten
  end

  private

  def shard_for(id)
    # Retrieve shard assignment from directory
    shard_id = ShardDirectory.lookup(id)
    @shards[shard_id]
  end

  def determine_shard(key)
    shard_index = key.hash.abs % @shards.length
    @shards[shard_index]
  end

  def load_shard_connections
    shard_configs.map do |config|
      Class.new(ActiveRecord::Base) do
        establish_connection(config)
        self.abstract_class = true
      end
    end
  end
end

Connection Management

Each shard requires its own connection pool. Rails connection pooling must be configured per shard to prevent connection exhaustion:

# config/database.yml
shard_1:
  adapter: postgresql
  pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %>
  host: shard1.db.com

shard_2:
  adapter: postgresql
  pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %>
  host: shard2.db.com

# Establish connections in initializer
# config/initializers/shards.rb
SHARDS = {
  shard_1: Class.new(ActiveRecord::Base) do
    establish_connection(:shard_1)
    self.abstract_class = true
  end,
  shard_2: Class.new(ActiveRecord::Base) do
    establish_connection(:shard_2)
    self.abstract_class = true
  end
}

class ShardedUser < ActiveRecord::Base
  self.abstract_class = true
  
  def self.on_shard(shard_name)
    SHARDS[shard_name].connection_pool.with_connection do
      yield
    end
  end
end

Handling Migrations

Database migrations must run on all shards. Custom rake tasks coordinate migrations across shards:

# lib/tasks/sharded_migrations.rake
namespace :db do
  namespace :sharded do
    desc 'Migrate all shards'
    task migrate: :environment do
      SHARDS.each do |name, connection|
        puts "Migrating #{name}..."
        ActiveRecord::Base.establish_connection(name)
        ActiveRecord::Migrator.migrate('db/migrate')
      end
      
      # Restore default connection
      ActiveRecord::Base.establish_connection(:default)
    end
  end
end

Query Distribution

Queries must explicitly specify shard routing. For transparent routing, monkey-patch Active Record finder methods:

module ShardedFinder
  def find(*args)
    options = args.extract_options!
    
    if options[:shard_key]
      shard = determine_shard(options[:shard_key])
      shard.find(*args)
    else
      # Scatter-gather across all shards
      SHARDS.flat_map do |_, shard|
        begin
          shard.find(*args)
        rescue ActiveRecord::RecordNotFound
          nil
        end
      end.compact
    end
  end
end

ActiveRecord::Base.extend(ShardedFinder)

Performance Considerations

Sharding directly impacts application performance through query distribution, network latency, and resource utilization. Understanding these characteristics guides optimization strategies.

Query Performance

Single-shard queries execute with similar performance to non-sharded databases. The routing overhead adds negligible latency—typically under 1ms. Performance improves when load distributes evenly across shards, as each database handles a fraction of total queries.

Cross-shard queries suffer from scatter-gather latency. A query requiring data from all shards takes as long as the slowest shard plus aggregation overhead:

# Single-shard query: fast
Octopus.using(:shard_1) do
  User.where(country: 'US').limit(100)
end
# ~10ms

# Cross-shard query: slower
[:shard_1, :shard_2, :shard_3].flat_map do |shard|
  Octopus.using(shard) do
    User.where(country: 'US').limit(100)
  end
end
# ~40ms (3x shard time + aggregation + network)

Parallel execution reduces cross-shard query latency. Thread pools or async operations query shards concurrently:

require 'concurrent'

def parallel_cross_shard_query
  futures = shards.map do |shard|
    Concurrent::Future.execute do
      shard.users.where(active: true).count
    end
  end
  
  futures.map(&:value).sum
end
# Sequential: 30ms, Parallel: 10ms

Index Performance

Each shard maintains its own indexes. Smaller shard sizes improve index efficiency—B-tree lookups complete faster in smaller indexes. A 1 billion row table split across 10 shards means each index covers 100 million rows, reducing lookup depth.

However, indexes on non-shard-key columns cannot optimize queries spanning shards. A query filtering by email requires scanning indexes on all shards if email is not the shard key.

Write Performance

Sharding distributes write load across multiple databases. Applications writing 100,000 records per second can distribute writes across 10 shards, each handling 10,000 writes. This scales write throughput linearly with shard count.

Hotspot prevention requires even distribution. Sequential ID generation can concentrate writes on a single shard. Distributed ID generation or UUIDs distribute writes evenly:

# Hotspot: sequential IDs concentrate on one shard
User.create(id: next_id, name: 'Alice')
# If next_id = 1,000,000, always routes to same shard

# Better: UUID distributes evenly
User.create(id: SecureRandom.uuid, name: 'Alice')
# Random hash distributes across shards

Connection Pooling

Each shard requires a connection pool. Applications connecting to N shards multiply connection count by N. A web server with 25 threads connecting to 8 shards opens 200 database connections total. Monitor connection pool usage to prevent exhaustion:

# Monitor shard connection pools
SHARDS.each do |name, shard|
  pool = shard.connection_pool
  puts "#{name}: #{pool.connections.size}/#{pool.size} connections"
end

Memory Considerations

Sharding reduces per-shard memory requirements. Smaller working sets fit in buffer pools, improving cache hit rates. However, application servers maintain connection state for each shard, increasing memory usage. Each connection holds several megabytes of memory for query buffers and result caching.

Rebalancing Cost

Adding or removing shards requires data movement. The cost depends on the sharding strategy:

• Hash with modulo: Adding shard 4 to a 3-shard system requires moving ~75% of data
• Consistent hashing: Moving ~1/N of data where N is new shard count
• Range-based: May require no data movement if ranges split cleanly

Rebalancing operations can take hours or days for terabyte-scale databases. Plan rebalancing during low-traffic periods and use tools to throttle data movement:

class ShardRebalancer
  def rebalance(source_shard, target_shard, batch_size: 1000)
    offset = 0
    
    loop do
      records = source_shard.users
                           .where('id > ?', offset)
                           .limit(batch_size)
                           .to_a
      
      break if records.empty?
      
      records.each do |record|
        if should_move?(record, target_shard)
          target_shard.users.create(record.attributes)
          source_shard.users.delete(record.id)
        end
      end
      
      offset = records.last.id
      sleep 0.1  # Throttle to reduce load
    end
  end
end

Common Pitfalls

Sharding introduces failure modes and edge cases that manifest in production environments. Awareness of these issues prevents outages and data inconsistencies.

Cross-Shard Transactions

Transactions spanning multiple shards cannot use database ACID guarantees. A two-phase commit protocol provides atomicity but reduces availability—if one shard fails during commit, the entire transaction blocks:

# Problematic: transfer between users on different shards
def transfer_funds(from_user_id, to_user_id, amount)
  from_shard = get_shard(from_user_id)
  to_shard = get_shard(to_user_id)
  
  # These run on different databases - not atomic!
  from_shard.transaction do
    from_shard.users.find(from_user_id).decrement!(:balance, amount)
  end
  
  to_shard.transaction do
    to_shard.users.find(to_user_id).increment!(:balance, amount)
  end
  # If second transaction fails, first already committed
end

Compensating transactions or saga patterns handle distributed transactions at the application level. Design operations to be idempotent and implement retry logic with rollback capability.

Shard Key Immutability

Changing a record's shard key requires moving the record between shards. This operation is expensive and error-prone. Applications should prevent shard key updates or implement explicit resharding operations:

class User < ActiveRecord::Base
  before_update :prevent_shard_key_change

  private

  def prevent_shard_key_change
    if user_id_changed?
      errors.add(:user_id, 'cannot be changed - shard key is immutable')
      throw :abort
    end
  end
end

Hotspot Creation

Poor shard key selection creates hotspots where one shard handles disproportionate load. Time-based sharding often creates hotspots on the most recent shard:

# Hotspot: all new records go to latest shard
def get_shard(created_at)
  month = created_at.strftime('%Y-%m')
  shards[month]
end
# Current month shard handles all writes

Monitor per-shard query rates and storage growth. Significant imbalance indicates hotspots requiring shard key reevaluation.

Cross-Shard Foreign Keys

Foreign key relationships spanning shards cannot be enforced at the database level. Applications must implement referential integrity checks:

class Order < ActiveRecord::Base
  validates :user_id, presence: true
  validate :user_exists_on_shard

  private

  def user_exists_on_shard
    shard = get_shard(user_id)
    unless shard.users.exists?(user_id)
      errors.add(:user_id, 'user does not exist')
    end
  end
end

Cascading deletes require application-level coordination. Deleting a user necessitates finding and deleting related records across all shards.

Query Planning Without Shard Key

Queries lacking the shard key require scatter-gather across all shards. These queries scale poorly as shard count increases:

# Missing shard key: must query all shards
User.where(email: 'alice@example.com').first

# Each shard must be queried
shards.each do |shard|
  shard.users.where(email: 'alice@example.com').first
end

Index secondary attributes on each shard, but recognize cross-shard queries remain expensive. Consider denormalizing frequently queried attributes into shard key selection.

Connection Pool Exhaustion

Each shard maintains a connection pool. A 10-shard system with 25 connections per pool requires 250 total database connections. Connection pool exhaustion manifests as timeout errors:

# Monitor connection pool saturation
SHARDS.each do |name, shard|
  pool = shard.connection_pool
  utilization = pool.connections.size.to_f / pool.size
  
  if utilization > 0.8
    Rails.logger.warn("Shard #{name} connection pool at #{utilization * 100}%")
  end
end

Shard Count Changes

Modulo-based hash sharding breaks when shard count changes. The hash value user_id % 3 produces different results than user_id % 4, requiring data movement across all shards. Consistent hashing or directory-based mapping minimize data movement:

# Modulo sharding: adding shard 4 changes routing
user_id = 12345
12345 % 3  # => 0 (shard 0)
12345 % 4  # => 1 (shard 1) - different shard!

# Most records must move when shard count changes

Backup and Recovery Complexity

Each shard requires independent backup and point-in-time recovery. Restoring consistent state across shards requires coordinated snapshots. Ensure backup timing aligns across shards:

# Coordinated backup across shards
def backup_all_shards
  timestamp = Time.now.utc
  
  shards.each do |shard|
    # Trigger backup at same logical point
    shard.connection.execute(
      "SELECT pg_create_restore_point('backup_#{timestamp}')"
    )
  end
end

Monitoring and Observability

Sharded systems require monitoring each database independently. Aggregate metrics hide per-shard issues. A single slow shard degrades overall performance but may not appear in system-wide averages:

# Per-shard metrics
shards.each do |name, shard|
  metrics = {
    query_time: shard.connection.execute('SELECT avg(query_time) FROM pg_stat_statements'),
    connection_count: shard.connection_pool.connections.size,
    disk_usage: shard.connection.execute('SELECT pg_database_size(current_database())')
  }
  
  StatsD.gauge("shard.#{name}.query_time", metrics[:query_time])
  StatsD.gauge("shard.#{name}.connections", metrics[:connection_count])
  StatsD.gauge("shard.#{name}.disk_usage", metrics[:disk_usage])
end

Reference

Sharding Strategies

Query Pattern Performance

Connection Pool Sizing

Rebalancing Data Movement

Common Shard Key Characteristics

Ruby Sharding Libraries

Monitoring Metrics

Migration Checklist