Join Execution: Nested Loop, Hash and Merge Joins

Three join algorithms, one of which will be responsible for your worst query. Here is how each behaves under memory pressure.

Three algorithms, one job: take two subtrees and produce every pair of tuples satisfying a set of clauses. The plan already chose the algorithm, and §15.3 and §17.3 covered the cost reasoning behind that choice. What the executor adds is what the cost model only models — a concrete control flow, a concrete memory footprint, and, in Cloudberry, a concrete hazard, because either input may be arriving over a Motion from three segments at once. Everything below turns on one asymmetry: the inner side is the side that gets re-read or buffered, while the outer side is streamed exactly once. Nested loop rescans the whole inner subtree per outer tuple; merge join walks two sorted streams in lockstep but must be able to back the inner one up; hash join reads the inner side to completion and keeps it. That is why almost every knob in this section — Materialize, batch files, prefetch — is about the inner side. And a Cloudberry surprise before we start: two of the three algorithms ship switched off.

The join GUCs as this cluster boots them — look at boot_val, not just setting (guc_tables.c:952 and :962; upstream PostgreSQL has both as true)

SELECT name, setting, boot_val, source
  FROM pg_settings
 WHERE name IN ('enable_hashjoin','enable_mergejoin','enable_nestloop',
                'enable_memoize','gp_hashjoin_tuples_per_bucket');

             name              | setting | boot_val | source
-------------------------------+---------+----------+---------
 enable_hashjoin               | on      | on       | default
 enable_memoize                | on      | on       | default
 enable_mergejoin              | off     | off    | default
 enable_nestloop               | off     | off    | default
 gp_hashjoin_tuples_per_bucket | 5       | 5        | default
(5 rows)

18.4.1 Nested loop: rescanning the inner side

nodeNestloop.c is the shortest of the three at 770 lines and reads exactly like its name. Pull an outer tuple; walk the inner subtree to exhaustion, emitting every pair that passes joinqual; pull the next outer tuple; ExecReScan the inner subtree; repeat. The one subtlety sits between “next outer tuple” and “rescan”: if the plan attached nestParams to the node, the outer tuple’s values are pushed into PARAM_EXEC slots and flagged as changed, which is what turns the inner subtree’s index scan into a different index scan on every iteration.

src/backend/executor/nodeNestloop.c:223

foreach(lc, nl->nestParams)
{
    NestLoopParam *nlp = (NestLoopParam *) lfirst(lc);
    ParamExecData *prm = &(econtext->ecxt_param_exec_vals[nlp->paramno]);

    prm->value = slot_getattr(outerTupleSlot,
                              nlp->paramval->varattno, &(prm->isnull));
    /* Flag parameter value as changed */
    innerPlan->chgParam = bms_add_member(innerPlan->chgParam, nlp->paramno);
}
/* now rescan the inner plan */
ExecReScan(innerPlan);

That is the parameterized, or index, nested loop — the shape that makes the algorithm worth having. Below it runs against a 50-row outer table and an indexed 300 000-row inner. The number to read is on the inner scan: loops=5. Take §18.1’s rule with you — actual rows is the per-loop average, and both figures come from one segment (the one that produced the most tuples), not from the cluster. Five outer tuples reached this segment’s join node, the bitmap scan ran five times, and 103 × 5 = 515 is exactly what the join reports.

A parameterized nested loop: the inner subtree runs once per outer tuple (PostgreSQL planner)

SET optimizer = off;
SET enable_hashjoin = off; SET enable_mergejoin = off; SET enable_nestloop = on;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM m184_s s, m184_f f WHERE f.did = s.sid AND s.sid < 6;

 Aggregate (actual rows=1 loops=1)
   ->  Gather Motion 3:1  (slice1; segments: 3) (actual rows=1500 loops=1)
         ->  Nested Loop (actual rows=515 loops=1)
               ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=5 loops=1)
                     ->  Seq Scan on m184_s s (actual rows=3 loops=1)
                           Filter: (sid < 6)
                           Rows Removed by Filter: 15
               ->  Bitmap Heap Scan on m184_f f (actual rows=103 loops=5)
                     Recheck Cond: ((did = s.sid) AND (did < 6))
                     ->  Bitmap Index Scan on m184_f_did_idx (actual rows=103 loops=5)
                           Index Cond: ((did = s.sid) AND (did < 6))
 Optimizer: Postgres query optimizer

