Executor Nodes That Exist Only in Cloudberry

If a node in your EXPLAIN output does not appear in the PostgreSQL documentation, it is almost certainly in this chapter.

Everything in §18.1–§18.6 was upstream PostgreSQL’s executor, read in a Cloudberry context. The Plan→PlanState mirroring, the pull loop, nodeSeqscan.c, nodeAgg.c, nodeHashjoin.c, execExprInterp.c — same files, same algorithms, same struct names as PostgreSQL 16. What this section covers is the delta: the handful of operators that have no upstream counterpart at all, because they answer questions that only arise when a table is split across segments and no process may write to another process’s storage.

The delta is small and it is easy to count. ExecInitNode’s dispatch switch (execProcnode.c:236–:531) has 58 case T_… arms. 42 of them are PostgreSQL’s. The other 16 are Cloudberry’s, and — a nice accident of how the port was maintained — most of them sit in one contiguous block at the very end of the switch, after T_Limit and T_Motion (execProcnode.c:508–:531), as if somebody had appended a chapter to a finished book. Which is roughly what happened.

Three of the sixteen are taught elsewhere and only cross-referenced here. Motion is large enough, and interesting enough, to get Chapter 19 to itself together with the interconnect that carries it. The six Dynamic* scans belong to §18.2. TupleSplit belongs to §18.3. What is left is eight operators, and each of the next six subsections is organised around the problem, not the file.

18.7.1 Split Update: a row that has to change segments

A heap UPDATE in PostgreSQL is a local affair: mark the old tuple dead, write the new version, ideally on the same page (§4.3, §4.5). Now suppose the table is DISTRIBUTED BY (a) and you update a. The new row’s hash no longer maps to this segment. But a QE has no way to write into another segment’s files — there is no shared storage and no remote-write path. So the update cannot be an update. It has to become a DELETE here and an INSERT over there, and the tuple has to travel.

That decomposition is what Split Update (nodeSplitUpdate.c) does, and the planner decides whether it is needed from the target list alone — check_splitupdate(), memoised into root->is_split_update at preptlist.c:129. The contrast is visible in one pair of EXPLAINs:

Updating the distribution key gets a Split Update and a Motion; updating a payload column gets neither (GPORCA, the default on this cluster).

-- m187_t (a int, b int, c text) DISTRIBUTED BY (a), 1000 rows
EXPLAIN (COSTS OFF) UPDATE m187_t SET a = a + 1     WHERE b = 5;   -- touches the key
EXPLAIN (COSTS OFF) UPDATE m187_t SET c = c || 'x'  WHERE b = 5;   -- does not

                            QUERY PLAN
------------------------------------------------------------------
 Update on m187_t
   ->  Result
         One-Time Filter: true
         ->  Result
               ->  Redistribute Motion 3:3  (slice1; segments: 3)
                     Hash Key: a
                     ->  Split Update
                           ->  Seq Scan on m187_t
                                 Filter: (b = 5)
 Optimizer: GPORCA
(10 rows)

        QUERY PLAN
--------------------------
 Update on m187_t
   ->  Seq Scan on m187_t
         Filter: (b = 5)
 Optimizer: GPORCA
(4 rows)

The node itself is almost embarrassingly simple, and its simplicity is the point: it is a one-in, two-out operator. Each input tuple is expanded into two virtual tuples — one carrying the old column values plus DML_DELETE, one carrying the new values plus DML_INSERT — and the node hands them upward on alternate calls, tracked by a single bool:

