Database Sharding Manager

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: Dict[int, ShardConfig] = {}  # Hash -> Shard mapping
        self.shards: Dict[int, ShardConfig] = {}  # Shard ID -> Config

    def add_shard(self, shard: ShardConfig) -> None:
        """Add shard to consistent hash ring with virtual nodes."""
        self.shards[shard.shard_id] = shard

        # Create virtual nodes weighted by shard capacity
        num_vnodes = self.virtual_nodes * shard.weight

        for i in range(num_vnodes):
            # Create unique hash for each virtual node
            vnode_key = f"{shard.shard_id}:{shard.host}:{i}"
            hash_value = self._hash(vnode_key)

            # Insert into sorted ring
            bisect.insort(self.ring, hash_value)
            self.ring_map[hash_value] = shard

        logger.info(
            f"Added shard {shard.shard_id} ({shard.host}) with {num_vnodes} virtual nodes"
        )

    def remove_shard(self, shard_id: int) -> None:
        """Remove shard from hash ring."""
        if shard_id not in self.shards:
            raise ValueError(f"Shard {shard_id} not found")

        shard = self.shards[shard_id]

        # Remove all virtual nodes for this shard
        num_vnodes = self.virtual_nodes * shard.weight
        removed_count = 0

        for i in range(num_vnodes):
            vnode_key = f"{shard.shard_id}:{shard.host}:{i}"
            hash_value = self._hash(vnode_key)

            if hash_value in self.ring_map:
                self.ring.remove(hash_value)
                del self.ring_map[hash_value]
                removed_count += 1

        del self.shards[shard_id]

        logger.info(
            f"Removed shard {shard_id} ({removed_count} virtual nodes)"
        )

    def get_shard(self, key: str) -> Optional[ShardConfig]:
        """
        Find shard for given key using consistent hashing.

        Args:
            key: Sharding key (user_id, tenant_id, etc.)

        Returns:
            ShardConfig for the shard responsible for this key
        """
        if not self.ring:
            raise ValueError("No shards available in hash ring")

        key_hash = self._hash(key)

        # Find first hash value >= key_hash (clockwise search)
        idx = bisect.bisect_right(self.ring, key_hash)

        # Wrap around to beginning if at end of ring
        if idx == len(self.ring):
            idx = 0

        shard = self.ring_map[self.ring[idx]]

        # Skip if shard is in maintenance
        if shard.status == 'maintenance':
            logger.warning(f"Shard {shard.shard_id} in maintenance, finding alternate")
            return self._find_next_active_shard(idx)

        return shard

    def _find_next_active_shard(self, start_idx: int) -> Optional[ShardConfig]:
        """Find next active shard in ring, skipping maintenance shards."""
        for i in range(len(self.ring)):
            idx = (start_idx + i) % len(self.ring)
            shard = self.ring_map[self.ring[idx]]

            if shard.status == 'active':
                return shard

        raise ValueError("No active shards available")

    def _hash(self, key: str) -> int:
        """
        Generate consistent hash value for key.

        Uses MD5 for speed. SHA256 is more secure but slower.
        """
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def get_shard_distribution(self) -> Dict[int, int]:
        """Analyze key distribution across shards (for testing)."""
        distribution = {shard_id: 0 for shard_id in self.shards}

        # Sample 10000 keys to estimate distribution
        for i in range(10000):
            shard = self.get_shard(str(i))
            distribution[shard.shard_id] += 1

        return distribution

    def rebalance_check(self) -> Dict[str, Any]:
        """
        Check if shards are balanced and recommend rebalancing.

        Returns:
            Dict with balance metrics and recommendations
        """
        distribution = self.get_shard_distribution()

        total = sum(distribution.values())
        expected_per_shard = total / len(self.shards)

        imbalance = {}
        for shard_id, count in distribution.items():
            deviation = abs(count - expected_per_shard) / expected_per_shard * 100
            imbalance[shard_id] = {
                'count': count,
                'expected': expected_per_shard,
                'deviation_percent': round(deviation, 2)
            }

        max_deviation = max(s['deviation_percent'] for s in imbalance.values())

        return {
            'balanced': max_deviation < 10,  # <10% deviation is acceptable
            'max_deviation_percent': max_deviation,
            'shard_distribution': imbalance,
            'recommendation': (
                'Rebalancing recommended' if max_deviation > 20
                else 'Distribution acceptable'
            )
        }

