Motion Nodes: Gather, Broadcast and Redistribute

Motion is where MPP queries get slow. Reading Motion nodes correctly is the single highest-return EXPLAIN skill on this platform.

Every plan in Chapters 13 through 18 carried lines like Gather Motion 3:1 and Redistribute Motion 3:3, and every time we read them as annotations. Here they become code. A Motion is an ordinary Plan node with an ordinary ExecProcNode callback (nodeMotion.c:702), and it is initialised on every process that participates in the query — but it behaves completely differently depending on which side of the wire that process is on. §19.1 framed the duality and §19.3 owns the identity question (which slice am I?); this section is the machinery.

There are three layers stacked here and it pays to keep them apart. nodeMotion.c (1405 lines) is the executor node: it decides sender-versus-receiver, computes a destination route per tuple, and runs the merge heap for a sorted Gather. cdbmotion.c (1400 lines) is the Motion layer: it serialises a slot into a byte stream, cuts the stream into TupleChunks, reassembles chunks back into tuples per sender, counts end-of-stream tokens, and keeps the per-node statistics. Below it sits one function-pointer call, CurrentMotionIPCLayer->SendTupleChunkToAMS(), and everything past that is §19.5.

Look at the left-hand column again. The sender’s ExecMotion returns NULL, always — never a tuple. It is a legitimate plan node whose ExecProcNode produces no output at all: it drains its child, pushes every tuple sideways onto the wire, sends an end-of-stream token, and reports “no rows” upward so that the slice’s top-level ExecutePlan loop terminates. Every other node in the book returns tuples to its parent. This one is a sink.

19.4.1 One node, two implementations

ExecMotion is a three-way branch on MotionState.mstype, and a second branch on whether the plan asked for order preservation.

src/backend/executor/nodeMotion.c:125

	if (node->mstype == MOTIONSTATE_RECV)
	{
		...
		if (motion->sendSorted)
			tuple = execMotionSortedReceiver(node);
		else
			tuple = execMotionUnsortedReceiver(node);
		...
	}
	else if (node->mstype == MOTIONSTATE_SEND)
		return execMotionSender(node);
	else
		elog(ERROR, "cannot execute inactive Motion");

mstype is decided once, in ExecInitMotion. On a QE it is a straight comparison of LocallyExecutingSliceIndex(estate) against the two slice indexes the node straddles (nodeMotion.c:751-760): equal to the parent slice means receive, equal to the sending slice means send, and neither means this process is an alien to this Motion and the node stays MOTIONSTATE_NONE. On the QD, a Gather’s receiving slice is the root slice, so the QD is always the receiver (nodeMotion.c:720-741). That is the whole of the sender/receiver duality: no negotiation, no runtime role election, just an integer comparison against the slice table §19.3 shipped with the plan.

The sender is a tight loop with three exits — child exhausted, receiver stopped, or an error:

src/backend/executor/nodeMotion.c:251

		if (done || TupIsNull(outerTupleSlot))
		{
			doSendEndOfStream(motion, node);
			done = true;
		}
		...
		else
		{
			doSendTuple(motion, node, outerTupleSlot);
			if (node->stopRequested)
			{
				ExecSquelchNode(outerNode, true);   /* tell the children */
				done = true;
			}
		}

Note what is not there: on the stop path the sender exits without sending end-of-stream. The receiver already said it does not want any more, so there is nobody left who cares. The loop’s postcondition is the invariant Assert(node->stopRequested || node->numTuplesFromChild == node->numTuplesToAMS) (nodeMotion.c:299) — either every tuple pulled from the child went to the wire, or we were stopped. ExecSquelchNode is §18.1’s; the Motion node is simply the place where a network-level stop turns back into an executor-level squelch.

19.4.2 The four motion types, and how a route is computed

A route is a small integer: the index of a receiving process within the parent slice’s primaryProcesses list. doSendTuple (nodeMotion.c:1181) exists to turn a tuple into a route, and it is a plain if-chain over motion->motionType. There are six enum values (plannodes.h:1878-1887), of which MOTIONTYPE_OUTER_QUERY is a planner placeholder resolved before execution, so five reach the executor.

EXPLAIN label	MotionType	how the destination route is computed	source
Gather Motion	MOTIONTYPE_GATHER	Fixed: route 0. The parent slice has exactly one process, so the code does not even consult it.	nodeMotion.c:1197-1207
Explicit Gather Motion	MOTIONTYPE_GATHER_SINGLE	Route 0 as well, but only one segment actually sends: the one where GpIdentity.segindex equals gp_session_id modulo numInputSegs. The others execute the subplan and throw the tuples away.	nodeMotion.c:256-264
Redistribute Motion	MOTIONTYPE_HASH	cdbhash over the motion’s hashExprs, reduced with cdbhashreduce to a segment index; with intra-segment parallelism a second hash picks the worker.	nodeMotion.c:1265-1297
Broadcast Motion	MOTIONTYPE_BROADCAST	The pseudo-route BROADCAST_SEGIDX, which is -2 so it can never collide with a real route. One SendTuple call; the transport does the fan-out.	nodeMotion.c:1209-1212, tupchunk.h:59
Broadcast Workers Motion	MOTIONTYPE_BROADCAST_WORKERS	The node itself loops over receiving segments, one SendTuple each, picking a worker inside each segment with random() modulo parallel_workers.	nodeMotion.c:1213-1263
Explicit Redistribute Motion	MOTIONTYPE_EXPLICIT	slot_getattr of the junk column at motion->segidColIdx — the destination travels inside the tuple.	nodeMotion.c:1299-1309