src/backend/executor/nodeSplitUpdate.c:181

	/* Returns INSERT TupleTableSlot. */
	if (!node->processInsert)
	{
		result = node->insertTuple;

		node->processInsert = true;
	}
	else
	{
		/* Creates both TupleTableSlots. Returns DELETE TupleTableSlots.*/
		slot = ExecProcNode(outerNode);

SplitTupleTableSlot() (:72) fills the two slots by walking the target list; the DMLActionExpr entry is where the action code is stamped (:93), and evalHashKey() (:40) runs cdbhash over the new values to work out where the INSERT half must land. So the row count doubles exactly at this node — which EXPLAIN ANALYZE shows without ambiguity:

The row count doubles at Split Update, and the verbose target list shows the extra DMLAction column the node stamps on each half. (PostgreSQL-planner path; row counts are per-segment maxima — §18.1.)

SET optimizer = off;
EXPLAIN (COSTS OFF, VERBOSE) UPDATE m187_t SET a = a + 1 WHERE b = 5;
BEGIN;
EXPLAIN (COSTS OFF, ANALYZE, TIMING OFF) UPDATE m187_t SET a = a + 1 WHERE b = 5;
ROLLBACK;

                              QUERY PLAN
-----------------------------------------------------------------------
 Update on public.m187_t
   ->  Explicit Redistribute Motion 3:3  (slice1; segments: 3)
         Output: ((a + 1)), b, c, ctid, gp_segment_id, (DMLAction)
         ->  Split Update
               Output: ((a + 1)), b, c, ctid, gp_segment_id, DMLAction
               ->  Seq Scan on public.m187_t
                     Output: (a + 1), b, c, ctid, gp_segment_id
                     Filter: (m187_t.b = 5)
 Optimizer: Postgres query optimizer

                                       QUERY PLAN
----------------------------------------------------------------------------------------
 Update on m187_t (actual rows=0 loops=1)
   ->  Explicit Redistribute Motion 3:3  (slice1; segments: 3) (actual rows=78 loops=1)
         ->  Split Update (actual rows=84 loops=1)
               ->  Seq Scan on m187_t (actual rows=42 loops=1)
                     Filter: (b = 5)
                     Rows Removed by Filter: 298
 Optimizer: Postgres query optimizer

Notice that the two optimizers put a different Motion above the node, and for a reason worth knowing. ORCA emits a plain Redistribute Motion with Hash Key: a and lets the Motion compute the target segment; the PostgreSQL planner emits an Explicit Redistribute Motion, which reads a target segment number straight out of the tuple’s gp_segment_id junk column — the number evalHashKey() computed. The comment at createplan.c:3569 spells the divergence out, and explains why the planner leaves deleteColIdx filled with -1 while ORCA populates it with the old key values (its Redistribute Motion needs them, to send the DELETE half back to the segment that owns the old row).

And the payoff, which is the entire reason the node exists — a row physically changing owner. The QD took its RowExclusiveLock on the table before any of this (§11.1), and both halves land inside one distributed transaction (§9.2, §10.5), so at no point is the row visible twice:

Row v11 leaves segment 2 and arrives on segment 1. That is a DELETE on one QE and an INSERT on another, inside one transaction.

BEGIN;
SELECT gp_segment_id, a, c FROM m187_t WHERE a = 11;
UPDATE m187_t SET a = a + 1 WHERE a = 11;
SELECT gp_segment_id, a, c FROM m187_t WHERE c = 'v11';
ROLLBACK;

 gp_segment_id | a  |  c
---------------+----+-----
           2 | 11 | v11
(1 row)

UPDATE 1
 gp_segment_id | a  |  c
---------------+----+-----
           1 | 12 | v11
(1 row)

MERGE has the same problem, one degree harder: a single statement mixes INSERT, UPDATE and DELETE actions decided per row, so some rows must move and others must not. That is nodeSplitMerge.c, and it is the bigger file of the pair (662 lines against 292). It carries a per-row action column with a third value, SPLITMERGE_ACTION_PASSTHROUGH = -1 (:34), meaning “leave this one to ModifyTable’s ordinary ExecMerge path”, and only splits into DELETE+INSERT for matched rows when hasSplitUpdate is set (:397). The planner decides that in preptlist.c:147–:156, by running the same check_splitupdate() over every WHEN MATCHED THEN UPDATE action list. ORCA does not participate at all: it has no MERGE translation, so every MERGE falls back.

MERGE falls back to the PostgreSQL planner, which produces Split Merge under an Explicit Redistribute Motion. Matched rows whose key changes really do move.

SET optimizer_trace_fallback = on;
EXPLAIN (COSTS OFF) MERGE INTO m187_t t USING m187_src s ON t.a = s.a
  WHEN MATCHED     THEN UPDATE SET a = s.a + 10000
  WHEN NOT MATCHED THEN INSERT (a,b,c) VALUES (s.a, s.b, 'new');

INFO:  GPORCA failed to produce a plan, falling back to Postgres-based planner
DETAIL:  Falling back to Postgres-based planner because GPORCA does not
         support the following feature: MERGE command
                          QUERY PLAN
---------------------------------------------------------------
 Merge on m187_t t
   ->  Explicit Redistribute Motion 3:3  (slice1; segments: 3)
         ->  Split Merge
               ->  Hash Right Join
                     Hash Cond: (t.a = s.a)
                     ->  Seq Scan on m187_t t
                     ->  Hash
                           ->  Seq Scan on m187_src s
 Optimizer: Postgres query optimizer
(9 rows)

18.7.2 Assert: proving at run time what SQL promised at parse time

SELECT (SELECT c FROM t WHERE b = 7) FROM t contains a scalar subquery, and SQL says a scalar subquery returning more than one row is an error. PostgreSQL keeps that guarantee structurally: the subquery becomes a SubPlan/InitPlan (§18.6) and ExecSetParamPlan raises the error the moment it fetches a second row. ORCA does not keep the structure. Decorrelation (§17.3) rewrites the subquery into a join, and a join is perfectly happy to produce several rows — the cardinality guarantee evaporates in the rewrite.

ORCA’s answer is to re-materialise the guarantee as an operator. It numbers the subquery’s rows with a window function and inserts an Assert node that checks the number as the tuples stream past:

ORCA turns “at most one row” into a predicate over row_number() and an Assert node to evaluate it.

EXPLAIN (COSTS OFF) SELECT (SELECT c FROM m187_t WHERE b = 7) FROM m187_t LIMIT 1;

                                        QUERY PLAN
------------------------------------------------------------------------------------------
 Limit
   ->  Gather Motion 3:1  (slice1; segments: 3)
         ->  Limit
               ->  Nested Loop Left Join
                     Join Filter: true
                     ->  Seq Scan on m187_t m187_t_1
                     ->  Assert
                           Assert Cond: ((row_number() OVER (?)) = 1)
                           ->  Materialize
                                 ->  Broadcast Motion 1:3  (slice2)
                                       ->  WindowAgg
                                             ->  Gather Motion 3:1  (slice3; segments: 3)
                                                   ->  Seq Scan on m187_t
                                                         Filter: (b = 7)
 Optimizer: GPORCA
(15 rows)

The node is 151 lines and does exactly one thing per tuple: evaluate ps.qual, and if it comes back false, ereport. What makes it general rather than a hard-wired cardinality check is that both the SQLSTATE and the message text travel in the plan: ORCA’s DXL assert operator carries a five-character SQL state which TranslateDXLAssert turns into an errcode at CTranslatorDXLToPlStmt.cpp:5718, and the message list comes from the DXL constraint. So one C node serves every runtime assertion ORCA can express.

src/backend/executor/nodeAssertOp.c:46

	if (!isNull && !DatumGetBool(expr_value))
	{
		String *valErrorMessage = list_nth(plannode->errmessage, 0);
		...
		appendStringInfo(&errorString, "%s\n", strVal(valErrorMessage));

		ereport(ERROR,
				(errcode(plannode->errcode),
				 errmsg("one or more assertions failed"),
				 errdetail("%s", errorString.data)));
	}

Which is much more convincing if you make it fire. b = 7 matches 100 rows, so the subquery is illegal; drop the LIMIT so that the tuples actually flow, and run the two optimizers side by side:

The same illegal query, two optimizers, two entirely different error paths: an Assert node on a segment (P0003) versus the SubPlan machinery on the QD (21000).

\set VERBOSITY verbose
SELECT (SELECT c FROM m187_t WHERE b = 7) FROM m187_t;      -- optimizer = on
SET optimizer = off;
SELECT (SELECT c FROM m187_t WHERE b = 7) FROM m187_t;      -- same query

ERROR:  P0003: one or more assertions failed  (seg1 slice1 10.100.0.3:7103 pid=22587)
DETAIL:  Expected no more than one row to be returned by expression
LOCATION:  CheckForAssertViolations, nodeAssertOp.c:55

SET
ERROR:  21000: more than one row returned by a subquery used as an expression
LOCATION:  ExecSetParamPlan, nodeSubplan.c:1322

Read the two lines carefully, because the difference is not cosmetic. ORCA’s error is P0003 (ERRCODE_TOO_MANY_ROWS, errcodes.h:360 — the code PL/pgSQL uses for STRICT violations) and it is raised on segment 1, by an operator, in the middle of a data flow. The planner’s is 21000 (cardinality_violation) and it is raised on the QD, by control logic in nodeSubplan.c. Same SQL, same violation, different SQLSTATE and different reporter — worth knowing if you have application code that switches on error codes.

An honest footnote from this cluster: with the LIMIT 1 still in place, the ORCA plan does not raise the error at all — it returns a row, where the planner path errors. Nothing is broken in the node; the Limit above it simply stops pulling after the first tuple, so the second row that would have carried row_number() = 2 is never fetched, and an operator can only assert about tuples it sees. It is a small, precise illustration of what “volcano, demand-pull” (§18.1) actually costs you: a check implemented as a node is only as complete as the demand above it.

18.7.3 ShareInputScan and Sequence: compute once, read from several processes

A CTE referenced twice, or ORCA’s multi-DQA rewrite, produces a plan with one expensive subtree and two or more consumers. Materialize (§18.6) solves the single-consumer, single-process version of this. It cannot solve the Cloudberry version, because in an MPP plan the consumers may live in different slices — that is, in different QE processes — and a private tuplestore in one backend’s memory is invisible to another. ShareInputScan (nodeShareInputScan.c, 1064 lines, the largest of the native nodes) exists for exactly that gap, and it comes in two flavours which the plan node distinguishes with a single bool cross_slice (plannodes.h:1307).

Start with the easy flavour. All three Shared Scan nodes below report the same share slice:id 1:0, and explain.c:2245 tells you how to read that: the first number is the slice the EXPLAIN walk is currently in, the second is sisc->share_id. Same slice on all three ⇒ one process ⇒ a local share, no IPC at all:

A CTE read twice under ORCA: Sequence over a producer Shared Scan and two consumer Shared Scans, all in slice 1. The aggregate runs once.

EXPLAIN (COSTS OFF, ANALYZE, TIMING OFF)
WITH c AS (SELECT b, count(*) n FROM m187_t GROUP BY b)
SELECT * FROM c x JOIN c y ON x.b = y.b;

 Gather Motion 3:1  (slice1; segments: 3) (actual rows=10 loops=1)
   ->  Sequence (actual rows=5 loops=1)
         ->  Shared Scan (share slice:id 1:0) (actual rows=0 loops=1)
               ->  Finalize HashAggregate (actual rows=5 loops=1)
                     Group Key: m187_t.b
                     ->  Redistribute Motion 3:3  (slice2; segments: 3) (actual rows=15 loops=1)
                           Hash Key: m187_t.b
                           ->  Streaming Partial HashAggregate (actual rows=10 loops=1)
                                 ->  Seq Scan on m187_t (actual rows=340 loops=1)
         ->  Hash Join (actual rows=5 loops=1)
               Hash Cond: (share0_ref3.b = share0_ref2.b)
               ->  Shared Scan (share slice:id 1:0) (actual rows=5 loops=1)
               ->  Hash (actual rows=5 loops=1)
                     ->  Shared Scan (share slice:id 1:0) (actual rows=5 loops=1)
 Optimizer: GPORCA

Two details in that output repay a second look. The consumers are named share0_ref2 and share0_ref3 in the hash condition — ORCA’s aliases for “reference 2 and 3 of share 0”. And the producer reports actual rows=0 while each consumer reports 5. That is not a bug: the producer’s job is to fill the store, not to emit, so discard_output is set for an ORCA CTE producer (plannodes.h:1333) and ExecShareInputScan returns NULL immediately after materialising (nodeShareInputScan.c:376). The five aggregate rows were computed once and read twice.

Now change the join key so that each consumer needs its own redistribution, and the shape changes completely. The producer stays in slice 1; the consumers move to slices 5 and 6. Three processes, one tuplestore:

Joining the CTE on a non-key column pushes the two consumers into their own slices — a genuine cross-slice share.

EXPLAIN (COSTS OFF)
WITH c AS (SELECT b, count(*) n FROM m187_t GROUP BY b)
SELECT count(*) FROM c x JOIN c y ON x.n = y.n;

         ->  Sequence
               ->  Shared Scan (share slice:id 1:0)      <-- producer
                     ->  Finalize HashAggregate
                           ...
               ->  Redistribute Motion 1:3  (slice3)
                     ->  Finalize Aggregate
                           ->  Gather Motion 3:1  (slice4; segments: 3)
                                 ->  Partial Aggregate
                                       ->  Hash Join
                                             Hash Cond: (share0_ref3.n = share0_ref2.n)
                                             ->  Redistribute Motion 3:3  (slice5; segments: 3)
                                                   ->  Shared Scan (share slice:id 5:0)
                                             ->  Hash
                                                   ->  Redistribute Motion 3:3  (slice6; segments: 3)
                                                         ->  Shared Scan (share slice:id 6:0)
 Optimizer: GPORCA

Cross-process sharing needs three things the local case does not: somewhere to put the tuples that another process can open, a way to say ready, and a way to say done reading so the producer knows when it may delete the files. Cloudberry’s answers are, in order: a BufFile in a SharedFileSet, named SIRW_⟨session_id⟩_⟨command_count⟩_⟨share_id⟩ (nodeShareInputScan.c:692); a shareinput_Xslice_state entry in a shared-memory hash table (:101) holding two atomics, ready and ndone; and one ConditionVariable per entry, ready_done_cv, used in both directions.

You do not have to take that diagram on faith. debug_shareinput_xslice makes every step of the handshake log, and with client_min_messages = log the QE lines come back over dispatch, tagged with segment, slice and pid:

The handshake, live, trimmed to one segment’s worth of lines. Consumers block before the producer has even created the store; the producer’s notify_ready releases all of them; nconsumers = 2 releases the producer.

SET debug_shareinput_xslice = on;
SET client_min_messages = log;
WITH c AS (SELECT b, count(*) n FROM m187_t GROUP BY b)
SELECT count(*) FROM c x JOIN c y ON x.n = y.n;

LOG:  SISC (shareid=0, slice=5): initialized xslice state         (seg0 slice5 pid=25319)
LOG:  SISC READER (shareid=0, slice=5): Waiting for producer      (seg0 slice5 pid=25319)
LOG:  SISC READER (shareid=0, slice=6): Waiting for producer        (seg0 slice6 pid=25325)
LOG:  SISC WRITER (shareid=0, slice=1): No tuplestore yet, creating tuplestore  (seg0 slice1 pid=25291)
LOG:  SISC WRITER (shareid=0, slice=1): wrote notify_ready        (seg0 slice1 pid=25291)
LOG:  SISC READER (shareid=0, slice=5): Wait ready got writer's handshake       (seg0 slice5 pid=25319)
LOG:  SISC READER (shareid=0, slice=6): Wait ready got writer's handshake       (seg0 slice6 pid=25325)
LOG:  SISC READER (shareid=0, slice=5): wrote notify_done           (seg0 slice5 pid=25319)
LOG:  SISC READER (shareid=0, slice=6): wrote notify_done           (seg0 slice6 pid=25325)
LOG:  SISC WRITER (shareid=0, slice=1): got DONE message from 2 readers       (seg0 slice1 pid=25291)
LOG:  SISC (shareid=0, slice=1): decreased xslice state refcount to 0           (seg0 slice1 pid=25291)
LOG:  SISC (shareid=0, slice=1): removed xslice state              (seg0 slice1 pid=25291)

src/backend/executor/nodeShareInputScan.c:940

static void
shareinput_reader_waitready(shareinput_Xslice_reference *ref)
{
	shareinput_Xslice_state *state = ref->xslice_state;

	for (;;)
	{
		/* pg_atomic_exchange_u32() in the writer acts as a memory barrier */
		int ready = pg_atomic_read_u32(&state->ready);
		if (ready)
			break;
		ConditionVariableSleep(&state->ready_done_cv, WAIT_EVENT_SHAREINPUT_SCAN);
	}
	ConditionVariableCancelSleep();
}

None of this would work under the ordinary demand-pull discipline, because nothing above the producer ever asks it for a tuple — the consumers are elsewhere in the tree. That is what Sequence (nodeSequence.c, 168 lines) contributes, and why it always appears above a shared scan. It is the one node in the executor that runs its children eagerly and in order: on first call it drives every subplan except the last to exhaustion with completeSubplan(), throwing the output away, and only then starts pulling from the last one (:100–:115). Producer first, consumers after; and if a consumer’s slice starts before the producer has finished, shareinput_reader_waitready holds it. Sequence is not exclusive to sharing — §18.2 uses it to sequence a PartitionSelector ahead of a Dynamic* scan — and Chapter 19 explains how those slices come to be running at once in the first place.

18.7.4 Partition Selector: pruning decided while the query runs

Static partition pruning happens in the planner: a constant in the WHERE clause eliminates partitions before the plan is finished, and Append never even lists them (§18.6). But the interesting case in a star schema is not a constant — it is a join. WHERE dim.proj_id = 1 selects a handful of dimension rows, whose did values determine which fact partitions can possibly match, and nobody knows those values until the dimension side has been read. Cloudberry solves this with a pair of cooperating nodes: a PartitionSelector on the inner (hash-build) side, which watches every row go past and accumulates the set of partitions it implies, and a partition-aware scan on the outer side that consults the result. Both optimizers can build this shape, and they build it differently.

PostgreSQL-planner form: Partition Selector on the build side publishes into $0; the Append on the probe side reads it and leaves four of five partitions never executed.

SET optimizer = off;
EXPLAIN (COSTS OFF, ANALYZE, TIMING OFF)
SELECT count(*) FROM m187_fact f, m187_dim d WHERE f.did = d.did AND d.proj_id = 1;

               ->  Hash Join (actual rows=16772 loops=1)
                     Hash Cond: (f.did = d.did)
                     ->  Append (actual rows=16772 loops=1)
                           Partition Selectors: $0
                           ->  Seq Scan on m187_fact_1_prt_2 f_1 (never executed)
                           ->  Seq Scan on m187_fact_1_prt_3 f_2 (actual rows=16772 loops=1)
                           ->  Seq Scan on m187_fact_1_prt_4 f_3 (never executed)
                           ->  Seq Scan on m187_fact_1_prt_5 f_4 (never executed)
                           ->  Seq Scan on m187_fact_1_prt_other f_5 (never executed)
                     ->  Hash (actual rows=10 loops=1)
                           ->  Partition Selector (selector id: $0) (actual rows=10 loops=1)
                                 ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=10 loops=1)
                                       ->  Seq Scan on m187_dim d (actual rows=5 loops=1)
                                             Filter: (proj_id = 1)
                                             Rows Removed by Filter: 8
 Optimizer: Postgres query optimizer

