Partitioning & Sharding Strategies

Partitioned Table Architecture:

Logical Table: orders (1 billion rows)
┌─────────────────────────────────────────────────┐
│         Application sees: SELECT * FROM orders  │
└─────────────────────────────────────────────────┘
                         │
        ┌────────────────┴────────────────┐
        │    Database Query Planner       │
        │  (Partition Pruning)            │
        └────────────────┬────────────────┘
                         │
        ┌────────────────┴────────────────────┐
        │                                     │
        ▼                                     ▼
┌──────────────────┐              ┌──────────────────┐
│ orders_2023      │              │ orders_2024      │
│ 400M rows        │     ...      │ 500M rows        │
│ (Jan-Dec 2023)   │              │ (Jan-Dec 2024)   │
└──────────────────┘              └──────────────────┘

Benefits:
• Query only scans relevant partition(s)
• Indexes are smaller (per partition)
• Maintenance easier (DROP old partitions)
• Parallel processing (scan partitions in parallel)

-- Create parent partitioned table
CREATE TABLE orders (
    order_id BIGSERIAL,
    customer_id INTEGER NOT NULL,
    order_date DATE NOT NULL,
    total_amount DECIMAL(10, 2),
    status VARCHAR(20)
) PARTITION BY RANGE (order_date);

-- Create partitions for each year
CREATE TABLE orders_2022 PARTITION OF orders
    FOR VALUES FROM ('2022-01-01') TO ('2023-01-01');

CREATE TABLE orders_2023 PARTITION OF orders
    FOR VALUES FROM ('2023-01-01') TO ('2024-01-01');

CREATE TABLE orders_2024 PARTITION OF orders
    FOR VALUES FROM ('2024-01-01') TO ('2025-01-01');

-- Create default partition for future dates
CREATE TABLE orders_default PARTITION OF orders DEFAULT;

import psycopg2
from datetime import datetime

conn = psycopg2.connect("dbname=ecommerce user=postgres")
cursor = conn.cursor()

# Insert works transparently - database routes to correct partition
cursor.execute("""
    INSERT INTO orders (customer_id, order_date, total_amount, status)
    VALUES (%s, %s, %s, %s)
""", (12345, datetime(2024, 6, 15), 299.99, 'completed'))

conn.commit()

# Query with date filter - partition pruning in action
cursor.execute("""
    SELECT COUNT(*), AVG(total_amount)
    FROM orders
    WHERE order_date >= '2024-01-01' AND order_date < '2025-01-01'
""")

count, avg_amount = cursor.fetchone()
print(f"2024 orders: {count}, Average: ${avg_amount:.2f}")

# EXPLAIN shows partition pruning
cursor.execute("""
    EXPLAIN (ANALYZE, VERBOSE)
    SELECT * FROM orders
    WHERE order_date BETWEEN '2024-06-01' AND '2024-06-30'
""")

for line in cursor.fetchall():
    print(line[0])

# Automated partition creation script
import psycopg2
from datetime import datetime, timedelta

def create_monthly_partitions(year, month_count=12):
    """Create monthly partitions for the given year"""
    conn = psycopg2.connect("dbname=ecommerce user=postgres")
    cursor = conn.cursor()

    for month in range(1, month_count + 1):
        start_date = datetime(year, month, 1)

        # Calculate next month
        if month == 12:
            end_date = datetime(year + 1, 1, 1)
        else:
            end_date = datetime(year, month + 1, 1)

        partition_name = f"orders_{year}_{month:02d}"

        cursor.execute(f"""
            CREATE TABLE IF NOT EXISTS {partition_name}
            PARTITION OF orders
            FOR VALUES FROM ('{start_date.date()}') TO ('{end_date.date()}')
        """)

        print(f"Created partition: {partition_name}")

    conn.commit()
    cursor.close()
    conn.close()

# Create partitions for 2025
create_monthly_partitions(2025)