The interesting one is MOTIONTYPE_HASH, because it raises a question the book has been circling since §16.1: is a Redistribute Motion’s hash the same hash the storage layer uses to place rows? §16.4 proved direct dispatch uses cdbhash. Here is the answer, and it is unambiguous.

src/backend/executor/nodeMotion.c:1105

		cdbhashinit(h);
		i = 0;
		foreach(hk, hashkeys)
		{
			keyval = ExecEvalExpr((ExprState *) lfirst(hk), econtext, &isNull);
			cdbhash(h, i + 1, keyval, isNull);
			i++;
		}
		target_seg = cdbhashreduce(h);

That is evalHashKey, and it is exactly the sequence cdbtargeteddispatch.c:333 (direct dispatch, §16.4), copyfrom.c:4096 (COPY’s row placement) and nodeSplitUpdate.c:64 all run. The CdbHash is built by makeCdbHash(numHashSegments, nkeys, node->hashFuncs) at nodeMotion.c:834, and hashFuncs was filled in at plan time by cdb_hashproc_in_opfamily() (cdbmutate.c:122) — the same opclass hash procedure the distribution policy resolves. Not a look-alike: the same functions, the same reduction (cdbhash.c:253). Which means a Redistribute Motion on column k deposits each row on precisely the segment where a table DISTRIBUTED BY (k) would have stored it. We can watch that happen: gp_execution_segment() evaluated below the Gather reports the segment that actually processed the row.

Left: where a hash-distributed table physically stores each key. Right: where a Redistribute Motion sends the same keys, read out of a randomly-distributed table. Identical, row for row.

-- storage placement
SELECT gp_segment_id, k FROM m194_hash ORDER BY k;

-- routing decided by Redistribute Motion (m194_rand is DISTRIBUTED RANDOMLY)
EXPLAIN (costs off) SELECT gp_execution_segment() AS seg, k
  FROM (SELECT k FROM m194_rand GROUP BY k) t;
SELECT gp_execution_segment() AS seg, k
  FROM (SELECT k FROM m194_rand GROUP BY k) t ORDER BY k;

 gp_segment_id | k  
---------------+----
             1 |  1
             0 |  2
             0 |  3
             0 |  4
             2 |  5
             2 |  6
             0 |  7
             0 |  8
             2 |  9
             2 | 10
             2 | 11
             1 | 12
(12 rows)

                            QUERY PLAN                            
------------------------------------------------------------------
 Gather Motion 3:1  (slice1; segments: 3)
   ->  GroupAggregate
         Group Key: k
         ->  Sort
               Sort Key: k
               ->  Redistribute Motion 3:3  (slice2; segments: 3)
                     Hash Key: k
                     ->  Seq Scan on m194_rand
 Optimizer: GPORCA
(9 rows)

 seg | k  
-----+----
   1 |  1
   0 |  2
   0 |  3
   0 |  4
   2 |  5
   2 |  6
   0 |  7
   0 |  8
   2 |  9
   2 | 10
   2 | 11
   1 | 12
(12 rows)

MOTIONTYPE_EXPLICIT is the opposite design: instead of computing a destination, the sender reads one out of the tuple. §18.7 showed where the value comes from — a Split Update emits the row’s original gp_segment_id as a junk column — and EXPLAIN VERBOSE shows it riding along in the Motion’s output list.

An Explicit Redistribute Motion. The target table is DISTRIBUTED RANDOMLY, so a modified row has to go back to the segment it came from; gp_segment_id is carried through the join for exactly that purpose.

SET optimizer=off;
EXPLAIN (costs off, verbose) UPDATE m194_rand r SET v = h.v FROM m194_hash h WHERE r.k = h.k;

                            QUERY PLAN                            
------------------------------------------------------------------
 Update on public.m194_rand r
   ->  Explicit Redistribute Motion 3:3  (slice1; segments: 3)
         Output: h.v, r.ctid, r.gp_segment_id, h.ctid
         ->  Hash Join
               Output: h.v, r.ctid, r.gp_segment_id, h.ctid
               Hash Cond: (r.k = h.k)
               ->  Redistribute Motion 3:3  (slice2; segments: 3)
                     Output: r.ctid, r.gp_segment_id, r.k
                     Hash Key: r.k
                     ->  Seq Scan on public.m194_rand r
               ->  Hash
                     Output: h.v, h.ctid, h.k
                     ->  Seq Scan on public.m194_hash h
(17 rows)