Same query, same Partition Selector — but ORCA pairs it with a Dynamic Seq Scan (§18.2), which reports the surviving partition count per segment instead of marking children never executed.

SET optimizer = on;
EXPLAIN (COSTS OFF, ANALYZE, TIMING OFF)
SELECT count(*) FROM m187_fact f, m187_dim d WHERE f.did = d.did AND d.proj_id = 1;

               ->  Hash Join (actual rows=16772 loops=1)
                     Hash Cond: (f.did = d.did)
                     ->  Dynamic Seq Scan on m187_fact f (actual rows=16772 loops=1)
                           Number of partitions to scan: 5 (out of 5)
                           Partitions scanned:  Avg 1.0 x 3 workers.  Max 1 parts (seg0).
                     ->  Hash (actual rows=10 loops=1)
                           ->  Partition Selector (selector id: $0) (actual rows=10 loops=1)
                                 ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=10 loops=1)
                                       ->  Seq Scan on m187_dim d (actual rows=5 loops=1)
                                             Filter: (proj_id = 1)
 Optimizer: GPORCA

Read the two instrumentation styles as the same fact stated twice: the planner’s Append shows four never executed children, while ORCA’s Dynamic Seq Scan shows Number of partitions to scan: 5 (out of 5) — the plan-time count, all five still possible — followed by what actually happened, Max 1 parts (seg0). One partition out of five was opened. The (out of 5) figure comes from countLeafPartTables() at EXPLAIN time (explain.c:2561), so it is a statement about the plan, not the run; the Partitions scanned line is the run.

