The physical layout decides which bytes a query reads. The same logical Listens table appears in two of them. Row-oriented storage packs whole rows into a page, so reading row 5 returns all four columns in one read and a write is one append. Columnar storage gives each column its own file, so AVG(rating) touches only the rating file and skips the other three. Which layout wins depends on the query: SELECT AVG(rating) favors columnar (row layout wastes about 75% of the I/O on this 4-column toy, far more on wide tables), while SELECT * WHERE listen_id = 42 and inserts or updates favor row, since one record is one packed page rather than a seek-and-reassemble across four files.

A 64 MB slotted page stores variable-length records from the top downward and stores a fixed-size slot array from the bottom upward. Free space remains between them. Each slot stores an (offset, length) pointer, so S0 can point to Row A and S1 can point to Row B. An in-place update writes the row without moving it when the row keeps the same size. When a row grows, the page moves the row to a new location and updates the slot, so references like page_id, slot_id remain stable.

Parquet stores columnar data inside row-groups in one file. A pure columnar layout stores one file per column, so reading one row performs a seek per column and a wide partitioned table produces roughly 100,000 files. Parquet keeps per-column compression and scan behavior, and it stores the columns as chunks inside each row-group. A row stays within one row-group. Row-groups support writes and parallel scans. The footer stores per-row-group min/max statistics, and predicates can skip groups without reading their column chunks. Encodings stop at row-group boundaries, which reduces compression slightly.

BigQuery answers a query over trillions of rows with columnar storage and tree execution. A query for average rating per genre reads the rating and genre column files (green) and skips user_id, song_id, and listen_time (grey). Thousands of leaf workers run while the query runs, each scans a slice of those columns, and partial results flow up a tree to a root that aggregates the answer. BigQuery decouples storage and compute, so what took hours on row storage finishes in seconds.

listen_time provides a partitioning dimension. The system writes fresh listens into small row-shaped parts (grey) to accept inserts at about 1M/sec by writing each row contiguously. A background merge rewrites older parts into columnar segments (green) after activity slows. Analytics scan one column and skip others at roughly 10:1 compression. One query engine reads both layouts through the same SQL surface, without a second system, without ETL lag, and without manual layout choice.

Hash partitioning makes one pass over a too-big-for-RAM file using a small RAM workspace. A page streams into the input slot, each row hashes by h(user_id) % B and routes to one of ~B output buffers, and a full buffer flushes to its partition file on disk. Rows with the same key always land in the same partition, so later stages can sort, join, or group one partition at a time. Cost is below; see also the IO Cost Summary.

Phase 1 reads four pages into memory, quicksorts them in place, and writes the pages back as one sorted run, repeating until the file becomes a set of sorted runs. Phase 2 keeps the head page of each run in memory, repeatedly selects the smallest remaining value, and writes it to the output to merge the runs into one sorted file. Each tile is a page shaded by its sort key, light for small and dark for large. The red box is RAM (volatile). The green is disk (persistent).

Block Nested Loop Join, R ⋈ S. Read the smaller table R from disk and hold it in RAM for the whole join. The larger table S streams past in 4 pages, each loaded into a one-page slot, probed against R, then discarded, producing 8 joined rows (2 per S page here). Best case, when R fits in B pages, is P(R) + P(S) + OUT; if R is bigger than B it becomes P(R) + ⌈P(R)/B⌉ × P(S) + OUT, because S is re-scanned once per R-block. Always make the smaller table the outer one.

Hash Partition Join, R ⋈ S on user_id. Phase 1 hashes both tables with the same function so matching user_ids always land in the same partition number. Phase 2 joins each (Rᵢ, Sᵢ) pair independently in RAM, building a hash table on R and probing with S, three parallelizable joins that yield 4 + 2 + 2 = 8 rows. Best-case cost is 3(N+M) + OUT (read each table, write partitions, read them back, write the output). The worst case is hash skew, where a hot key overflows a partition past RAM and recursive re-partitioning or a sort-merge fallback adds another pass.

Sort-Merge Join, R ⋈ S on user_id (Phase 2 shown). Phase 1 (not shown) BigSorts both R and S on the join key (see the big sort page). Phase 2 scans the two sorted runs with two pointers. Equal keys form a matching key group, and the operator emits every pair in the group. Unequal keys advance the smaller side. The example advances through user_id 7 → 2, 13 → 2, 42 → 3, 99 → 1, total 8 rows. A key that appears on only one side emits nothing under inner join semantics. A key with m matches on one side and n on the other emits the m × n cross-product for that key. The output stream stays sorted on the join key, so a downstream operator that requires that order does not add a sort. Best case is P(R) + P(S) + OUT for one linear scan per run, plus the two Phase-1 BigSort costs. Worst case is O(P(R) × P(S)) + OUT when one key dominates and the merge performs a per-key cross-product. OUT can dominate when match rates are high.

