Databases Advanced: Distributed Systems & Warehousing

A distributed database spreads data across multiple nodes for scale and resilience. Two core strategies: replication (copies of the same data on different nodes — read scale + failover) and sharding (different data on different nodes — write scale). Every distributed system is built from these primitives, combined and traded off. The fundamental tension: you gain scale and availability but pay in complexity and consistency.

# Replication: copies of SAME data
#   node_A: users 1-1000 (primary)
#   node_B: users 1-1000 (replica)
# Sharding: DIFFERENT data
#   shard_0: users 1-500k
#   shard_1: users 500k-1M

Replication copies the same data; sharding splits different data across nodes.

Consistent hashing maps keys to nodes on a ring, so adding/removing a node only remaps a fraction of keys — unlike naive modulo hashing where every key moves. Virtual nodes (many points per physical node) balance the load. This is how distributed caches and databases (Redis Cluster, Cassandra, DynamoDB) distribute data. It's THE algorithm for elastic scaling.

import hashlib

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

# Keys map to positions on a ring; each node owns a range
print(ring_position('user:42') % 100)

Every key remaps when node count changes, causing mass cache misses.

Two-Phase Commit atomically commits a transaction across multiple databases. Phase 1 (prepare): coordinator asks all nodes 'can you commit?' — each locks and votes. Phase 2 (commit/abort): all commit or all abort. It guarantees atomicity across nodes, but blocks during failures and is slow — the classic reason distributed systems prefer eventual consistency for most operations.

# Phase 1: COORDINATOR -> 'PREPARE?' -> NODE A, NODE B
# Phase 2: all 'YES' -> 'COMMIT' ; any 'NO' -> 'ABORT'
# Blocks if a node is unreachable (coordinator waits)

Prepare (vote) and Commit/Abort.

Consensus algorithms let a cluster agree on a value even with node failures. Raft (more understandable than Paxos) elects a leader, replicates a log of commands to followers, and commits when a majority agree. This is how etcd, Consul, and modern distributed databases maintain consistency. Consensus is the foundation of reliable distributed systems — the 'brain' of the cluster.

# Raft: leader election -> log replication -> commit on majority
# Leader receives write, replicates to followers
# Commit when majority (quorum) acknowledge
# If leader fails -> new election

A majority of nodes (>50%) agreeing on a value.

Eventual consistency means replicas converge over time — a read may briefly return stale data after a write. It enables massive scale (no coordination latency) at the cost of temporary inconsistency. Patterns: read-your-writes (user sees own writes), monotonic reads (never see time go backward), and conflict resolution (last-write-wins vs merge). Most internet-scale systems are eventually consistent.

# Write to primary, async replication to replicas
# Replica lag = staleness window (usually < 1s)
# Read-your-writes: route user's reads to primary after their writes

No coordination overhead — massively higher write throughput and availability.

CRDTs are data structures that merge deterministically regardless of update order — two nodes can update independently and always converge. A grow-only counter, an add-wins set. They power collaborative editing (Google Docs-style), offline-first apps, and multi-region sync. The magic: the merge operation is commutative and associative, so order doesn't matter.

# Grow-only counter: only increments, never conflicts
# Node A: count = 3 (saw +1, +2)
# Node B: count = 5 (saw +1, +2, +3, -2 ignored)
# Merge: max(A, B) = 5 — converged

Their merge is commutative/associative — order doesn't matter, so no conflict.

A vector clock tracks, for each node, how many events it has seen — enabling detection of causal relationships between events. Two events are concurrent if their vector clocks are incomparable (neither dominates). This is how systems detect conflicts and enforce causal consistency. Vector clocks trade size (one entry per node) for precise causality tracking.

# Node A clock: {A: 3, B: 1}
# Node B clock: {A: 1, B: 4}
# Incomparable => concurrent writes => conflict

The events are concurrent — no causal ordering.

The Saga pattern manages a distributed transaction as a sequence of local transactions, each with a compensating action (undo). If a step fails, previously-completed steps are rolled back via their compensations. Unlike 2PC, sagas don't block — they're eventually consistent. This is how microservices handle multi-step operations like 'place order + charge card + ship item'.

# Saga steps:
# 1. Reserve inventory  -> compensate: release inventory
# 2. Charge payment     -> compensate: refund payment
# 3. Ship order         -> compensate: cancel shipment
# If step 3 fails, run compensations for 2 then 1

An operation that undoes a previously-completed saga step.