The channel between the two nodes is an ordinary executor Param slot used as a mailbox. The selector prunes per tuple, accumulating into part_prune_result, and only when its input is exhausted — when the set is final — does it publish a pointer to itself into es_param_exec_vals[paramid]. The consumer, Append or Dynamic*, dereferences that Param instead of running pruning steps of its own. It works only because the hash side is guaranteed to be fully drained before the probe side is read, which is exactly the ordering a hash join gives you for free (§18.4):

src/backend/executor/nodePartitionSelector.c:132

	if (TupIsNull(inputSlot))
	{
		/* The set of surviving partitions is now complete. */
		ParamExecData *param;

		param = &(estate->es_param_exec_vals[ps->paramid]);
		param->value = PointerGetDatum(node);
		return NULL;
	}
	econtext->ecxt_outertuple = inputSlot;
	node->part_prune_result = ExecAddMatchingSubPlans(node->prune_state,
													  node->part_prune_result);

Both optimizers emit this node, which is unusual for the native set. The PostgreSQL-planner path lives in joinpartprune.c — a wholly Cloudberry file inside the upstream planner — and its 70-line header comment is the clearest description of the mechanism anywhere in the tree, including the three-phase construction during create_plan(): push_partition_selector_candidate_for_join(), then make_partition_join_pruneinfos() when the Append is built, then pop_and_inject_partition_selectors(). ORCA reaches the same node through CTranslatorDXLToPlStmt.cpp:531.

