Key Code Examples
1. Database Schema Design — E-Commerce
-- Complete e-commerce database schema
CREATE DATABASE ecommerce;
USE ecommerce;
-- Users and authentication
CREATE TABLE users (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
first_name VARCHAR(100) NOT NULL,
last_name VARCHAR(100) NOT NULL,
is_active BOOLEAN DEFAULT TRUE,
role ENUM('customer', 'admin', 'manager') DEFAULT 'customer',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_email (email),
INDEX idx_role (role)
);
-- Products catalog
CREATE TABLE categories (
id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
slug VARCHAR(100) UNIQUE NOT NULL,
parent_id INT UNSIGNED NULL,
description TEXT,
FOREIGN KEY (parent_id) REFERENCES categories(id),
INDEX idx_slug (slug)
);
CREATE TABLE products (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
slug VARCHAR(255) UNIQUE NOT NULL,
description TEXT,
price DECIMAL(10, 2) NOT NULL CHECK (price >= 0),
compare_at_price DECIMAL(10, 2),
sku VARCHAR(100) UNIQUE,
stock_quantity INT UNSIGNED DEFAULT 0,
weight_kg DECIMAL(6, 2),
category_id INT UNSIGNED,
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (category_id) REFERENCES categories(id),
INDEX idx_category (category_id),
INDEX idx_price (price),
INDEX idx_active_price (is_active, price),
FULLTEXT INDEX idx_search (name, description)
);
-- Orders system
CREATE TABLE orders (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
order_number VARCHAR(50) UNIQUE NOT NULL,
user_id BIGINT UNSIGNED NOT NULL,
status ENUM('pending', 'processing', 'shipped', 'delivered', 'cancelled')
DEFAULT 'pending',
subtotal DECIMAL(10, 2) NOT NULL,
tax_amount DECIMAL(10, 2) DEFAULT 0.00,
shipping_amount DECIMAL(10, 2) DEFAULT 0.00,
discount_amount DECIMAL(10, 2) DEFAULT 0.00,
total DECIMAL(10, 2) NOT NULL,
shipping_address JSON,
billing_address JSON,
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id),
INDEX idx_user (user_id),
INDEX idx_status (status),
INDEX idx_created (created_at)
);
CREATE TABLE order_items (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
order_id BIGINT UNSIGNED NOT NULL,
product_id BIGINT UNSIGNED NOT NULL,
product_name VARCHAR(255) NOT NULL, -- denormalized for history
quantity INT UNSIGNED NOT NULL CHECK (quantity > 0),
unit_price DECIMAL(10, 2) NOT NULL,
subtotal DECIMAL(10, 2) NOT NULL,
FOREIGN KEY (order_id) REFERENCES orders(id) ON DELETE CASCADE,
FOREIGN KEY (product_id) REFERENCES products(id),
INDEX idx_order (order_id)
);
-- Shopping cart (can convert to order)
CREATE TABLE cart_items (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
user_id BIGINT UNSIGNED NOT NULL,
product_id BIGINT UNSIGNED NOT NULL,
quantity INT UNSIGNED DEFAULT 1,
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY unique_cart_item (user_id, product_id),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
FOREIGN KEY (product_id) REFERENCES products(id)
);
2. Complex SQL Queries
-- 1. Top 10 customers by total spend
SELECT
u.id,
u.first_name,
u.last_name,
u.email,
COUNT(DISTINCT o.id) AS total_orders,
SUM(o.total) AS total_spent,
AVG(o.total) AS avg_order_value,
MAX(o.created_at) AS last_order_date
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE o.status != 'cancelled'
GROUP BY u.id
ORDER BY total_spent DESC
LIMIT 10;
-- 2. Monthly revenue with year-over-year comparison
SELECT
DATE_FORMAT(created_at, '%Y-%m') AS month,
COUNT(*) AS orders,
SUM(total) AS revenue,
SUM(total) - LAG(SUM(total), 12) OVER (ORDER BY DATE_FORMAT(created_at, '%Y-%m'))
AS yoy_change,
ROUND(
(SUM(total) - LAG(SUM(total), 12) OVER (ORDER BY DATE_FORMAT(created_at, '%Y-%m')))
/ LAG(SUM(total), 12) OVER (ORDER BY DATE_FORMAT(created_at, '%Y-%m')) * 100, 2
) AS yoy_percent
FROM orders
WHERE status = 'delivered'
GROUP BY month
ORDER BY month DESC;
-- 3. Product performance with inventory status
SELECT
p.id,
p.name,
p.sku,
c.name AS category,
p.stock_quantity,
COALESCE(SUM(oi.quantity), 0) AS units_sold,
COALESCE(SUM(oi.subtotal), 0) AS total_revenue,
CASE
WHEN p.stock_quantity = 0 THEN 'Out of Stock'
WHEN p.stock_quantity < 10 THEN 'Low Stock'
ELSE 'In Stock'
END AS stock_status
FROM products p
LEFT JOIN categories c ON p.category_id = c.id
LEFT JOIN order_items oi ON p.id = oi.product_id
LEFT JOIN orders o ON oi.order_id = o.id AND o.status = 'delivered'
WHERE p.is_active = TRUE
GROUP BY p.id
ORDER BY total_revenue DESC;
-- 4. Customer segmentation (RFM Analysis)
WITH customer_stats AS (
SELECT
u.id,
u.first_name,
DATEDIFF(CURDATE(), MAX(o.created_at)) AS recency,
COUNT(DISTINCT o.id) AS frequency,
SUM(o.total) AS monetary
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE o.status = 'delivered'
GROUP BY u.id
),
rfm_scores AS (
SELECT
*,
NTILE(5) OVER (ORDER BY recency DESC) AS r_score,
NTILE(5) OVER (ORDER BY frequency) AS f_score,
NTILE(5) OVER (ORDER BY monetary) AS m_score
FROM customer_stats
)
SELECT
*,
r_score + f_score + m_score AS rfm_total,
CASE
WHEN r_score >= 4 AND f_score >= 4 THEN 'Champions'
WHEN r_score >= 3 AND f_score >= 3 THEN 'Loyal Customers'
WHEN r_score >= 4 AND f_score <= 2 THEN 'New Customers'
WHEN r_score <= 2 AND f_score >= 3 THEN 'At Risk'
WHEN r_score <= 2 AND f_score <= 2 THEN 'Lost'
ELSE 'Average'
END AS segment
FROM rfm_scores
ORDER BY rfm_total DESC;
3. Database Transactions & Stored Procedures
-- Stored procedure for placing an order
DELIMITER //
CREATE PROCEDURE PlaceOrder(
IN p_user_id BIGINT UNSIGNED,
IN p_shipping_address JSON,
IN p_billing_address JSON,
OUT p_order_id BIGINT UNSIGNED,
OUT p_success BOOLEAN,
OUT p_message VARCHAR(255)
)
BEGIN
DECLARE v_subtotal DECIMAL(10, 2) DEFAULT 0;
DECLARE v_tax DECIMAL(10, 2) DEFAULT 0;
DECLARE v_total DECIMAL(10, 2) DEFAULT 0;
DECLARE v_order_number VARCHAR(50);
DECLARE v_product_id BIGINT UNSIGNED;
DECLARE v_quantity INT UNSIGNED;
DECLARE v_price DECIMAL(10, 2);
DECLARE v_stock INT UNSIGNED;
DECLARE v_item_subtotal DECIMAL(10, 2);
DECLARE done INT DEFAULT FALSE;
-- Cursor for cart items
DECLARE cart_cursor CURSOR FOR
SELECT product_id, quantity FROM cart_items WHERE user_id = p_user_id;
DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = TRUE;
-- Start transaction
START TRANSACTION;
-- Generate order number
SET v_order_number = CONCAT('ORD-', DATE_FORMAT(NOW(), '%Y%m%d-'),
LPAD(FLOOR(RAND() * 10000), 4, '0'));
-- Calculate subtotal from cart
SELECT COALESCE(SUM(p.price * ci.quantity), 0)
INTO v_subtotal
FROM cart_items ci
JOIN products p ON ci.product_id = p.id
WHERE ci.user_id = p_user_id;
IF v_subtotal = 0 THEN
SET p_success = FALSE;
SET p_message = 'Cart is empty';
ROLLBACK;
ELSE
SET v_tax = ROUND(v_subtotal * 0.08, 2); -- 8% tax
SET v_total = v_subtotal + v_tax;
-- Create order
INSERT INTO orders (order_number, user_id, status, subtotal,
tax_amount, total, shipping_address, billing_address)
VALUES (v_order_number, p_user_id, 'pending', v_subtotal,
v_tax, v_total, p_shipping_address, p_billing_address);
SET p_order_id = LAST_INSERT_ID();
-- Process cart items and update inventory
OPEN cart_cursor;
read_loop: LOOP
FETCH cart_cursor INTO v_product_id, v_quantity;
IF done THEN
LEAVE read_loop;
END IF;
-- Check stock
SELECT price, stock_quantity INTO v_price, v_stock
FROM products WHERE id = v_product_id FOR UPDATE;
IF v_stock < v_quantity THEN
SET p_success = FALSE;
SET p_message = CONCAT('Insufficient stock for product ID ', v_product_id);
ROLLBACK;
LEAVE read_loop;
END IF;
-- Add order item
SET v_item_subtotal = v_price * v_quantity;
INSERT INTO order_items (order_id, product_id, product_name,
quantity, unit_price, subtotal)
SELECT p_order_id, v_product_id, p.name,
v_quantity, v_price, v_item_subtotal
FROM products p WHERE p.id = v_product_id;
-- Update inventory
UPDATE products
SET stock_quantity = stock_quantity - v_quantity
WHERE id = v_product_id;
END LOOP;
CLOSE cart_cursor;
-- Clear cart
DELETE FROM cart_items WHERE user_id = p_user_id;
COMMIT;
SET p_success = TRUE;
SET p_message = 'Order placed successfully';
END IF;
END //
DELIMITER ;
-- Usage
CALL PlaceOrder(1, '{"street": "123 Main St"}',
'{"street": "123 Main St"}',
@order_id, @success, @message);
SELECT @order_id, @success, @message;
4. ORM Usage (SQLAlchemy / Prisma)
# Python SQLAlchemy ORM Example
from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey, DateTime
from sqlalchemy.orm import declarative_base, relationship, Session
from sqlalchemy.sql import func
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
email = Column(String(255), unique=True, nullable=False)
first_name = Column(String(100), nullable=False)
last_name = Column(String(100), nullable=False)
created_at = Column(DateTime, default=func.now())
# Relationships
orders = relationship("Order", back_populates="user",
cascade="all, delete-orphan")
def __repr__(self):
return f""
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
status = Column(String(20), default='pending')
total = Column(Float, nullable=False)
created_at = Column(DateTime, default=func.now())
# Relationships
user = relationship("User", back_populates="orders")
items = relationship("OrderItem", back_populates="order",
cascade="all, delete-orphan")
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String(255), nullable=False)
price = Column(Float, nullable=False)
stock = Column(Integer, default=0)
class OrderItem(Base):
__tablename__ = 'order_items'
id = Column(Integer, primary_key=True)
order_id = Column(Integer, ForeignKey('orders.id'))
product_id = Column(Integer, ForeignKey('products.id'))
quantity = Column(Integer, nullable=False)
unit_price = Column(Float, nullable=False)
order = relationship("Order", back_populates="items")
product = relationship("Product")
# Usage
engine = create_engine("postgresql://user:pass@localhost/ecommerce")
Base.metadata.create_all(engine)
with Session(engine) as session:
# Create user with order
user = User(email="alice@example.com", first_name="Alice", last_name="Smith")
product = session.query(Product).filter_by(name="Laptop").first()
order = Order(user=user, total=product.price, status="pending")
order.items.append(OrderItem(product=product, quantity=1, unit_price=product.price))
session.add(order)
session.commit()
# Query with joins
results = session.query(User, Order).\
join(Order).\
filter(Order.status == 'pending').\
all()
for user, order in results:
print(f"{user.email}: Order #{order.id} - ${order.total}")
5. MongoDB — NoSQL Document Design
// MongoDB Schema Design and Queries
// User collection with embedded preferences
db.users.insertOne({
_id: ObjectId(),
email: "alice@example.com",
passwordHash: "$2b$12$...",
profile: {
firstName: "Alice",
lastName: "Smith",
avatar: "https://cdn.example.com/avatars/alice.jpg"
},
preferences: {
theme: "dark",
language: "en",
notifications: { email: true, push: false }
},
addresses: [
{
type: "shipping",
street: "123 Main St",
city: "New York",
state: "NY",
zip: "10001",
isDefault: true
}
],
createdAt: new Date(),
lastLogin: new Date()
});
// Product collection with variants
db.products.insertOne({
_id: ObjectId(),
name: "Running Shoes",
slug: "running-shoes-x1",
description: "Lightweight running shoes",
basePrice: 129.99,
category: { name: "Footwear", slug: "footwear" },
variants: [
{ color: "black", size: "9", stock: 15, sku: "RS-X1-BLK-9" },
{ color: "black", size: "10", stock: 8, sku: "RS-X1-BLK-10" },
{ color: "white", size: "9", stock: 0, sku: "RS-X1-WHT-9" }
],
ratings: { average: 4.5, count: 128 },
tags: ["running", "lightweight", "breathable"],
createdAt: new Date()
});
// Order collection with embedded items
db.orders.insertOne({
_id: ObjectId(),
orderNumber: "ORD-20260115-0001",
userId: ObjectId("..."),
status: "shipped",
items: [
{
productId: ObjectId("..."),
name: "Running Shoes",
variant: { color: "black", size: "9" },
quantity: 1,
unitPrice: 129.99,
subtotal: 129.99
}
],
shipping: {
address: { street: "123 Main St", city: "New York" },
method: "express",
trackingNumber: "1Z999AA1234567890"
},
payment: { method: "credit_card", last4: "4242" },
subtotal: 129.99,
tax: 10.40,
shipping: 15.00,
total: 155.39,
createdAt: new Date(),
updatedAt: new Date()
});
// Aggregation pipeline: Revenue by category
db.orders.aggregate([
{ $match: { status: { $in: ["shipped", "delivered"] } } },
{ $unwind: "$items" },
{
$lookup: {
from: "products",
localField: "items.productId",
foreignField: "_id",
as: "product"
}
},
{ $unwind: "$product" },
{
$group: {
_id: "$product.category.name",
totalRevenue: { $sum: "$items.subtotal" },
orderCount: { $sum: 1 },
avgOrderValue: { $avg: "$items.subtotal" }
}
},
{ $sort: { totalRevenue: -1 } }
]);
6. Index Optimization
-- Analyzing query performance
EXPLAIN ANALYZE
SELECT u.first_name, u.last_name, COUNT(o.id) AS order_count, SUM(o.total) AS total
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE o.created_at >= '2026-01-01'
AND o.status = 'delivered'
GROUP BY u.id
ORDER BY total DESC;
-- Index strategies for the above query
-- 1. Composite index for the WHERE clause + JOIN
CREATE INDEX idx_orders_user_status_date
ON orders(user_id, status, created_at);
-- 2. Covering index (includes all needed columns)
CREATE INDEX idx_orders_covering
ON orders(user_id, status, created_at, total);
-- 3. Partial index (only delivered orders)
CREATE INDEX idx_orders_delivered
ON orders(user_id, created_at)
WHERE status = 'delivered';
-- When to use each:
-- - Basic composite: good general purpose
-- - Covering: best when query selects few columns (avoids table lookup)
-- - Partial: best when querying subset frequently (smaller index)
-- Finding unused indexes
SELECT
table_name,
index_name,
rows_selected,
rows_inserted,
rows_deleted
FROM performance_schema.table_io_waits_summary_by_index_usage
WHERE index_name IS NOT NULL
ORDER BY rows_selected ASC;
-- Finding missing indexes (slow queries)
SELECT
digest_text AS query,
count_star AS executions,
ROUND(avg_timer_wait / 1000000000, 2) AS avg_ms,
sum_rows_examined AS total_rows_examined
FROM events_statements_summary_by_digest
ORDER BY avg_timer_wait DESC
LIMIT 20;
7. Database Migration Script
-- Flyway migration: V3__add_product_reviews.sql
CREATE TABLE reviews (
id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
product_id BIGINT UNSIGNED NOT NULL,
user_id BIGINT UNSIGNED NOT NULL,
rating TINYINT UNSIGNED NOT NULL CHECK (rating BETWEEN 1 AND 5),
title VARCHAR(200),
body TEXT,
is_verified_purchase BOOLEAN DEFAULT FALSE,
helpful_count INT UNSIGNED DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (product_id) REFERENCES products(id) ON DELETE CASCADE,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
UNIQUE KEY unique_user_product_review (product_id, user_id),
INDEX idx_product_rating (product_id, rating),
INDEX idx_user (user_id)
);
-- Update product rating summary
UPDATE products p
SET
p.avg_rating = (
SELECT ROUND(AVG(rating), 2)
FROM reviews WHERE product_id = p.id
),
p.review_count = (
SELECT COUNT(*) FROM reviews WHERE product_id = p.id
)
WHERE EXISTS (SELECT 1 FROM reviews WHERE product_id = p.id);
-- Create trigger to auto-update product rating
DELIMITER //
CREATE TRIGGER update_product_rating_after_insert
AFTER INSERT ON reviews
FOR EACH ROW
BEGIN
UPDATE products
SET
avg_rating = (
SELECT ROUND(AVG(rating), 2) FROM reviews
WHERE product_id = NEW.product_id
),
review_count = (
SELECT COUNT(*) FROM reviews WHERE product_id = NEW.product_id
)
WHERE id = NEW.product_id;
END //
CREATE TRIGGER update_product_rating_after_update
AFTER UPDATE ON reviews
FOR EACH ROW
BEGIN
UPDATE products
SET
avg_rating = (
SELECT ROUND(AVG(rating), 2) FROM reviews
WHERE product_id = NEW.product_id
),
review_count = (
SELECT COUNT(*) FROM reviews WHERE product_id = NEW.product_id
)
WHERE id = NEW.product_id;
END //
DELIMITER ;
8. Redis Caching Layer
import redis
import json
from functools import wraps
from datetime import timedelta
class CacheManager:
def __init__(self, host='localhost', port=6379, db=0):
self.redis = redis.Redis(host=host, port=port, db=db,
decode_responses=True)
def cached(self, ttl=300, key_prefix='cache'):
"""Decorator to cache function results."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
cache_key = f"{key_prefix}:{func.__name__}:{str(args)}:{str(kwargs)}"
# Try cache first
cached = self.redis.get(cache_key)
if cached:
return json.loads(cached)
# Execute function
result = func(*args, **kwargs)
# Store in cache
self.redis.setex(
cache_key,
timedelta(seconds=ttl),
json.dumps(result)
)
return result
return wrapper
return decorator
def invalidate(self, pattern):
"""Invalidate cache keys matching pattern."""
for key in self.redis.scan_iter(match=pattern):
self.redis.delete(key)
def cache_product(self, product_id, product_data):
"""Cache product data."""
key = f"product:{product_id}"
self.redis.hset(key, mapping={
'data': json.dumps(product_data),
'cached_at': str(datetime.now())
})
self.redis.expire(key, 3600) # 1 hour
def get_product(self, product_id):
"""Get cached product or None."""
key = f"product:{product_id}"
cached = self.redis.hget(key, 'data')
return json.loads(cached) if cached else None
def rate_limit(self, key, limit=100, window=60):
"""Simple rate limiter using Redis."""
pipe = self.redis.pipeline()
current = pipe.get(key)
pipe.expire(key, window)
pipe.execute()
count = int(current[0]) if current else 0
if count >= limit:
return False
self.redis.incr(key)
return True
# Usage
cache = CacheManager()
@cache.cached(ttl=600, key_prefix='products')
def get_expensive_products():
"""This will be cached for 10 minutes."""
# Expensive database query
return db.query("SELECT * FROM products ORDER BY sales DESC LIMIT 100")
# Invalidate when product is updated
def update_product(product_id, data):
db.update(product_id, data)
cache.invalidate(f"product:{product_id}")
# Cache-aside pattern for user sessions
def get_user_session(session_id):
key = f"session:{session_id}"
session = cache.redis.get(key)
if session:
return json.loads(session)
# Fetch from DB
session = db.get_session(session_id)
if session:
cache.redis.setex(key, timedelta(hours=24), json.dumps(session))
return session
9. Backup & Recovery Script
#!/bin/bash
# backup_database.sh — Automated database backup
DB_HOST="localhost"
DB_PORT="3306"
DB_NAME="ecommerce"
DB_USER="backup_user"
DB_PASS="secure_password"
BACKUP_DIR="/backups/mysql"
RETENTION_DAYS=30
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="${BACKUP_DIR}/${DB_NAME}_${DATE}.sql.gz"
# Create backup directory
mkdir -p ${BACKUP_DIR}
# Full backup with mysqldump
echo "Starting backup of ${DB_NAME}..."
mysqldump \
--host=${DB_HOST} \
--port=${DB_PORT} \
--user=${DB_USER} \
--password=${DB_PASS} \
--single-transaction \
--routines \
--triggers \
--events \
--hex-blob \
${DB_NAME} | gzip > ${BACKUP_FILE}
if [ $? -eq 0 ]; then
echo "Backup completed: ${BACKUP_FILE}"
# Upload to S3
aws s3 cp ${BACKUP_FILE} s3://eduverse-backups/mysql/
# Remove old local backups
find ${BACKUP_DIR} -name "*.sql.gz" -mtime +${RETENTION_DAYS} -delete
# Verify backup
gunzip -t ${BACKUP_FILE} && echo "Backup verified"
else
echo "Backup FAILED!" >&2
# Send alert
echo "Database backup failed for ${DB_NAME}" | mail -s "BACKUP FAILURE" admin@eduverse.com
exit 1
fi
# ===== Point-in-Time Recovery =====
# 1. Restore from full backup
# gunzip < backup.sql.gz | mysql -u root -p ecommerce
#
# 2. Apply binary logs up to specific point
# mysqlbinlog --start-datetime="2026-01-15 10:00:00" \
# --stop-datetime="2026-01-15 10:30:00" \
# /var/lib/mysql/binlog.000123 | mysql -u root -p
# ===== Snapshot Backup (LVM) =====
# lvcreate --size 10G --snapshot --name mysql_snap /dev/vg0/mysql_data
# mount /dev/vg0/mysql_snap /mnt/snapshot
# rsync -av /mnt/snapshot/ /backups/snapshot/
# umount /mnt/snapshot && lvremove /dev/vg0/mysql_snap
10. Database Connection Pooling
import psycopg2
from psycopg2.pool import ThreadedConnectionPool
from contextlib import contextmanager
import threading
class DatabasePool:
"""Thread-safe PostgreSQL connection pool."""
def __init__(self, min_conn=2, max_conn=10, **db_config):
self.pool = ThreadedConnectionPool(
minconn=min_conn,
maxconn=max_conn,
**db_config
)
self._local = threading.local()
print(f"Pool created: {min_conn}-{max_conn} connections")
@contextmanager
def get_connection(self):
"""Get a connection from the pool."""
conn = self.pool.getconn()
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
self.pool.putconn(conn)
@contextmanager
def get_cursor(self, name=None):
"""Get a named cursor (for server-side cursors)."""
with self.get_connection() as conn:
cursor = conn.cursor(name=name)
try:
yield cursor
finally:
cursor.close()
def execute(self, query, params=None):
"""Execute a query and return results."""
with self.get_cursor() as cursor:
cursor.execute(query, params)
if cursor.description: # Has result set
return cursor.fetchall()
return cursor.rowcount
def execute_many(self, query, params_list):
"""Execute a query with multiple parameter sets."""
with self.get_cursor() as cursor:
cursor.executemany(query, params_list)
return cursor.rowcount
def close(self):
"""Close all connections."""
self.pool.closeall()
# Usage
db = DatabasePool(
min_conn=2, max_conn=20,
host='localhost', port=5432,
dbname='ecommerce', user='app_user',
password='password'
)
# Simple query
results = db.execute(
"SELECT id, name, price FROM products WHERE stock > %s",
(0,)
)
# Transaction
with db.get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"UPDATE products SET stock = stock - %s WHERE id = %s",
(2, 101)
)
cursor.execute(
"INSERT INTO order_items (order_id, product_id, quantity) VALUES (%s, %s, %s)",
(1001, 101, 2)
)
# Batch insert
products = [(f"Product {i}", 9.99) for i in range(1000)]
db.execute_many(
"INSERT INTO products (name, price) VALUES (%s, %s)",
products
)
db.close()
Core Concepts
Relational vs Non-Relational Databases
RDBMS (SQL): Tables with predefined schemas, ACID transactions, powerful queries (PostgreSQL, MySQL, Oracle).
NoSQL: Flexible schemas, horizontal scaling, eventual consistency.
- Document (MongoDB): JSON-like documents, flexible schema
- Key-Value (Redis): Simple lookups, caching, sessions
- Column-Family (Cassandra): Massive write throughput, time-series data
- Graph (Neo4j): Relationship-heavy data (social networks, recommendations)
ACID Properties
Atomicity: All operations in a transaction succeed or none do ("all or nothing").
Consistency: Database transitions from one valid state to another (constraints, triggers enforced).
Isolation: Concurrent transactions don't interfere with each other.
Durability: Committed data survives system crashes (write-ahead logging).
Isolation Levels (weakest to strongest): Read Uncommitted → Read Committed → Repeatable Read → Serializable
Normalization (Normal Forms)
Definition: Process of organizing data to reduce redundancy and improve integrity.
- 1NF: Atomic values, no repeating groups (each cell has one value)
- 2NF: 1NF + no partial dependencies (all non-key columns depend on full primary key)
- 3NF: 2NF + no transitive dependencies (non-key columns depend only on primary key)
- BCNF: Every determinant is a candidate key (stricter 3NF)
- Denormalization: Intentionally adding redundancy for read performance
Indexing & Query Optimization
Definition: Data structures (typically B-trees) that improve data retrieval speed at the cost of storage and write performance.
- B-Tree Index: Balanced tree for equality and range queries
- Hash Index: O(1) lookups for exact matches only
- Composite Index: Index on multiple columns (column order matters!)
- Covering Index: Contains all columns needed by a query (avoids table access)
- When to Index: WHERE, JOIN, ORDER BY, GROUP BY columns
- When NOT to Index: Low cardinality columns, frequently updated tables
Transactions & Concurrency Control
Problems with Concurrency:
- Lost Update: Two transactions overwrite each other's changes
- Dirty Read: Reading uncommitted data
- Non-Repeatable Read: Same query returns different results
- Phantom Read: New rows appear between reads
Solutions: Locking (shared/exclusive), MVCC (Multi-Version Concurrency Control), Optimistic concurrency control
Database Replication
Master-Slave Replication: Writes to master, reads from slaves. Improves read scalability.
Master-Master Replication: Multiple writable nodes. Better availability but conflict resolution needed.
Synchronous vs Asynchronous:
- Synchronous: Zero data loss, higher latency
- Asynchronous: Lower latency, possible data loss on failover
CAP Theorem
Definition: In a distributed system, you can only guarantee 2 of 3 properties:
- Consistency: Every read gets the most recent write
- Availability: Every request gets a response (may be stale)
- Partition Tolerance: System continues despite network partitions
Implication: During a network partition, choose between CP (consistency) or AP (availability).
Sharding & Partitioning
Horizontal Partitioning (Sharding): Split rows across multiple tables/servers based on shard key.
Vertical Partitioning: Split columns into separate tables (hot vs cold data).
- Range-based: Data by range (e.g., dates, IDs)
- Hash-based: Even distribution using hash function
- Directory-based: Lookup service maps keys to shards
- Challenges: Cross-shard queries, rebalancing, distributed transactions
Worked Examples
Example 1: Design a Social Media Database
Problem: Design a database for a social media platform with users, posts, comments, likes, and follows.
Solution:
CREATE TABLE users (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
display_name VARCHAR(100),
bio TEXT,
avatar_url VARCHAR(500),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE posts (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
user_id BIGINT NOT NULL,
content TEXT NOT NULL,
media_url VARCHAR(500),
like_count INT DEFAULT 0,
comment_count INT DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id),
INDEX idx_user_created (user_id, created_at DESC)
);
CREATE TABLE comments (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
post_id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (post_id) REFERENCES posts(id),
FOREIGN KEY (user_id) REFERENCES users(id)
);
CREATE TABLE follows (
follower_id BIGINT NOT NULL,
following_id BIGINT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (follower_id, following_id),
FOREIGN KEY (follower_id) REFERENCES users(id),
FOREIGN KEY (following_id) REFERENCES users(id)
);
-- Feed query: Get posts from followed users
SELECT p.*, u.username, u.display_name, u.avatar_url
FROM posts p
JOIN users u ON p.user_id = u.id
WHERE p.user_id IN (
SELECT following_id FROM follows WHERE follower_id = ?
)
ORDER BY p.created_at DESC
LIMIT 20;
Example 2: Normalize a Denormalized Table
Problem: Normalize this denormalized table to 3NF.
Before (unnormalized):
orders_raw:
| order_id | customer_name | customer_email | product_name | product_price | quantity |
| 1 | Alice | alice@mail.com | Laptop | 999.99 | 1 |
| 1 | Alice | alice@mail.com | Mouse | 29.99 | 2 |
| 2 | Bob | bob@mail.com | Keyboard | 79.99 | 1 |
After (3NF):
-- Table: customers
| customer_id | name | email |
| 1 | Alice | alice@mail.com |
| 2 | Bob | bob@mail.com |
-- Table: products
| product_id | name | price |
| 1 | Laptop | 999.99 |
| 2 | Mouse | 29.99 |
| 3 | Keyboard | 79.99 |
-- Table: orders
| order_id | customer_id | order_date |
| 1 | 1 | 2026-01-15 |
| 2 | 2 | 2026-01-15 |
-- Table: order_items
| order_id | product_id | quantity | unit_price |
| 1 | 1 | 1 | 999.99 |
| 1 | 2 | 2 | 29.99 |
| 2 | 3 | 1 | 79.99 |
Example 3: Optimize a Slow Query
Problem: This query takes 15 seconds to run. Optimize it.
Original (slow):
-- No index on created_at, using functions on columns
SELECT * FROM orders
WHERE YEAR(created_at) = 2026 AND MONTH(created_at) = 1
ORDER BY total DESC
LIMIT 100;
Optimized:
-- Add composite index
CREATE INDEX idx_orders_created_total ON orders(created_at, total DESC);
-- Rewrite query to use range instead of functions
SELECT * FROM orders
WHERE created_at >= '2026-01-01' AND created_at < '2026-02-01'
ORDER BY total DESC
LIMIT 100;
-- Further optimization: covering index for pagination
CREATE INDEX idx_orders_covering ON orders(created_at, total DESC, id)
INCLUDE (user_id, status, order_number);
Result: Query time reduced from 15s to 50ms (300x faster).
Example 4: Implement Database Sharding
Problem: Shard a users table across 4 database servers by user_id.
Solution:
// Consistent hashing for shard selection
function getShard(user_id, num_shards = 4) {
return user_id % num_shards;
}
// Shard mapping
const shardMap = {
0: { host: 'db-shard-0.internal', database: 'app_shard_0' },
1: { host: 'db-shard-1.internal', database: 'app_shard_1' },
2: { host: 'db-shard-2.internal', database: 'app_shard_2' },
3: { host: 'db-shard-3.internal', database: 'app_shard_3' }
};
// Connection pools per shard
const shardPools = {};
for (let i = 0; i < 4; i++) {
shardPools[i] = new Pool({ ...shardMap[i], max: 10 });
}
// Route query to correct shard
async function getUserById(user_id) {
const shard = getShard(user_id);
const result = await shardPools[shard].query(
'SELECT * FROM users WHERE id = $1', [user_id]
);
return result.rows[0];
}
// Cross-shard query (scatter-gather)
async function getUsersByIds(user_ids) {
const shardGroups = {};
user_ids.forEach(id => {
const shard = getShard(id);
if (!shardGroups[shard]) shardGroups[shard] = [];
shardGroups[shard].push(id);
});
const queries = Object.entries(shardGroups).map(([shard, ids]) =>
shardPools[shard].query(
'SELECT * FROM users WHERE id = ANY($1)', [ids]
)
);
const results = await Promise.all(queries);
return results.flatMap(r => r.rows);
}
Example 5: Database Deadlock Resolution
Problem: Two transactions are deadlocked. Analyze and resolve.
Scenario:
-- Transaction A -- Transaction B
BEGIN; BEGIN;
UPDATE accounts SET UPDATE accounts SET
balance = balance - 100 balance = balance - 50
WHERE id = 1; WHERE id = 2;
-- Waiting for row 2 -- Waiting for row 1
UPDATE accounts SET UPDATE accounts SET
balance = balance + 100 balance = balance + 50
WHERE id = 2; -- BLOCKED WHERE id = 1; -- BLOCKED
-- DEADLOCK!
Solutions:
- Prevention: Always lock rows in consistent order (e.g., lower ID first)
- Detection: Database automatically detects deadlocks and kills one transaction
- Mitigation: Use shorter transactions, appropriate isolation levels
- Retry Logic: Application retries failed transactions automatically
Practice Problems
Problem 1: Write SQL for Library Management
Design a database for a library with books, members, and loans. Write SQL to find overdue books.
CREATE TABLE books (
id INT PRIMARY KEY AUTO_INCREMENT,
isbn VARCHAR(13) UNIQUE NOT NULL,
title VARCHAR(255) NOT NULL,
author VARCHAR(100) NOT NULL,
genre VARCHAR(50),
copies_total INT DEFAULT 1,
copies_available INT DEFAULT 1
);
CREATE TABLE members (
id INT PRIMARY KEY AUTO_INCREMENT,
name VARCHAR(100) NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL,
joined_date DATE DEFAULT (CURRENT_DATE)
);
CREATE TABLE loans (
id INT PRIMARY KEY AUTO_INCREMENT,
book_id INT NOT NULL,
member_id INT NOT NULL,
borrowed_date DATE NOT NULL,
due_date DATE NOT NULL,
returned_date DATE NULL,
FOREIGN KEY (book_id) REFERENCES books(id),
FOREIGN KEY (member_id) REFERENCES members(id)
);
-- Find overdue books
SELECT b.title, m.name, m.email, l.due_date,
DATEDIFF(CURRENT_DATE, l.due_date) AS days_overdue
FROM loans l
JOIN books b ON l.book_id = b.id
JOIN members m ON l.member_id = m.id
WHERE l.returned_date IS NULL AND l.due_date < CURRENT_DATE
ORDER BY days_overdue DESC;
Problem 2: Implement Materialized View
Create a materialized view for daily sales summary and set up automatic refresh.
-- PostgreSQL Materialized View
CREATE MATERIALIZED VIEW daily_sales AS
SELECT
DATE(created_at) AS sale_date,
COUNT(*) AS order_count,
SUM(subtotal) AS gross_revenue,
SUM(tax_amount) AS total_tax,
SUM(total) AS net_revenue,
COUNT(DISTINCT user_id) AS unique_customers
FROM orders
WHERE status != 'cancelled'
GROUP BY DATE(created_at)
ORDER BY sale_date DESC;
CREATE UNIQUE INDEX idx_daily_sales_date ON daily_sales(sale_date);
-- Refresh (can be scheduled via cron)
REFRESH MATERIALIZED VIEW CONCURRENTLY daily_sales;
-- Query (fast, pre-computed)
SELECT * FROM daily_sales
WHERE sale_date >= CURRENT_DATE - INTERVAL '30 days'
ORDER BY sale_date DESC;
Problem 3: Write a Recursive CTE
Find all subordinates (direct and indirect) of a manager using a recursive CTE.
WITH RECURSIVE subordinates AS (
-- Base case: direct reports
SELECT id, name, manager_id, 1 AS level
FROM employees
WHERE manager_id = 1 -- CEO's ID
UNION ALL
-- Recursive case: indirect reports
SELECT e.id, e.name, e.manager_id, s.level + 1
FROM employees e
JOIN subordinates s ON e.manager_id = s.id
)
SELECT id, name, level, manager_id
FROM subordinates
ORDER BY level, name;
-- Result shows the full org hierarchy under manager 1
Problem 4: Implement Soft Delete
Add soft delete functionality to a table with audit trail.
ALTER TABLE products ADD COLUMN deleted_at TIMESTAMP NULL DEFAULT NULL;
ALTER TABLE products ADD COLUMN deleted_by BIGINT NULL;
-- Soft delete query
UPDATE products
SET deleted_at = NOW(), deleted_by = ?
WHERE id = ? AND deleted_at IS NULL;
-- Active products (exclude soft-deleted)
SELECT * FROM products WHERE deleted_at IS NULL;
-- Recently deleted (for admin review)
SELECT p.*, u.name AS deleted_by_name
FROM products p
LEFT JOIN users u ON p.deleted_by = u.id
WHERE p.deleted_at IS NOT NULL
ORDER BY p.deleted_at DESC;
-- Permanent delete (after retention period)
DELETE FROM products
WHERE deleted_at < DATE_SUB(NOW(), INTERVAL 90 DAY);
-- Restore soft-deleted
UPDATE products SET deleted_at = NULL, deleted_by = NULL WHERE id = ?;
Problem 5: Database Connection in a Serverless Environment
Explain the connection pooling challenge in serverless and implement RDS Proxy usage.
Answer:
- Problem: Each Lambda instance opens new connections → exhausts database connections
- Solution: Use RDS Proxy — a managed connection pooler
- Benefits: Connection multiplexing, failover support, IAM authentication
// Using RDS Proxy endpoint instead of direct DB endpoint
const pool = new Pool({
host: 'my-proxy.proxy-xyz.us-east-1.rds.amazonaws.com',
port: 5432,
database: 'ecommerce',
user: 'app_user',
password: authToken, // IAM auth token
max: 2 // Small pool — RDS Proxy handles multiplexing
});
Problem 6: Implement Event Sourcing
Design an event-sourced system for a banking account.
CREATE TABLE events (
id BIGSERIAL PRIMARY KEY,
aggregate_id UUID NOT NULL,
aggregate_type VARCHAR(50) NOT NULL,
event_type VARCHAR(100) NOT NULL,
event_data JSONB NOT NULL,
version INT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE (aggregate_id, version)
);
-- Event types: AccountOpened, MoneyDeposited, MoneyWithdrawn, AccountClosed
-- Rebuild account state from events
SELECT
aggregate_id,
MAX(version) as current_version,
SUM((event_data->>'amount')::DECIMAL) FILTER (
WHERE event_type = 'MoneyDeposited') -
SUM((event_data->>'amount')::DECIMAL) FILTER (
WHERE event_type = 'MoneyWithdrawn') AS balance
FROM events
WHERE aggregate_id = 'account-uuid'
GROUP BY aggregate_id;
Problem 7: Time-Series Data Storage
Design a schema for IoT sensor data with 1M writes/day and efficient time-range queries.
-- Optimized time-series schema (TimescaleDB)
CREATE TABLE sensor_readings (
time TIMESTAMPTZ NOT NULL,
sensor_id INT NOT NULL,
temperature DOUBLE PRECISION,
humidity DOUBLE PRECISION,
pressure DOUBLE PRECISION,
battery_level INT
);
-- Convert to hypertable (automatic partitioning by time)
SELECT create_hypertable('sensor_readings', 'time',
chunk_time_interval => INTERVAL '1 day');
-- Indexes for common queries
CREATE INDEX idx_sensor_time ON sensor_readings(sensor_id, time DESC);
-- Efficient time-range query
SELECT sensor_id,
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
COUNT(*) as readings
FROM sensor_readings
WHERE time >= NOW() - INTERVAL '24 hours'
AND sensor_id = 42
GROUP BY sensor_id;
-- Continuous aggregates (materialized views)
CREATE MATERIALIZED VIEW sensor_hourly
WITH (timescaledb.continuous) AS
SELECT
time_bucket('1 hour', time) AS bucket,
sensor_id,
AVG(temperature) as avg_temp,
MIN(temperature) as min_temp,
MAX(temperature) as max_temp
FROM sensor_readings
GROUP BY bucket, sensor_id;
Problem 8: Implement Full-Text Search
Add full-text search to a product catalog with relevance scoring.
-- MySQL Full-Text Search
ALTER TABLE products ADD FULLTEXT INDEX ft_search(name, description);
-- Search with relevance
SELECT
id, name, price,
MATCH(name, description) AGAINST('wireless headphones' IN NATURAL LANGUAGE MODE)
AS relevance_score
FROM products
WHERE MATCH(name, description) AGAINST('wireless headphones' IN NATURAL LANGUAGE MODE)
ORDER BY relevance_score DESC
LIMIT 20;
-- Boolean mode (advanced)
SELECT id, name FROM products
WHERE MATCH(name, description) AGAINST('+wireless +noise -bluetooth' IN BOOLEAN MODE);
-- PostgreSQL equivalent
SELECT id, name,
ts_rank(search_vector, query) AS rank
FROM products,
plainto_tsquery('english', 'wireless headphones') query
WHERE search_vector @@ query
ORDER BY rank DESC;
Problem 9: Database Schema Migration Strategy
Design a zero-downtime migration strategy for adding a NOT NULL column to a large table.
Answer:
- Step 1: Add column as NULLABLE (fast, no table rebuild)
- Step 2: Backfill existing rows in batches
- Step 3: Add NOT NULL constraint (validated)
- Step 4: Update application code to always set the column
- Step 5: Deploy code changes
-- Step 1: Add nullable column
ALTER TABLE orders ADD COLUMN priority INT NULL;
-- Step 2: Backfill in batches (avoid locking)
UPDATE orders SET priority = 0
WHERE id BETWEEN 1 AND 100000 AND priority IS NULL;
-- Repeat for each batch...
-- Step 3: Add constraint
ALTER TABLE orders MODIFY COLUMN priority INT NOT NULL DEFAULT 0;
Problem 10: Implement Multi-Tenant Database
Design a multi-tenant SaaS database with data isolation between tenants.
Answer (3 approaches):
- Shared Database, Shared Schema: tenant_id column on every table (simplest, least isolation)
- Shared Database, Separate Schema: Each tenant has own schema in same database
- Separate Database: Each tenant has own database (best isolation, highest cost)
-- Shared schema approach with Row Level Security (PostgreSQL)
ALTER TABLE products ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON products
USING (tenant_id = current_setting('app.current_tenant')::INT);
-- Set tenant context per request
SET app.current_tenant = '123';
-- Now all queries automatically filtered to tenant 123
-- Always include tenant_id in queries
CREATE INDEX idx_products_tenant ON products(tenant_id, id);