Massively Parallel Processing (MPP)

What is Massively Parallel Processing (MPP)

If you’ve ever wondered why some engines stay snappy under mixed, join-heavy workloads while others slow to a crawl, the difference often comes down to how they move and combine intermediate results. This is where MPP (massively parallel processing) earns its keep.

What MPP actually does:

1. Fragment the plan
   The optimizer breaks a query into fragments: scan, filter/project, join, aggregate, exchange. Think of each fragment as a small assembly line stage.
2. Fan out fragment instances
   Every fragment is instantiated many times and scheduled across the cluster. These instances are the smallest units of parallel work and let the engine match compute to available cores.
3. Pipeline and vectorize
   Inside each instance, operators run in a pipeline (Scan → Filter/Project → Join/Aggregate → Exchange). Vectorization keeps CPU hot by processing data in columnar batches instead of row-by-row function calls. Pipelines also provide backpressure so fast stages don’t overrun slow ones.
4. Exchange data in memory
   When two big sides must meet (distributed JOIN, global GROUP BY), rows are hashed on the key and routed directly over the network to peers. Well-behaved runs never touch disk for these exchanges; spilling is a safety valve, not the plan.
5. Distribute finalization
   Partials are merged in a tree across multiple nodes—partial → merge → final—so “the last step” isn’t a single machine. The planner/coordinator orchestrates but doesn’t become the bottleneck.

MPP vs Other Compute Architectures

Modern distributed query engines typically adopt one of three architectural paradigms to execute analytical workloads: scatter-gather, stage-based (MapReduce-style), and in-memory MPP. Each has distinct trade-offs in terms of scalability, latency, and workload suitability.

1. Scatter-Gather Architecture

How it works
A coordinator breaks a query into sub-tasks and dispatches them to worker shards. Each shard scans and filters its local data and often performs partial aggregations. The coordinator merges those partials, applies any remaining GROUP BY/ORDER BY/LIMIT work, and returns the final result. Most of the heavy lifting happens close to the data; the network carries compact intermediate results instead of raw rows.

Used by
ClickHouse and similar coordinator-merge systems.

Strengths
• Simple mental model—easy to reason about where work runs and to debug slow queries.
• Very fast for single-table scans with filters, projections, and rollups that push down to shards.
• Network-efficient—shards return partial aggregates or compressed blocks, not full row sets.
• Scales reads predictably by adding shards for more parallel scan bandwidth.

Limitations
• Coordinator hot spot—final merges and aggregations centralize some work; at high concurrency this can bottleneck.
• Joins are constrained—common patterns are broadcasting a small table to shards or co-locating large tables on the same shard keys; big-table-to-big-table repartitioned joins are not the default.
• Skew sensitivity—uneven key distribution can create stragglers and long tails.
• Write-time layout matters—performance depends on good partitioning, sorting, and projections.

Best for
Operational analytics on wide, flat fact tables; single-table access with selective filters and simple to moderate aggregations; high-throughput dashboards where joins are limited or dimensions are small enough to broadcast.

2. Stage-Based Execution (MapReduce, DAG)

How it works
Queries are decomposed into multiple stages separated by shuffles. A shuffle redistributes data across executors and creates a stage boundary; upstream tasks must finish before downstream tasks begin. Intermediate data typically lands on local disk between stages. Optimizers can combine, reorder, or adapt plans (e.g., dynamic partition pruning, adaptive query execution), but the fundamental rhythm is “execute → materialize → execute.”

Used by: Apache Spark, Databricks Photon, Flink (in batch mode), and similar DAG-based systems.

Strengths
• Very flexible—handles SQL, DataFrames, ETL, ML, and batch streaming patterns.
• High fault tolerance—failed tasks are retried from materialized boundaries.
• Elastic scaling—great for cloud autoscaling and spot/ephemeral compute pools.
• Rich ecosystem—connectors, libraries, and tooling are mature.

Limitations
• Higher latency—disk I/O and serialized stage transitions add seconds to minutes.
• Fragmented pipelines—more moving parts (shuffle, spill, serialization) to tune.
• Operational complexity—resource sizing, partitioning, and shuffle tuning matter a lot for large joins/aggregations.

