interviewpath
  • 0day streak
  • 10total XP
  • Level 1

Focus mode

OLTP vs OLAP, distributed joins and distributed transactions

Read alongside

CMU 15-445/645 Intro to Database Systems: Fall 2025 Lecture #24: Distributed Database Systems II (notes)

Atomic commit protocols and distributed join algorithms; Lecture #23 covers architectures and partitioning.

Objectives

  • Contrast OLTP and OLAP workloads by access pattern, storage layout, indexing and concurrency needs.
  • Choose a partitioning scheme (hash, range, by tenant or entity) and explain which queries and transactions it keeps on one node.
  • Estimate network cost for co-located, broadcast and shuffle joins and choose one for given table sizes and partitioning.
  • Walk through two-phase commit, including the prepared state, the coordinator’s decision record and the behavior on each failure.
  • Explain why two-phase commit blocks when the coordinator fails after participants vote yes, and how replicating coordinator and participants addresses it.
  • Explain why analytics usually runs on a separate columnar system fed by ETL or change data capture, and the freshness tradeoff.

Why it matters

"Design an analytics pipeline for X" and "the database has outgrown one machine" are standard system-design prompts. Good answers rest on recognizing the workload's shape, choosing a partition key that keeps work local, knowing what a distributed join costs, and being honest about what a cross-partition transaction requires.

Core ideas

Two workload shapes. OLTP: many concurrent short transactions, point reads and writes by key, strict latency targets, row storage with B+ tree indexes. OLAP: a few large queries that scan, filter, join and aggregate millions or billions of rows; columnar storage, compression, vectorized and parallel execution. Running heavy analytics on the OLTP primary competes for memory, I/O and CPU, and long-running queries hold old snapshots that delay version cleanup. Analytical data is therefore usually copied to a warehouse or lake through ETL or change data capture, trading freshness for isolation. HTAP systems try to serve both, often with separate row and column copies.

Warehouse modeling. A star schema has a large fact table (one row per event, with numeric measures) surrounded by smaller dimension tables (customer, product, date). Typical queries filter dimensions, join them to facts and aggregate, which is exactly what columnar scans and hash joins do well.

Partitioning. Hash partitioning spreads load evenly but turns range queries into scatter-gather; range partitioning keeps ranges local but creates hot spots on monotonically increasing keys. Partitioning by entity (tenant, user, account) keeps most transactions on one node. A global secondary index on another attribute is itself partitioned, so maintaining it can require cross-node writes.

Distributed joins. If both tables are partitioned on the join key, the join is co-located and moves no data. Otherwise:

  • Broadcast join: send the small table to every node holding part of the large one; cost is roughly the small table's size times the number of nodes.
  • Shuffle (repartition) join: rehash both inputs on the join key so matching rows meet; cost is roughly the size of both inputs. Broadcast wins when one side is small after filtering; shuffle wins when both are large. A heavily skewed key overloads the node that receives it, so engines split heavy hitters or broadcast only their matching rows. Semi-join reduction ships just the join keys of one side, and a Bloom filter of those keys is smaller still, so the other side can be filtered before it moves. Aggregation runs in two phases: partial aggregates per node, then a final merge.

Atomic commit across partitions. In two-phase commit (2PC) the coordinator asks every participant to prepare; each makes its changes durable, keeps its locks and votes. If all vote yes, the coordinator durably records commit and informs everyone; any no vote or timeout means abort. A participant that voted yes cannot decide alone afterwards: if the coordinator crashes before announcing the outcome, the participant waits with its locks held. That is the blocking problem. Production systems shrink it by replicating the coordinator's decision and each participant with consensus (each participant becomes a Paxos or Raft group), so a failed leader is replaced rather than waited for. PostgreSQL exposes the participant side as PREPARE TRANSACTION, and an orphaned prepared transaction keeps its locks until someone commits or rolls it back.

Atomicity is not isolation. 2PC makes all participants commit or abort together; it does not stop concurrent transactions from seeing a partial state. Distributed systems still need concurrency control spanning partitions, and reading a consistent snapshot across partitions needs coordinated timestamps.

Common misconceptions

  • "2PC is a consensus protocol." 2PC needs every participant's vote and blocks on coordinator failure; consensus needs a majority and survives minority failures. Practical systems combine the two.
  • "Sharding makes everything faster." Single-partition operations scale; cross-partition joins and transactions get slower and more failure-prone.
  • "Broadcast joins are always cheaper." Only when the broadcast side is small; otherwise network traffic and memory multiply.
  • "Atomic commit gives isolation." It gives all-or-nothing; isolation needs concurrency control across partitions.
  • "Analytics can just run on a read replica." A replica removes write contention but keeps the row layout and indexes built for OLTP; scan-heavy work is usually far cheaper on columnar storage.

How to approach it

Use one running example (orders, customers, products) and answer four questions: which partition key, which queries stay local, what a cross-partition join moves, and what a transfer between two customers on different partitions requires. Practice the 2PC failure cases as timelines, then run the PREPARE TRANSACTION exercise to see a prepared transaction hold locks. In interviews, quantify data movement and name the failure behavior of whatever commit protocol you propose.

Self-check before you finish

  1. What exactly is a participant waiting for after it votes yes, and why can it not decide alone?
  2. How does replicating each participant with consensus change 2PC’s failure behavior?
  3. When does a broadcast join beat a shuffle join, and what can make it fail?
  4. Why does range partitioning on a timestamp create a hot spot?
  5. Why is atomic commit not enough to give isolation across partitions?
  6. What is a fact table, and what queries is a star schema designed for?