Massively Parallel Processing (MPP)

Massively parallel processing (MPP) is an architecture in which a single query is divided into pieces that many nodes execute at the same time, each working on its own portion of the data. A coordinator plans the query and hands out the work; the workers process their share in parallel; partial results are brought together into one answer.

It is the design behind most analytical warehouses and distributed SQL engines, and it is the reason adding nodes can make a large scan finish sooner.

How the work is divided

Amazon Redshift’s documentation describes a representative version. A leader node parses the query, builds the execution plan, compiles code for its steps, and sends that code to the compute nodes, assigning each a portion of the data. Each compute node has its own CPU and memory and is further divided into slices that work in parallel. The compute nodes run their part and send intermediate results back for final aggregation.

A simple aggregation — revenue per category for one month — typically runs in four stages:

1. Parallel scan. Each worker reads its own files or blocks and applies the filter. Nothing needs to be exchanged yet.
2. Local aggregation. Each worker sums revenue per category over the rows it read, shrinking its output to at most one row per category.
3. Redistribution. Partial results are sent over the network so that all partials for one category land on the same worker.
4. Final aggregation. The partials are combined and the result is returned.

The example is illustrative; real plans vary by engine and query. The shape, though, is common: stages that work on local data scale well, and the stage that moves data between nodes does not.

Where MPP slows down

Moving data. Joins and aggregations need matching rows on the same node. Redshift’s documentation describes the optimizer redistributing rows for this, either by sending specific rows to the nodes where they will be joined or by broadcasting an entire table to every node, and warns that redistribution can account for a substantial portion of a query plan’s cost and that its network traffic can slow other work on the system. Much of MPP design is about avoiding that step: filtering early, aggregating locally before exchanging, and placing data in advance so that rows that join together already live together. Redshift exposes the last of these as a table’s distribution style and key.

Uneven division. A stage is finished only when its slowest worker finishes. If one node receives far more data than the others — for example, when a join key has a few values that account for most rows — that node sets the pace for the whole query. The same documentation names this distribution skew and states that it forces some nodes to do more work than others, impairing performance. This is data skew, and adding nodes does not fix it, because the overloaded node’s share does not shrink.

Coordination overhead. Planning, distributing code, and gathering results cost something regardless of data size. For small data that fits comfortably on one machine, this overhead can outweigh the benefit of parallelism.

How it relates to other ideas

• Vectorized execution makes a single worker efficient; MPP divides work among many workers. They are complementary, and a slow query can have either problem. See vectorized execution.
• Storage and compute separation changes where data lives, not how a query is parallelized. Classic MPP systems kept data on the compute nodes’ own disks, so adding compute meant moving data. Many current systems keep data in shared storage and let the number of workers change independently — Redshift, for instance, describes a managed storage tier that lets compute and storage scale separately.
• General distributed processing frameworks such as Spark follow the same scan, exchange, and combine pattern; the shuffle there is the redistribution stage here, and a broadcast join is the same trade of copying a small table to avoid moving a large one.

When an MPP query is slow, the useful first question is which stage holds it up: the scan, the exchange, or one overloaded worker. Each has a different remedy. How parallelism fits with column layout, data skipping, and the execution model is worked through in Analytical Engines: Where the Speed Actually Comes From.

References: Amazon Redshift Documentation, Data warehouse system architecture; Amazon Redshift Documentation, Data distribution for query optimization.