Best for
Large-scale ETL, periodic backfills, feature engineering, and offline transformations of structured or semi-structured data where minute-level latency is acceptable.

3. In-Memory MPP Execution

How it works
Nodes execute fragments of a physical plan concurrently and exchange intermediate results directly over the network using in-memory data streams. Operators are pipelined and vectorized for high CPU efficiency. Distributed hash joins and aggregations run across nodes; engines spill to disk only when needed. Techniques like runtime filters, dictionary encoding, and code generation keep hot paths tight.

Used by
StarRocks, Apache Impala, Greenplum, and other modern MPP databases.

Strengths
• Low-latency execution for complex, multi-table queries—JOINs, GROUP BYs, windows—at interactive speeds.
• Fully parallel joins—supports repartition (hash) joins, broadcast joins, and co-located joins.
• Real-time friendly—ingests continuously and makes fresh data queryable quickly for user-facing analytics.
• Concurrency headroom—built for many short queries from dashboards and applications.

Limitations
• Needs solid infra—sufficient memory and high-throughput networking are important.
• Requires discipline—statistics, partitioning/bucketing, and materialized views should be managed to sustain performance as data and concurrency grow.

Best for
Interactive dashboards, customer-facing analytics, anomaly/fraud detection loops, and high-cardinality aggregations with JOIN-heavy query patterns.

Summary Comparison Table

Core Features of an MPP Database

To understand why MPP engines deliver high performance and scale for interactive analytics, it helps to unpack the design patterns they use:

1. Independent node design
   Each node runs its own compute with local memory and manages its slice of the data. This minimizes contention and lets the system scale out by adding nodes, each contributing parallel scan, join, and aggregate capacity.
2. Data partitioning
   Tables are horizontally partitioned (hash, range, or buckets) so each node processes just its portion. Good partition keys align with common filters and joins to reduce cross-node traffic and hotspots.
3. Query fragmentation and scheduling
   The optimizer breaks a plan into fragments—scans, joins, aggregations, exchanges—which are scheduled across nodes. Multiple instances of the same fragment run in parallel to saturate CPU and I/O. Pipelines keep operators busy without unnecessary materialization.
4. Shuffles and join strategies
   When a true distributed join or GROUP BY is required, the engine repartitions data (hash shuffle) so matching keys meet on the same node. Engines also support broadcast joins (ship the small side to all nodes) and co-located joins (plan tables on compatible partition keys to avoid shuffles). Modern systems prefer in-memory exchanges and only spill when necessary.
5. Coordinator role
   A lightweight coordinator parses SQL, builds the physical plan, and orchestrates fragment instances. Unlike coordinator-merge systems, the “last mile” aggregation is typically distributed (often a tree/partial-final pattern), so the coordinator doesn’t become a bottleneck as concurrency grows.

A Closer Look: MPP in Practice with StarRocks

StarRocks is an MPP engine by design. Its execution core leans into MPP’s strengths while sidestepping common bottlenecks in other architectures.

Fragmentation and Parallel Execution

When a query is submitted, StarRocks decomposes it into logical fragments. Each fragment covers one part of the plan—scan, project/filter, join, aggregate, exchange—and is instantiated into multiple fragment instances, the smallest schedulable units.

Instances run in parallel on backend (BE) nodes. Operators are pipelined (for example, Scan → Project/Filter → Join/Aggregate → Exchange) and vectorized so CPUs process columnar batches efficiently instead of row by row. The engine chooses the degree of parallelism per fragment based on available cores, table partitioning/bucketing, and runtime limits, so the cluster’s compute is saturated without introducing artificial stage barriers.

.

Figure: SELECT COUNT(*) FROM table GROUP BY id is split into three fragments. Fragment 2 scans, projects, and performs a partial aggregate. The results are hash-repartitioned by id and sent to Fragment 1, which merges the partials. Fragment 0 performs the distributed final aggregation. Each fragment is instantiated many times and scheduled across BE nodes.