Change Data Capture captures row-level changes (inserts/updates/deletes) from the database's transaction log and streams them to other systems — data warehouses, caches, search indexes. Debezium (on top of Kafka) is the standard tool. CDC keeps systems in sync in near-real-time without batch ETL, and without touching the source app. It's the backbone of event-driven data pipelines.

# CDC: read WAL (write-ahead log)
# INSERT users -> {user_id: 1, name: 'Ali'}
# Stream to: search index, cache, warehouse

The database's transaction log (WAL).

A materialized view stores the RESULT of a query physically — so expensive aggregations run once, not every read. Dashboards over millions of rows become instant. The trade-off: results get stale between refreshes. Refresh strategies: on-demand, scheduled, or (in some DBs) incremental. Materialized views are the bridge between raw data and fast analytics.

CREATE MATERIALIZED VIEW daily_sales AS
SELECT date_trunc('day', order_date) AS day, SUM(total) AS revenue
FROM orders
GROUP BY day;

REFRESH MATERIALIZED VIEW daily_sales;

To precompute expensive aggregations so queries are instant.

A data warehouse is a dedicated analytical database optimized for complex queries over historical data — separate from the operational (OLTP) database so analytics doesn't slow down the app. It stores denormalized, subject-oriented, time-variant data. The separation is key: OLTP optimizes fast writes; the warehouse optimizes fast reads over millions of rows. This is the foundation of business intelligence.

# OLTP: normalized, fast writes, current data
# Warehouse: denormalized, fast reads, historical data
# ETL moves data from OLTP -> warehouse

Analytical queries would compete with transactional writes, slowing the app.

Analytical schemas organize data into fact tables (measurements: sales, events) and dimension tables (descriptors: customer, product, date). The star schema keeps dimensions denormalized (flat, fast); the snowflake schema normalizes dimensions (nested, less redundancy but more joins). Star is the default for query speed; snowflake saves space. This is the classic dimensional modeling pattern.

# STAR: fact_sales <-> dim_customer, dim_product, dim_date
# SNOWFLAKE: dim_customer -> dim_city -> dim_country (normalized)

A table of quantitative measures/events (e.g., sales transactions).

ETL (Extract, Transform, Load) transforms data BEFORE loading into the warehouse — traditional, good for strict governance. ELT (Extract, Load, Transform) loads raw data first, transforms inside the warehouse — modern, leverages warehouse compute power and flexibility. Cloud warehouses (Snowflake, BigQuery) made ELT dominant. The key shift: transformation happens in the warehouse, not a separate tool.

# ETL: extract -> transform (in ETL tool) -> load clean
# ELT: extract -> load raw (into warehouse) -> transform (SQL in warehouse)

Cloud warehouses are powerful enough to transform data in-place, making pre-load transformation unnecessary.

OLTP (transactional) optimizes for many small, fast writes — bank transactions, shopping carts. OLAP (analytical) optimizes for few large, complex reads — reports, dashboards. The optimizations conflict: row stores for OLTP, column stores for OLAP. Using the wrong system for a workload causes massive performance problems. This distinction drives almost every database architecture decision.

# OLTP: row store, many small writes
#   UPDATE account SET balance = ... WHERE id = 123
# OLAP: column store, few large reads
#   SELECT SUM(revenue) FROM sales WHERE year = 2026

OLTP — many small, fast writes.

Columnar storage stores each column's data together (instead of each row). For analytical queries touching few columns over millions of rows, this is transformative: scan only needed columns, compress similar values (same column = same type = great compression), and vectorize operations. Columnar DBs (ClickHouse, BigQuery, Redshift) are orders of magnitude faster for aggregations than row stores.

# Row store: [id, name, age], [1, 'Ali', 25], [2, 'Sara', 30]
# Column store: id[1,2], name['Ali','Sara'], age[25,30]
# Query 'AVG(age)' reads ONLY the age column

Same column = same data type = similar values compress extremely well.

The query optimizer transforms your SQL into an efficient execution plan. It estimates costs for different strategies (which index, which join order, which join algorithm — nested loop, hash join, merge join) and picks the cheapest. Understanding the optimizer's choices explains why some queries are fast and others slow. It's the 'compiler' of the database.

# Optimizer choices:
# 1. Which index to use (or full scan)
# 2. Join order (A⋈B vs B⋈A)
# 3. Join algorithm (nested loop / hash / merge)
# Picks lowest estimated cost

Estimated execution cost (I/O + CPU).

EXPLAIN ANALYZE shows both estimated and ACTUAL rows/time — the discrepancy between them reveals stale statistics. Profiling breaks down where time goes (scan, join, sort). The workflow: run EXPLAIN ANALYZE, look for row-count mismatches (planner guessed wrong), expensive nodes (sorts, hash joins), and fix accordingly. This turns slow-query debugging from guessing into a precise process.