Take the index away and the node degenerates into the textbook double loop, with two consequences visible in the plan. The clause moves out of the inner scan and becomes a Join Filter evaluated per pair — three outer tuples × 196 inner rows = 588 pairs considered, 147 kept, Rows Removed by Join Filter: 441. And the planner slips a Materialize (§18.6) above the inner subtree, whose own loops reading is the interesting one: Materialize shows loops=3 because it was rescanned once per outer tuple, while its child Seq Scan shows loops=1 because the tuplestore absorbed the rescans and never asked the scan to run again. That contrast is the whole reason for putting it there.

The unparameterized nested loop: a Materialize absorbs the rescans and the clause becomes a per-pair Join Filter

SET enable_indexscan = off; SET enable_bitmapscan = off; SET enable_indexonlyscan = off;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM m184_s s, m184_d d WHERE d.grp = s.sid AND s.sid < 4;

         ->  Nested Loop (actual rows=147 loops=1)
               Join Filter: (s.sid = d.grp)
               Rows Removed by Join Filter: 441
               ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=3 loops=1)
                     ->  Seq Scan on m184_s s (actual rows=2 loops=1)
                           Filter: (sid < 4)
               ->  Materialize (actual rows=196 loops=3)
                     ->  Seq Scan on m184_d d (actual rows=196 loops=1)
                           Filter: (grp < 4)
 Optimizer: Postgres query optimizer

18.4.2 Merge join: two streams in lockstep

nodeMergejoin.c (1827 lines) never buffers a whole side. It keeps one cursor into each sorted stream, compares the merge keys, and advances whichever side is behind; equal keys produce output. The complication is duplicates — if the outer stream has two tuples with key 7 and the inner has three, the second outer tuple must revisit all three — so before consuming a duplicate group the node calls ExecMarkPos on the inner subtree and later ExecRestrPos to rewind to that mark (execAmi.c:419 and :480; call sites at nodeMergejoin.c:1288 and :1175). Not every node supports mark and restore, which is one more reason a Sort or an ordered index tends to sit directly beneath the inner side. All of it is written as an explicit state machine — eleven states at nodeMergejoin.c:108-119, driven by a switch inside one long loop. The state names are the algorithm:

State	What happens in it
EXEC_MJ_INITIALIZE_OUTER	First call: fetch the first outer tuple and evaluate its merge keys.
EXEC_MJ_INITIALIZE_INNER	Fetch the first inner tuple; it becomes the initial mark.
EXEC_MJ_SKIP_TEST	Compare the two current keys and decide which side is behind.
EXEC_MJ_SKIPOUTER_ADVANCE	Outer key is smaller: emit a null-extended row for a left join, then advance outer.
EXEC_MJ_SKIPINNER_ADVANCE	Inner key is smaller: emit a null-extended row for a right join, then advance inner.
EXEC_MJ_JOINTUPLES	Keys match: test the remaining quals and emit the joined tuple.
EXEC_MJ_NEXTINNER	Advance inside the current inner duplicate group.
EXEC_MJ_NEXTOUTER	Inner group exhausted: advance the outer stream.
EXEC_MJ_TESTOUTER	New outer tuple compared against the marked inner tuple; if equal, ExecRestrPos and replay the group.
EXEC_MJ_ENDOUTER	Outer exhausted: drain unmatched inner tuples (right and full joins).
EXEC_MJ_ENDINNER	Inner exhausted: drain unmatched outer tuples (left and full joins).

Live, with a Sort on each side. The number worth staring at is on the inner Sort: it emitted 99 861 rows while its own child scan produced 100 172. Nothing was lost — merge join simply stopped asking. The outer stream’s largest key was 997, so once the inner cursor passed it the join was provably finished and the last few hundred inner tuples were never pulled. Neither hash join nor nested loop can do that; it is the one place where the lockstep advance pays back the cost of the sorts.

A merge join with a Sort on both inputs — and an inner side deliberately left unfinished

SET optimizer = off;
SET enable_hashjoin = off; SET enable_nestloop = off; SET enable_mergejoin = on;
SET enable_indexscan = off; SET enable_bitmapscan = off; SET enable_indexonlyscan = off;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM m184_f f JOIN m184_d d ON f.did = d.did WHERE d.grp = 3;

         ->  Partial Aggregate (actual rows=1 loops=1)
               ->  Merge Join (actual rows=14323 loops=1)
                     Merge Cond: (d.did = f.did)
                     ->  Sort (actual rows=143 loops=1)
                           Sort Key: d.did
                           Sort Method:  quicksort  Memory: 75kB
                           ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=143 loops=1)
                                 ->  Seq Scan on m184_d d (actual rows=54 loops=1)
                                       Filter: (grp = 3)
                     ->  Sort (actual rows=99861 loops=1)
                           Sort Key: f.did
                           Sort Method:  quicksort  Memory: 9219kB
                           ->  Seq Scan on m184_f f (actual rows=100172 loops=1)
 Optimizer: Postgres query optimizer