# Usage example
if __name__ == "__main__":
    # Initialize router
    router = ConsistentHashRouter(virtual_nodes=150)

    # Add shards
    router.add_shard(ShardConfig(
        shard_id=1,
        host='shard1.db.example.com',
        port=5432,
        database='myapp_shard1',
        weight=1
    ))

    router.add_shard(ShardConfig(
        shard_id=2,
        host='shard2.db.example.com',
        port=5432,
        database='myapp_shard2',
        weight=2  # Double capacity
    ))

    router.add_shard(ShardConfig(
        shard_id=3,
        host='shard3.db.example.com',
        port=5432,
        database='myapp_shard3',
        weight=1
    ))

    # Route queries
    user_id = "user_12345"
    shard = router.get_shard(user_id)
    print(f"User {user_id} → Shard {shard.shard_id} ({shard.host})")

    # Check balance
    balance_report = router.rebalance_check()
    print(f"\nBalance report:")
    print(f"  Balanced: {balance_report['balanced']}")
    print(f"  Max deviation: {balance_report['max_deviation_percent']}%")

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
            });

            this.shards.set(config.shardId, {
                config,
                pool,
                stats: {
                    queries: 0,
                    errors: 0,
                    avgLatency: 0
                }
            });

            this.addToRing(config);
        });

        console.log(`Initialized ${this.shards.size} shards with ${this.ring.length} virtual nodes`);
    }

    addToRing(config) {
        const numVNodes = this.virtualNodes * (config.weight || 1);

        for (let i = 0; i < numVNodes; i++) {
            const vnodeKey = `${config.shardId}:${config.host}:${i}`;
            const hash = this.hash(vnodeKey);

            this.ring.push(hash);
            this.ringMap.set(hash, config.shardId);
        }

        // Sort ring for binary search
        this.ring.sort((a, b) => a - b);
    }

    hash(key) {
        return parseInt(
            crypto.createHash('md5').update(key).digest('hex').substring(0, 8),
            16
        );
    }

    getShardId(key) {
        if (this.ring.length === 0) {
            throw new Error('No shards available');
        }

        const keyHash = this.hash(key);

        // Binary search for next hash >= keyHash
        let idx = this.ring.findIndex(h => h >= keyHash);

        if (idx === -1) {
            idx = 0;  // Wrap around
        }

        return this.ringMap.get(this.ring[idx]);
    }

    async query(shardKey, sql, params = []) {
        const shardId = this.getShardId(shardKey);
        const shard = this.shards.get(shardId);

        if (!shard) {
            throw new Error(`Shard ${shardId} not found`);
        }

        const startTime = Date.now();

        try {
            const result = await shard.pool.query(sql, params);

            // Update stats
            shard.stats.queries++;
            const latency = Date.now() - startTime;
            shard.stats.avgLatency =
                (shard.stats.avgLatency * (shard.stats.queries - 1) + latency) /
                shard.stats.queries;

            return result;

        } catch (error) {
            shard.stats.errors++;
            console.error(`Query error on shard ${shardId}:`, error);
            throw error;
        }
    }

    async queryMultipleShards(sql, params = []) {
        /**
         * Execute query across all shards and merge results.
         * Use sparingly - cross-shard queries are expensive.
         */
        const promises = Array.from(this.shards.values()).map(async shard => {
            try {
                const result = await shard.pool.query(sql, params);
                return {
                    shardId: shard.config.shardId,
                    rows: result.rows,
                    success: true
                };
            } catch (error) {
                return {
                    shardId: shard.config.shardId,
                    error: error.message,
                    success: false
                };
            }
        });

        const results = await Promise.all(promises);

        // Merge rows from all shards
        const allRows = results
            .filter(r => r.success)
            .flatMap(r => r.rows);

        return {
            rows: allRows,
            shardResults: results
        };
    }

    async transaction(shardKey, callback) {
        /**
         * Execute transaction on specific shard.
         * Cross-shard transactions require 2PC (not implemented).
         */
        const shardId = this.getShardId(shardKey);
        const shard = this.shards.get(shardId);

        const client = await shard.pool.connect();

        try {
            await client.query('BEGIN');
            const result = await callback(client);
            await client.query('COMMIT');
            return result;
        } catch (error) {
            await client.query('ROLLBACK');
            throw error;
        } finally {
            client.release();
        }
    }

    getStats() {
        const stats = {};

        for (const [shardId, shard] of this.shards) {
            stats[shardId] = {
                ...shard.stats,
                poolSize: shard.pool.totalCount,
                idleConnections: shard.pool.idleCount,
                waitingClients: shard.pool.waitingCount
            };
        }

        return stats;
    }

    async close() {
        for (const shard of this.shards.values()) {
            await shard.pool.end();
        }
    }
}