EXPLAIN ANALYZE
SELECT o.id, c.name
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.total > 1000;

Stale table statistics — the optimizer made a bad plan.

Cost-based optimization estimates each plan's I/O and CPU cost using table statistics — row counts, distinct values, histograms of value distribution. Accurate statistics = good plans; stale statistics = the planner picks wrong. The fix: keep ANALYZE/VACUUM up to date. This is why 'the database got slow' after bulk data loads — statistics are stale.

ANALYZE orders;  -- refresh statistics
SELECT * FROM pg_stats WHERE tablename = 'orders';

The optimizer picks suboptimal execution plans based on wrong row estimates.

Three ways to read data: Sequential Scan (full table, best for large fraction of rows), Index Scan (random access via index, best for few rows), Bitmap Index Scan (combines multiple indexes, best for medium selectivity). The optimizer picks based on row estimates and row order correlation. Spotting the wrong scan in EXPLAIN is the core query-tuning skill.

EXPLAIN SELECT * FROM users WHERE age > 30 AND city = 'Lahore';
-- Bitmap Heap Scan on users
--   Bitmap Index Scan (age + city indexes combined)

When returning most of the table — random index access is then slower than reading sequentially.

Sharding splits data across nodes by a shard key. Range sharding (by date, id ranges) is intuitive but risks hot spots (recent data on one node). Hash sharding (hash(key) mod N) balances uniformly but makes range queries expensive. The shard key choice is the hardest decision in sharding — it determines access patterns forever. Choose based on your dominant query pattern.

# Range: shard by user_id 0-1M, 1M-2M (hot spots possible)
# Hash: shard = hash(user_id) % 4 (uniform, no range queries)
# Consider: hot user sharding (isolate heavy users)

Most traffic concentrating on a single shard, causing imbalance.

Read/write splitting routes writes to the primary and reads to replicas — scaling read-heavy apps (most web apps are read-dominant). The challenge: replication lag means replicas may briefly return stale data. Solutions: route critical reads to primary, use read-your-writes consistency, and monitor lag. This is the standard scaling pattern for PostgreSQL/MySQL.

# writes -> primary
# reads -> replica (if staleness tolerable)
# critical reads (after write) -> primary

Replication lag — reads may briefly return stale data.