18.7.5 RuntimeFilter: a Bloom filter in front of the fact table

The partition selector prunes at the granularity of a whole partition. RuntimeFilter (nodeRuntimeFilter.c) prunes at the granularity of a row, and for the same reason: in a star join, the overwhelming majority of fact rows will never match the small, filtered dimension side, yet each one is read, formed into a slot, hashed and probed. If the hash table is already built, we can answer “is this key definitely absent?” far more cheaply than a probe — with a Bloom filter. That is the whole node: 327 lines wrapping the outer input of a Hash Join, rejecting rows before the join sees them.

It is off by default (gp_enable_runtime_filter = off) and it is a PostgreSQL-planner-only node — the decision is made in try_runtime_filter(), called from joinpath.c:2332 when hash clauses are found. Turning it on rewrites the plan:

One GUC, one extra node. m187_frf has 100 000 rows, m187_drf 10 000, and the dimension predicate keeps 20% of them.

SET optimizer = off;                         -- required: ORCA never emits this node
SET gp_enable_runtime_filter = off;
EXPLAIN (COSTS OFF) SELECT COUNT(*) FROM m187_frf, m187_drf
  WHERE m187_frf.did = m187_drf.did AND proj_id < 2;
SET gp_enable_runtime_filter = on;
EXPLAIN (COSTS OFF) SELECT COUNT(*) FROM m187_frf, m187_drf
  WHERE m187_frf.did = m187_drf.did AND proj_id < 2;

               ->  Hash Join
                     Hash Cond: (m187_frf.did = m187_drf.did)
                     ->  Seq Scan on m187_frf
                     ->  Hash
                           ->  Broadcast Motion 3:3  (slice2; segments: 3)
                                 ->  Seq Scan on m187_drf
                                       Filter: (proj_id < 2)
 Optimizer: Postgres query optimizer

               ->  Hash Join
                     Hash Cond: (m187_frf.did = m187_drf.did)
                     ->  RuntimeFilter
                           ->  Seq Scan on m187_frf
                     ->  Hash
                           ->  Broadcast Motion 3:3  (slice2; segments: 3)
                                 ->  Seq Scan on m187_drf
                                       Filter: (proj_id < 2)
 Optimizer: Postgres query optimizer