# Drop old partitions (archive first!)
def archive_and_drop_partition(partition_name):
    conn = psycopg2.connect("dbname=ecommerce user=postgres")
    cursor = conn.cursor()

    # Export to archive table first
    cursor.execute(f"""
        CREATE TABLE archive.{partition_name} AS
        SELECT * FROM {partition_name}
    """)

    # Detach partition (makes it a regular table)
    cursor.execute(f"ALTER TABLE orders DETACH PARTITION {partition_name}")

    # Drop the detached partition
    cursor.execute(f"DROP TABLE {partition_name}")

    conn.commit()
    cursor.close()
    conn.close()
    print(f"Archived and dropped: {partition_name}")

# Archive data older than 2 years
archive_and_drop_partition("orders_2022")

-- Partition by country (geography-based)
CREATE TABLE users (
    user_id BIGSERIAL,
    email VARCHAR(255),
    country_code CHAR(2) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) PARTITION BY LIST (country_code);

-- North America partition
CREATE TABLE users_north_america PARTITION OF users
    FOR VALUES IN ('US', 'CA', 'MX');

-- Europe partition
CREATE TABLE users_europe PARTITION OF users
    FOR VALUES IN ('GB', 'DE', 'FR', 'IT', 'ES', 'NL');

-- Asia partition
CREATE TABLE users_asia PARTITION OF users
    FOR VALUES IN ('CN', 'JP', 'IN', 'KR', 'SG');

-- Default partition for other countries
CREATE TABLE users_other PARTITION OF users DEFAULT;

import psycopg2

conn = psycopg2.connect("dbname=ecommerce user=postgres")
cursor = conn.cursor()

# Insert users from different countries
users = [
    ('alice@example.com', 'US'),
    ('bob@example.co.uk', 'GB'),
    ('chen@example.cn', 'CN'),
    ('diego@example.mx', 'MX')
]

for email, country in users:
    cursor.execute("""
        INSERT INTO users (email, country_code)
        VALUES (%s, %s)
    """, (email, country))

conn.commit()

# Query European users - scans only users_europe partition
cursor.execute("""
    SELECT COUNT(*) as euro_users
    FROM users
    WHERE country_code IN ('GB', 'DE', 'FR', 'IT', 'ES')
""")

euro_count = cursor.fetchone()[0]
print(f"European users: {euro_count}")

# Get partition distribution
cursor.execute("""
    SELECT
        tableoid::regclass AS partition_name,
        COUNT(*) as user_count
    FROM users
    GROUP BY tableoid
    ORDER BY user_count DESC
""")

print("\nUsers per partition:")
for partition, count in cursor.fetchall():
    print(f"  {partition}: {count} users")

-- Hash partition by user_id for even distribution
CREATE TABLE sessions (
    session_id UUID DEFAULT gen_random_uuid(),
    user_id BIGINT NOT NULL,
    login_time TIMESTAMP,
    logout_time TIMESTAMP,
    ip_address INET
) PARTITION BY HASH (user_id);

-- Create 8 hash partitions (power of 2 is recommended)
CREATE TABLE sessions_part_0 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 0);

CREATE TABLE sessions_part_1 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 1);

CREATE TABLE sessions_part_2 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 2);

CREATE TABLE sessions_part_3 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 3);

CREATE TABLE sessions_part_4 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 4);

CREATE TABLE sessions_part_5 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 5);

CREATE TABLE sessions_part_6 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 6);

CREATE TABLE sessions_part_7 PARTITION OF sessions
    FOR VALUES WITH (MODULUS 8, REMAINDER 7);

import psycopg2
from datetime import datetime, timedelta
import random

conn = psycopg2.connect("dbname=ecommerce user=postgres")
cursor = conn.cursor()

# Insert sessions for various users
print("Inserting 10,000 sessions...")
for i in range(10000):
    user_id = random.randint(1, 100000)
    login_time = datetime.now() - timedelta(days=random.randint(0, 30))

    cursor.execute("""
        INSERT INTO sessions (user_id, login_time)
        VALUES (%s, %s)
    """, (user_id, login_time))

