sharding

name	sharding
description	Implement horizontal database sharding for massive scale applications
shortcut	shar

Database Sharding Manager

Design and implement horizontal database sharding strategies to distribute data across multiple database instances, enabling applications to scale beyond single-server limitations with consistent hashing, automatic rebalancing, and cross-shard query coordination.

When to Use This Command

Use /sharding when you need to:

• Scale beyond single database server capacity (>10TB or >100k QPS)
• Distribute write load across multiple database servers
• Improve query performance through data locality
• Implement geographic data distribution for GDPR/data residency
• Reduce blast radius of database failures (isolate tenant data)
• Support multi-tenant SaaS with tenant-level isolation

DON'T use this when:

• Database is small (<1TB) and performing well
• Can solve with read replicas and caching instead
• Application can't handle distributed transactions complexity
• Team lacks expertise in distributed systems
• Cross-shard queries are majority of workload (use partitioning instead)

Design Decisions

This command implements consistent hashing with virtual nodes because:

• Minimizes data movement when adding/removing shards (only K/n keys move)
• Distributes load evenly across shards with virtual nodes
• Supports gradual shard addition without downtime
• Enables geographic routing for data residency compliance
• Provides automatic failover with shard replica promotion

Alternative considered: Range-based sharding

• Simple to implement and understand
• Predictable data distribution
• Prone to hotspots if key distribution uneven
• Recommended for time-series data with sequential IDs

Alternative considered: Directory-based sharding

• Flexible shard assignment with lookup table
• Easy to move individual records
• Single point of failure (directory lookup)
• Recommended for small-scale or initial implementations

Prerequisites

Before running this command:

1. Application supports sharding-aware database connections
2. Clear understanding of sharding key (immutable, high cardinality)
3. Strategy for handling cross-shard queries and joins
4. Monitoring infrastructure for shard health
5. Migration plan from single database to sharded architecture

Implementation Process

Step 1: Choose Sharding Strategy

Select sharding approach based on data access patterns and scale requirements.

Step 2: Design Shard Key

Choose immutable, high-cardinality key that distributes data evenly (user_id, tenant_id).

Step 3: Implement Shard Routing Layer

Build connection pooling and routing logic to direct queries to correct shard.

Step 4: Migrate Data to Shards

Perform zero-downtime migration from monolithic to sharded architecture.

Step 5: Monitor and Rebalance

Track shard load distribution and rebalance data as needed.

Output Format

The command generates:

• sharding/shard_router.py - Consistent hashing router implementation
• sharding/shard_manager.js - Shard connection pool manager
• migration/shard_migration.sql - Data migration scripts per shard
• monitoring/shard_health.sql - Per-shard metrics and health checks
• docs/sharding_architecture.md - Architecture documentation and runbooks

Code Examples

Example 1: Consistent Hashing Shard Router with Virtual Nodes

# sharding/consistent_hash_router.py
import hashlib
import bisect
from typing import List, Dict, Optional, Any
from dataclasses import dataclass
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ShardConfig:
    """Configuration for a database shard."""
    shard_id: int
    host: str
    port: int
    database: str
    weight: int = 1  # Relative weight for load distribution
    status: str = 'active'  # active, readonly, maintenance

