Partitioning & Sharding Strategies

-- Seed the partitions so the examples below have something to prune.
-- 450,000 orders spread evenly over 2022-01-01 .. 2024-12-31 (1096 days).
-- Deterministic: no random(), so you get exactly the row counts shown.
INSERT INTO orders (customer_id, order_date, total_amount, status)
SELECT
    (g % 50000) + 1,
    DATE '2022-01-01' + (g % 1096),
    ROUND((10 + (g * 37 % 49000) / 100.0)::numeric, 2),
    (ARRAY['completed','pending','shipped','cancelled'])[1 + (g % 4)]
FROM generate_series(0, 449999) g;

ANALYZE orders;

SELECT tableoid::regclass AS partition, count(*)
FROM orders GROUP BY 1 ORDER BY 1;

import psycopg
from datetime import datetime

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
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 maintenance.
from contextlib import suppress
import psycopg

DSN = "host=localhost dbname=ecommerce user=demo password=demo"


def create_monthly_partitions(year, month_count=12):
    """Create monthly partitions for a year. Safe to re-run."""
    with psycopg.connect(DSN) as conn, conn.cursor() as cur:
        for month in range(1, month_count + 1):
            start = f"{year}-{month:02d}-01"
            end = f"{year + 1}-01-01" if month == 12 else f"{year}-{month + 1:02d}-01"
            name = f"orders_{year}_{month:02d}"

            # Check first so the log tells the truth. CREATE TABLE IF NOT
            # EXISTS alone would let the loop print "Created" every month
            # forever, whether or not it created anything.
            cur.execute("SELECT to_regclass(%s)", (name,))
            if cur.fetchone()[0] is not None:
                print(f"already exists: {name}")
                continue

            cur.execute(f"""
                CREATE TABLE {name} PARTITION OF orders
                FOR VALUES FROM ('{start}') TO ('{end}')
            """)
            print(f"created:        {name}")


def archive_and_drop_partition(partition_name):
    """Detach a partition, copy it into the archive schema, then drop it."""
    # DETACH ... CONCURRENTLY cannot run inside a transaction block, so this
    # connection is autocommit. See the caveat below about default partitions.
    with psycopg.connect(DSN, autocommit=True) as conn, conn.cursor() as cur:
        cur.execute("CREATE SCHEMA IF NOT EXISTS archive")

        # Detach FIRST. If you copy while the partition is still attached,
        # rows inserted between the copy and the detach are lost.
        cur.execute(f"ALTER TABLE orders DETACH PARTITION {partition_name}")

        cur.execute(f"CREATE TABLE archive.{partition_name} "
                    f"AS SELECT * FROM {partition_name}")
        cur.execute(f"SELECT count(*) FROM archive.{partition_name}")
        n = cur.fetchone()[0]

        cur.execute(f"DROP TABLE {partition_name}")
        print(f"archived {n:,} rows, dropped {partition_name}")


create_monthly_partitions(2025)
with suppress(psycopg.errors.UndefinedTable):
    archive_and_drop_partition("orders_2022")

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
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
    -- The tiebreaker matters: three partitions hold one row each, and
    -- without a second sort key their relative order is unspecified.
    ORDER BY user_count DESC, partition_name