The same plan, executed. The scan below the filter produces 33 462 rows; only 6 682 reach the join — 80% rejected without a probe. (Ratios on one —enable-cassert build, not a benchmark.)

EXPLAIN (COSTS OFF, ANALYZE, TIMING OFF) SELECT COUNT(*) FROM m187_frf, m187_drf
  WHERE m187_frf.did = m187_drf.did AND proj_id < 2;

               ->  Hash Join (actual rows=6682 loops=1)
                     Hash Cond: (m187_frf.did = m187_drf.did)
                     Extra Text: (seg1)   Hash chain length 1.0 avg, 2 max, using 1996 of 524288 buckets.
                     ->  RuntimeFilter (actual rows=6682 loops=1)
                           Bloom Bits: 1048576
                           Extra Text: (seg1)   Inner Processed: 2000, Flase Positive Rate: 0.000000
                           ->  Seq Scan on m187_frf (actual rows=33462 loops=1)
                     ->  Hash (actual rows=2000 loops=1)
                           Buckets: 524288  Batches: 1  Memory Usage: 4167kB
                           ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=2000 loops=1)
                                 ->  Seq Scan on m187_drf (actual rows=689 loops=1)
                                       Filter: (proj_id < 2)
                                       Rows Removed by Filter: 2679
 Optimizer: Postgres query optimizer

Bloom Bits: 1048576 is not a coincidence, and neither is the sizing rule behind it. ExecInitRuntimeFilterFinish is called from ExecInitHashJoin (nodeHashjoin.c:1113) with the estimated inner row count, and bloom_create_aggresive (bloomfilter.c:325) clamps the bitset to [128 kB, 2 MB], rounds down to a power of two, and then refuses outright — returns NULL — if the result gives fewer than 1.6 bits per element. It picks 3 hash functions when it can afford 3.5 bits per element, otherwise 2. This is a deliberately cheap filter: small enough to stay in cache, sized to be fast rather than accurate, and paid for by the join re-checking every survivor anyway.