Which optimizer will give you a merge join? Only the PostgreSQL planner, in practice. ORCA has exactly one merge-join implementation rule in the entire xform library — CXformImplementFullOuterMergeJoin.cpp — so it can reach a merge join only for a FULL OUTER JOIN; and even then, on this cluster with optimizer_enable_mergejoin=on (a hidden developer GUC, guc_gp.c:2607) it still cost-preferred a Hash Full Join. Every merge join in this section therefore begins with optimizer=off, which matches §17.6’s observation.

18.4.3 Hash join: build, probe, and the numbers it reports

nodeHashjoin.c (2591 lines) plus nodeHash.c (4478 lines) implement the hybrid hash join of Zeller and Gray — the file header says so explicitly. The plan shape gives away the two phases: a Hash Join always has a Hash node as its inner child, and that Hash node is a blocking operator. Phase one, HJ_BUILD_HASHTABLE, drains it completely into a bucket array; phase two streams the outer side through, hashing each tuple and walking one bucket chain. Like merge join this is an explicit state machine — six states at nodeHashjoin.c:189-194 — and the shape of that loop is why semi joins, outer joins and batching all fall out of the same code. Every Hash node then reports three numbers under EXPLAIN ANALYZE, to which Cloudberry adds a fourth line of its own.

A one-batch hash join under GPORCA — a 300 000-row fact table probing a 143-row broadcast dimension. Buckets/Batches/Memory Usage come from upstream; the Extra Text line is Cloudberry’s.

SET optimizer = on;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM m184_f f JOIN m184_d d ON f.did = d.did WHERE d.grp = 3;

               ->  Hash Join (actual rows=14323 loops=1)
                     Hash Cond: (f.did = d.did)
                     Extra Text: (seg2)   Hash chain length 1.0 avg, 1 max, using 143 of 524288 buckets.
                     ->  Seq Scan on m184_f f (actual rows=100172 loops=1)
                     ->  Hash (actual rows=143 loops=1)
                           Buckets: 524288  Batches: 1  Memory Usage: 4102kB
                           ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=143 loops=1)
                                 ->  Seq Scan on m184_d d (actual rows=54 loops=1)
                                       Filter: (grp = 3)
   (slice1)    Executor memory: 4158K bytes avg x 3x(0) workers, 4158K bytes max (seg0).  Work_mem: 4102K bytes max.
 Optimizer: GPORCA

Be careful whose numbers those are. Buckets/Batches/Memory Usage are printed by upstream’s show_hash_info (explain.c:4041) out of a SharedHashInfo, which in PostgreSQL holds one entry per parallel worker. Cloudberry reuses the structure for segments: explain_gp.c:1246 ships each QE’s HashInstrumentation back and rebuilds the array — the comment there says so outright — and show_hash_info then takes the Max across the entries (explain.c:4075-4084). So Memory Usage: 4102kB is the peak on the worst segment, never a sum. The Extra Text line has a different provenance again: ExecHashTableExplainEnd (nodeHash.c:2718) produces it on each QE and explain_gp.c prints only the one segment it judged most interesting, hence the (seg2) tag. The bucket count is a genuine Cloudberry divergence too: upstream aims for roughly one tuple per bucket, whereas ExecChooseHashTableSize here treats gp_hashjoin_tuples_per_bucket (default 5) as a lower bound and, when the table is expected to fit, averages it with a bound proportional to the memory budget, floored at 1024 (nodeHash.c:1049-1075) — which is why the same 143-row table gets 524 288 buckets at a 128 MB budget and 2048 buckets at 512 kB. The of 524288 buckets denominator, finally, is nonemptybatches × nbuckets (nodeHash.c:2803), an identity we can check in a moment.

18.4.4 When the inner side does not fit: batches

If the inner side outgrows its budget the join neither fails nor thrashes: it partitions. Both sides are split into batches by hash bits; batch 0 stays resident and batches 1…n−1 are written to a pair of temporary files each, one for inner tuples and one for outer. After the first pass the join loops — load batch k’s inner file into the now-empty bucket array, stream batch k’s outer file through it, repeat. That is the HJ_NEED_NEW_BATCH edge in the diagram above, and the whole scheme rests on one small function.

src/backend/executor/nodeHash.c:2240

if (nbatch > 1)
{
    *bucketno = hashvalue & (nbuckets - 1);
    *batchno = pg_rotate_right32(hashvalue,
                                hashtable->log2_nbuckets) & (nbatch - 1);
}

Low bits pick the bucket, the next bits up pick the batch. Because the batch bits sit above the bucket bits, doubling nbatch consumes one more bit and can only move a tuple to a later batch, never an earlier one — which is what makes it safe to double the batch count halfway through the build, rescan the resident table, and flush the newly out-of-range tuples to their files (nodeHash.c:1257-1287).