conn.commit()

# Check distribution across partitions
cursor.execute("""
    SELECT
        tableoid::regclass AS partition,
        COUNT(*) as row_count,
        ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER(), 2) as percentage
    FROM sessions
    GROUP BY tableoid
    ORDER BY partition
""")

print("\nHash partition distribution:")
print(f"{'Partition':<25} {'Rows':<10} {'%'}")
print("-" * 45)
for partition, count, pct in cursor.fetchall():
    print(f"{partition:<25} {count:<10} {pct}%")

Horizontal Partitioning (Divide Rows):

Original Table: products (10 million rows)
┌──────────┬────────────┬───────┬──────────────┐
│ id       │ name       │ price │ category     │
├──────────┼────────────┼───────┼──────────────┤
│ 1        │ Widget A   │ 10.99 │ electronics  │
│ 2        │ Gadget B   │ 25.50 │ electronics  │
│ 3        │ Book C     │ 15.00 │ books        │
│ ...      │ ...        │ ...   │ ...          │
└──────────┴────────────┴───────┴──────────────┘

After Horizontal Partitioning (by category):
┌─────────────────────────┐  ┌─────────────────────────┐
│ products_electronics    │  │ products_books          │
│ 4M rows                 │  │ 3M rows                 │
├──────────┬────────┬─────┤  ├──────────┬────────┬─────┤
│ 1        │Widget A│10.99│  │ 3        │Book C  │15.00│
│ 2        │Gadget B│25.50│  │ ...      │...     │...  │
└──────────┴────────┴─────┘  └──────────┴────────┴─────┘

Benefits:
• Queries scan fewer rows
• Better parallelization
• Easier archiving/deletion

Vertical Partitioning (Divide Columns):

Original Table: users (frequently + rarely accessed columns)
┌──────┬────────┬────────┬──────────┬─────────────┬──────────┐
│ id   │ email  │ name   │ bio      │ preferences │ metadata │
│      │        │        │ (large)  │ (JSON)      │ (JSON)   │
└──────┴────────┴────────┴──────────┴─────────────┴──────────┘

After Vertical Partitioning:

Hot Table (frequently accessed):
┌──────┬────────────────┬────────┐
│ id   │ email          │ name   │  ← Fast queries, small indexes
└──────┴────────────────┴────────┘

Cold Table (rarely accessed):
┌──────┬──────────┬─────────────┬──────────┐
│ id   │ bio      │ preferences │ metadata │  ← Separate storage
└──────┴──────────┴─────────────┴──────────┘

Benefits:
• Smaller indexes on hot table
• Better cache utilization
• Reduced I/O for common queries

import psycopg2

conn = psycopg2.connect("dbname=ecommerce user=postgres")
cursor = conn.cursor()