Two asides on the exotic types. Broadcast and Broadcast Workers split the fan-out responsibility differently: plain broadcast makes one SendTuple call with route -2 and lets the transport replicate, while the parallel variant loops inside doSendTuple. A consequence is that broadcast can never use the zero-copy path of §19.4.3 — CandidateForSerializeDirect() refuses route -2 outright (tupser.c:334). And Explicit Gather Motion is genuinely rare: it is emitted only when the sender’s locus is Replicated (createplan.c:3492), and which segment does the sending is chosen by gp_session_id % numInputSegs — a load-spreading trick that makes the same query send from a different segment in a different session.

19.4.3 Serialization and chunking — and who chooses the chunk size

SendTuple (cdbmotion.c:426) has a fast path and a slow path, and it tries the fast one first. It asks the transport for a pointer straight into the outgoing packet buffer (GetTransportDirectBuffer, cdbmotion.c:458), and if the whole tuple fits there, SerializeTuple writes the chunk header and the tuple body in place and returns a byte count — no chunk list, no intermediate copy.

src/backend/cdb/motion/tupser.c:450

	if (CandidateForSerializeDirect(targetRoute, b) &&
		tuplen + TUPLE_CHUNK_HEADER_SIZE <= b->prilen)
	{
		memcpy(b->pri + TUPLE_CHUNK_HEADER_SIZE, &tupbodylen, sizeof(tupbodylen));
		memcpy(b->pri + TUPLE_CHUNK_HEADER_SIZE + sizeof(int), tupbody, tupbodylen);
		dataSize += tuplen;
		SetChunkType(b->pri, TC_WHOLE);
		SetChunkDataSize(b->pri, dataSize - TUPLE_CHUNK_HEADER_SIZE);
		return dataSize;
	}

The serialized form of a tuple is disarmingly simple: a 4-byte body length, then the raw MinimalTuple body (everything from MINIMAL_TUPLE_DATA_OFFSET onward). Toasted attributes are detoasted first (tupser.c:402-440) — a TOAST pointer would be meaningless in another process, the same class of problem tupleremap.c solves for record typmods in §19.4.5. When the tuple does not fit the direct buffer, the stream is cut into TupleChunks by addByteStringToChunkList (tupser.c:190), and each chunk gets a 4-byte header: a uint16 payload length and a uint16 type tag.

src/include/cdb/tupchunk.h:21

typedef enum TupleChunkType
{
	TC_WHOLE,			/* Contains a whole tuple. */
	TC_PARTIAL_START,	/* Contains the starting portion of a tuple. */
	TC_PARTIAL_MID,		/* Contains a middle part of a tuple. */
	TC_PARTIAL_END,		/* Contains the final portion of a tuple. */
	TC_END_OF_STREAM,	/* Indicates "end of tuples" from this source. */
	TC_EMPTY,			/* Empty tuple */
	TC_MAXVAL
} TupleChunkType;

The maximum chunk size is not a constant and not a GUC: it is asked of the transport, once per query, through the vtable. Gp_max_tuple_chunk_size = CurrentMotionIPCLayer->GetMaxTupleChunkSize() (cdbmotion.c:169). udpifc answers Gp_max_packet_size - sizeof(struct icpkthdr) - 4 (ic_udpifc.c:8179) and tcp answers Gp_max_packet_size - PACKET_HEADER_SIZE - 4 (ic_tcp.c:3365). With gp_max_packet_size = 8192 on this cluster, and sizeof(icpkthdr) = 88 against TCP’s 4-byte length prefix, that is 8100 bytes over UDP and 8184 over TCP. The same tuple therefore fragments differently depending on which transport is loaded — the clearest illustration in the chapter of what the vtable seam actually buys.

TupleChunk header — 4 bytes, and the length does NOT include the header (tupchunk.h:34-49)

uint16  payload length

uint16  TupleChunkType

↑0

1

2↑

3

the serialized tuple, before chunking — 16 310 bytes

int32 = 16306

MinimalTuple body — 16 306 bytes

cut to Gp_max_tuple_chunk_size = 8100  (8192 − 88 − 4)

8096

TC_PARTIAL_START

len field 4 B

body [0 … 8091]  8092 B

∑ 8100 B

8096

TC_PARTIAL_MID

body [8092 … 16187]  8096 B

∑ 8100 B

118

TC_PARTIAL_END

body [16188 … 16305]

∑ 122 B

0

TC_END_OF_STREAM

header only — a static buffer, cdbmotion.c:51 and :176

∑ 4 B

⟦ total on the wire: 16 326 B ⟧

over tcp, capacity 8184: 8184 + 8134 + 4 = ⟦ 16 322 B in 3 chunks ⟧

To make that measurable, a table with one wide, uncompressed row, sent over each transport in turn (gp_interconnect_type has backend context, so PGOPTIONS switches it per connection with no restart), with gp_log_interconnect = verbose so the Motion layer dumps its per-node counters at teardown.

The same tuple, the same 16 310 bytes of tuple data — but four chunks over udpifc and three over tcp, because the transport, not the executor, sets the chunk size. Segments 0 and 2 hold no matching row and send nothing but a 4-byte end-of-stream chunk.