Multi-region replication places data near users worldwide, cutting latency dramatically. The trade-off: cross-region writes pay network latency, and consistency becomes eventual (regions can't stay strongly consistent across oceans). Strategies: single-primary multi-region (writes to one region), multi-primary (writes anywhere, conflict resolution), and read-only replicas per region. This is global-scale database design.

# Region US: primary
# Region EU: read replica (latency for EU users ~10ms)
# Region APAC: read replica
# Cross-region writes: async replication (eventual)

Cross-region network latency (speed of light) makes synchronous replication too slow.

Point-In-Time Recovery (PITR) restores a database to any moment using backups + transaction logs (WAL). Two metrics define recovery quality: RPO (Recovery Point Objective — how much data you can lose, in time) and RTO (Recovery Time Objective — how fast you recover). Continuous WAL archiving gives near-zero RPO. This is the difference between 'we lost data' and 'we restored to 30 seconds ago'.

# Full backup + continuous WAL archiving
# Restore: load last full backup, replay WAL to desired point
# RPO: seconds (WAL shipped continuously)
# RTO: minutes-hours

Maximum acceptable data loss, measured in time (e.g., 5 minutes).

Databases have connection limits (often 100s). Apps serving thousands of users need connection pooling and multiplexing: PgBouncer pools connections, multiplexing many clients over few DB connections. Without it, 'too many connections' crashes apps under load. This is non-negotiable production infrastructure for any serious web app.

# PgBouncer: 10k app clients -> 200 DB connections
# Transaction pooling: connection released after each transaction
# Session pooling: held for the session (needed for some features)

Each connection costs memory/CPU; DBs have hard limits and degrade beyond them.

A caching layer (Redis/Memcached) absorbs repeated reads so the database only handles writes and cache misses. Redis is richer (data structures, persistence, pub/sub); Memcached is simpler (pure cache, multi-threaded). Cache stampede prevention (when many clients miss simultaneously and hammer the DB) uses locking or TTL jitter. Caching is the cheapest, highest-impact performance win.

# Cache: key -> value in Redis
# get -> hit? return : query DB, set cache, return
# Stampede fix: lock on miss, or randomized TTL

Many clients miss the cache at once and overload the database.

Encryption at rest protects data on disk (full-disk encryption, Transparent Data Encryption); encryption in transit protects data moving over the network (TLS/SSL). Both are required for security compliance (PCI, HIPAA, GDPR). Key management (where keys live, who can access) is the hard part. This is baseline database security for any sensitive data.

# At rest: TDE (transparent data encryption) encrypts data files
# In transit: SSL/TLS encrypts client-database connections
# Key management: keys in KMS, rotated regularly

At rest = stored data; in transit = data moving over the network.

Auditing tracks WHO did WHAT to data and WHEN — essential for compliance (GDPR, HIPAA, SOC 2) and security forensics. Database audit logs, trigger-based audit tables, and dedicated audit tools (pgAudit) capture this. GDPR adds requirements: right to access, right to erasure ('delete my data'), data minimization. Auditing is how you prove compliance.

CREATE EXTENSION pgaudit;
SET pgaudit.log = 'write, ddl';
-- Now every write and DDL is logged with user + timestamp

Users can request complete deletion of their personal data.

Backup strategy: full backups (everything, periodic), incremental (changes since last backup, frequent), differential (changes since last FULL backup). The 3-2-1 rule: 3 copies, 2 media types, 1 offsite. Critical practice: regularly TEST restores — a backup you can't restore is worthless. RPO/RTO drive the strategy choice. This is operational database maturity.

# Full: weekly, complete snapshot
# Incremental: daily, changes since last backup
# 3-2-1: 3 copies, 2 media, 1 offsite
# TEST restore monthly

3 copies, 2 different media types, 1 offsite copy.

Performance monitoring uses built-in instrumentation (PostgreSQL's pg_stat_* views, MySQL's performance_schema) to track: query latency, lock waits, cache hit ratio, connection usage, and I/O. Prometheus + Grafana turn these into dashboards with alerts. Key insight: you can't fix what you don't measure. Monitoring turns 'users say it's slow' into precise data.

SELECT * FROM pg_stat_activity;  -- current queries
SELECT * FROM pg_stat_database;  -- cache hits, commits
SELECT * FROM pg_locks;          -- lock contention

Low hit ratio means insufficient memory — data being read from disk instead of cache.

The slow-query workflow: (1) identify slow queries (log_min_duration_statement, pg_stat_statements), (2) run EXPLAIN ANALYZE, (3) find the bottleneck (seq scan? bad join? sort?), (4) fix (add index, rewrite query, update stats), (5) verify improvement. This disciplined loop beats guessing — each step is evidence-driven. It's the difference between 'I tried stuff' and 'I fixed it'.

# 1. Find slow queries
SELECT query, mean_exec_time FROM pg_stat_statements
ORDER BY mean_exec_time DESC LIMIT 10;
# 2. EXPLAIN ANALYZE the slowest
# 3. Fix (index, rewrite)
# 4. Verify improvement

Run EXPLAIN ANALYZE to see the actual execution plan.

Advanced partitioning: RANGE (time-based), LIST (category-based), HASH (uniform distribution). Partition pruning means queries touching one partition only scan that one — massive speedup. Management benefits: drop old partitions instantly (vs slow deletes), archive easily. For billion-row tables, partitioning is mandatory. Choose the partition key that matches your query filter patterns.

CREATE TABLE events (...) PARTITION BY RANGE (event_date);
CREATE TABLE events_2026_01 PARTITION OF events
FOR VALUES FROM ('2026-01-01') TO ('2026-02-01');
-- Drop Jan data = DROP TABLE events_2026_01 (instant)

The optimizer only scans partitions relevant to the query, skipping the rest.

Full-text search matches KEYWORDS (exact/stemmed terms). Vector search matches MEANING (embeddings — similar sentences have close vectors). Keyword is precise (great for exact terms); vector finds related content even with different words ('car' ≈ 'automobile'). Modern search systems use HYBRID — keyword + vector — for both precision and semantic recall. Vector databases (pgvector, Pinecone, Weaviate) power RAG and semantic search.

CREATE EXTENSION vector;
CREATE TABLE articles (id INT, content TEXT, embedding vector(384));
-- Semantic query
SELECT id FROM articles
ORDER BY embedding <-> query_embedding LIMIT 10;

Semantic meaning via embedding similarity.

Graph databases store data as nodes and relationships — perfect for highly-connected data: social networks, fraud detection, recommendation engines. Cypher (Neo4j's query language) expresses relationship queries elegantly ('friends of friends') that would be painful recursive SQL. Graph DBs excel at traversal depth, pattern matching, and relationship-centric queries where relational DBs struggle.

// Cypher: friends of friends
MATCH (u:User {name: 'Ali'})-[:FRIENDS]->(f)-[:FRIENDS]->(fof)
WHERE fof <> u
RETURN fof.name;

Relationship traversal and pattern matching across connected data.

Time-series DBs (TimescaleDB, InfluxDB) specialize in timestamp-ordered data: sensor readings, metrics, financial ticks. Key features: hypertables (auto-partition by time), continuous aggregates (precomputed rollups), compression (columnar, huge savings), retention policies (auto-delete old data). Write-heavy, time-range-query workloads are their sweet spot — relational DBs degrade on these.

SELECT create_hypertable('metrics', 'time');

-- Continuous aggregate (precomputed hourly rollup)
CREATE MATERIALIZED VIEW hourly_avg
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 hour', time) AS hour, AVG(value)
FROM metrics GROUP BY hour;

Automatic deletion of data older than a threshold.

A data lake stores raw data in cheap object storage (S3) in open formats (Parquet, Iceberg) — any format, any size. A lakehouse adds warehouse features (ACID transactions, schema, SQL) on top of the lake — the best of both: cheap storage + warehouse performance. Iceberg/Delta Lake table formats enable this. The lakehouse is the modern data architecture, replacing rigid warehouses.

# Data lake: raw Parquet files in S3 (any data, cheap)
# Lakehouse: Iceberg table format (ACID, schema, time-travel)
#   SELECT * FROM iceberg_table WHERE ... (SQL on the lake)

Cheap object storage with warehouse-grade querying and ACID.

Data governance is the discipline of managing data quality, security, lineage, and cataloging across an organization. Components: data catalogs (what data exists), data lineage (where it came from), data quality rules, access policies. It's the framework that makes data trustworthy and compliant. Without governance, data becomes an unmanaged liability; with it, a reliable asset.

# Governance pillars:
# Catalog: what data exists (Atlas, Unity Catalog)
# Lineage: where data came from (dbt lineage, OpenLineage)
# Quality: validation rules (Great Expectations)
# Access: who can see what (RBAC/ABAC)

Tracking where data came from and how it's transformed.

Trino (formerly Presto) is a distributed query engine that runs SQL across MULTIPLE data sources at once — PostgreSQL, S3, Kafka, MySQL — without moving the data. It federates queries, splitting work across worker nodes and pushing filters down to sources. This enables 'one SQL query across your whole data estate'. It's the query layer on top of a data lake/lakehouse.

-- Trino: query Postgres AND S3 in one query
SELECT u.name, o.total
FROM postgres.users u
JOIN s3.orders o ON u.id = o.user_id;

Querying multiple data sources in a single SQL query.

OLAP cubes precompute aggregates across multiple dimensions (product, region, time) so any slice-and-dice query is instant. SQL's GROUP BY ROLLUP (hierarchical subtotals) and CUBE (all combinations) generate these. The trade-off: precomputation takes storage and compute, but query latency drops to milliseconds. This is the engine behind pivot tables and BI dashboards.

SELECT region, product, SUM(sales) AS total
FROM sales
GROUP BY ROLLUP(region, product);
-- gives: per region+product, per region, grand total

ROLLUP gives hierarchical subtotals; CUBE gives all dimension combinations.

Database-as-a-Service (RDS, Aurora, Cloud SQL) handles backups, patching, replication, failover — you focus on schema and queries. Serverless databases (Aurora Serverless, Neon) scale compute automatically, billing per usage — ideal for variable workloads. The trade-off: less control, vendor lock-in. For most teams, managed DBs are a massive operational win: reliability without the ops team.

# Managed: backups, replication, failover handled
# Serverless: scales to zero when idle, bursts under load
# You: schema + queries + application logic

Automatic scaling and pay-per-use — no idle capacity costs.

Your capstone: architect a complete data platform. Design a data warehouse (star schema) fed by CDC streaming, a caching layer for hot queries, materialized views for dashboards, and a sharding/replication plan for scale. Document RPO/RTO, consistency choices, and security. This integrates everything — the deliverable is a production-grade architecture, portfolio-ready.

# Architecture:
# OLTP (Postgres) --CDC--> Kafka --ETL--> Warehouse (star schema)
# Warehouse -> materialized views -> dashboards
# Redis cache for hot queries
# Replication for read scale, sharding for write scale

OLTP source, CDC/streaming, warehouse/lakehouse, caching, BI/dashboards.