# Create hot table (frequently accessed columns)
cursor.execute("""
    CREATE TABLE users_core (
        user_id BIGSERIAL PRIMARY KEY,
        email VARCHAR(255) UNIQUE NOT NULL,
        username VARCHAR(50) NOT NULL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# Create cold table (rarely accessed columns)
cursor.execute("""
    CREATE TABLE users_extended (
        user_id BIGINT PRIMARY KEY REFERENCES users_core(user_id),
        bio TEXT,
        avatar_url VARCHAR(500),
        preferences JSONB,
        metadata JSONB
    )
""")

# Insert user data
cursor.execute("""
    INSERT INTO users_core (email, username)
    VALUES ('alice@example.com', 'alice')
    RETURNING user_id
""")
user_id = cursor.fetchone()[0]

cursor.execute("""
    INSERT INTO users_extended (user_id, bio, preferences)
    VALUES (%s, %s, %s)
""", (user_id, 'Software engineer from California',
      '{"theme": "dark", "notifications": true}'))

conn.commit()

# Fast query (uses only hot table - small, fast indexes)
cursor.execute("""
    SELECT user_id, email, username
    FROM users_core
    WHERE email = 'alice@example.com'
""")
print("User core data:", cursor.fetchone())

# Full profile query (joins both tables)
cursor.execute("""
    SELECT c.email, c.username, e.bio, e.preferences
    FROM users_core c
    LEFT JOIN users_extended e ON c.user_id = e.user_id
    WHERE c.user_id = %s
""", (user_id,))
print("Full profile:", cursor.fetchone())

Sharding Architecture:

┌─────────────────────────────────────────┐
│         Application Layer               │
│  (Shard Router - determines which DB)   │
└──────────────┬──────────────────────────┘
               │
       ┌───────┴────────┬─────────────┐
       │                │             │
       ▼                ▼             ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│  Shard 1     │ │  Shard 2     │ │  Shard 3     │
│  DB Server   │ │  DB Server   │ │  DB Server   │
│  users 1-1M  │ │users 1M-2M   │ │users 2M-3M   │
└──────────────┘ └──────────────┘ └──────────────┘

Each shard:
• Separate PostgreSQL instance
• Independent storage and resources
• Contains subset of total data
• No cross-shard transactions

import psycopg2
from typing import Dict

class GeographyShardRouter:
    """Route queries to the appropriate geographic shard"""

    def __init__(self):
        # Map regions to database connections
        self.shards: Dict[str, psycopg2.extensions.connection] = {
            'us': psycopg2.connect(
                host='db-us-east.example.com',
                dbname='users_us',
                user='app'
            ),
            'eu': psycopg2.connect(
                host='db-eu-west.example.com',
                dbname='users_eu',
                user='app'
            ),
            'asia': psycopg2.connect(
                host='db-asia-southeast.example.com',
                dbname='users_asia',
                user='app'
            )
        }

        # Map country codes to regions
        self.country_to_region = {
            'US': 'us', 'CA': 'us', 'MX': 'us',
            'GB': 'eu', 'DE': 'eu', 'FR': 'eu', 'IT': 'eu',
            'CN': 'asia', 'JP': 'asia', 'IN': 'asia', 'KR': 'asia'
        }

    def get_shard(self, country_code: str):
        """Get database connection for a country"""
        region = self.country_to_region.get(country_code, 'us')  # Default to US
        return self.shards[region]

    def insert_user(self, email: str, country_code: str):
        """Insert user into the appropriate shard"""
        conn = self.get_shard(country_code)
        cursor = conn.cursor()

        cursor.execute("""
            INSERT INTO users (email, country_code)
            VALUES (%s, %s)
            RETURNING user_id
        """, (email, country_code))

        user_id = cursor.fetchone()[0]
        conn.commit()
        cursor.close()

        return user_id

    def get_user(self, email: str, country_code: str):
        """Retrieve user from the appropriate shard"""
        conn = self.get_shard(country_code)
        cursor = conn.cursor()

        cursor.execute("""
            SELECT user_id, email, country_code, created_at
            FROM users
            WHERE email = %s
        """, (email,))

        result = cursor.fetchone()
        cursor.close()
        return result

# Usage
router = GeographyShardRouter()

# Insert users in their home regions
us_user_id = router.insert_user('alice@example.com', 'US')
eu_user_id = router.insert_user('bob@example.co.uk', 'GB')
asia_user_id = router.insert_user('chen@example.cn', 'CN')

print(f"US user created: {us_user_id} (stored in US shard)")
print(f"EU user created: {eu_user_id} (stored in EU shard)")
print(f"Asia user created: {asia_user_id} (stored in Asia shard)")

# Retrieve user from their shard
user = router.get_user('alice@example.com', 'US')
print(f"\nRetrieved user: {user}")

import psycopg2
from typing import Dict, Optional

class TenantShardRouter:
    """Route queries based on tenant (organization) ID"""

    def __init__(self):
        # Shard configuration
        self.shards = {
            'shard_1': psycopg2.connect(
                host='shard1.example.com',
                dbname='saas_shard1',
                user='app'
            ),
            'shard_2': psycopg2.connect(
                host='shard2.example.com',
                dbname='saas_shard2',
                user='app'
            ),
            'shard_3': psycopg2.connect(
                host='shard3.example.com',
                dbname='saas_shard3',
                user='app'
            )
        }

        # Tenant-to-shard mapping (stored in routing database)
        self.routing_db = psycopg2.connect(
            host='routing.example.com',
            dbname='routing',
            user='app'
        )

    def get_tenant_shard(self, tenant_id: int) -> str:
        """Look up which shard contains this tenant's data"""
        cursor = self.routing_db.cursor()
        cursor.execute("""
            SELECT shard_name
            FROM tenant_shards
            WHERE tenant_id = %s
        """, (tenant_id,))

        result = cursor.fetchone()
        cursor.close()

        if not result:
            raise ValueError(f"Tenant {tenant_id} not found in routing table")

        return result[0]

    def assign_tenant_to_shard(self, tenant_id: int, tenant_name: str) -> str:
        """Assign new tenant to least-loaded shard"""
        cursor = self.routing_db.cursor()

        # Find shard with fewest tenants
        cursor.execute("""
            SELECT shard_name, COUNT(*) as tenant_count
            FROM tenant_shards
            GROUP BY shard_name
            ORDER BY tenant_count ASC
            LIMIT 1
        """)

        result = cursor.fetchone()
        shard_name = result[0] if result else 'shard_1'

        # Record tenant assignment
        cursor.execute("""
            INSERT INTO tenant_shards (tenant_id, tenant_name, shard_name)
            VALUES (%s, %s, %s)
        """, (tenant_id, tenant_name, shard_name))

        self.routing_db.commit()
        cursor.close()

        return shard_name

    def create_workspace(self, tenant_id: int, workspace_name: str):
        """Create workspace in tenant's shard"""
        shard_name = self.get_tenant_shard(tenant_id)
        conn = self.shards[shard_name]
        cursor = conn.cursor()

        cursor.execute("""
            INSERT INTO workspaces (tenant_id, name)
            VALUES (%s, %s)
            RETURNING workspace_id
        """, (tenant_id, workspace_name))

        workspace_id = cursor.fetchone()[0]
        conn.commit()
        cursor.close()

        return workspace_id, shard_name

    def get_tenant_workspaces(self, tenant_id: int):
        """Get all workspaces for a tenant"""
        shard_name = self.get_tenant_shard(tenant_id)
        conn = self.shards[shard_name]
        cursor = conn.cursor()

        cursor.execute("""
            SELECT workspace_id, name, created_at
            FROM workspaces
            WHERE tenant_id = %s
            ORDER BY created_at DESC
        """, (tenant_id,))

        workspaces = cursor.fetchall()
        cursor.close()

        return workspaces, shard_name

# Usage
router = TenantShardRouter()

# Assign new tenants to shards
shard1 = router.assign_tenant_to_shard(1001, 'Acme Corp')
shard2 = router.assign_tenant_to_shard(1002, 'TechStart Inc')
shard3 = router.assign_tenant_to_shard(1003, 'Global Enterprises')

print(f"Acme Corp assigned to {shard1}")
print(f"TechStart Inc assigned to {shard2}")
print(f"Global Enterprises assigned to {shard3}")

# Create workspaces for each tenant
ws_id1, shard = router.create_workspace(1001, 'Engineering Team')
print(f"\nCreated workspace {ws_id1} for tenant 1001 in {shard}")

ws_id2, shard = router.create_workspace(1001, 'Marketing Team')
print(f"Created workspace {ws_id2} for tenant 1001 in {shard}")

# Retrieve tenant's workspaces
workspaces, shard = router.get_tenant_workspaces(1001)
print(f"\nTenant 1001 workspaces (from {shard}):")
for ws_id, name, created_at in workspaces:
    print(f"  - {name} (ID: {ws_id})")

import psycopg2
import hashlib
from typing import List

class HashShardRouter:
    """Route queries using hash-based sharding with consistent hashing"""

    def __init__(self, shard_count: int = 4):
        self.shard_count = shard_count

        # Initialize database connections
        self.shards = {
            i: psycopg2.connect(
                host=f'shard{i}.example.com',
                dbname=f'tweets_shard{i}',
                user='app'
            )
            for i in range(shard_count)
        }

    def get_shard_id(self, key: str) -> int:
        """Determine shard ID using hash function"""
        # Use MD5 hash for consistent distribution
        hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
        return hash_value % self.shard_count

    def insert_tweet(self, user_id: int, tweet_text: str):
        """Insert tweet into the shard determined by user_id hash"""
        shard_id = self.get_shard_id(str(user_id))
        conn = self.shards[shard_id]
        cursor = conn.cursor()

        cursor.execute("""
            INSERT INTO tweets (user_id, tweet_text)
            VALUES (%s, %s)
            RETURNING tweet_id
        """, (user_id, tweet_text))

        tweet_id = cursor.fetchone()[0]
        conn.commit()
        cursor.close()

        return tweet_id, shard_id

    def get_user_tweets(self, user_id: int):
        """Get all tweets for a user from their shard"""
        shard_id = self.get_shard_id(str(user_id))
        conn = self.shards[shard_id]
        cursor = conn.cursor()

        cursor.execute("""
            SELECT tweet_id, tweet_text, created_at
            FROM tweets
            WHERE user_id = %s
            ORDER BY created_at DESC
            LIMIT 20
        """, (user_id,))

        tweets = cursor.fetchall()
        cursor.close()

        return tweets, shard_id

    def get_tweet_by_id(self, tweet_id: int):
        """Get tweet by ID - requires searching all shards (slow!)"""
        for shard_id, conn in self.shards.items():
            cursor = conn.cursor()
            cursor.execute("""
                SELECT tweet_id, user_id, tweet_text, created_at
                FROM tweets
                WHERE tweet_id = %s
            """, (tweet_id,))

            result = cursor.fetchone()
            cursor.close()

            if result:
                return result, shard_id

        return None, None

# Usage
router = HashShardRouter(shard_count=4)

# Insert tweets for different users
users = [
    (12345, "Hello from user 12345!"),
    (67890, "Tweet from user 67890"),
    (11111, "User 11111 posting"),
    (22222, "Another tweet from 22222"),
    (12345, "Second tweet from 12345")
]

print("Inserting tweets:")
for user_id, text in users:
    tweet_id, shard_id = router.insert_tweet(user_id, text)
    print(f"  User {user_id} → Shard {shard_id} (Tweet ID: {tweet_id})")

# Verify hash distribution
print("\nVerifying shard distribution:")
shard_counts = {}
for user_id, _ in users:
    shard_id = router.get_shard_id(str(user_id))
    shard_counts[shard_id] = shard_counts.get(shard_id, 0) + 1

for shard_id in range(router.shard_count):
    count = shard_counts.get(shard_id, 0)
    print(f"  Shard {shard_id}: {count} tweets")

# Retrieve user's tweets (fast - single shard query)
tweets, shard_id = router.get_user_tweets(12345)
print(f"\nUser 12345 tweets (from shard {shard_id}):")
for tweet_id, text, created_at in tweets:
    print(f"  - {text}")

Query Routing Patterns:

1. Single-Shard Query (Fast ✓):
   Query: "Get user 12345's profile"
   → Hash(12345) = Shard 2
   → Query only Shard 2
   → Return result
   Latency: ~10ms

2. Scatter-Gather Query (Slow ⚠):
   Query: "Get all users with premium status"
   → Query ALL shards (scatter)
   → Merge results (gather)
   → Sort/paginate
   Latency: ~100ms (10x slower)

3. Cross-Shard Join (Avoid! ✗):
   Query: "Join users with their orders"
   → Users in one set of shards
   → Orders in another set
   → Must fetch from all shards and join in application
   Latency: ~500ms+ (very slow)

Best Practice: Design shard key so related data lives together

import psycopg2
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Tuple

class ShardedQueryExecutor:
    """Execute queries across multiple shards efficiently"""

    def __init__(self, shards: dict):
        self.shards = shards
        self.executor = ThreadPoolExecutor(max_workers=len(shards))

    def scatter_gather_query(
        self,
        query: str,
        params: tuple = None,
        limit: int = 100
    ) -> List[Tuple]:
        """Execute query on all shards and merge results"""

        def query_shard(shard_id, conn):
            """Query a single shard"""
            cursor = conn.cursor()
            cursor.execute(query, params or ())
            results = cursor.fetchall()
            cursor.close()
            return shard_id, results

        # Scatter: submit queries to all shards in parallel
        futures = {
            self.executor.submit(query_shard, shard_id, conn): shard_id
            for shard_id, conn in self.shards.items()
        }

        # Gather: collect results from all shards
        all_results = []
        for future in as_completed(futures):
            shard_id, results = future.result()
            all_results.extend(results)
            print(f"  Shard {shard_id}: {len(results)} rows")

        # Sort and limit (application-level)
        all_results.sort(reverse=True, key=lambda x: x[2])  # Sort by timestamp
        return all_results[:limit]

    def search_across_shards(self, search_term: str):
        """Search for users across all shards (scatter-gather)"""
        query = """
            SELECT user_id, email, username, created_at
            FROM users
            WHERE username LIKE %s OR email LIKE %s
            ORDER BY created_at DESC
            LIMIT 50
        """

        print(f"Searching for '{search_term}' across all shards...")
        results = self.scatter_gather_query(
            query,
            (f'%{search_term}%', f'%{search_term}%'),
            limit=20
        )

        return results

# Usage example
shards = {
    0: psycopg2.connect(host='shard0.example.com', dbname='users0', user='app'),
    1: psycopg2.connect(host='shard1.example.com', dbname='users1', user='app'),
    2: psycopg2.connect(host='shard2.example.com', dbname='users2', user='app'),
    3: psycopg2.connect(host='shard3.example.com', dbname='users3', user='app')
}

executor = ShardedQueryExecutor(shards)

# Execute scatter-gather search
results = executor.search_across_shards('john')

print(f"\nFound {len(results)} matching users across all shards:")
for user_id, email, username, created_at in results[:5]:
    print(f"  - {username} ({email})")

# BAD: Cross-shard join (requires application-level join)
# Users sharded by user_id, Orders sharded by user_id
# To get user with their orders, need 2 queries:

def get_user_with_orders_bad(user_id: int):
    """Inefficient: separate queries"""
    # Query 1: Get user from users shard
    user_shard = get_user_shard(user_id)
    user = query_shard(user_shard, "SELECT * FROM users WHERE user_id = %s", (user_id,))

    # Query 2: Get orders from orders shard
    orders_shard = get_orders_shard(user_id)
    orders = query_shard(orders_shard, "SELECT * FROM orders WHERE user_id = %s", (user_id,))

    return user, orders


# GOOD: Denormalize to keep related data together
# Store both user and orders in same shard (co-located)

def get_user_with_orders_good(user_id: int):
    """Efficient: single shard query with denormalized data"""
    shard = get_shard(user_id)  # Same shard for user and their orders

    cursor = shard.cursor()

    # Single query - everything in one shard
    cursor.execute("""
        SELECT
            u.user_id, u.email, u.username,
            COALESCE(json_agg(
                json_build_object(
                    'order_id', o.order_id,
                    'total', o.total_amount,
                    'status', o.status
                ) ORDER BY o.created_at DESC
            ) FILTER (WHERE o.order_id IS NOT NULL), '[]') as orders
        FROM users u
        LEFT JOIN orders o ON u.user_id = o.user_id
        WHERE u.user_id = %s
        GROUP BY u.user_id, u.email, u.username
    """, (user_id,))

    result = cursor.fetchone()
    cursor.close()

    return result

# Usage
user_data = get_user_with_orders_good(12345)
print(f"User: {user_data[2]}")
print(f"Orders: {user_data[3]}")