CREATE TABLE m194_wide(id int, payload text) DISTRIBUTED BY (id);
ALTER TABLE m194_wide ALTER COLUMN payload SET STORAGE EXTERNAL;
INSERT INTO m194_wide SELECT 1, repeat('ab', 8148);   -- length 16296

-- run 1: default interconnect
SET gp_log_interconnect='verbose';
SELECT payload FROM m194_wide;
-- run 2: PGOPTIONS='-c gp_interconnect_type=tcp' psql -p 7100 ... same query

-- then, from the segment logs:
$ grep 'Interconnect seg' .../dbfast*/demoDataDir*/log/gpdb-*.csv

-- udpifc  (Gp_max_tuple_chunk_size = 8100)
Interconnect seg0 slice1 sent 0 tuples,     4 total bytes,     0 tuple bytes, 1 chunks.
Interconnect seg1 slice1 sent 1 tuples, 16326 total bytes, 16310 tuple bytes, 4 chunks.
Interconnect seg2 slice1 sent 0 tuples,     4 total bytes,     0 tuple bytes, 1 chunks.

-- tcp  (Gp_max_tuple_chunk_size = 8184)
Interconnect seg0 slice1 sent 0 tuples,     4 total bytes,     0 tuple bytes, 1 chunks.
Interconnect seg1 slice1 sent 1 tuples, 16322 total bytes, 16310 tuple bytes, 3 chunks.
Interconnect seg2 slice1 sent 0 tuples,     4 total bytes,     0 tuple bytes, 1 chunks.

The arithmetic closes exactly. Over UDP: 16 306 body bytes plus the 4-byte length field spread over three chunks of 8100, 8100 and 122 bytes, plus a 4-byte EOS chunk, is 16 326. Over TCP the larger capacity absorbs the body in two chunks, 8184 + 8134, plus EOS, is 16 322. The 4-byte difference between the two totals is one chunk header, and nothing else changed — the executor produced an identical byte stream both times.

The opposite extreme is TC_EMPTY: a tuple with no columns at all. A count(*) over a semi join needs no values from the segments, only the fact that a row qualified, so the Gather Motion’s target list is empty and each qualifying row costs a bare 4-byte header.

A row with zero columns on the wire. The Gather has no Output list at all; two rows qualify, one on each of two segments, and each is a single header-only TC_EMPTY chunk (tupser.c:368-375).

SET optimizer=off; SET gp_log_interconnect='verbose';
EXPLAIN (costs off, verbose) SELECT count(*) FROM m194_hash h
  WHERE EXISTS (SELECT 1 FROM m194_dim d WHERE d.d = h.k);
SELECT count(*) FROM m194_hash h
  WHERE EXISTS (SELECT 1 FROM m194_dim d WHERE d.d = h.k);

                      QUERY PLAN                       
-------------------------------------------------------
 Aggregate
   Output: count(*)
   ->  Gather Motion 3:1  (slice1; segments: 3)      <-- no Output line
         ->  Hash Semi Join
               Hash Cond: (h.k = d.d)
               ->  Seq Scan on public.m194_hash h
                     Output: h.k, h.v
               ->  Hash
                     Output: d.d
                     ->  Seq Scan on public.m194_dim d

-- segment logs:
Interconnect seg0 slice1 sent 1 tuples, 12 total bytes, 4 tuple bytes, 2 chunks.
Interconnect seg1 slice1 sent 1 tuples, 12 total bytes, 4 tuple bytes, 2 chunks.
Interconnect seg2 slice1 sent 0 tuples,  4 total bytes,  0 tuple bytes, 1 chunks.

Careful reading that 4 tuple bytes: it is 4 more than the truth. On the direct-buffer path SendTuple sets serialized_data_length = sent, and sent includes the chunk header (cdbmotion.c:473-474), so tuple bytes is inflated by 4 for every direct-path send. The comment above the stats helpers admits as much — “the only fields that are required to be valid are num_chunks and serialized_data_length” (cdbmotion.c:1126-1131). A zero-column tuple carries literally no payload: SetChunkDataSize(b->pri, 0).

19.4.4 The receiver: per-sender state, and counting end-of-stream

RecvTupleFrom (cdbmotion.c:550) is a loop over a queue. If a completed tuple is sitting in the FIFO, return it. Otherwise, if this source is already at end-of-stream, return NULL. Otherwise call processIncomingChunks and try again. All the real work is in addChunkToSorter (cdbmotion.c:982), which is a switch over the chunk type.

The state that matters is ChunkSorterEntry (cdbinterconnect.h:34), and there is one per source route, not one per Motion node. The reason is in the comment on TupleChunkListData: chunks carry no tuple identity — “they are expected to be sent and received in order, and without loss” (tupchunklist.h:57-62). Ordering is only guaranteed within one sender’s stream. If a single global chunk list were used, a TC_PARTIAL_MID from segment 0 could land between segment 1’s start and end fragments and silently corrupt both. So each sender gets its own partial-tuple accumulator, and addChunkToSorter enforces the grammar loudly — receiving a TC_WHOLE while partial data is pending is an ERRCODE_GP_INTERCONNECTION_ERROR, not a recoverable condition.