bits 31 .. 16

unused

bits 15 .. 11

batch, 5 bits

32 batches

bits 10 .. 0

bucket, 11 bits

2048 buckets

bucketno = hashvalue & 2047     batchno = rotate_right32(hashvalue, 11) & 31

batch 0
resident: buckets in memory

batch 1
innerBatchFile[1]
outerBatchFile[1]

batch 2
innerBatchFile[2]
outerBatchFile[2]

…

batch 31
innerBatchFile[31]
outerBatchFile[31]

2 files x 31 non-resident batches = 62 temporary files per segment

One self-join at 128 MB, 2 MB and 512 kB. Note the knob: on this cluster gp_resource_manager is queue, so per-operator memory comes from statement_mem and setting work_mem alone changes nothing — nodeHash.c compares against PlanStateOperatorMemKB. Statement-memory policy is Chapter 21’s subject.

SET optimizer = off;
SET statement_mem = '128MB';   -- then '2MB', then '512kB'
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM m184_f a JOIN m184_f b ON a.id = b.id AND a.did = b.did;

-- statement_mem = 128MB
                     Extra Text: (seg2)   Hash chain length 1.1 avg, 5 max, using 91229 of 524288 buckets.
                     ->  Hash (actual rows=100172 loops=1)
                           Buckets: 524288  Batches: 1  Memory Usage: 8009kB
   (slice1)    Executor memory: 8061K bytes avg x 3x(0) workers, 8072K bytes max (seg1).  Work_mem: 8009K bytes max.
 Memory used:  128000kB

-- statement_mem = 2MB
                     Extra Text: (seg2)   Initial batch 0:
 (seg2)     Wrote 2016K bytes to inner workfile.
 (seg2)     Wrote 3168K bytes to outer workfile.
 (seg2)   Initial batches 1..3:
 (seg2)     Read 2060K bytes from inner workfile: 687K avg x 3 nonempty batches, 690K max.
 (seg2)     Read 3210K bytes from outer workfile: 1070K avg x 3 nonempty batches, 1075K max.
 (seg2)   Hash chain length 2.0 avg, 8 max, using 51189 of 65536 buckets.
                     ->  Hash (actual rows=100172 loops=1)
                           Buckets: 16384  Batches: 4  Memory Usage: 1112kB
 * (slice1)    Executor memory: 1351K bytes avg x 3x(0) workers, 1351K bytes max (seg0).  Work_mem: 1112K bytes max, 4041K bytes wanted.
 Memory used:  2048kB
 Memory wanted:  4640kB

-- statement_mem = 512kB
                     Extra Text: (seg2)   Initial batches 1..31:
 (seg2)     Read 2653K bytes from inner workfile: 86K avg x 31 nonempty batches, 88K max.
 (seg2)     Read 4134K bytes from outer workfile: 134K avg x 31 nonempty batches, 137K max.
 (seg2)   Hash chain length 2.0 avg, 8 max, using 51189 of 65536 buckets.
                     ->  Hash (actual rows=100172 loops=1)
                           Buckets: 2048  Batches: 32  Memory Usage: 141kB
 * (slice1)    Executor memory: 2185K bytes avg x 3x(0) workers, 2185K bytes max (seg0).  Work_mem: 141K bytes max, 3929K bytes wanted.
 Memory used:  512kB
 Memory wanted:  4528kB

Two things to check by hand. The bucket identity first: 4 × 16 384 = 65 536 and 32 × 2048 = 65 536, and both runs report of 65536 buckets — exactly nonemptybatches × nbuckets. Second, Work_mem: … bytes wanted appears only when nbatch > 1: ExecHashTableExplainEnd sums the final hash space of the completed batches plus the sizes of inner workfiles it never got to read, and reports the total only if it exceeds the operator’s budget (nodeHash.c:2749-2777). It answers “how much memory would have made this a one-batch join”, not “how much did we use”.

The batch files, caught from a second session while a 32-batch join is probing: 62 = 2 x (32 - 1), because batch 0 never gets files

-- session 1: SET statement_mem='512kB'; a deliberately slow probe of the same join
-- session 2, while it runs:
SELECT segid, slice, optype, numfiles, size
  FROM gp_toolkit.gp_workfile_entries ORDER BY segid;

 segid | slice |  optype  | numfiles |  size
-------+-------+----------+----------+---------
     0 |     1 | HashJoin |     62 | 4784128
     1 |     1 | HashJoin |       62 | 5079040
     2 |     1 | HashJoin |       62 | 4030464
