SQL 117 🛢️ Massively Parallel Processing (MPP) Architecture
MPP architecture is the structural answer to a single question: what happens when the data is too large and the query too complex for one machine to process in a reasonable time? The answer is to spread both the data and the work across many independent nodes, each with its own CPU, memory, and disk, connected by a network rather than a shared bus. This is the defining property of MPP: a shared-nothing architecture where no node can read another node’s memory or disk directly . Every query that touches distributed data becomes a coordination problem—which node holds which rows, which node computes which fragment, and when must data move between nodes to complete the work.
This chapter covers the MPP architecture as it applies to analytical databases and data warehouses. It examines the shared-nothing foundation, the coordinator-and-compute-node topology, the two table distribution strategies that determine whether a query stays local or requires a network shuffle, and the execution model that splits a single SQL statement into parallel fragments across nodes.
Key point: MPP performance is determined less by the raw speed of individual nodes than by how well the data distribution matches the query patterns. A query that filters or joins on the distribution key executes locally on each node. A query that does not requires a shuffle—and the shuffle is where the time goes.
Why MPP architecture exists
The single-machine ceiling problem. A symmetric multiprocessing (SMP) database runs on one server and scales by adding CPUs, memory, and storage to that server. The ceiling is physical: a motherboard holds a finite number of sockets, memory slots, and drive bays. When the data outgrows the machine, the only options are to archive data, accept slower queries, or move to a different architecture . MPP exists because analytical workloads—scanning billions of rows, joining large fact tables, aggregating across dimensions—outgrow single machines faster than transactional workloads do.
The shared-resource contention problem. In an SMP system, every query competes for the same memory bus, the same disk controllers, and the same CPU cache. Adding more concurrent queries degrades every query’s performance. MPP eliminates this contention by isolating resources per node: each node’s CPU and disk serve only its own portion of the data and its own query fragments . The trade-off is that coordination across nodes becomes the new bottleneck, replacing the memory bus with the network interconnect.
The data distribution problem. An MPP system must decide where each row lives. If a fact table is spread across nodes by a hash of customer_id, then a query filtering on customer_id reads only the rows on the nodes that hold that customer. A query filtering on order_date reads every node, because no node owns a date range exclusively. The choice of distribution key is therefore a performance decision made at table creation time, and it determines which queries will be fast and which will require data movement .
The execution coordination problem. A single SQL statement cannot be sent to one node in an MPP system and executed there. The coordinator must parse the query, determine which fragments can run locally and which require redistribution, generate a distributed plan, dispatch fragments to compute nodes, and assemble the results. This coordination layer is what distinguishes an MPP database from a collection of independent databases .
a. Shared-Nothing Topology and the Coordinator
The shared-nothing architecture means each node is a self-contained computer with its own operating system, database instance, and storage. Nodes communicate only through a high-bandwidth interconnect, typically InfiniBand or a high-speed Ethernet network . There is no shared disk array and no shared memory segment. The isolation is what allows linear scaling: adding a node adds CPU, memory, and disk without the contention that would occur if all nodes shared a single resource.
The topology separates roles. The control node (or coordinator) receives queries from clients, parses and optimizes them, generates the distributed execution plan, and coordinates execution across the compute nodes . It also holds metadata about table distribution, statistics used by the optimizer, and the catalog of database objects. The compute nodes hold the actual data partitions and execute the query fragments assigned to them. Each compute node runs its own instance of the database engine, processing only the data stored locally .
Microsoft’s Parallel Data Warehouse documentation describes this separation explicitly: the control node “creates parallel query plans and coordinates the parallel query execution across the compute nodes,” while compute nodes are “parallel units of data processing and storage” with direct-attached storage . The control node does not store user data in this architecture. It is the brain; the compute nodes are the muscle.
The coordinator’s workload is not trivial. It must estimate the cost of different join orders and distribution strategies, decide whether to broadcast a small dimension table to every node or to shuffle both sides of a join, and manage the memory and concurrency limits of each compute node. Modern MPP optimizers use cost-based planning: they consult statistics about table sizes and data distributions to choose between broadcast joins (replicating a small table to all nodes) and shuffle joins (repartitioning both tables by the join key) .
b. Table Distribution: Replicated vs. Distributed
An MPP database has two fundamental ways to place a table’s rows across compute nodes. The choice is made at table creation time and has permanent consequences for query performance.
Distributed tables are partitioned across the compute nodes using a distribution key—usually a hash of one or more columns. Each row is assigned to exactly one node based on the hash value of its key. Large fact tables are distributed because they are too large to replicate, and because queries typically filter or join them on the distribution key, allowing local execution . If the orders table is distributed by customer_id, then a query joining orders to a customers table that is also distributed by customer_id can execute entirely locally on each node: the matching rows are already co-located.
Replicated tables are copied in their entirety to every compute node. Dimension and lookup tables are typically replicated because they are small enough that the storage cost is acceptable, and because replication eliminates the need for a shuffle when joining a distributed fact table to a dimension . A countries table with 200 rows is a candidate for replication; a transactions table with billions of rows is not.
The performance consequence is direct. A join between two distributed tables on the distribution key is a local join: each node joins its own rows without network traffic. A join between a distributed table and a replicated table is also local: each node has the full replicated table and its own partition of the distributed table. A join between two distributed tables on a non-distribution key requires a shuffle: both tables must be repartitioned by the join key across the network before the join can proceed . The shuffle is the expensive operation, and the distribution key exists to avoid it for the most important queries.
c. Query Execution and Data Movement
A query submitted to an MPP database is transformed into a distributed execution plan. The coordinator parses the SQL, validates it against the catalog, and generates a plan tree. The optimizer chooses join orders, join algorithms (broadcast or shuffle), and aggregation strategies. The plan is then split into fragments, each assigned to a compute node.
The execution of a simple aggregation illustrates the pattern. A GROUP BY query is processed in two phases. Each compute node performs a local pre-aggregation on its partition of the data, reducing the number of rows that must be sent to the coordinator. The coordinator (or a designated reducer node) then performs a global aggregation on the partial results from each node . This two-phase pattern is fundamental: it minimizes network traffic by aggregating before moving data.
For joins that require a shuffle, the data movement is the dominant cost. The coordinator generates a redistribution plan that specifies which rows move from which node to which node. A hash-based shuffle applies a hash function to the join key and routes each row to the node responsible for that hash bucket. An optimized redistribution matrix can reduce the volume of data moved by permuting the target nodes based on the distribution of values . The goal is always the same: co-locate matching rows on the same node so the join can execute locally.
Data skew complicates this. If the distribution key has a few values that appear in millions of rows—a single customer with 40% of the transactions, for example—the node that owns that key value becomes a hotspot while other nodes idle. MPP systems detect skew and handle it through partial redistribution: the skewed values are broadcast to all nodes while the non-skewed values are shuffled normally, so that no single node bears the entire burden of the popular key .
Complete Example Session
-- ============================================
-- PART 1: SHARED-NOTHING TOPOLOGY
-- ============================================
-- Control Node: receives queries, optimizes, coordinates
-- Compute Nodes: hold data partitions, execute fragments
-- Interconnect: the only path between nodes
-- ============================================
-- PART 2: DISTRIBUTED TABLE (FACT)
-- ============================================
-- Rows are hashed across compute nodes by customer_id
-- Each node owns a subset of customers' orders
CREATE TABLE orders (
order_id BIGINT,
customer_id BIGINT,
order_date DATE,
amount DECIMAL(12,2)
) DISTRIBUTE BY HASH(customer_id);
-- ============================================
-- PART 3: REPLICATED TABLE (DIMENSION)
-- ============================================
-- Full copy of the small table on every compute node
CREATE TABLE customers (
customer_id BIGINT,
name STRING,
region STRING
) REPLICATE;
-- ============================================
-- PART 4: LOCAL JOIN — NO SHUFFLE
-- ============================================
-- Both tables are distributed by customer_id
-- Each node joins its local orders with its local customers
SELECT c.region, SUM(o.amount)
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
GROUP BY c.region;
-- ============================================
-- PART 5: SHUFFLE JOIN — DISTRIBUTION MISMATCH
-- ============================================
-- orders is distributed by customer_id
-- products is distributed by product_id
-- Joining on product_id requires repartitioning orders
SELECT p.category, SUM(o.amount)
FROM orders o
JOIN products p ON o.product_id = p.product_id
GROUP BY p.category;
-- The optimizer will shuffle orders by product_id
-- before the join, or broadcast products if it is small
-- ============================================
-- PART 6: TWO-PHASE AGGREGATION
-- ============================================
-- Phase 1: each node computes a partial SUM for its rows
-- Phase 2: coordinator sums the partial results
SELECT region, SUM(amount) FROM orders GROUP BY region;
-- The plan: local partial aggregate → shuffle by region
-- → global aggregate on the coordinator
-- ============================================
-- PART 7: BROADCAST JOIN DECISION
-- ============================================
-- If products is small, the optimizer broadcasts it
-- to every node instead of shuffling orders
-- Each node joins its local orders to the full products table
-- No rows of orders move across the network
-- ============================================
-- PART 8: SKEW DETECTION
-- ============================================
-- If customer_id = 42 has 40% of all orders,
-- the node owning hash(42) becomes a hotspot
-- The optimizer may broadcast the skewed rows
-- and shuffle only the remaining rows
-- ============================================
-- PART 9: REDISTRIBUTION COST
-- ============================================
-- The cost of a shuffle is proportional to
-- the number of rows that must move and the
-- bandwidth of the interconnect
-- A distributed plan that avoids a shuffle
-- is almost always faster than one that does not
-- ============================================
-- PART 10: DISTRIBUTION KEY SELECTION
-- ============================================
-- Choose the key used in the most frequent joins
-- Choose a key with high cardinality (many distinct values)
-- Choose a key with uniform distribution (no hotspots)
-- The wrong key converts local joins into shufflesThe ten parts trace the architecture from topology through table design to query execution. Parts 2 and 3 show the two distribution strategies. Parts 4 and 5 contrast a local join with a shuffle join. Part 6 shows the two-phase aggregation pattern. The remaining parts address the optimizer’s decisions and the cost of getting the distribution key wrong.
Quick Reference
MPP vs. SMP
| Property | SMP | MPP |
|---|---|---|
| Architecture | Single server, shared resources | Multiple servers, shared-nothing |
| Scaling | Vertical (bigger server) | Horizontal (more nodes) |
| Resource contention | All queries compete | Isolated per node |
| Ceiling | Physical motherboard limits | Interconnect bandwidth |
| Typical use | Transactional workloads | Analytical workloads |
Table Distribution Strategies
| Strategy | Storage | Best For | Join Behavior |
|---|---|---|---|
| Distributed (hash) | One copy, partitioned by key | Large fact tables | Local if joined on key; shuffle otherwise |
| Replicated | Full copy on every node | Small dimension tables | Always local |
Execution Patterns
| Pattern | Mechanism | Cost |
|---|---|---|
| Local join | Matching rows co-located | No network |
| Broadcast join | Small table replicated | Low (small table) |
| Shuffle join | Both tables repartitioned | High (network) |
| Two-phase aggregate | Local partial, global final | Moderate (partial results) |
Best Practices
✅ Do This:
-- Distribute fact tables by the key used in the most common joins
DISTRIBUTE BY HASH(customer_id) -- ✅-- Replicate small dimension tables to eliminate shuffle
REPLICATE -- ✅-- Choose distribution keys with high cardinality and uniform spread
-- (customer_id, not status) -- ✅-- Filter on the distribution key to enable partition pruning
WHERE customer_id = 42 -- ✅❌ Don’t Do This:
-- Distribute a fact table by a low-cardinality column
DISTRIBUTE BY HASH(status) -- ❌ (skew)-- Replicate a large table
REPLICATE on a billion-row table -- ❌ (storage explosion)-- Join two distributed tables on a non-distribution key
JOIN products ON orders.product_id = products.product_id -- ❌ (shuffle)-- Assume every query benefits from MPP
-- Small tables and simple queries do not -- ❌Common Pitfalls
| Pitfall | Why It Happens | Fix |
|---|---|---|
| Query is slower than on SMP | Distribution key mismatch forces shuffle | Redistribute or replicate the smaller table |
| One node does all the work | Data skew in distribution key | Choose a higher-cardinality key; detect skew |
| Storage costs exceed expectations | Large tables replicated | Distribute large tables; replicate only small |
| Joins always shuffle | Fact tables distributed by different keys | Co-locate or replicate one side |
| Coordinator becomes bottleneck | Too many small queries | Batch or route small queries elsewhere |
Real-World Examples
1. Distribute Fact Table by Customer
CREATE TABLE orders (...) DISTRIBUTE BY HASH(customer_id);2. Replicate Dimension Table
CREATE TABLE customers (...) REPLICATE;3. Local Join on Distribution Key
SELECT * FROM orders o JOIN customers c ON o.customer_id = c.customer_id;4. Two-Phase Aggregation
SELECT region, SUM(amount) FROM orders GROUP BY region;5. Broadcast Join for Small Table
-- Optimizer broadcasts products if it is small
SELECT * FROM orders o JOIN products p ON o.product_id = p.product_id;6. Partition Pruning
SELECT * FROM orders WHERE customer_id = 42;7. Skew Detection
-- Check distribution: SELECT customer_id, COUNT(*) FROM orders GROUP BY customer_id ORDER BY 2 DESC;8. Control Node Query
-- EXPLAIN shows the distributed plan with shuffle operations
EXPLAIN SELECT region, SUM(amount) FROM orders GROUP BY region;9. Co-Located Join
-- Both tables distributed by customer_id
-- No shuffle required10. Redistribute Existing Table
ALTER TABLE orders DISTRIBUTE BY HASH(order_date);Visual
Shared-Nothing Topology
┌──────────────────────────────────────────────────────────────┐
│ CONTROL NODE │
│ │
│ ┌─────────────────────────────────────────┐ │
│ │ Parse SQL → Optimize → Distribute plan │ │
│ │ Hold metadata and statistics │ │
│ └──────────────────┬──────────────────────┘ │
│ │ │
│ ┌───────────┼───────────┐ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ COMPUTE │ │ COMPUTE │ │ COMPUTE │ │
│ │ NODE 1 │ │ NODE 2 │ │ NODE 3 │ │
│ │ │ │ │ │ │ │
│ │ CPU │ │ CPU │ │ CPU │ │
│ │ Memory │ │ Memory │ │ Memory │ │
│ │ Disk │ │ Disk │ │ Disk │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │
│ No shared memory. No shared disk. │
│ Nodes communicate only through the interconnect. │
│ │
└──────────────────────────────────────────────────────────────┘Distributed vs. Replicated Tables
┌──────────────────────────────────────────────────────────────┐
│ DISTRIBUTED TABLE (orders) REPLICATED TABLE (customers) │
│ │
│ Node 1: rows with hash 0-N Node 1: FULL table │
│ Node 2: rows with hash N+1-M Node 2: FULL table │
│ Node 3: rows with hash M+1-K Node 3: FULL table │
│ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Partition of │ │ Complete copy │ │
│ │ the fact table │ │ of dimension │ │
│ └─────────────────┘ └─────────────────┘ │
│ │
│ Large, grows unbounded Small, stable │
│ Hash-distributed by key Copied to every node │
│ Joins on key are local Joins are always local │
│ │
└──────────────────────────────────────────────────────────────┘Local Join vs. Shuffle Join
┌──────────────────────────────────────────────────────────────┐
│ LOCAL JOIN (both by customer_id) │
│ │
│ Node 1: orders(cust=1) JOIN customers(cust=1) │
│ Node 2: orders(cust=2) JOIN customers(cust=2) │
│ Node 3: orders(cust=3) JOIN customers(cust=3) │
│ │
│ No rows cross the network. │
│ │
├──────────────────────────────────────────────────────────────┤
│ SHUFFLE JOIN (orders by customer, products by product) │
│ │
│ Node 1: orders rows ──shuffle──► products rows │
│ Node 2: orders rows ──shuffle──► products rows │
│ Node 3: orders rows ──shuffle──► products rows │
│ │
│ Both tables are repartitioned by product_id. │
│ The network transfer is the dominant cost. │
│ │
└──────────────────────────────────────────────────────────────┘Two-Phase Aggregation
┌──────────────────────────────────────────────────────────────┐
│ QUERY: SELECT region, SUM(amount) FROM orders GROUP BY region│
│ │
│ ┌─────────────────┐ │
│ │ Node 1: │ Partial: East=100, West=200 │
│ │ scan local rows │ │
│ └────────┬────────┘ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Node 2: │ Partial: East=50, West=75 │
│ │ scan local rows │ │
│ └────────┬────────┘ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Node 3: │ Partial: East=25, West=150 │
│ │ scan local rows │ │
│ └────────┬────────┘ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ Coordinator: │ Global: East=175, West=425 │
│ │ merge partials │ │
│ └─────────────────┘ │
│ │
│ Local aggregation reduces the rows that cross the network. │
│ │
└──────────────────────────────────────────────────────────────┘Summary
| Item | Value |
|---|---|
| Architecture | Shared-nothing; each node has private CPU, memory, disk |
| Control node | Parses, optimizes, coordinates query execution |
| Compute nodes | Hold data partitions, execute fragments |
| Distributed table | Hash-partitioned by a distribution key |
| Replicated table | Full copy on every compute node |
| Local join | Co-located rows; no network transfer |
| Shuffle join | Repartitioning across network; expensive |
| Two-phase aggregation | Local partial, then global final |
| Data skew | Uneven distribution key; hotspot on one node |
| Cost driver | Network shuffle, not CPU |
Key takeaways:
- MPP is shared-nothing by definition. Each node has its own resources, and communication happens only through the interconnect. This isolation is what allows linear scaling, but it makes data distribution the central design decision .
- The control node coordinates; the compute nodes execute. The coordinator holds metadata, optimizes queries, and dispatches fragments. Compute nodes hold data partitions and return partial results. Neither role can be collapsed into the other without losing the parallelism that defines MPP .
- Distribution key determines query performance. A join on the distribution key is local and fast. A join on any other key requires a shuffle, which moves data across the network and dominates execution time. The key should be chosen for the most important queries, not for convenience .
- Replicated tables eliminate shuffles for dimension joins. Copying a small table to every node is cheaper than shuffling a large fact table. The threshold between “small enough to replicate” and “too large to replicate” is a storage and network trade-off, not a fixed rule.
- Two-phase aggregation reduces network traffic. Local pre-aggregation on each node shrinks the result set before it crosses the network. The coordinator performs the final aggregation on the reduced data .
- Data skew turns parallelism into a bottleneck. A distribution key with a few very common values sends a disproportionate share of rows to one node. Skew detection and partial redistribution are the mechanisms MPP systems use to mitigate this .
Remember: MPP architecture is not simply “more machines doing the same thing.” It is a different execution model in which the location of data determines the cost of every operation. The coordinator’s job is to minimize data movement, and the distribution key is the primary lever for doing so. A query that stays local on every node runs in parallel across the cluster. A query that requires a shuffle runs at the speed of the network. The architecture rewards deliberate table design and penalizes assumptions carried over from single-server databases. Understanding where the data lives—and why—is the foundation for understanding everything else about MPP performance.
Stop using slow, ad-bloated tool sites! 🤮
🔎 Search “KandZ Tools” on Google to use many professional utilities for free.
KandZ.me is the ultimate minimalist hub for:
✅ Finance (Mortgage, Interest, Inflation)
✅ Tech (Base64, JSON, Dev Suite, IP)
✅ Health (BMI, BMR, TDEE)
✅ Productivity (Timer, Workspace, QR)
⚡️ Fast & Private
🔒 No data leaves your device
💎 100% Free
🔗 Use it now: https://tools.kandz.me
🔖 Bookmark it—you’ll need it later!