How Distributed SQL Execution Engines Really Work

Pillar Post: Query Shape & Execution Reality. This is a deep dive into how data moves across a cluster to satisfy a query. For how these systems fail, see Why HTAP Systems Fail Quietly.

Most engineers treat distributed SQL databases as "PostgreSQL, but faster and spread across many nodes." This mental model fails the moment you hit production and discover that a simple JOIN that ran in 100ms on your laptop now takes 30 seconds at scale.

The execution engine is why.

TL;DR

• Distributed execution has coordinators and workers—aggregators plan and merge, leaves execute
• Query planning is cheap; data movement is expensive—the network is your bottleneck
• Joins explode when data isn't co-located—mismatched shard keys = reshuffling gigabytes
• Aggregations fan-in—partial results on leaves, final merge on aggregator
• "It worked in staging" means nothing—execution behavior changes with data size and distribution
• Most "slow database" complaints are actually data movement problems

Understanding execution gives you predictive power over query performance.

---

The Mental Model: Coordinators vs Workers

In PostgreSQL, there's one process executing your query.

In a distributed system like SingleStore, your query runs across two layers:

Aggregators (Coordinators)

Role: Planning and merging

What they do:

• Parse SQL
• Build execution plan
• Dispatch subqueries to leaves
• Merge partial results
• Return final result to client

What they DON'T do:

• Store data (usually)
• Execute full table scans
• Do heavy computation (unless necessary)

Leaves (Workers)

Role: Data storage and local execution

What they do:

• Store data shards
• Execute local scans
• Perform local joins (if data is co-located)
• Compute partial aggregations
• Send results to aggregator

Key insight: Each leaf only sees its own shard. It has no idea what data is on other leaves.

---

Query Planning vs Query Execution

This is where most engineers get confused.

Planning (Cheap)

The aggregator:

1. Parses your SQL
2. Figures out which tables are involved
3. Determines which shards have the data
4. Decides how to split the work
5. Generates a plan

Cost: Milliseconds
Network traffic: Zero

Execution (Expensive)

The real work:

1. Aggregator sends plan to leaves
2. Leaves scan their local data
3. Leaves execute local operations (filters, joins, aggregations)
4. Leaves send results back to aggregator
5. Aggregator merges results

Cost: Seconds to minutes
Network traffic: Gigabytes (potentially)

Total time: ~60ms
Planning: 3ms
Execution: 57ms

The plan is not the problem. The data movement is.

---

