-- Good shard key examples
-- For user-centric applications
CREATE TABLE users (
    user_id BIGINT PRIMARY KEY,
    email VARCHAR(255),
    created_at TIMESTAMP
) PARTITION BY HASH(user_id);

-- For time-series data
CREATE TABLE events (
    event_id BIGINT,
    user_id BIGINT,
    event_time TIMESTAMP,
    data JSONB
) PARTITION BY RANGE(event_time);

-- Composite shard key for tenant isolation
CREATE TABLE tenant_data (
    tenant_id INT,
    record_id BIGINT,
    data JSONB,
    PRIMARY KEY (tenant_id, record_id)
) PARTITION BY HASH(tenant_id);

    for shard in shards:
        self.add_shard(shard)

def add_shard(self, shard):
    for i in range(self.replicas):
        key = self._hash(f"{shard}:{i}")
        self.ring[key] = shard
    self.sorted_keys = sorted(self.ring.keys())

def get_shard(self, key):
    if not self.ring:
        return None

    hash_key = self._hash(key)
    for ring_key in self.sorted_keys:
        if hash_key <= ring_key:
            return self.ring[ring_key]
    return self.ring[self.sorted_keys[0]]

def _hash(self, key):
    return int(hashlib.md5(str(key).encode()).hexdigest(), 16)

## Маршрутизация запросов и управление подключениями

### Маршрутизация на уровне приложения
```python
class ShardRouter:
    def __init__(self):
        self.connections = {
            'shard_0': create_connection('db-shard-0'),
            'shard_1': create_connection('db-shard-1'),
            'shard_2': create_connection('db-shard-2')
        }
        self.hash_ring = ConsistentHashRing(list(self.connections.keys()))

    def execute_query(self, shard_key, query, params=None):
        shard = self.hash_ring.get_shard(shard_key)
        connection = self.connections[shard]
        return connection.execute(query, params)

    def execute_cross_shard_query(self, query, params=None):
        results = []
        for shard, conn in self.connections.items():
            try:
                result = conn.execute(query, params)
                results.extend(result)
            except Exception as e:
                logger.error(f"Query failed on {shard}: {e}")
        return results

    def transaction_across_shards(self, operations):
        # Two-phase commit for cross-shard transactions
        prepared_transactions = []

        # Phase 1: Prepare
        for shard_key, operation in operations:
            shard = self.hash_ring.get_shard(shard_key)
            conn = self.connections[shard]
            tx_id = conn.begin_transaction()
            conn.execute(operation['query'], operation['params'])
            prepared_transactions.append((shard, tx_id, conn))

        # Phase 2: Commit or Rollback
        try:
            for shard, tx_id, conn in prepared_transactions:
                conn.commit_transaction(tx_id)
        except Exception as e:
            for shard, tx_id, conn in prepared_transactions:
                conn.rollback_transaction(tx_id)
            raise e

class ShardMigrator:
    def __init__(self, source_shard, target_shard):
        self.source = source_shard
        self.target = target_shard
        self.migration_state = {}

    def migrate_data(self, table_name, batch_size=1000):
        # Step 1: Create shadow table in target shard
        self.target.execute(f"CREATE TABLE {table_name}_shadow LIKE {table_name}")

        # Step 2: Copy existing data in batches
        offset = 0
        while True:
            rows = self.source.execute(
                f"SELECT * FROM {table_name} ORDER BY id LIMIT {batch_size} OFFSET {offset}"
            )

            if not rows:
                break

            self.target.execute_batch(
                f"INSERT INTO {table_name}_shadow VALUES (%s)", rows
            )
            offset += batch_size

        # Step 3: Set up change capture for ongoing writes
        self.setup_change_capture(table_name)

        # Step 4: Atomic switch
        self.atomic_switch(table_name)

    def setup_change_capture(self, table_name):
        # Use database-specific change capture (CDC, triggers, etc.)
        trigger_sql = f"""
        CREATE TRIGGER {table_name}_migration_trigger
        AFTER INSERT OR UPDATE OR DELETE ON {table_name}
        FOR EACH ROW EXECUTE FUNCTION replicate_to_target_shard();
        """
        self.source.execute(trigger_sql)

class ShardMonitor:
    def __init__(self, shards):
        self.shards = shards
        self.metrics = {}

    def collect_metrics(self):
        for shard_name, connection in self.shards.items():
            metrics = {
                'connection_count': self.get_connection_count(connection),
                'query_latency': self.measure_latency(connection),
                'storage_size': self.get_storage_size(connection),
                'query_rate': self.get_query_rate(connection),
                'hot_partitions': self.detect_hot_partitions(connection)
            }
            self.metrics[shard_name] = metrics

    def detect_imbalance(self):
        sizes = [m['storage_size'] for m in self.metrics.values()]
        avg_size = sum(sizes) / len(sizes)

        imbalanced_shards = []
        for shard, metrics in self.metrics.items():
            if metrics['storage_size'] > avg_size * 1.5:  # 50% above average
                imbalanced_shards.append(shard)

        return imbalanced_shards

    def recommend_rebalancing(self):
        imbalanced = self.detect_imbalance()
        if imbalanced:
            return {
                'action': 'rebalance',
                'shards': imbalanced,
                'strategy': 'split_hot_ranges'
            }
        return {'action': 'none'}

-- Good: Query includes shard key
SELECT * FROM orders 
WHERE customer_id = 12345 AND status = 'pending';

-- Bad: Missing shard key requires scatter-gather
SELECT * FROM orders WHERE status = 'pending';

-- Solution: Maintain lookup tables or use search indexes
CREATE TABLE order_status_index (
    status VARCHAR(50),
    customer_id BIGINT,
    order_id BIGINT,
    shard_location VARCHAR(50)
);