""")

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

-- A Brazilian signs up. No BR partition exists, so the row lands in the
-- default and nothing looks wrong yet.
INSERT INTO users (email, country_code) VALUES ('bruno@example.br', 'BR');

-- Order by the partition NAME, not by tableoid: ORDER BY 1 here would sort
-- by OID, which is creation order rather than anything you can predict.
SELECT tableoid::regclass AS partition, email, country_code
FROM users ORDER BY tableoid::regclass::text, email;

-- Months later, South America has earned its own partition.
CREATE TABLE users_south_america PARTITION OF users
    FOR VALUES IN ('BR', 'AR', 'CL');

-- Detach the default, create the real partition, move the rows back through
-- the parent so they route themselves, then reattach.
BEGIN;
ALTER TABLE users DETACH PARTITION users_other;

CREATE TABLE users_south_america PARTITION OF users
    FOR VALUES IN ('BR', 'AR', 'CL');

INSERT INTO users SELECT * FROM users_other
    WHERE country_code IN ('BR', 'AR', 'CL');
DELETE FROM users_other
    WHERE country_code IN ('BR', 'AR', 'CL');

ALTER TABLE users ATTACH PARTITION users_other DEFAULT;
COMMIT;

SELECT tableoid::regclass AS partition, email, country_code
FROM users ORDER BY tableoid::regclass::text, email;

import psycopg
from datetime import datetime, timedelta
import random

random.seed(42)   # so you get exactly the distribution printed below

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
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}%")

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
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())

import psycopg
from typing import Dict

# One DSN per shard. In production these are three separate servers; with
# the single-container setup from "Try it locally" they are three databases
# on the same server, which is enough to show the routing behaviour.
SHARD_DSN = {
    'us':   "host=localhost dbname=users_us   user=demo password=demo",
    'eu':   "host=localhost dbname=users_eu   user=demo password=demo",
    'asia': "host=localhost dbname=users_asia user=demo password=demo",
}


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

    def __init__(self):
        self.shards: Dict[str, psycopg.Connection] = {
            region: psycopg.connect(dsn) for region, dsn in SHARD_DSN.items()
        }

        # 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)
        with conn.cursor() as 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()
        return user_id

    def get_user(self, email: str, country_code: str):
        """Retrieve user from the shard implied by country_code"""
        conn = self.get_shard(country_code)
        with conn.cursor() as cursor:
            cursor.execute("""
                SELECT user_id, email, country_code, created_at
                FROM users
                WHERE email = %s
            """, (email,))
            return cursor.fetchone()


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)")

# Look Bob up with the right key, then with the wrong one.
print(f"\nLookup with 'GB': {router.get_user('bob@example.co.uk', 'GB')}")
print(f"Lookup with 'US': {router.get_user('bob@example.co.uk', 'US')}")

import psycopg
from typing import Dict, Optional

# One DSN per shard, plus the small routing database that says which shard
# owns which tenant. With the single-container setup from "Try it locally"
# these are four databases on the same server.
SHARD_DSN = {
    'shard_1': "host=localhost dbname=saas_shard1 user=demo password=demo",
    'shard_2': "host=localhost dbname=saas_shard2 user=demo password=demo",
    'shard_3': "host=localhost dbname=saas_shard3 user=demo password=demo",
}
ROUTING_DSN = "host=localhost dbname=routing user=demo password=demo"


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

    def __init__(self):
        self.shards: Dict[str, psycopg.Connection] = {
            name: psycopg.connect(dsn) for name, dsn in SHARD_DSN.items()
        }
        self.routing_db = psycopg.connect(ROUTING_DSN)
    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.
        # The shard list has to come from the LEFT side: grouping
        # tenant_shards alone would only ever return shards that already
        # HAVE tenants, so an empty shard could never win and every new
        # tenant would pile onto the first one.
        cursor.execute("""
            SELECT s.shard_name, COUNT(t.tenant_id) AS tenant_count
            FROM (VALUES ('shard_1'), ('shard_2'), ('shard_3'))
                 AS s(shard_name)
            LEFT JOIN tenant_shards t ON t.shard_name = s.shard_name
            GROUP BY s.shard_name
            ORDER BY tenant_count ASC, s.shard_name
            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
            -- created_at alone can tie at microsecond resolution, so pin
            -- the order with the primary key as well.
            ORDER BY created_at DESC, workspace_id DESC
        """, (tenant_id,))

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

        return workspaces, shard_name

# Usage
router = TenantShardRouter()

# Assign new tenants to shards
for tenant_id, name in [(101, 'Acme Corp'), (102, 'Globex'), (103, 'Initech'),
                        (104, 'Umbrella'), (105, 'Soylent')]:
    print(f"{name:<9} assigned to {router.assign_tenant_to_shard(tenant_id, name)}")

# Create a workspace in the tenant's shard
ws_id, shard = router.create_workspace(101, 'Engineering Team')
print(f"\nWorkspace {ws_id} created for tenant 101 on {shard}")

# An unknown tenant is refused loudly rather than defaulting somewhere
try:
    router.create_workspace(999, 'Ghost Team')
except ValueError as e:
    print(f"ValueError: {e}")

import psycopg
import hashlib
from typing import List

class HashShardRouter:
    """Route queries using hash-based (modulo) sharding.

    NOTE: This uses modulo hashing (hash % shard_count) for simplicity.
    It distributes evenly, but changing shard_count remaps almost every
    key. Production systems use CONSISTENT HASHING (a hash ring) so that
    adding/removing a shard only remaps ~1/N of the keys. See the note
    below the code.
    """

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

        # Initialize database connections
        self.shards = {
            i: psycopg.connect(f"host=localhost dbname=tweets_shard{i} "
                               f"user=demo password=demo")
            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
            -- created_at alone can tie for tweets posted in the same
            -- instant, so pin the order with the id as well.
            ORDER BY created_at DESC, tweet_id 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}")

import psycopg
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Tuple

SHARD_DSN = {
    'shard_1': "host=localhost dbname=users0 user=demo password=demo",
    'shard_2': "host=localhost dbname=users1 user=demo password=demo",
    'shard_3': "host=localhost dbname=users2 user=demo password=demo",
}


class ShardedQueryExecutor:
    """Execute a query across every shard and merge the results."""

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

    @staticmethod
    def _query_shard(shard_id, conn, query, params):
        with conn.cursor() as cur:
            cur.execute(query, params or ())
            return shard_id, cur.fetchall()

    def scatter_gather(self, query, params=None, limit=100, verbose=True):
        """Scatter to every shard in parallel, gather, merge, sort, truncate."""
        futures = {
            self.executor.submit(self._query_shard, sid, conn, query, params): sid
            for sid, conn in self.shards.items()
        }

        merged = []
        for future in as_completed(futures):
            shard_id, rows = future.result()
            merged.extend(rows)
            if verbose:
                print(f"  {shard_id}: {len(rows):,} rows")

        # The global sort has to happen here: no shard can order rows it
        # cannot see. Index 3 is created_at.
        merged.sort(key=lambda r: r[3], reverse=True)
        return merged[:limit]

    def sequential(self, query, params=None, limit=100):
        """Same work, one shard at a time. For comparison only."""
        merged = []
        for shard_id, conn in self.shards.items():
            merged.extend(self._query_shard(shard_id, conn, query, params)[1])
        merged.sort(key=lambda r: r[3], reverse=True)
        return merged[:limit]


SEARCH = """
    SELECT user_id, email, username, created_at
    FROM users
    WHERE username LIKE %s OR email LIKE %s