src/backend/executor/nodeRuntimeFilter.c:242

	node->bf = bloom_create_aggresive((int64) inner_rows, work_mem, random());
	if (node->bf == NULL)
		node->build_suspend = true;
	/* we can't handle inner table which is bigger than this threshold */
	node->inner_threshold = bloom_total_bits(node->bf) / 1.6;

Which leads to the part of the design most worth copying: the node distrusts its own estimate and can switch itself off. inner_threshold is a runtime cap of bits / 1.6; every inner row added bumps inner_processed, and the moment it crosses the cap the node sets build_suspend = true (:311). ExecRuntimeFilter then degenerates into a pass-through — if (!node->build_finish || node->build_suspend) return ExecProcNode(outerPlan); (:60) — so a badly underestimated hash side costs a wasted Bloom build and nothing more. EXPLAIN ANALYZE prints Suspend in that case (:216). Note also the two deliberate holes: NULL keys are passed through unfiltered (:132), and only bloom_lacks_element is trusted — a hit means “maybe”, so every survivor still gets a real probe.

The planner’s own judgement is worth exercising too, because it declines more often than it accepts. try_runtime_filter (costsize.c:5613) bails out on four separate grounds, and the gp_runtime_filter regression test is built around them:

Three refusals. A less selective dimension predicate, a non-eliminable LEFT JOIN, and ORCA — none of them get a RuntimeFilter, with gp_enable_runtime_filter still on.

SET gp_enable_runtime_filter = on;   SET optimizer = off;
EXPLAIN (COSTS OFF) SELECT COUNT(*) FROM m187_frf, m187_drf
  WHERE m187_frf.did = m187_drf.did AND proj_id < 7;              -- bad filter rate
EXPLAIN (COSTS OFF) SELECT COUNT(*) FROM m187_frf LEFT JOIN m187_drf
  ON m187_frf.did = m187_drf.did WHERE proj_id IS NULL OR proj_id < 2;
SET optimizer = on;
EXPLAIN (COSTS OFF) SELECT COUNT(*) FROM m187_frf, m187_drf
  WHERE m187_frf.did = m187_drf.did AND proj_id < 2;              -- the winning shape

               ->  Hash Join                              <-- no RuntimeFilter
                     Hash Cond: (m187_frf.did = m187_drf.did)
                     ->  Seq Scan on m187_frf
                     ...  Filter: (proj_id < 7)

               ->  Hash Left Join                        <-- no RuntimeFilter
                     Hash Cond: (m187_frf.did = m187_drf.did)
                     Filter: ((m187_drf.proj_id IS NULL) OR (m187_drf.proj_id < 2))
                     ->  Seq Scan on m187_frf

               ->  Hash Join                              <-- no RuntimeFilter
                     Hash Cond: (m187_frf.did = m187_drf.did)
                     ->  Seq Scan on m187_frf
 Optimizer: GPORCA

The four bail-outs, in the order the function applies them. Join type: only INNER, RIGHT and SEMI (:5632) — a build-side Bloom filter can only tell you a row has no match, which for a left or anti join means early output, not early rejection, and that is not implemented (the comment at nodeHashjoin.c:2309 makes the argument). Absolute size: if the filter would spare fewer than 10 000 outer rows it is not worth the cycles (:5673). Selectivity: the surviving row count must be under RUNTIME_FILTER_RATE_THRESHOLD = 0.6 of the outer rows — it has to reject at least 40% (:5681, costsize.c:109) — which is what proj_id < 7 fails. By-reference types are refused outright (:5658), since the node hashes fixed-width Datums. And the third refusal is the categorical one: ORCA has no runtime-filter translation, so on this cluster it never emits the node, whatever the GUC says.

There is a second, quite separate mechanism hiding behind a similar name. gp_enable_runtime_filter_pushdown adds no plan node at all: at ExecInitHashJoin time, on the QE only, CreateRuntimeFilter (nodeHashjoin.c:2296) walks down the outer side looking for a SeqScanState or DynamicSeqScanState and hangs an AttrFilter — its own Bloom filter — directly on the scan, which then tests keys inside its own tuple loop, before projection (nodeSeqscan.c:83). The filtering moves one level lower, and the accounting moves with it:

Pushdown instead of a node: the Seq Scan itself now reports 6 682 rows out and its own removal counter. Compare with the previous panel, where the scan still emitted 33 462.

SET gp_enable_runtime_filter = off;            -- no RuntimeFilter node
SET gp_enable_runtime_filter_pushdown = on;   -- filter inside the scan instead
EXPLAIN (COSTS OFF, ANALYZE, TIMING OFF) SELECT COUNT(*) FROM m187_frf, m187_drf
  WHERE m187_frf.did = m187_drf.did AND proj_id < 2;

               ->  Hash Join (actual rows=6682 loops=1)
                     Hash Cond: (m187_frf.did = m187_drf.did)
                     ->  Seq Scan on m187_frf (actual rows=6682 loops=1)
                           Rows Removed by Pushdown Runtime Filter: 26645
                     ->  Hash (actual rows=2000 loops=1)
                           Buckets: 524288  Batches: 1  Memory Usage: 4167kB
                           ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=2000 loops=1)
                                 ->  Seq Scan on m187_drf (actual rows=689 loops=1)
                                       Filter: (proj_id < 2)
 Optimizer: Postgres query optimizer