class ConsistentHashRouter:
    """
    Consistent hashing implementation with virtual nodes.

    Virtual nodes ensure even distribution even with heterogeneous shard sizes.
    Adding/removing shards only affects K/n keys where n = number of shards.
    """

    def __init__(self, virtual_nodes: int = 150):
        """
        Initialize consistent hash ring.

        Args:
            virtual_nodes: Number of virtual nodes per physical shard.
                          More nodes = better distribution, higher memory usage.
        """
        self.virtual_nodes = virtual_nodes
        self.ring: List[int] = []  # Sorted hash values
        self.ring_map: [, ShardConfig] = {}  
        .shards: [, ShardConfig] = {}  

     () -> :
        
        .shards[shard.shard_id] = shard

        
        num_vnodes = .virtual_nodes * shard.weight

         i  (num_vnodes):
            
            vnode_key = 
            hash_value = ._(vnode_key)

            
            bisect.insort(.ring, hash_value)
            .ring_map[hash_value] = shard

        logger.info(
            
        )

     () -> :
        
         shard_id   .shards:
             ValueError()

        shard = .shards[shard_id]

        
        num_vnodes = .virtual_nodes * shard.weight
        removed_count = 

         i  (num_vnodes):
            vnode_key = 
            hash_value = ._(vnode_key)

             hash_value  .ring_map:
                .ring.remove(hash_value)
                 .ring_map[hash_value]
                removed_count += 

         .shards[shard_id]

        logger.info(
            
        )

     () -> [ShardConfig]:
        
          .ring:
             ValueError()

        key_hash = ._(key)

        
        idx = bisect.bisect_right(.ring, key_hash)

        
         idx == (.ring):
            idx = 

        shard = .ring_map[.ring[idx]]

        
         shard.status == :
            logger.warning()
             ._find_next_active_shard(idx)

         shard

     () -> [ShardConfig]:
        
         i  ((.ring)):
            idx = (start_idx + i) % (.ring)
            shard = .ring_map[.ring[idx]]

             shard.status == :
                 shard

         ValueError()

     () -> :
        
         (hashlib.md5(key.encode()).hexdigest(), )

     () -> [, ]:
        
        distribution = {shard_id:   shard_id  .shards}

        
         i  ():
            shard = .get_shard((i))
            distribution[shard.shard_id] += 

         distribution

     () -> [, ]:
        
        distribution = .get_shard_distribution()

        total = (distribution.values())
        expected_per_shard = total / (.shards)

        imbalance = {}
         shard_id, count  distribution.items():
            deviation = (count - expected_per_shard) / expected_per_shard * 
            imbalance[shard_id] = {
                : count,
                : expected_per_shard,
                : (deviation, )
            }

        max_deviation = (s[]  s  imbalance.values())

         {
            : max_deviation < ,  
            : max_deviation,
            : imbalance,
            : (
                  max_deviation > 
                 
            )
        }


 __name__ == :
    
    router = ConsistentHashRouter(virtual_nodes=)

    
    router.add_shard(ShardConfig(
        shard_id=,
        host=,
        port=,
        database=,
        weight=
    ))

    router.add_shard(ShardConfig(
        shard_id=,
        host=,
        port=,
        database=,
        weight=  
    ))

    router.add_shard(ShardConfig(
        shard_id=,
        host=,
        port=,
        database=,
        weight=
    ))

    
    user_id = 
    shard = router.get_shard(user_id)
    ()

    
    balance_report = router.rebalance_check()
    ()
    ()
    ()

Example 2: Shard-Aware Database Connection Pool

// sharding/shard_connection_pool.js
const { Pool } = require('pg');
const crypto = require('crypto');

class ShardConnectionPool {
    constructor(shardConfigs) {
        this.shards = new Map();
        this.virtualNodes = 150;
        this.ring = [];
        this.ringMap = new Map();

        // Initialize connection pools for each shard
        shardConfigs.forEach(config => {
            const pool = new Pool({
                host: config.host,
                port: config.port,
                database: config.database,
                user: config.user,
                password: config.password,
                max: 20,  // Max connections per shard
                idleTimeoutMillis: 30000,
                connectionTimeoutMillis: 2000
            });

            ..(config., {
                config,
                pool,
                : {
                    : ,
                    : ,
                    : 
                }
            });

            .(config);
        });

        .();
    }

    () {
         numVNodes = . * (config. || );

         ( i = ; i < numVNodes; i++) {
             vnodeKey = ;
             hash = .(vnodeKey);

            ..(hash);
            ..(hash, config.);
        }

        
        ..( a - b);
    }

    () {
         (
            crypto.().(key).().(, ),
            
        );
    }

    () {
         (.. === ) {
              ();
        }

         keyHash = .(key);

        
         idx = ..( h >= keyHash);

         (idx === -) {
            idx = ;  
        }

         ..(.[idx]);
    }

     () {
         shardId = .(shardKey);
         shard = ..(shardId);

         (!shard) {
              ();
        }

         startTime = .();

         {
             result =  shard..(sql, params);

            
            shard..++;
             latency = .() - startTime;
            shard.. =
                (shard.. * (shard.. - ) + latency) /
                shard..;

             result;

        }  (error) {
            shard..++;
            .(, error);
             error;
        }
    }

     () {
        
         promises = .(..()).( shard => {
             {
                 result =  shard..(sql, params);
                 {
                    : shard..,
                    : result.,
                    : 
                };
            }  (error) {
                 {
                    : shard..,
                    : error.,
                    : 
                };
            }
        });

         results =  .(promises);

        
         allRows = results
            .( r.)
            .( r.);

         {
            : allRows,
            : results
        };
    }

     () {
        
         shardId = .(shardKey);
         shard = ..(shardId);

         client =  shard..();

         {
             client.();
             result =  (client);
             client.();
             result;
        }  (error) {
             client.();
             error;
        }  {
            client.();
        }
    }

    () {
         stats = {};

         ( [shardId, shard]  .) {
            stats[shardId] = {
                ...shard.,
                : shard..,
                : shard..,
                : shard..
            };
        }

         stats;
    }

     () {
         ( shard  ..()) {
             shard..();
        }
    }
}


 shardPool =  ([
    {
        : ,
        : ,
        : ,
        : ,
        : ,
        : ,
        : 
    },
    {
        : ,
        : ,
        : ,
        : ,
        : ,
        : ,
        : 
    }
]);


 userId = ;
 user =  shardPool.(
    userId,
    ,
    [userId]
);


 allActiveUsers =  shardPool.(
    ,
    []
);