// Usage example
const shardPool = new ShardConnectionPool([
    {
        shardId: 1,
        host: 'shard1.db.example.com',
        port: 5432,
        database: 'myapp_shard1',
        user: 'app_user',
        password: 'password',
        weight: 1
    },
    {
        shardId: 2,
        host: 'shard2.db.example.com',
        port: 5432,
        database: 'myapp_shard2',
        user: 'app_user',
        password: 'password',
        weight: 2
    }
]);

// Single-shard query
const userId = 'user_12345';
const user = await shardPool.query(
    userId,
    'SELECT * FROM users WHERE user_id = $1',
    [userId]
);

// Cross-shard query (expensive - avoid if possible)
const allActiveUsers = await shardPool.queryMultipleShards(
    'SELECT * FROM users WHERE status = $1',
    ['active']
);

console.log(`Found ${allActiveUsers.rows.length} active users across all shards`);

// Transaction on specific shard
await shardPool.transaction(userId, async (client) => {
    await client.query(
        'UPDATE users SET balance = balance - $1 WHERE user_id = $2',
        [100, userId]
    );

    await client.query(
        'INSERT INTO transactions (user_id, amount, type) VALUES ($1, $2, $3)',
        [userId, -100, 'withdrawal']
    );
});

// Monitor shard health
setInterval(() => {
    const stats = shardPool.getStats();
    console.log('Shard statistics:', JSON.stringify(stats, null, 2));
}, 60000);

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."""
        if shard.region not in self.region_shards:
            self.region_shards[shard.region] = []

        self.region_shards[shard.region].append(shard)
        print(f"Added shard {shard.shard_id} for region {shard.region.value}")

    def assign_user_region(self, user_id: str, region: Region) -> None:
        """Assign user to geographic region (permanent assignment)."""
        if user_id in self.user_region_map:
            raise ValueError(
                f"User {user_id} already assigned to {self.user_region_map[user_id]}"
            )

        self.user_region_map[user_id] = region
        print(f"Assigned user {user_id} to region {region.value}")

    def get_shard_for_user(self, user_id: str) -> Optional[GeoShardConfig]:
        """Get shard for user based on regional assignment."""
        region = self.user_region_map.get(user_id)

        if not region:
            raise ValueError(f"User {user_id} not assigned to any region")

        shards = self.region_shards.get(region)

        if not shards:
            raise ValueError(f"No shards available for region {region.value}")

        # Simple round-robin across shards in region
        # Could use consistent hashing within region for better distribution
        shard_idx = hash(user_id) % len(shards)
        return shards[shard_idx]

    def validate_data_residency(self, user_id: str, shard: GeoShardConfig) -> bool:
        """Ensure data residency compliance before query execution."""
        user_region = self.user_region_map.get(user_id)

        if user_region != shard.region:
            raise ValueError(
                f"Data residency violation: User {user_id} in {user_region.value} "
                f"attempting access to shard in {shard.region.value}"
            )

        return True

# Usage
geo_router = GeographicShardRouter()

# Add region-specific shards
geo_router.add_region_shard(GeoShardConfig(
    region=Region.US_EAST,
    shard_id=1,
    host='us-east-shard1.db.example.com',
    port=5432,
    database='myapp_us_east'
))

geo_router.add_region_shard(GeoShardConfig(
    region=Region.EU_WEST,
    shard_id=2,
    host='eu-west-shard1.db.example.com',
    port=5432,
    database='myapp_eu_west',
    data_residency_compliant=True
))

# Assign users to regions (based on signup location)
geo_router.assign_user_region('user_us_12345', Region.US_EAST)
geo_router.assign_user_region('user_eu_67890', Region.EU_WEST)

# Route queries to correct regional shard
us_user_shard = geo_router.get_shard_for_user('user_us_12345')
print(f"US user → {us_user_shard.host}")

eu_user_shard = geo_router.get_shard_for_user('user_eu_67890')
print(f"EU user → {eu_user_shard.host}")

Error Handling

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