How Distributed Query Execution Works in Cloudberry

Slices and gangs are the two words you need before anything else in distributed execution makes sense.

Every plan in the last six chapters had one of these in it — Gather Motion 3:1, Redistribute Motion 3:3, Broadcast Motion 1:3, Explicit Redistribute Motion — and every time we read it as an annotation: a note in the margin saying and then the data moves. It is not an annotation. A Motion is a plan node. It has an ExecInitMotion, an ExecMotion and an ExecEndMotion, it sits in the tree exactly where a Sort or a Hash Join would sit, and it is the only reason any of the rest of this works.

That design choice is the whole architecture. §18.1 described the executor as a demand-pull volcano: a parent calls ExecProcNode on its child and gets one tuple, or NULL for end-of-stream. Cloudberry does not extend that protocol with a send this somewhere else primitive. It inserts a node whose child happens to live in a different process. A Motion’s parent calls ExecProcNode and gets a tuple, and cannot tell that the tuple arrived over a socket. A Motion’s child is called by a parent that will ship the result sideways instead of returning it upward, and cannot tell either. Every other node in the tree stays completely oblivious, which is why Chapters 13–18 could be written at all.

Three consequences follow immediately, and they are the vocabulary for the rest of the chapter. First, the plan is cut at Motion boundaries into slices — §16.3 built the cut at planning time, §19.3 has the data structures. Second, each slice is run by a gang: one process per segment, all of them running the same slice of the same plan on different data (§19.2). Third — and this is the single most important idea here — once the QD has dispatched, nothing schedules anything. Each process runs its own volcano loop, blocking when its input is empty and blocking when its output buffer is full. There is no global coordinator during execution.

The chapter’s running example: one aggregate, two motions, three slices

CREATE TABLE m191_sales (id int, region int, amt numeric) DISTRIBUTED BY (id);
INSERT INTO m191_sales SELECT g, g % 97, (g % 1000)::numeric FROM generate_series(1,4000000) g;
ANALYZE m191_sales;

EXPLAIN (costs off) SELECT region, count(*) FROM m191_sales GROUP BY region;

CREATE TABLE
INSERT 0 4000000
ANALYZE
                         QUERY PLAN                         
------------------------------------------------------------
 Gather Motion 3:1  (slice1; segments: 3)
   ->  Finalize HashAggregate
         Group Key: region
         ->  Redistribute Motion 3:3  (slice2; segments: 3)
               Hash Key: region
               ->  Streaming Partial HashAggregate
                     Group Key: region
                     ->  Seq Scan on m191_sales
 Optimizer: GPORCA
(9 rows)

region is not the distribution key, so the groups are scattered across all three segments and a two-stage aggregate is the only way to finish them (§16.4 planned this shape, §18.3 executed it). Two motions cut the plan into three slices: the scan plus the partial aggregate; the finalize aggregate; and the root slice that the QD runs. The slice number printed on a Motion node is always the number of the sender slice below it, so the root is the unnumbered slice0 — §19.3 owns that numbering. Here is the same plan drawn against the processes that actually run it.

Nine arrows between slice2 and slice1 is not artistic licence: a Redistribute Motion 3:3 really is all-to-all, every sender potentially talking to every receiver. That is the fact §19.5 has to build a transport for. To prove the process picture, run the same query long enough to photograph it — we inflate each row 120× with a generate_series join so the statement lives for about forty seconds, which changes nothing about the slice structure — and take a census from a second session.

Seven processes, one distributed statement. The QD row comes from the coordinator’s own pg_stat_activity, the six segment rows from gp_dist_random.

-- session A, held open ~40 s (same three slices, 120x more input rows):
--   SELECT region, count(*) FROM m191_sales, generate_series(1,120) g GROUP BY region;

-- session B, while A runs:
SELECT 'seg'||gp_execution_segment() AS host, sess_id, pid, state, left(query,26) AS query
  FROM gp_dist_random('pg_stat_activity') WHERE sess_id = 188
UNION ALL
SELECT 'QD  ', sess_id, pid, state, left(query,26) FROM pg_stat_activity WHERE sess_id = 188
ORDER BY 1, 3;

 host | sess_id | pid  | state  |           query            