(3 rows)

The slice column is what lets you attribute spill in a plan with several joins; the sibling views gp_workfile_usage_per_query, gp_workfile_usage_per_segment and gp_workfile_mgr_used_diskspace aggregate the same data, and §18.3 uses them for sort and aggregate spill. Batching does have one adversary, though: a single hot join key. Tuples sharing a key share a hash value, so no number of batch bits can separate them. Below, the inner side is 150 000 rows with one distinct join key at a 512 kB budget. The executor doubles the batch count from 32 to 64, discovers that the offending batch retained all of its tuples, and gives up — the resident table finishes at 5274 kB against a 512 kB budget, in a single bucket, with a chain 150 000 long.

Skew defeats batching: one hot inner key, a batch doubling that changed nothing, and a chain length of 150000

SET optimizer = off; SET statement_mem = '512kB';
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM m184_f f JOIN m184_k2 k ON k.did = f.did;   -- every k.did = 5000

               ->  Hash Join (actual rows=0 loops=1)
                     Hash Cond: (f.did = k.did)
                     Extra Text: (seg1)   Initial batch 0:
 (seg1)     Wrote 3488K bytes to inner workfile.
 (seg1)   Initial batches 1..31:
 (seg1)     Read 3516K bytes from inner workfile.
 (seg1)     Wrote 3488K bytes to inner workfile.
 (seg1)   Secondary Overflow batches 32..63:
 (seg1)     Read 3516K bytes from inner workfile.
 (seg1)   Hash chain length 150000.0 avg, 150000 max, using 1 of 27648 buckets.  Skipped 37 empty batches.
                     ->  Redistribute Motion 3:3  (slice2; segments: 3) (actual rows=96600 loops=1)
                           ->  Seq Scan on m184_f f (actual rows=100172 loops=1)
                     ->  Hash (actual rows=150000 loops=1)
                           Buckets: 1024 (originally 1024)  Batches: 64 (originally 32)  Memory Usage: 5274kB
 * (slice1)    Executor memory: 2235K bytes avg x 3x(0) workers, 6401K bytes max (seg1).  Work_mem: 5274K bytes max, 5282K bytes wanted.
 Memory wanted:  5981kB

Upstream does have skew handling, and it is reachable here — but it does not help this case. ExecHashBuildSkewHash (nodeHash.c:3064) sets aside up to SKEW_HASH_MEM_PERCENT = 2 % of the budget for dedicated buckets holding the inner tuples that match the outer relation’s most common values, so those never spill. It requires nbatch > 1 at build time (nodeHash.c:788), a single hash clause whose outer side is a plain Var on a base relation (createplan.c:6166-6188 is where skewTable is set), MCV statistics for that column, and top MCVs covering at least SKEW_MIN_OUTER_FRACTION = 1 % of the outer — all true in the plan above. But the hot key here is on the inner side, which the mechanism says nothing about. Note also that the ORCA translator never sets skewTable at all, so under optimizer=on the skew optimization is unreachable by construction.

Two different “hash table” lines. A HashAggregate prints an Extra Text line too, and it looks confusingly similar: hash table(s): 1; chain length 2.4 avg, 5 max; using 337 of 512 buckets; total 0 expansions. That one comes from agg_hash_explain_extra_message (nodeAgg.c:5199) and belongs to §18.3 — total N expansions counts growth events in a simplehash aggregate table and has no hash-join equivalent. The join’s line always begins Hash chain length and comes from nodeHash.c:2806.

18.4.5 Semi, anti and outer joins

The states in the diagram already cover every join flavour; the variants differ only in which of them are enabled. Two macros decide it (nodeHashjoin.c:197 and :199): HJ_FILL_OUTER is true when the node was initialised with a null inner tuple slot — unmatched outer rows must still be emitted — and HJ_FILL_INNER is true when there is a null outer slot. hj_MatchedOuter records whether the current outer tuple ever found a partner. A semi join sets js.single_match, which makes HJ_SCAN_BUCKET stop at the first match instead of walking the chain; an anti join emits the outer tuple precisely when hj_MatchedOuter is still false. All of it is a dozen lines:

src/backend/executor/nodeHashjoin.c:749

case HJ_FILL_OUTER_TUPLE:
    node->hj_JoinState = HJ_NEED_NEW_OUTER;

    if (!node->hj_MatchedOuter && HJ_FILL_OUTER(node))
    {
        /* Generate a fake join tuple with nulls for the inner tuple */
        econtext->ecxt_innertuple = node->hj_NullInnerTupleSlot;

        if (otherqual == NULL || ExecQual(otherqual, econtext))
            return ExecProject(node->js.ps.ps_ProjInfo);
    }
    break;