HPJ-best stays near 54K because partitioning does one read and one write independent of B. BNLJ drops from 2.6M to 18K because larger B reduces rescans of the inner table. SMJ-best drops as larger initial runs reduce merge passes. The crossover marks the point where BNLJ becomes cheaper than HPJ at this scale. SMJ with many duplicates can rescan during merge. HPJ with heavy skew can create oversized partitions, each up to 2 to 10x worse.

The Songs table stores whole 10 MB rows (10 TB, 160,000 pages). The artist index stores 20-byte pointers (20 MB, one page), 500,000 times smaller. WHERE artist = 'Beatles' without an index scans 160,000 pages. With an index, the database reads one index page, learns Beatles is on page 47, then reads page 47. The query uses two IOs instead of 160,000, 80,000 times fewer for the same answer.

A hash index on People.name stores (value, page#) entries in buckets chosen by hash(value). A unique key like Alice requires 2 IOs: one index-page read and one data-page read. A duplicate key like Dan requires 4 IOs here: one index-page read plus reads of data pages 0, 1, and 4. The constant factor grows with the number of duplicate keys for the search value.

Data pages on disk store the Songs rows. The B+Tree on disk stores a root and internal nodes that route lookups to sorted leaves. Leaves store (key, data-page pointer) pairs and link left-to-right. For release_date = 12890 ("As It Was"), the root selects 5000..14999, the internal node selects 11234..13788, the leaf resolves 12890 → Page 3, and Page 3 contains the row. The lookup reads four pages.

For 12000 ≤ release_date ≤ 14000, the query visits the solid green nodes, skips the dashed grey ones, and fetches only the data pages that contain matches (Page 3 and Page 4). The descent runs root → middle internal → Leaf 3, then follows leaf links. Leaf 3 yields 12890 and 13456. Leaf 4 yields 13789 and 13980. Leaf 4 ends at 13980, so the scan reads one more leaf to check the boundary. Leaf 5 begins with 14456, which exceeds 14000, so the scan stops. The internal node stores 14456 as the separator key for Leaf 5. The query reads 5 index pages and 2 data pages, 7 IOs total, versus roughly 160,000 for a full scan, about 23,000× faster.

An SSTable is one immutable on-disk file with three sections: a bloom filter in RAM that answers "is key K here?" in O(1) and can skip a disk read on a miss, a sparse index block that binary-searches to the target data block, and sorted, compressed data blocks. Multiple SSTables can contain the same key; a read returns the newest version, here user100's v3 in sstable_3.db over v1 and v2.

LSM write path at 1:44 PM, RAM on top and disk below, time flowing downward. Incoming plays land in the in-RAM MemTable (fast, no seek, lost on crash unless replayed from the WAL). Every 15 minutes the engine flushes it to an immutable Level-0 SSTable on disk (1:00, 1:15, 1:30). Hourly compaction merges those into one larger sorted, de-duplicated Level-1+ file. Total plays for S75 spread across tiers (+2 in the MemTable, +10, +5, and +3 in Level-0, 5,420 in compacted history) and the read path sums them to 5,440 (merge-on-read). The merge operator is configurable: "add" for counters, "newest wins" for last-write-wins keys, "set union" for tags, "max" for high-water-marks. The merge section walks through more examples.

Merge-on-read computes S75's play count at read time instead of storing it, by summing every tier that holds a fragment. The live in-RAM MemTable (red) adds +2, the three flushed Level-0 SSTables on disk (green) add +10 · +5 · +3 = +18, and the compacted Level-1+ SSTable on disk (green) adds +5,420, for a total of 2 + 18 + 5,420 = 5,440 (the merge function is "add" because plays are a counter). Each SSTable carries a bloom filter, so the reader skips any tier that cannot hold the key with no disk read, making most reads far cheaper than this worst case.

A range query for rating 4 or 5 matches four rows. A clustered table stores rows sorted by rating, so the matches occupy two contiguous pages and require two sequential reads. The index stores one entry per page (sparse). An unclustered table stores rows in insert order, so the four matches occupy four pages and require one random read per matching row. The index stores one entry per row (dense). Green marks an in-range row and a page that gets read.

An inverted index over the Songs table uses a term dictionary on disk, implemented as a B+Tree over every unique word, and postings lists on disk, stored as sorted packed arrays of song IDs per word. For "Crazy Heart" the engine looks up each word, intersects the two sorted postings lists in RAM to {12, 108}, and reads those 2 data pages, about 10 IOs. A full-table scan reads all 100 million rows, so the index uses roughly 10,000,000× fewer reads.

Spotify avoids seeks on the write path: each like, play, or pause appends to an in-memory MemTable, then flushes sequentially to an immutable SSTable on disk. Background compaction merges SSTables through the merge function and drops superseded versions and tombstones. A read scans newest to oldest, starting with the MemTable, and stops at the first match.

Wrapped runs the expensive computation overnight against an index that supports sequential reads. Chronological storage (grey) scatters one user's listens across the disk. Clustering Listens by (user_id, listen_time) places that user's year on one page, so the nightly batch reads it as a sequential pass and writes one summary row per user (green). On December 1 the card fetch uses a single hash-index lookup on user_id.

Four stages of the same weekly aggregation. The baseline Hash Partition Join scans the full 1 TB at 49,200 IOs and about $550,000 a year. A B+Tree index cuts rows scanned (×90). Columnar storage cuts columns read (×9). Compression cuts bytes stored (×2). The combined effect reaches roughly 1,400× and about $400 a year.

SQL is declarative: it states the result, not the procedure. Parsing and binding produce a validated tree; the logical plan then applies equivalence-preserving relational-algebra rewrites (predicate pushdown, projection pruning, subquery flattening). "Defines the result" means exactly this: every rewrite, and every plan the next stage considers, returns the same rows, so the answer is now pinned and only running time can still change. The physical-plan cost search decides speed: it chooses a join order, join algorithm, and access path, and because trying all n! orders is infeasible it uses dynamic programming, computing the cheapest sub-plan for each subset of tables once and reusing it (so n! work collapses to about 2ⁿ), with each candidate costed from table statistics. Only the winner executes, routinely about 100 times faster than the naive ordering.

The optimizer assembles a physical plan from four families of interchangeable blocks: access methods (sequential, index, and bitmap scans), partition and sort (Hash Partition, BigSort), aggregation (Hash Aggregate, Sort Aggregate), and join algorithms (Hash Join, Sort-Merge Join, Block Nested Loop). Any plan is one choice from each family wired together, so plan selection is a cost comparison across those combinations. Each block is a primitive with its own page.

Reservoir sampling is the default when you need an unbiased uniform sample in one streaming pass over a table too big to shuffle. Keep the first k rows; for each later row i, swap it into a random slot with probability k/i, otherwise discard it. The result is exact: every one of the N rows is in the final sample with equal probability k/N, using only O(k) memory and no second pass. Two cheaper alternatives trade exactness for speed: block sampling reads whole pages (fast I/O, but biased when rows are clustered on disk), and progressive sampling starts small and grows until the estimate's variance is low enough (cheap when the data is easy, more passes when it is not).

An equi-width histogram splits the value range into equal slices; it is trivial to build but on skewed data one bucket swallows almost all the rows, so any selectivity read off it is wildly wrong. A height-balanced histogram instead gives every bucket the same row count (each holds 1/B of the rows), so bucket ranges automatically shrink where data is dense, which keeps estimates accurate exactly where skew would otherwise hurt. A third kind, the wavelet histogram, stores a multi-resolution summary for complex or multi-dimensional distributions.

PostgreSQL's actual EXPLAIN output for the popular-songs query, reproduced verbatim from the course colab and read bottom-up: a Seq Scan on songs builds a hash, a Seq Scan on listens probes the Hash Join on song_id, a HashAggregate groups by title and artist, a Sort orders by count, and a Limit takes the top ten. Each line shows the operator (green) and the optimizer's own cost=startup..total and rows= estimate (grey); these are PostgreSQL's estimates on the loaded table, so read the shape of the plan rather than the absolute numbers.

Rows begin scattered across multiple nodes. Hashing each row on the join key picks its target machine, so every row sharing a key travels over the network to the same node. Once the keys sit together, each node runs an ordinary local join on the rows it now owns. This is exactly hash partitioning with the network as the destination, so the cost is the same 2N IO plus the network transfer.

A single tap on play fans out into three systems in sequence. Act 1 is the OLTP write that appends the play to an LSM tree; Act 2 is the ETL pipeline where Spark hash-partitions and BigSorts the events, a broadcast join enriches them, and Spark writes the result columnar; Act 3 is the OLAP query that scans the columnar star schema with a broadcast join to answer the dashboard in about two seconds. Every block in this path is a primitive built earlier in the course, now wired end to end.

The five workloads below and the structure each one calls for, from a B+Tree for date ranges to hash partitioning for a viral surge. Every row pairs a Spotify problem with the tool that fits its access pattern and the one reason it fits. Green marks the chosen tool.