------+---------+------+--------+----------------------------
 QD   |     188 | 9968 | active | select region, count(*) fr
 seg0 |     188 | 9971 | active | select region, count(*) fr
 seg0 |     188 | 9977 | active | select region, count(*) fr
 seg1 |     188 | 9972 | active | select region, count(*) fr
 seg1 |     188 | 9978 | active | select region, count(*) fr
 seg2 |     188 | 9973 | active | select region, count(*) fr
 seg2 |     188 | 9979 | active | select region, count(*) fr
(7 rows)

Seven processes, one sess_id, one statement: the QD plus one gang of three per non-root slice. The operating system agrees, and it labels them for us — a Cloudberry QE writes its slice number into its process title, so ps reads out the slice tree directly. Watch the C column (the kernel’s CPU-usage estimate, fourth field of ps -ef).

ps sees the slice tree. The three slice2 processes are pegged; the three slice1 processes are burning nothing at all.

$ ps -ef | grep -E "con188 (seg|cmd)" | grep -v grep | sed 's/  */ /g' | sed 's/^gpadmin //'

9968 19803 1 01:54 ? 00:00:00 postgres: 7100, gpadmin postgres [local] con188 cmd5 SELECT
9971 19768 0 01:54 ? 00:00:00 postgres: 7102, gpadmin postgres 10.100.0.3(57978) con188 seg0 cmd6 slice1 MPPEXEC SELECT
9972 19772 0 01:54 ? 00:00:00 postgres: 7103, gpadmin postgres 10.100.0.3(41800) con188 seg1 cmd6 slice1 MPPEXEC SELECT
9973 19773 0 01:54 ? 00:00:00 postgres: 7104, gpadmin postgres 10.100.0.3(38304) con188 seg2 cmd6 slice1 MPPEXEC SELECT
9977 19768 93 01:54 ? 00:00:07 postgres: 7102, gpadmin postgres 10.100.0.3(57984) con188 seg0 cmd6 slice2 MPPEXEC SELECT
9978 19772 92 01:54 ? 00:00:07 postgres: 7103, gpadmin postgres 10.100.0.3(41806) con188 seg1 cmd6 slice2 MPPEXEC SELECT
9979 19773 91 01:54 ? 00:00:07 postgres: 7104, gpadmin postgres 10.100.0.3(38310) con188 seg2 cmd6 slice2 MPPEXEC SELECT

This capture is the no-global-coordinator claim. Nobody put the three slice1 processes to sleep and nobody will wake them: each one is sitting inside its Redistribute Motion receiver, blocked because its input is empty, and it will resume the instant a tuple lands. Nobody is throttling the three slice2 processes either; they scan until an output buffer fills. The QD is not a scheduler in this picture — it is simply the process that happens to run the root slice, plus the dispatcher that got everyone started (§13.3, §19.2). Treat the numbers as one snapshot on one machine, not a measurement: this build is —enable-cassert.

19.1.1 One plan node, two roles

A distributed plan is serialized once on the QD and shipped verbatim to every QE (§13.3 for the dispatch, §19.2 for what physically goes on the wire). So every process holds the same tree, including the Motion nodes belonging to slices it will never touch. What differs is where each process enters the tree. A QE looks up the Motion node that is the top of its slice and makes that node its plan root.