Completed tuples, by contrast, may or may not share a queue. An unordered Motion points every route’s ready_tuples at one shared htup_fifo, because the receiver does not care who sent what; an order-preserving Motion gives each route its own FIFO, because the merge in §19.4.6 must be able to ask for the next tuple from sender k specifically (cdbmotion.c:917-939). htupfifo.c is 176 lines of singly-linked list with a free list, and that is all it needs to be.

Termination is a count. UpdateMotionExpectedReceivers sets num_senders to the number of live CdbProcess entries in the child slice (cdbmotion.c:338) — live, because direct dispatch leaves NULL placeholders for segments that were never asked to run. Each TC_END_OF_STREAM bumps a tally, and only when the tally reaches num_senders does the node declare the stream finished:

src/backend/cdb/motion/cdbmotion.c:1100

			/* Mark the state as "end of stream." */
			chunkSorterEntry->end_of_stream = true;
			pMNEntry->num_stream_ends_recvd++;

			if (pMNEntry->num_stream_ends_recvd == pMNEntry->num_senders)
				pMNEntry->moreNetWork = false;

			CurrentMotionIPCLayer->DeregisterReadInterest(transportStates, motNodeID,
														  srcRoute, "end of stream");

The per-route ChunkSorterEntry made visible. Three source routes, three entries, printed at teardown for an order-preserving Gather over three segments.

SET optimizer=off;
SET gp_log_interconnect='debug';
SET client_min_messages='debug4';
SELECT k FROM m194_hash ORDER BY k;   -- output elided

DEBUG:  Chunk-sorter entry [route=0,node=1] statistics:
	Available Tuples High-Watermark: 10
DEBUG:  Chunk-sorter entry [route=1,node=1] statistics:
	Available Tuples High-Watermark: 10
DEBUG:  Chunk-sorter entry [route=2,node=1] statistics:
	Available Tuples High-Watermark: 10

That capture also exposes a small piece of bit rot worth knowing about before you trust the number. The high-watermark is printed once per route, but all three lines report 10 — even though the three senders contributed 5, 2 and 5 tuples. The comment in statNewTupleArrived promises “if the motion node is order-preserving, we track a per-sender high-watermark as well” (cdbmotion.c:1198-1203), but the function only maintains pMNEntry->stat_tuples_available_hwm, which is node-wide. The ChunkSorterEntry argument is unused. So the figure is real, but it is the same node-level figure repeated, not a per-sender breakdown.

19.4.5 tupleremap — a typmod is a process-local name

Here is a problem no single-node database has. When PostgreSQL evaluates ROW(k, v) it produces a value of type RECORD, and it labels that value with a transient typmod — a small integer allocated from a backend-local counter, which indexes into that backend’s own cache of anonymous tuple descriptors. It is a pointer, written as an integer, into private memory. Segment 0’s typmod 3 and the coordinator’s typmod 3 have nothing to do with each other.

So a record value cannot simply be serialized and sent: the tuple header carries a number that will be nonsense at the other end, and the receiver has no way to discover what shape the record was. Both halves of that have to be fixed, and tupleremap.c (773 lines, adapted from upstream’s parallel-worker tqueue.c) is the fix.

The protocol is sender-push, incremental, and per-connection. Before every SendTuple, doSendTuple calls CheckAndSendRecordCache (nodeMotion.c:1313). That function compares the sender’s current transient-typmod counter against a high-water mark stored on the connection (conn->sent_record_typmod, reached through CurrentMotionIPCLayer->GetMotionSentRecordTypmod, ic_common.c:540), and if the counter has moved, it serialises every newly-registered tuple descriptor and pushes it down the same chunk stream as the data — tagged with a magic length so the receiver can tell it apart from a tuple:

src/backend/cdb/motion/tupser.c:285

	typelist = build_tuple_node_list(sent_record_typmod);
	buf = serializeNode((Node *) typelist, &size, NULL);
	list_free_deep(typelist);

	/* we use magic tuplen to identify that this chunk (or list of chunks)
	 * actually carries the serialized record cache table. */
	int tupbodylen = RECORD_CACHE_MAGIC_TUPLEN;    /* -1 */
	addByteStringToChunkList(tcList, (char *) &tupbodylen, sizeof(int), ...);

The gate is cheap: ShouldSendRecordCache returns false unless the Motion’s tuple descriptor contains at least one non-base type at all (pSerInfo->has_record_types, set in InitSerTupInfo at tupser.c:131), so a Motion carrying only int and text never pays for any of this. The receiver spots the magic length in CvtChunksToTup (tupser.c:651), hands the list to TRHandleTypeLists, registers each descriptor in its own cache, and records the translation:

src/backend/cdb/motion/tupleremap.c:249

		/* assign_record_type_typmod() will update tdtypmod to the local typmod */
		assign_record_type_typmod(descnode->tuple);
		local_typmod = descnode->tuple->tdtypmod;

		remapper->typmodmap[remote_typmod] = local_typmod;

		if (!remapper->remap_needed && local_typmod != remote_typmod)
			remapper->remap_needed = true;