Where Data Actually Moves (And Why That's Expensive)

In a single-node database, data moves from disk to CPU to RAM.

In a distributed database, data also moves across the network.

Three Types of Data Movement

1. Local scan (no network)

sqlCopy
SELECT * FROM orders WHERE order_date > '2026-01-01';

Each leaf scans its own shard. Results go to aggregator.

Network cost: Only the results (could be small)

2. Broadcast (aggregator → all leaves)

sqlCopy
SELECT o.*, u.name 
FROM orders o 
JOIN users u ON o.user_id = u.user_id;

If users is small and not sharded, the aggregator broadcasts the entire table to all leaves.

Network cost: size(users) × number_of_leaves

3. Reshuffle (leaf → leaf)

sqlCopy
SELECT o.*, p.name
FROM orders o
JOIN products p ON o.product_id = p.product_id;

If orders is sharded by order_id and products is sharded by product_id, data must be reshuffled.

Network cost: size(orders) + size(products) (potentially)

The Hidden Cost

Moving 1GB across a network:

• 10 Gbps network: ~1 second
• 1 Gbps network: ~10 seconds

But you're not just moving data once. You're:

• Serializing it
• Compressing it
• Sending it
• Decompressing it
• Deserializing it

Real time: 2-5x the naive calculation.

This is why queries that work fine in staging (small data, single node) explode in production.

---

Joins in a Distributed System

Co-Located Join (Fast)

Both tables are sharded on the join key:

sqlCopy
-- orders sharded by user_id
-- user_activity sharded by user_id
SELECT o.order_id, a.activity
FROM orders o
JOIN user_activity a ON o.user_id = a.user_id;

Network cost: Minimal (just results)
Execution: Parallel on each leaf

This is the ideal case.

Reshuffle Join (Slow)

Tables are sharded on different keys:

sqlCopy
-- orders sharded by order_id
-- products sharded by product_id
SELECT o.order_id, p.name
FROM orders o
JOIN products p ON o.product_id = p.product_id;

What happens:

1. Each leaf scans its local orders
2. Data is reshuffled across the network by product_id
3. Each leaf now has orders + products for the same product IDs
4. Join happens locally
5. Results go to aggregator

Network cost: Entire dataset moves across the network
Time: 10-100x slower than co-located join

This is the failure mode.

Broadcast Join (Medium)

One table is small:

sqlCopy
-- orders is large, sharded
-- order_status is tiny (10 rows)
SELECT o.*, s.description
FROM orders o
JOIN order_status s ON o.status_id = s.status_id;

What happens:

1. Aggregator broadcasts order_status (tiny) to all leaves
2. Each leaf joins locally

Network cost: size(small_table) × number_of_leaves
Time: Fast (if the small table is actually small)

---

Aggregations and Fan-In

Aggregations in distributed systems use partial → final pattern.

sqlCopy
SELECT user_id, COUNT(*), SUM(amount)
FROM orders
GROUP BY user_id;

Execution:

On each leaf:

sqlCopy
SELECT user_id, COUNT(*) as cnt, SUM(amount) as total
FROM orders
WHERE <local_shard>
GROUP BY user_id;

On aggregator:

sqlCopy
-- Merge partial results
SELECT user_id, SUM(cnt), SUM(total)
FROM <partial_results>
GROUP BY user_id;

Why This Matters

Aggregations that work: SUM, COUNT, AVG, MIN, MAX

Aggregations that don't work well: DISTINCT, MEDIAN, PERCENTILE

Why?

COUNT DISTINCT can't be easily combined from partial results. The aggregator needs to see all unique values.

Example:

sqlCopy
SELECT COUNT(DISTINCT user_id) FROM orders;

What happens:

1. Each leaf sends all unique user_id values to aggregator
2. Aggregator deduplicates

Network cost: Potentially gigabytes (all unique IDs)

---

Why "It Worked in Staging" Means Nothing

Your staging environment:

• Data: 1GB
• Nodes: 1 leaf
• Query: Fast

Production:

• Data: 1TB
• Nodes: 10 leaves
• Query: Timeout

What changed?

1. Data Movement Scales Non-Linearly

Staging:

• No reshuffling (single node)
• No network overhead
• All data in memory

Production:

• Reshuffling 100GB across network
• Network becomes bottleneck
• Data doesn't fit in memory

2. Query Plans Change

The query planner chooses different strategies based on:

• Table sizes
• Data distribution
• Available memory

Staging plan: Broadcast join
Production plan: Reshuffle join (because the "small" table is now 50GB)

3. Sharding Matters

Staging: Single shard, no coordination
Production: 10 shards, coordination overhead, potential skew

---

The Diagram That Makes Everything Click

This is the execution model you need in your head:

Key insight: Steps 3b-3c (reshuffle) are optional—but when they happen, they dominate execution time.

Your goal: Design schemas and queries that avoid reshuffling.

---

Failure Modes Engineers Misattribute to "Slow Databases"

Failure Mode 1: "The Database Is Slow"

Symptom: Query takes 30 seconds

Actual cause: Reshuffle join moving 50GB across network

Fix: Shard tables on the join key

Failure Mode 2: "High CPU on Database"

Symptom: Database CPU at 90%

Actual cause: Aggregator is merging 10M rows from leaves

Fix: Add LIMIT clause, or filter earlier

Failure Mode 3: "Queries Work Locally But Not in Production"

Symptom: Local query: 100ms. Production: timeout.

Actual cause: Query plan changed because table sizes are different

Fix: Test with production-scale data

Failure Mode 4: "Adding Indexes Didn't Help"

Symptom: Added index, query still slow

Actual cause: Query is network-bound, not CPU-bound

Fix: Optimize data movement, not index structure

Failure Mode 5: "RAM Upgrade Didn't Help"

Symptom: Doubled RAM, no improvement

Actual cause: Query is bottlenecked on network, not memory

Fix: Reduce data movement

---

Sideways Failure: One Query, Cluster-Wide Impact

The execution engine has shared resources:

What happens:

1. One expensive query starts reshuffling 100GB
2. Network saturates
3. Other queries queue, waiting for network bandwidth
4. Application sees "slow database"
5. Alert fires
6. Engineer investigates the symptom (slow queries)
7. Root cause is the expensive query that started it all

The pattern: In distributed systems, symptoms appear far from causes.

---

What I'd Do Differently Next Time

1. Test Queries at Production Scale

Don't trust staging. Use production data size, even if it's in a dev environment.

Tooling:

• EXPLAIN on production (read-only)
• Synthetic data generators
• Cloned production database (sanitized)

2. Choose Shard Keys Based on Join Patterns

Shard key is the most important schema decision.

Ask:

• What are the most common joins?
• Can I co-locate data for those joins?
• What's the cardinality of the shard key?

Bad: Shard on order_id when you frequently join with user_id
Good: Shard on user_id if most queries filter or join on users

3. Monitor Data Movement, Not Just CPU

Track:

• Network bytes sent/received per query
• Reshuffle frequency
• Broadcast sizes

Alert when:

• A query moves >10GB
• Network bandwidth exceeds 70%
• Reshuffle operations spike

4. Use EXPLAIN Before Every Deploy

sqlCopy
EXPLAIN FORMAT=JSON 
SELECT o.*, p.name 
FROM orders o 
JOIN products p ON o.product_id = p.product_id;

Look for:

• "exchange" (data movement between nodes)
• "broadcast" (full table sent to all nodes)
• "repartition" (data reshuffled)

If you see any of these, understand the cost.

5. Add Query Limits by Default

sqlCopy
-- Bad: unbounded
SELECT * FROM orders WHERE user_id = 123;

-- Good: explicit limit
SELECT * FROM orders WHERE user_id = 123 LIMIT 1000;

Limits prevent accidental full-table scans.

6. Separate OLTP and OLAP Workloads

Don't mix:

• Transactional queries (point lookups)
• Analytical queries (full scans, aggregations)

Why: They have different execution patterns and resource needs.

Solution: Replicate data to separate clusters.

---

Final Takeaway

Understanding the execution engine gives you predictive power.

You can look at a query and know:

• Will it reshape data?
• How much network traffic will it generate?
• Why it might work locally but fail in production?

The execution engine isn't magic.
It's coordinators dispatching work, workers executing locally, and network moving data.

Optimize for that, and your queries will scale.

---

Further Reading

• Lessons Learned Running SingleStore in Production
• SingleStore vs PostgreSQL: When Distributed SQL Actually Wins
• How SingleStore Handles Real-Time Analytics at Scale

---

Have execution engine questions? Email me or connect on LinkedIn.