The two do not compose, and the reason is structural rather than intentional: FindTargetNodes (nodeHashjoin.c:2488) only descends through HashJoinState, ResultState and AppendState, so a RuntimeFilter sitting between the join and the scan stops the search dead. With both GUCs on, the plan gets its RuntimeFilter node and no pushdown — verified on this cluster. Pushdown is also skipped entirely under intra-segment parallelism (estate->useMppParallelMode, :1118). Which gives four sibling mechanisms for “the scan learns something at run time”, worth laying side by side:

18.7.6 Table Function Scan: a function whose argument is a table

The last of the eight is the oldest and the least used. Cloudberry supports functions declared over the pseudo-type anytable, called as f( TABLE(subquery SCATTER BY …) ). Because the argument is a whole relation rather than a scalar, the parser converts the range-table entry to RTE_TABLEFUNCTION (parse_relation.c:1925) and the plan gets a TableFunctionScan with the subquery as its child — the distinction nodeTableFunction.c’s header draws between a normal table function (a FunctionScan, no input) and an enhanced one (a TableFunctionScan over a subplan). The function body pulls its input through the AnyTable handle in tablefuncapi.h, so this is a C-only interface in practice; src/test/regress/regress.so has the reference implementations.

An enhanced table function over m187_t. SCATTER BY is the MPP half of the feature: it decides how the input is distributed to the QEs that will run the function body.

SET compute_query_id = off;   -- see note
CREATE FUNCTION m187_multiset(a anytable) RETURNS TABLE(a int, b text)
  AS '.../src/test/regress/regress.so', 'multiset_example'
  LANGUAGE C READS SQL DATA;
EXPLAIN (COSTS OFF) SELECT * FROM
  m187_multiset( TABLE(SELECT a, c FROM m187_t WHERE b = 3 SCATTER BY a) );
EXPLAIN (COSTS OFF) SELECT count(*) FROM
  m187_multiset( TABLE(SELECT a, c FROM m187_t WHERE b = 3 SCATTER BY c) );

                 QUERY PLAN
--------------------------------------------
 Gather Motion 3:1  (slice1; segments: 3)
   ->  Table Function Scan on m187_multiset
         ->  Seq Scan on m187_t
               Filter: (b = 3)

 Finalize Aggregate
   ->  Gather Motion 3:1  (slice1; segments: 3)
         ->  Partial Aggregate
               ->  Table Function Scan on m187_multiset
                     ->  Redistribute Motion 3:3  (slice2; segments: 3)
                           Hash Key: m187_t.c
                           ->  Seq Scan on m187_t
                                 Filter: (b = 3)
 Optimizer: Postgres query optimizer

SCATTER BY a asks for the input grouped by the distribution key, which it already is, so the function runs on three segments over local data and no Motion appears. SCATTER BY c asks for it grouped by something else, and a Redistribute Motion materialises to satisfy the request. That is the whole reason the clause exists: a table function that computes something per group needs each group to arrive complete at one QE, and SCATTER BY is how the query author states which grouping the function body assumes.

A real bug found while writing this section, reported honestly: with the cluster’s default compute_query_id and pg_stat_statements loaded, any query containing a TABLE(…) argument fails with ERROR: unrecognized RTE kind: 10 (queryjumble.c:324). Upstream’s query-jumbling code has no arm for RTE_TABLEFUNCTION. SET compute_query_id = off works around it for the session. The node itself is fine — the plan and the result above are real.

18.7.7 Closing the chapter

That is the whole delta. Forty-two of the executor’s fifty-eight node types are upstream PostgreSQL’s, running unchanged; §18.1–§18.6 described them, and the only thing Cloudberry contributed there was context — one executor per slice, per segment, all at once, comparing its actuals against estimates that were made on a different machine. The sixteen native nodes exist for reasons you can now state in one sentence each: a row that must change owner (Split Update, Split Merge), a guarantee that decorrelation destroyed (Assert), a subtree several processes must read (ShareInputScan, with Sequence to make it run first), a pruning decision that could not be made until run time (Partition Selector, and §18.2’s Dynamic* scans), a row rejected before it costs a probe (RuntimeFilter), a function whose argument is a distributed relation (TableFunctionScan), and §18.3’s two-stage TupleSplit.

Every one of them, notice, is downstream of a single fact: the data is spread across segments, and a segment can only touch its own. Which leaves the sixteenth, and the largest — Motion, the operator that does the actual moving. nodeMotion.c is 1 405 lines and it has appeared in almost every plan in this chapter, quietly, as Gather Motion 3:1, Redistribute Motion 3:3, Broadcast Motion 1:3, Explicit Redistribute Motion. We have read those labels as annotations. Chapter 19 reads them as machinery: what a slice is, how a gang of QEs is assembled to run one, how a tuple is serialised and put on the wire, what the interconnect does when the receiver is slower than the sender, and why a plan with two Motions feeding one join can deadlock unless somebody prefetches (§18.4) or squelches (§18.1). The plan is a prediction; the executor tests it; the interconnect is what makes the test distributed.