Semi, anti and left variants over the same two tables, the same 143-row hash table and the same 100 172-row outer scan, so only the label and the count move. EXISTS gives 14 323, NOT EXISTS gives 85 849, and 14 323 + 85 849 = 100 172 — every outer row goes to exactly one of them.

SET optimizer = off;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT count(*) FROM m184_f f
  WHERE EXISTS (SELECT 1 FROM m184_d d WHERE d.did = f.did AND d.grp = 3);
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT count(*) FROM m184_f f
  WHERE NOT EXISTS (SELECT 1 FROM m184_d d WHERE d.did = f.did AND d.grp = 3);
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT count(d.dname) FROM m184_f f
  LEFT JOIN m184_d d ON f.did = d.did AND d.grp = 3;

-- EXISTS
               ->  Hash Semi Join (actual rows=14323 loops=1)
                     Hash Cond: (f.did = d.did)
                     ->  Seq Scan on m184_f f (actual rows=100172 loops=1)
                     ->  Hash (actual rows=143 loops=1)
                           Buckets: 524288  Batches: 1  Memory Usage: 4102kB
-- NOT EXISTS
               ->  Hash Anti Join (actual rows=85849 loops=1)
                     Hash Cond: (f.did = d.did)
                     ->  Seq Scan on m184_f f (actual rows=100172 loops=1)
                     ->  Hash (actual rows=143 loops=1)
-- LEFT JOIN
               ->  Hash Left Join (actual rows=100172 loops=1)
                     Hash Cond: (f.did = d.did)
                     ->  Seq Scan on m184_f f (actual rows=100172 loops=1)
                     ->  Hash (actual rows=143 loops=1)

Two labels you will meet that are not upstream’s. The first is PostgreSQL 16’s Hash Right Anti Join (JOIN_RIGHT_ANTI, nodes.h:1029): when the hashed side is the one being filtered, the planner commutes the join and lets HJ_FILL_INNER_TUPLES do the work — the same NOT EXISTS written the other way round produces it. The second is Cloudberry’s own JOIN_LASJ_NOTIN (nodes.h:1026), printed as Left Anti Semi (Not-In) (explain.c:2184): a left anti semi join with SQL NOT IN semantics, in which a single NULL on the inner side must suppress all output. nodeHashjoin.c:660 and nodeNestloop.c:152 carry the extra NULL checks that implement it, and ORCA emits it for NOT IN:

A Cloudberry-only join type: NOT IN becomes a Left Anti Semi (Not-In) Join under GPORCA

SET optimizer = on;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(*) FROM m184_f f WHERE f.did NOT IN (SELECT did FROM m184_d WHERE grp = 3);

               ->  Hash Left Anti Semi (Not-In) Join (actual rows=85849 loops=1)
                     Hash Cond: (f.did = m184_d.did)
                     Extra Text: (seg2)   Hash chain length 1.0 avg, 1 max, using 143 of 524288 buckets.
                     ->  Seq Scan on m184_f f (actual rows=100172 loops=1)
                     ->  Hash (actual rows=143 loops=1)
                           ->  Broadcast Motion 3:3  (slice2; segments: 3) (actual rows=143 loops=1)
                                 ->  Seq Scan on m184_d (actual rows=54 loops=1)
 Optimizer: GPORCA

18.4.6 Memoize: a cache in front of a parameterized inner side

nodeMemoize.c (1239 lines) exists for one situation: a parameterized nested loop whose outer side repeats parameter values. Rather than rescanning the inner subtree for the tenth occurrence of did = 7, Memoize keeps an LRU hash table keyed on the parameters and replays the cached tuples. It is worth having only because the planner can estimate the number of distinct parameter values; when that estimate is wrong the cache is pure overhead. Cloudberry inherits the node unchanged, and the PostgreSQL planner does pick it here.

A live Memoize over a parameterized index scan — read the two loops counts against each other

SET optimizer = off;
SET enable_hashjoin = off; SET enable_mergejoin = off;
SET enable_nestloop = on; SET enable_memoize = on;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF)
SELECT count(d.dname) FROM m184_f a, m184_d d WHERE d.did = a.did AND a.id < 20000;

               ->  Nested Loop (actual rows=6800 loops=1)
                     ->  Redistribute Motion 3:3  (slice2; segments: 3) (actual rows=6800 loops=1)
                           Hash Key: a.did
                           ->  Seq Scan on m184_f a (actual rows=6780 loops=1)
                                 Filter: (id < 20000)
                     ->  Memoize (actual rows=1 loops=6800)
                           Cache Key: a.did
                           Cache Mode: logical
                           ->  Index Scan using m184_d_did_idx on m184_d d (actual rows=1 loops=340)
                                 Index Cond: (did = a.did)
 Optimizer: Postgres query optimizer