src/backend/executor/execMain.c:1934

	if (estate->eliminateAliens)
	{
		Motion *m = findSenderMotion(plannedstmt, LocallyExecutingSliceIndex(estate));

		if (NULL != m)
		{
			start_plan_node = (Plan *) m;
			ExecSlice *sendSlice = &estate->es_sliceTable->slices[m->motionID];
			estate->currentSliceId = sendSlice->parentIndex;

findSenderMotion searches the whole plan for the Motion whose motionID equals this process’s slice index — Motion.motionID is the sender slice’s index (nodeMotion.c:710) — and everything outside that subtree is an alien, never initialized (the execute_pruned_plan developer GUC, on by default here, gates this; execMain.c:564). A receiver Motion does not even initialize its child: nodeMotion.c:796 skips the recursive ExecInitNode when the node is not a sender. Then, inside the one Motion node they share, sender and receiver are told apart by a single comparison.

src/backend/executor/nodeMotion.c:751

		if (LocallyExecutingSliceIndex(estate) == recvSlice->sliceIndex)
		{
			/* this is recv */
			motionstate->mstype = MOTIONSTATE_RECV;
		}
		else if (LocallyExecutingSliceIndex(estate) == sendSlice->sliceIndex)
		{
			/* this is send */
			motionstate->mstype = MOTIONSTATE_SEND;
		}

From that one field everything else follows. ExecMotion branches on it before it does anything at all (nodeMotion.c:125); the hash-key expressions and the CdbHash object are built only if mstype == MOTIONSTATE_SEND (nodeMotion.c:816), so a receiver never even owns a hash function; the merge comparator for a sorted gather is built only for MOTIONSTATE_RECV (nodeMotion.c:858). The complementary decision on the QD side is §18.1’s getGpExecIdentity (execUtils.c:2064), which returns GP_ROOT_SLICE when this process runs the top of the tree, GP_NON_ROOT_ON_QE when it runs a slice capped by a sender Motion — the call returns no tuples at all, they all leave over the interconnect — and GP_IGNORE when the process has nothing to run locally.

slice — A maximal piece of the plan tree containing no Motion boundary. The unit of parallelism, and the unit of dispatch. Cut by the planner (§16.3, §17.5), numbered in EXPLAIN as slice0 (root) upward (§19.3).

gang — The set of processes that runs one slice — normally one QE per segment, sometimes exactly one QE anywhere. Allocation, reuse and death are §19.2.

sender / receiver slice — For each Motion: the slice below it sends, the slice containing it receives. Motion.motionID names the sender slice; PlanSlice.parentIndex names the receiver.

route — A sender’s local index for one destination process, 0..N-1, plus the pseudo-route BROADCAST_SEGIDX = -2 meaning everyone (src/include/cdb/tupchunk.h:59). Choosing the route is what distinguishes the motion types.

19.1.2 Three shapes of data

§16.1 gave the planner’s side of this as a locus algebra. On the execution side there are only three physical shapes a rowset can have, and a Motion exists to convert between them.

partitioned — N processes, N disjoint row sets whose union is the answer. A DISTRIBUTED BY or DISTRIBUTED RANDOMLY table, and the output of any Redistribute. Most rowsets in most plans.

replicated — N processes, N identical row sets. A DISTRIBUTED REPLICATED table on disk — and, transiently, whatever a Broadcast Motion has just delivered. Cheap to join against, expensive to produce.

singleton — Exactly one process holds everything: the QD, a one-QE reader gang on a segment, or an entry-db reader on the coordinator. The client only ever sees a singleton — which is why nearly every plan ends in a Gather.

A replicated table needs no motion to join, and needs only one sender to gather

-- m191_dim: 97 rows, DISTRIBUTED BY (region)
CREATE TABLE m191_rep (region int, rname text) DISTRIBUTED REPLICATED;
INSERT INTO m191_rep SELECT * FROM m191_dim;

EXPLAIN (costs off) SELECT s.id, r.rname FROM m191_sales s JOIN m191_rep r
                     ON s.region = r.region WHERE s.id < 5;
EXPLAIN (costs off) SELECT count(*) FROM m191_rep;

CREATE TABLE
INSERT 0 97
                QUERY PLAN                
------------------------------------------
 Gather Motion 3:1  (slice1; segments: 3)
   ->  Hash Join
         Hash Cond: (s.region = r.region)
         ->  Seq Scan on m191_sales s
               Filter: (id < 5)
         ->  Hash
               ->  Seq Scan on m191_rep r
 Optimizer: GPORCA
(8 rows)

                   QUERY PLAN                   
------------------------------------------------
 Finalize Aggregate
   ->  Gather Motion 1:1  (slice1; segments: 1)
         ->  Partial Aggregate
               ->  Seq Scan on m191_rep
 Optimizer: GPORCA
(5 rows)

Both plans are the shape algebra made visible. The join needs no motion whatsoever on the inner side — every segment already has all 97 dimension rows, so the hash table is built locally three times over. The aggregate is the opposite direction: counting a replicated table from three segments would count everything three times, so the planner gives the slice a one-segment gang and gathers 1:1. A replicated rowset converts to a singleton by ignoring two thirds of the cluster.

19.1.3 What each motion type converts

There are seven MotionType values (src/include/nodes/plannodes.h:1878), four of which you meet routinely. What separates them is a single question asked once per tuple in doSendTuple: which route does this row go out on, and who worked that out?

EXPLAIN node	MotionType	input shape → output shape	who computes the destination
Gather Motion N:1	MOTIONTYPE_GATHER	partitioned → singleton	Nobody. There is only one route, so the sender hard-wires targetRoute = 0 (nodeMotion.c:1197).
Explicit Gather Motion N:1	MOTIONTYPE_GATHER_SINGLE	replicated → singleton	Nobody. The subplan runs on N segments but only the planner-chosen one sends.
Redistribute Motion N:M	MOTIONTYPE_HASH	partitioned on key A → partitioned on key B	The sender, per tuple: a cdbhash over the Hash Key expressions (evalHashKey, nodeMotion.c:1272). §19.4 has the hash itself.
Broadcast Motion N:M	MOTIONTYPE_BROADCAST	partitioned or singleton → replicated	Nobody. The sender writes the pseudo-route BROADCAST_SEGIDX (-2) and the motion layer fans the tuple out to every receiver.
Broadcast Workers Motion	MOTIONTYPE_BROADCAST_WORKERS	partitioned → replicated once per segment, into one worker of each	The sender, at random inside each segment’s worker set (nodeMotion.c:1220) — an intra-segment-parallel variant, see §18.2.
Explicit Redistribute Motion N:M	MOTIONTYPE_EXPLICIT	partitioned → partitioned by a segment number carried in the row	Nobody. The destination is already a column of the tuple.
(never executed)	MOTIONTYPE_OUTER_QUERY	placeholder inside a correlated subplan	Resolved into a Gather or a Broadcast before the plan is finished.

A Broadcast, proved: one row per segment goes in, two rows per segment come out

EXPLAIN (analyze, timing off, costs off, summary off)
SELECT count(*) FROM m191_sales s JOIN m191_dim d ON s.region = d.region
 WHERE d.rname IN ('r5','r7');

                                              QUERY PLAN                                               
-------------------------------------------------------------------------------------------------------
 Finalize Aggregate (actual rows=1 loops=1)
   ->  Gather Motion 3:1  (slice1; segments: 3) (actual rows=3 loops=1)
         ->  Partial Aggregate (actual rows=1 loops=1)
               ->  Hash Join (actual rows=27610 loops=1)
                     Hash Cond: (s.region = d.region)
                     Extra Text: (seg1)   Hash chain length 1.0 avg, 1 max, using 2 of 524288 buckets.
                     ->  Seq Scan on m191_sales s (actual rows=1334931 loops=1)
                     ->  Hash (actual rows=2 loops=1)
                           Buckets: 524288  Batches: 1  Memory Usage: 4097kB
                           ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=2 loops=1)
                                 ->  Seq Scan on m191_sales d (actual rows=1 loops=1)
                                       Filter: (rname = ANY ('{r5,r7}'::text[]))
                                       Rows Removed by Filter: 63

Under §18.1’s reporting conventions those actual rows are per-segment figures, and they say exactly what the shape table claims: below the motion the busiest segment produced 1 matching dimension row (two rows exist in the whole cluster), above it every receiver holds 2. Partitioned went in, replicated came out, and the total row count on the wire is senders × rows, which is precisely why §16.2 charges a broadcast so much more than a redistribute. A Redistribute, by contrast, preserves the row count and only changes whose rows they are — and needs one slice per input, so a two-sided redistribute join costs one more gang.

Four slices, four gangs: 1 + 3 × 3 = 10 processes for one SELECT

-- m191_ship: 2 000 000 rows, DISTRIBUTED BY (sid)
EXPLAIN (costs off) SELECT count(*) FROM m191_sales s JOIN m191_ship p
                     ON s.region = p.region AND s.amt = p.qty;

-- then, from a second session while that query actually runs (sess_id 47):
SELECT count(*) AS processes FROM (
        SELECT 1 FROM gp_dist_random('pg_stat_activity') WHERE sess_id = 47
  UNION ALL
        SELECT 1 FROM pg_stat_activity                   WHERE sess_id = 47) z;

                                      QUERY PLAN                                       
---------------------------------------------------------------------------------------
 Finalize Aggregate
   ->  Gather Motion 3:1  (slice1; segments: 3)
         ->  Partial Aggregate
               ->  Hash Join
                     Hash Cond: ((s.region = p.region) AND (s.amt = (p.qty)::numeric))
                     ->  Redistribute Motion 3:3  (slice2; segments: 3)
                           Hash Key: s.region, s.amt
                           ->  Seq Scan on m191_sales s
                     ->  Hash
                           ->  Redistribute Motion 3:3  (slice3; segments: 3)
                                 Hash Key: p.region, (p.qty)::numeric
                                 ->  Seq Scan on m191_ship p
 Optimizer: GPORCA
(13 rows)

 processes 
-----------
        10

The last of the four is the strange one, and it is the reason DML on a distributed table works at all. When an UPDATE changes the distribution key, the row must move to a different segment — but the old version must be deleted where it already lives. §18.7 introduced the Split Update node that turns one update into a delete tuple and an insert tuple; the delete half already knows its home and must not be re-hashed. So the planner puts the segment number into the row itself.

Explicit Redistribute: gp_segment_id travels as a column of the tuple

SET optimizer = off;   -- ORCA renders this shape as Split Update + Redistribute
EXPLAIN (costs off, verbose) UPDATE m191_sales SET id = id + 1 WHERE region = 5;

SET
                                  QUERY PLAN                                   
-------------------------------------------------------------------------------
 Update on public.m191_sales
   ->  Explicit Redistribute Motion 3:3  (slice1; segments: 3)
         Output: ((id + 1)), region, amt, ctid, gp_segment_id, (DMLAction)
         ->  Split Update
               Output: ((id + 1)), region, amt, ctid, gp_segment_id, DMLAction
               ->  Seq Scan on public.m191_sales
                     Output: (id + 1), region, amt, ctid, gp_segment_id
                     Filter: (m191_sales.region = 5)
 Optimizer: Postgres query optimizer
 Settings: optimizer = 'off'

src/backend/executor/nodeMotion.c:1299

	else if (motion->motionType == MOTIONTYPE_EXPLICIT)
	{
		Datum		segidColIdxDatum;

		Assert(motion->segidColIdx > 0 && motion->segidColIdx <= list_length((motion->plan).targetlist));
		bool		is_null = false;

		segidColIdxDatum = slot_getattr(outerTupleSlot, motion->segidColIdx, &is_null);
		targetRoute = Int32GetDatum(segidColIdxDatum);
		Assert(!is_null);
	}

No hash function, no distribution policy, no key — one slot_getattr at the attribute number the planner recorded in Motion.segidColIdx (plannodes.h:1907), and that integer is the route. This is the only motion type whose routing is data rather than computation, and the reason it can be: the row was read from a specific segment, so it can be told where it came from and sent back there. ctid beside it in the target list is the other half of the trick — §18.7 has the DML side, §19.4 the sending side.

19.1.4 Reading the N:M label

The two numbers are senders : receivers, and they are gang sizes, not row counts. EXPLAIN computes them from the slice table at the moment it prints the node: the sender count is the size of the slice the Motion caps, the receiver count the size of its parent slice.

src/backend/commands/explain.c:1928

				motion_snd = list_length(es->currentSlice->segments);

				motion_recv = parentSlice == NULL ? 1 : list_length(parentSlice->segments);

				/* ... the motionType switch forces motion_recv = 1 for both
				 * gather flavours, and picks sname ... */

				pname = psprintf("%s %d:%d", sname, motion_snd, motion_recv);

So Gather Motion 3:1 is three sending gang members and, by definition, one receiver; Redistribute Motion 3:3 is three of each and the all-to-all mesh of the opening figure; Gather Motion 1:1 was the replicated table’s one-segment gang. The case that surprises people is 1:3, which needs a sender slice with exactly one process in it — for example a catalog scan, which can only run on the coordinator.

Broadcast Motion 1:3 — a single coordinator-side sender fanning out to the whole cluster

EXPLAIN (costs off) SELECT count(*) FROM m191_sales s, pg_class c WHERE s.region = c.relnatts;

                       QUERY PLAN                       
--------------------------------------------------------
 Finalize Aggregate
   ->  Gather Motion 3:1  (slice1; segments: 3)
         ->  Partial Aggregate
               ->  Hash Join
                     Hash Cond: (c.relnatts = s.region)
                     ->  Broadcast Motion 1:3  (slice2)
                           ->  Seq Scan on pg_class c
                     ->  Hash
                           ->  Seq Scan on m191_sales s
 Optimizer: Postgres query optimizer
(10 rows)

Note what is missing: every other motion in this section printed (sliceN; segments: N), and this one prints only (slice2). show_dispatch_info reports zero segments for a GANGTYPE_UNALLOCATED or GANGTYPE_ENTRYDB_READER slice (explain.c:1429, printed at :1450), and an entry-db reader is exactly what a coordinator-side catalog scan gets: one QE, on the coordinator, alongside the QD. Its data is a singleton, and the Broadcast converts singleton → replicated so the segments can join against it locally. The other way the arity moves is intra-segment parallelism, which multiplies the process count without changing the segment count.

With intra-segment parallelism the sender count is segments × workers — including the asymmetric 6:3

SET enable_parallel = on;
EXPLAIN (costs off) SELECT count(*) FROM m191_sales;
EXPLAIN (costs off) SELECT region, count(*) FROM m191_sales GROUP BY region;

SET
                    QUERY PLAN                     
---------------------------------------------------
 Finalize Aggregate
   ->  Gather Motion 6:1  (slice1; segments: 6)
         ->  Partial Aggregate
               ->  Parallel Seq Scan on m191_sales
 Optimizer: GPORCA
(5 rows)

                         QUERY PLAN                         
------------------------------------------------------------
 Gather Motion 3:1  (slice1; segments: 3)
   ->  Finalize HashAggregate
         Group Key: region
         ->  Redistribute Motion 6:3  (slice2; segments: 6)
               Hash Key: region
               ->  Streaming Partial HashAggregate
                     Group Key: region
                     ->  Parallel Seq Scan on m191_sales
 Optimizer: GPORCA
(9 rows)

Three segments, two workers each, so the sending gang has six members and the label says 6. Nothing about the label changed — it is still list_length(sendSlice->segments), and §19.3 explains why that list has six entries when there are three segments (numsegments × parallel_workers). The second plan is the one to remember: 6:3 is a perfectly legal motion in which six senders redistribute into three receivers, so the arity of a Motion is not a property of the cluster, it is a property of the two slices it joins. §18.2 owns why the planner chose six.

19.1.5 What the model costs

None of this is free, and it is worth being blunt about the bill. A motion is the only way to satisfy a join or a grouping whose keys do not line up with the distribution, which is why the planner has to reason about distribution at all (§16.1) and why it has a dedicated cost function for shipping tuples (§16.2’s cdbpath_cost_motion). Every motion boundary also multiplies processes: the ten-process census above is one SELECT count(*), and each of those processes carries its own work memory, its own snapshot and its own share of the resource budget (Chapter 21).

The subtler cost is structural, and it is the thread that runs through the rest of this chapter. A sender and a receiver are separate operating-system processes with finite buffers. A sender blocks when its outbound queue is full; a receiver blocks when nothing has arrived. Arrange four such processes in a cycle — two motions feeding the two sides of one join — and the executor can deadlock itself with no lock, no lock table and nothing for the deadlock detector of §11.5 to find. The book has already shown both mechanisms that prevent it: §18.1’s ExecSquelchNode (src/backend/executor/execAmi.c:756), which tells an abandoned sender to stop rather than fill its queue forever, and §18.4’s prefetch_inner (plannodes.h:1198), which forces a join to drain its inner motion before it touches the outer one. What neither section could show is the machinery underneath: how a receiver’s stop actually reaches a sender, and what happens on the wire while it is in flight.

That is the rest of the chapter. §19.2 takes the QD’s side — how gangs are allocated, reused and recycled, and what dispatch physically sends. §19.3 opens the slice table that this section kept pointing at, field by field. §19.4 opens nodeMotion.c: the sender loop, the receiver loop, chunking, and the stop protocol that answers the deadlock question. §19.5 goes below SendTupleChunkToAMS into the interconnect itself — which, it turns out, is not part of the backend at all. §19.6 closes Part III with the one feature that lets a client skip the Gather entirely and read from the segments directly.

(Demo objects m191_sales, m191_dim, m191_rep and m191_ship were dropped after the captures above.)