.();


 shardPool.(userId,  (client) => {
     client.(
        ,
        [, userId]
    );

     client.(
        ,
        [userId, -, ]
    );
});


( {
     stats = shardPool.();
    .(, .(stats, , ));
}, );

Example 3: Geographic Sharding with Data Residency

# sharding/geo_shard_router.py
from typing import Dict, Optional
from dataclasses import dataclass
from enum import Enum

class Region(Enum):
    """Geographic regions for data residency compliance."""
    US_EAST = 'us-east'
    US_WEST = 'us-west'
    EU_WEST = 'eu-west'
    ASIA_PACIFIC = 'asia-pacific'

@dataclass
class GeoShardConfig:
    region: Region
    shard_id: int
    host: str
    port: int
    database: str
    data_residency_compliant: bool = True

class GeographicShardRouter:
    """
    Route queries to region-specific shards for GDPR/data residency compliance.

    Each user/tenant is assigned to a geographic region and all their data
    resides in shards within that region.
    """

    def __init__(self):
        self.region_shards: Dict[Region, list[GeoShardConfig]] = {}
        self.user_region_map: Dict[str, Region] = {}  # user_id -> region

    def add_region_shard(self, shard: GeoShardConfig) -> None:
        """Add shard for specific geographic region."""
         shard.region   .region_shards:
            .region_shards[shard.region] = []

        .region_shards[shard.region].append(shard)
        ()

     () -> :
        
         user_id  .user_region_map:
             ValueError(
                
            )

        .user_region_map[user_id] = region
        ()

     () -> [GeoShardConfig]:
        
        region = .user_region_map.get(user_id)

          region:
             ValueError()

        shards = .region_shards.get(region)

          shards:
             ValueError()

        
        
        shard_idx = (user_id) % (shards)
         shards[shard_idx]

     () -> :
        
        user_region = .user_region_map.get(user_id)

         user_region != shard.region:
             ValueError(
                
                
            )

         


geo_router = GeographicShardRouter()


geo_router.add_region_shard(GeoShardConfig(
    region=Region.US_EAST,
    shard_id=,
    host=,
    port=,
    database=
))

geo_router.add_region_shard(GeoShardConfig(
    region=Region.EU_WEST,
    shard_id=,
    host=,
    port=,
    database=,
    data_residency_compliant=
))


geo_router.assign_user_region(, Region.US_EAST)
geo_router.assign_user_region(, Region.EU_WEST)


us_user_shard = geo_router.get_shard_for_user()
()

eu_user_shard = geo_router.get_shard_for_user()
()

Error Handling

Error	Cause	Solution
"No shards available"	All shards offline or empty ring	Add at least one shard, check shard health
"Cross-shard foreign key violation"	Reference to data on different shard	Denormalize data or use application-level joins
"Shard rebalancing in progress"	Data migration active	Retry query or route to new shard
"Distributed transaction failure"	2PC coordinator unreachable	Implement saga pattern or idempotent operations
"Hotspot detected on shard"	Uneven key distribution	Rebalance with more virtual nodes or reshard

Configuration Options

Sharding Strategies

• consistent_hash: Best for even distribution, minimal rebalancing
• range: Simple, good for time-series, prone to hotspots
• directory: Flexible, requires lookup table maintenance
• geographic: Data residency compliance, region isolation

Virtual Nodes

• 50-100: Faster routing, less even distribution
• 150-200: Balanced (recommended for production)
• 300+: Most even distribution, higher memory usage

Connection Pooling

• max_connections_per_shard: 10-50 depending on load
• idle_timeout: 30-60 seconds
• connection_timeout: 2-5 seconds

Best Practices

DO:

• Use immutable, high-cardinality shard keys (user_id, tenant_id)
• Implement connection pooling per shard
• Monitor shard load distribution continuously
• Design for cross-shard query minimization
• Use read replicas within shards for scale
• Plan shard capacity for 2-3 years growth

DON'T:

• Use mutable shard keys (email, username can change)
• Perform JOINs across shards (denormalize instead)
• Ignore shard imbalance (leads to hotspots)
• Add shards without capacity planning
• Skip monitoring per-shard metrics
• Use distributed transactions without strong justification

Performance Considerations

• Shard routing adds ~1-5ms latency per query
• Cross-shard queries 10-100x slower than single-shard
• Adding shard affects K/n keys where K=total keys, n=shard count
• Virtual nodes increase routing time O(log(v*n)) but improve distribution
• Connection pool per shard adds memory overhead (~10MB per pool)
• Rebalancing requires dual-write period (5-10% overhead)

Related Commands

• /database-partition-manager - Partition tables within shards
• /database-replication-manager - Set up replicas per shard
• /database-migration-manager - Migrate data between shards
• /database-health-monitor - Monitor per-shard health metrics

Version History

• v1.0.0 (2024-10): Initial implementation with consistent hashing and geographic routing
• Planned v1.1.0: Add automatic shard rebalancing and distributed transaction support