Every incoming tuple then passes through TRCheckAndRemap (cdbmotion.c:108), which short-circuits entirely when remap_needed is false. When it is needed, BuildFieldRemapInfo walks the tuple descriptor once and builds a tree of TupleRemapInfo nodes marking which attributes could possibly contain a composite — recursing through arrays, ranges and domains, because a record can hide arbitrarily deep (tupleremap.c:618-773). TRRemapRecord then deforms the value, replaces the typmod label, recursively fixes the fields, and reforms it — and only reforms if something actually changed (tupleremap.c:584).

This is measurable. Because CheckAndSendRecordCache routes its push through statSendTuple, the type table shows up in the counters as an extra send that is not a tuple.

The same 12 rows, twice. Adding a record column adds exactly one send and one chunk per sender — the type table — plus about 106 bytes, and makes each row bigger because a record datum carries its own tuple header.

SET optimizer=off; SET gp_log_interconnect='verbose';
SELECT k, v FROM m194_hash;          -- run 1
SELECT row(k,v) AS r FROM m194_hash; -- run 2

-- run 1: plain columns, no record types in the tupdesc
Interconnect seg0 slice1 sent 5 tuples, 129 total bytes, 105 tuple bytes, 6 chunks.
Interconnect seg1 slice1 sent 2 tuples,  55 total bytes,  43 tuple bytes, 3 chunks.
Interconnect seg2 slice1 sent 5 tuples, 131 total bytes, 107 tuple bytes, 6 chunks.

-- run 2: one record column
Interconnect seg0 slice1 sent 6 tuples, 342 total bytes, 314 tuple bytes, 7 chunks.
Interconnect seg1 slice1 sent 3 tuples, 205 total bytes, 189 tuple bytes, 4 chunks.
Interconnect seg2 slice1 sent 6 tuples, 344 total bytes, 316 tuple bytes, 7 chunks.

Five rows became six sends, two became three, five became six — one extra per segment, never more, because the watermark stops the second push. And the size backs it out: segment 0 sent 5 rows in 314 tuple bytes, segment 1 sent 2 in 189, so a row costs about 41.7 bytes and the fixed part is about 106 bytes on both — the serialized TupleDescNode list, sent once.

Record values surviving harder motions: a Broadcast Motion between segments, where neither end is the coordinator, and record values nested inside arrays and inside other records.

SET optimizer=off;
SELECT count(*) FROM (SELECT row(k,v) r FROM m194_rand) a
  JOIN (SELECT row(k,v) r FROM m194_hash) b ON a.r = b.r;

SELECT array[row(k,v)] AS arr, row(k, row(v,k)) AS nested
  FROM m194_hash ORDER BY k LIMIT 2;

 count 
-------
    12
(1 row)

    arr     |    nested    
------------+--------------
 {"(1,v1)"} | (1,"(v1,1)")
 {"(2,v2)"} | (2,"(v2,2)")
(2 rows)

Two structural details that follow from typmods being process-local. First, the TupleRemapper hangs off the connection, not the Motion node: processIncomingChunks fetches it with GetMotionConnTupleRemapper(transportStates, motNodeID, srcRoute) (cdbmotion.c:660, ic_common.c:524). It has to — two senders can hand out different typmods for the same shape, and each needs its own translation table. Second, a broadcast has no single route, so GetMotionSentRecordTypmod quietly folds BROADCAST_SEGIDX onto route 0 (ic_common.c:548) and uses that one connection’s watermark to speak for all of them.

19.4.6 Sorted Motion: an N-way merge in the receiver

A Gather Motion normally returns tuples in whatever order the chunks happen to arrive. But if each sender’s stream is already sorted, the receiver can produce a globally sorted result by repeatedly taking the smallest head among the senders — and then a distributed ORDER BY ... LIMIT needs no sort on the coordinator at all. The planner marks such a Motion sendSorted, and EXPLAIN prints a Merge Key: line for it (explain.c:2853-2866, explain_gp.c:2251).

The Merge Key line, and the per-segment Sort and Limit that make it legal. Each segment sorts and truncates locally; the coordinator only merges.

SET optimizer=off;
EXPLAIN (costs off) SELECT k FROM m194_hash ORDER BY k LIMIT 5;

                   QUERY PLAN                   
------------------------------------------------
 Limit
   ->  Gather Motion 3:1  (slice1; segments: 3)
         Merge Key: k
         ->  Limit
               ->  Sort
                     Sort Key: k
                     ->  Seq Scan on m194_hash

The merge is visible in the output itself. Add gp_execution_segment() and run the query with and without the ORDER BY: unsorted, the rows come out in per-segment blocks; sorted, the segments interleave under the control of the heap.

Left, an unsorted Gather: whole blocks per sender, in arrival order. Right, a sorted Gather: the senders interleave, and k is globally ordered — that interleaving is the binary heap choosing a route per row.

SELECT gp_execution_segment() AS seg, k FROM m194_hash;            -- unsorted
SELECT gp_execution_segment() AS seg, k FROM m194_hash ORDER BY k;  -- sorted

 seg | k  |     | seg | k  