Memoize was asked 6800 times; the index scan beneath it ran 340 times. The cache absorbed 6460 rescans — the same reading as the Materialize in §18.4.1, one level of indirection up. What is missing is upstream’s Hits: … Misses: … Evictions: … line, and the reason is instructive. show_memoize_info prints it from mstate->stats on the local MemoizeState, and skips the line entirely when cache_misses is zero (explain.c:4192-4195). In a dispatched plan the coordinator’s MemoizeState never executed anything — and unlike HashInstrumentation, Memoize’s counters are not among the fields explain_gp.c ships back from the QEs (explain_gp.c:880-896). So on Cloudberry you get Cache Key and Cache Mode, which come from the Plan, and the hit rate has to be inferred from the two loops values.

18.4.7 The MPP twist: prefetching the inner side

Now the part with no upstream counterpart. A join node’s instinct is to look at the outer side first — even hash join peeks at one outer tuple before building, so it can skip the build entirely if the outer turns out empty. In a single-process executor that peek is free. In Cloudberry, where either input may be a Motion fed by a whole gang of senders, it can hang the cluster. The reasoning is spelled out in a long comment at cdbllize.c:1459-1489. Motion senders block when their send buffers fill and nobody drains them. Segment A’s join has already taken an outer tuple and is now pulling from its inner sender; segment B’s join is still waiting for its first outer tuple. A cannot drain its outer sender because it is busy with the inner; B cannot drain its inner sender because it has not got past the outer. Four processes, one cycle:

prefetch_inner — Remove the half-and-half state: consume the inner side before touching the outer. For hash join this merely suppresses the empty-outer peek (nodeHashjoin.c:337), since the build already drains the inner. For nested loop (nodeNestloop.c:115) and merge join (nodeMergejoin.c:684) the node pulls one tuple from the inner subtree and then ExecReScans it.

prefetch_joinqual — The same hazard one level down: the join qual itself may hold a SubPlan with a Motion. ExecPrefetchQual (execUtils.c:1614) evaluates the qual once against fabricated all-NULL inner and outer tuples, purely to force that Motion to run and drain.

prefetch_qual — Identical treatment for the node’s non-join qual.

Why does pulling one tuple drain the inner side? Because the planner guarantees a blocking operator on top of it. create_mergejoin_plan demands “some kind of strict slackening operator … which consumes all of its input before producing a row of output” and installs a Sort, or a Material with cdb_strict set (createplan.c:5806-5822); ExecMaterial reacts to cdb_strict by looping the entire subplan into its tuplestore on the first call (nodeMaterial.c:108-121). One pull, whole Motion drained, cycle broken.

src/backend/executor/nodeMergejoin.c:684

if (node->prefetch_inner)
{
    innerTupleSlot = ExecProcNode(innerPlan);
    node->mj_InnerTupleSlot = innerTupleSlot;

    ExecReScan(innerPlan);
    ResetExprContext(econtext);

    node->prefetch_inner = false;
}

Who sets the three flags (plannodes.h:1198-1200)? Under the PostgreSQL planner, create_hashjoin_plan turns prefetch_inner on when either subpath has a motion hazard (createplan.c:6299-6304), merge join when both do (:5815), and nested loop only when it also managed to get a strict Material under the inner side (:5569-5572) — which is why a parameterized inner scan, whose inner side cannot run before the outer supplies values, comes out with prefetch_inner false even under a Motion. ORCA is blunter: its translator sets the flag unconditionally for every hash join (CTranslatorDXLToPlStmt.cpp:1554) and for every nested loop except an index nested loop, “because inner child depends on variables coming from outer child” (:2127). EXPLAIN prints none of this — the string prefetch does not occur anywhere in explain.c, with or without VERBOSE — but outfuncs.c:498-500 writes the fields, so debug_print_plan will show them.

The prefetch flags in the plan dump: a colocated join needs none, and one Motion under either side flips it

SET optimizer = off; SET client_min_messages = log; SET debug_print_plan = on;
-- (1) both sides distributed by id, joined on id: no Motion at all
EXPLAIN (COSTS OFF) SELECT count(*) FROM m184_f a JOIN m184_f b ON a.id = b.id;
-- (2) joined on a non-distribution column: the inner side gets a Broadcast Motion
EXPLAIN (COSTS OFF) SELECT count(*) FROM m184_f f JOIN m184_d d ON f.did = d.grp;