"""

# The same query, but with the ordering and the cut pushed down to each
# shard. A shard still cannot know the global answer, but it does know that
# only its own top `limit` rows can possibly survive the merge.
SEARCH_PUSHED_DOWN = SEARCH + """
    ORDER BY created_at DESC
    LIMIT %s
"""

LIMIT = 100
term = '%john%'
executor = ShardedQueryExecutor(
    {sid: psycopg.connect(dsn) for sid, dsn in SHARD_DSN.items()}
)

print("Naive scatter-gather: no LIMIT inside the shard query")
rows = executor.scatter_gather(SEARCH, (term, term), limit=LIMIT)
print(f"  -> merged down to {len(rows)} rows\n")

print("With ORDER BY + LIMIT pushed into each shard")
rows_pd = executor.scatter_gather(SEARCH_PUSHED_DOWN, (term, term, LIMIT), limit=LIMIT)
print(f"  -> merged down to {len(rows_pd)} rows")
print(f"  -> same answer as the naive version: {rows == rows_pd}\n")


def timeit(fn, reps=5):
    fn()                                                    # warm
    best = min(_time(fn) for _ in range(reps))
    return best


def _time(fn):
    t0 = time.perf_counter()
    fn()
    return (time.perf_counter() - t0) * 1000


par = timeit(lambda: executor.scatter_gather(SEARCH, (term, term), LIMIT, verbose=False))
seq = timeit(lambda: executor.sequential(SEARCH, (term, term), LIMIT))
print(f"Parallel scatter-gather: {par:6.1f} ms")
print(f"Sequential equivalent:   {seq:6.1f} ms")

import psycopg

# Users and orders are BOTH sharded on user_id, so a user and that user's
# orders always land on the same shard. That co-location is what makes the
# JOIN below legal at all.
SHARD_DSN = ["host=localhost dbname=shard0 user=demo password=demo",
             "host=localhost dbname=shard1 user=demo password=demo"]


def get_shard(user_id: int) -> psycopg.Connection:
    return psycopg.connect(SHARD_DSN[user_id % len(SHARD_DSN)])


# BAD: two round trips, then stitch the pieces together in Python.
def get_user_with_orders_bad(user_id: int):
    conn = get_shard(user_id)
    with conn.cursor() as cur:
        cur.execute("SELECT user_id, email, username FROM users WHERE user_id = %s",
                    (user_id,))
        user = cur.fetchone()

        cur.execute("""SELECT order_id, total_amount, status
                       FROM orders WHERE user_id = %s
                       ORDER BY created_at DESC""", (user_id,))
        orders = cur.fetchall()
    return user, orders


# GOOD: one round trip, the database does the nesting.
def get_user_with_orders_good(user_id: int):
    conn = get_shard(user_id)
    with conn.cursor() as cur:
        cur.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,))
        return cur.fetchone()


user, orders = get_user_with_orders_bad(12345)
print("BAD  (2 queries):")
print(f"  user   = {user}")
print(f"  orders = {orders}")

row = get_user_with_orders_good(12345)
print("\nGOOD (1 query):")
print(f"  user_id  = {row[0]}")
print(f"  username = {row[2]}")
print(f"  orders   = {row[3]}")

# A user with no orders still comes back, with an empty list rather than
# a row of NULLs. That is what the FILTER + COALESCE pair is buying.
print(f"\nUser with no orders: {get_user_with_orders_good(12347)}")