-----+----      -----+----
   0 |  2  |       1 |  1
   0 |  3  |        0 |  2
   0 |  4  |        0 |  3
   0 |  7  |        0 |  4
   0 |  8  |       2 |  5
   2 |  5  |        2 |  6
   2 |  6  |       0 |  7
   2 |  9  |        0 |  8
   2 | 10  |       2 |  9
   2 | 11  |        2 | 10
   1 |  1  |        2 | 11
   1 | 12  |       1 | 12
(12 rows)     (12 rows)

execMotionSortedReceiver (nodeMotion.c:433) implements it with lib/binaryheap.h over route numbers, not tuples. On the first call it primes the heap by calling RecvTupleFrom(..., iSegIdx) once for each live process in the sending slice, parking each tuple in a per-route slot and adding the route index to the heap unordered, then binaryheap_build once — cheaper than N inserts. Every later call is three lines of bookkeeping:

src/backend/executor/nodeMotion.c:574

		inputTuple = RecvTupleFrom(..., motion->motionID, node->routeIdNext);
		if (inputTuple)
		{
			ExecStoreMinimalTuple(inputTuple, node->slots[node->routeIdNext], true);
			slot_getsomeattrs(node->slots[node->routeIdNext], node->lastSortColIdx);
			binaryheap_replace_first(hp, Int32GetDatum(node->routeIdNext));
		}
		else
			binaryheap_remove_first(hp);   /* that sender hit EOS */

The comparator, CdbMergeComparator (nodeMotion.c:1039), reads the key columns straight out of the slots’ tts_values arrays — which is why the caller runs slot_getsomeattrs(slot, node->lastSortColIdx) first — and inverts the result with INVERT_COMPARE_RESULT because binaryheap is a max-heap and we want the minimum on top. When the heap empties, every sender has sent EOS and the node returns NULL. Structurally this is §18.6’s MergeAppend with the subplans replaced by network sources: same heap, same one-slot-per-input discipline, same amortised cost. The difference is where the inputs live — MergeAppend merges sorted subplans inside one process, a sorted Gather merges sorted processes.

The receiver-side prerequisite for all of this is set up much earlier: ExecInitMotion passes node->sendSorted to UpdateMotionLayerNode as preserveOrder (nodeMotion.c:898), which is what makes getChunkSorterEntry give every route its own htup_fifo instead of one shared queue (cdbmotion.c:917). Order preservation costs a queue per sender. And note the asymmetry the file’s own comment flags (nodeMotion.c:400-413): only the receiver has a sorted implementation. The senders of a sorted Gather run the ordinary unsorted path — their child already produced sorted output.

19.4.7 Stop, and end-of-stream

A Motion stream can end in two ways, and they are not symmetric. TC_END_OF_STREAM travels forward and means the sender is finished; a stop message travels backward and means the receiver has lost interest. §18.1 introduced ExecSquelchNode as the executor’s generic “stop early” traversal; the Motion node is where it crosses a process boundary.

On the receiving side, ExecSquelchMotion (nodeMotion.c:1389) sets stopRequested, calls SendStopMessage, and marks the node squelched. SendStopMessage sets pEntry->stopped in the Motion layer and then forwards through the vtable (cdbmotion.c:343-352); the transport’s job is to get the flag to the peer, which §19.5 covers. A squelched Motion that is somehow executed again raises an error rather than silently losing rows — ExecMotion checks for it at the very top (nodeMotion.c:114-119).

On the sending side there is no callback and no signal handler. The stop is discovered synchronously, as a return value: SendTupleChunkToAMS returns false, SendTuple converts that into STOP_SENDING (cdbmotion.c:494-498), doSendTuple sets node->stopRequested, and the sender loop squelches its own child subtree and quits — without sending EOS. That is the whole mechanism by which a satisfied LIMIT on the coordinator stops a scan on a segment, and it leaves a very recognisable fingerprint.

An EXISTS subplan whose Gather Motion needs exactly one row. Each segment holds around 100 000 rows and each began streaming them; the senders were stopped after about 1 351 rows apiece — roughly 1.4% of the table. That gap is the interconnect’s in-flight buffering, nothing more.

SET optimizer=off;
SELECT gp_segment_id, count(*) FROM m194_big GROUP BY 1 ORDER BY 1;
EXPLAIN (analyze, costs off, summary off)
  SELECT count(*) FROM m194_dim WHERE EXISTS (SELECT 1 FROM m194_big WHERE k > 0);

 gp_segment_id | count  
---------------+--------
             0 |  99877
             1 |  99951
             2 | 100172

                                         QUERY PLAN                                         
--------------------------------------------------------------------------------------------
 Aggregate (actual time=0.535..0.537 rows=1 loops=1)
   InitPlan 1 (returns $0)  (slice2)
     ->  Gather Motion 3:1  (slice3; segments: 3) (actual time=0.965..0.965 rows=1 loops=1)
           ->  Seq Scan on m194_big (actual time=0.063..0.253 rows=1351 loops=1)
                 Filter: (k > 0)
   ->  Gather Motion 3:1  (slice1; segments: 3) (actual time=0.445..0.524 rows=2 loops=1)
         ->  Result (actual time=0.056..0.057 rows=1 loops=1)
               One-Time Filter: $0
               ->  Seq Scan on m194_dim (actual time=0.055..0.056 rows=1 loops=1)