-- (1)
               {HASHJOIN
               :prefetch_inner false
               :prefetch_joinqual false
               :prefetch_qual false
   ->  Gather Motion 3:1  (slice1; segments: 3)
               ->  Hash Join

-- (2)
               {HASHJOIN
               :prefetch_inner true
               :prefetch_joinqual false
               :prefetch_qual false
   ->  Gather Motion 3:1  (slice1; segments: 3)
               ->  Hash Join
                           ->  Broadcast Motion 3:3  (slice2; segments: 3)

Motions on both sides of a merge join. gp_enable_motion_deadlock_sanity (guc_gp.c:852) traces the planner’s deadlock walk, and the prefetched inner Sort shows loops=2 — one pass for the prefetch, one for the join.

SET optimizer = off;
SET enable_hashjoin = off; SET enable_nestloop = off; SET enable_mergejoin = on;
SET gp_enable_motion_deadlock_sanity = on; SET client_min_messages = debug5;
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF) SELECT count(*)
  FROM m184_f a, m184_f b WHERE a.did = b.did AND a.id < 30 AND b.id > 299970;

DEBUG:  Motion Deadlock Sanity Check
DEBUG:      MJ going left
DEBUG:     found motion
DEBUG:      MJ going right
DEBUG:     found motion
DEBUG:      MJ branch point left 0x1 right 0x1
DEBUG:      MJ branch point left 0x1 right 0x1 res 0x1 deadlock safe: prefetching
...
               ->  Merge Join (actual rows=0 loops=1)
                     Merge Cond: (a.did = b.did)
                     ->  Sort (actual rows=12 loops=1)
                           ->  Redistribute Motion 3:3  (slice2; segments: 3) (actual rows=12 loops=1)
                                 ->  Seq Scan on m184_f a (actual rows=12 loops=1)
                     ->  Sort (actual rows=2 loops=2)
                           ->  Redistribute Motion 3:3  (slice3; segments: 3) (actual rows=13 loops=1)
                                 ->  Seq Scan on m184_f b (actual rows=13 loops=1)
 Optimizer: Postgres query optimizer

Two details in that output are worth the price of admission. motion_sanity_check runs at the end of planning when the GUC is on (planner.c:763) and would raise Post-planning sanity check detected motion deadlock had the plan been unsafe; the deadlock safe: prefetching verdict comes from cdbllize.c:1596. And loops=2 on the inner Sort is the prefetch’s fingerprint in ordinary EXPLAIN ANALYZE: the prefetch pass pulled from it once, and ExecReScan — armed with EXEC_FLAG_REWIND, which merge join requests only when prefetching (nodeMergejoin.c:1615) — rewound it for the real pass. The rarer prefetch_joinqual case needs a Motion inside the qual rather than under a child; test_print_prefetch_joinqual (guc_gp.c:1504) makes the executor announce it, and the announcement is a good reminder of what “one node” means here — the join exists once in the plan and four times at run time.

prefetch_joinqual firing once per QE: a Motion inside the join qual, drained by ExecPrefetchQual

SET optimizer = off; SET test_print_prefetch_joinqual = on;
SELECT count(*) FROM m184_d d LEFT JOIN m184_f f
    ON f.did = d.did
   AND NOT EXISTS (SELECT 1 FROM m184_s s WHERE s.sid = d.grp + f.did);

NOTICE:  prefetch join qual in slice 0 of plannode 3
NOTICE:  prefetch join qual in slice 1 of plannode 3  (seg0 slice1 10.100.0.3:7102 pid=9156)
NOTICE:  prefetch join qual in slice 1 of plannode 3  (seg1 slice1 10.100.0.3:7103 pid=9157)
NOTICE:  prefetch join qual in slice 1 of plannode 3  (seg2 slice1 10.100.0.3:7104 pid=9158)
 count
--------
 285947
(1 row)

Prefetch is one half of a matched pair. It stops a sender from blocking because nobody is reading yet; §18.1’s ExecSquelchNode (execAmi.c:756) stops a sender from blocking because nobody will read any more — a LIMIT satisfied early, an anti join short-circuited by a NULL. Both exist only because a subtree in this executor may be another process. Why a full send buffer blocks in the first place, how the ACK protocol works, and what the 3:3 in a Motion label costs are Chapter 19’s subject. One more join-adjacent node deserves a pointer: a RuntimeFilter built from a hash join’s inner side and pushed down into the outer scan (§18.7).

That closes the join family. nodeNestloop.c, nodeMergejoin.c, nodeHashjoin.c and nodeHash.c are 9666 lines of upstream PostgreSQL with three bool fields, one Extra Text line and one extra join type grafted on — and those small additions are exactly where the MPP executor lives. The next section drops a level, to the expression machinery every one of these nodes calls to evaluate a Hash Cond, a Merge Cond or a Join Filter.