Per §18.1’s conventions those rows= figures are averages across the three segments, and this build is —enable-cassert, so treat 1 351 as an order of magnitude and not a constant — it is a function of how much the sender managed to push into the send queue before the stop came back. What matters is the shape: 1 row delivered, 1 351 rows scanned, 100 000 rows available. And because a stopped sender skips EOS, the receiver’s teardown has to tolerate a missing token — hence the !pMNEntry->stopped && !pCSEntry->end_of_stream guard in EndMotionLayerNode (cdbmotion.c:750). Without stopped in that condition, every LIMIT query would log a complaint.

19.4.8 What a Motion node actually reports

The Motion layer counts a lot: sends, receives, chunks, wire bytes, tuple bytes, queue high-watermarks, both globally and per node. It is natural to assume some of that reaches EXPLAIN ANALYZE. It does not — and that is worth stating plainly, because it shapes how you debug an interconnect problem.

EXPLAIN ANALYZE VERBOSE across two Motion nodes. Everything printed on the Motion lines is generic instrumentation — rows, loops, time — plus the plan-time Hash Key. No chunks, no bytes, no acks. The slice footer reports executor memory, not interconnect traffic.

SET optimizer=off;
EXPLAIN (analyze, verbose, costs off, timing off)
  SELECT k, count(*) FROM m194_rand GROUP BY k;

 Gather Motion 3:1  (slice1; segments: 3) (actual rows=12 loops=1)
   Output: k, (count(*))
   ->  HashAggregate (actual rows=5 loops=1)
         Output: k, count(*)
         Group Key: m194_rand.k
         work_mem: 72kB  Segments: 3  Max: 24kB (segment 0)  Workfile: (0 spilling)
         ->  Redistribute Motion 3:3  (slice2; segments: 3) (actual rows=5 loops=1)
               Output: k
               Hash Key: k
               ->  Seq Scan on public.m194_rand (actual rows=6 loops=1)
                     Output: k
   (slice0)    Executor memory: 114K bytes.
 * (slice1)    Executor memory: 122K bytes avg x 3x(0) workers, 122K bytes max (seg0).
   (slice2)    Executor memory: 112K bytes avg x 3x(0) workers, 112K bytes max (seg0).

The T_Motion case in explain.c:2853 does exactly three things: call show_motion_keys for a sorted or hashed Motion, print Hash Module when numHashSegments differs from the receiver count, and fall through to the generic cdbexplain_showExecStats. show_motion_keys (explain_gp.c:2251) deparses plan-time expressions — it never touches runtime counters. And grepping the tree confirms it: stat_total_chunks_sent, stat_total_bytes_sent and their siblings are written in cdbmotion.c and read nowhere else; MotionState’s own numTuplesFromChild / numTuplesToAMS / numTuplesFromAMS / numTuplesToParent are declared in execnodes.h:3475-3478 and used only for assertions and for an elog behind the compile-time MEASURE_MOTION_TIME macro (nodeMotion.c:951), which is #undef’d here.

counter	what it counts	where you can actually see it
stat_total_sends	SendTuple calls — plus one for each record-cache push, which is why a record column adds a phantom tuple	gp_log_interconnect = verbose, at node teardown: cdbmotion.c:800
stat_total_chunks_sent	TupleChunks handed to the transport, including the EOS chunk	same log line
stat_total_bytes_sent / stat_tuple_bytes_sent	wire bytes with and without the 4-byte chunk headers; inflated by 4 per direct-path send	same log line
stat_total_chunks_recvd / _bytes_recvd / stat_total_recvs	the receive-side mirror, reported against the sending slice number	gp_log_interconnect = verbose: cdbmotion.c:813
stat_tuples_available_hwm	how deep the reassembly queue got	needs gp_log_interconnect >= debug AND a DEBUG4 log level, order-preserving nodes only: cdbmotion.c:744
MotionLayerState global totals	query-wide chunk and byte totals	nowhere — behind the AMS_VERBOSE_LOGGING compile-time macro: cdbmotion.c:133
numTuplesFromChild / ToAMS / FromAMS / ToParent	the executor node’s own tuple counts	nowhere — asserts, plus MEASURE_MOTION_TIME: nodeMotion.c:951

So the practical answer to “how many bytes did this Motion move?” is: set gp_log_interconnect to verbose and read the segment logs. Every byte figure quoted in this section came from that one elog at cdbmotion.c:800, and it is the only production-visible accounting the Motion layer offers. Everything else — the packet headers those chunks are wrapped in, the sequence numbers, the acknowledgements, the loss-based flow control that decided how much of m194_big a stopped sender managed to push before the stop arrived — lives on the far side of SendTupleChunkToAMS, in the pluggable interconnect modules of §19.5.