Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * nodeHashjoin.c
4 : : * Routines to handle hash join nodes
5 : : *
6 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
7 : : * Portions Copyright (c) 1994, Regents of the University of California
8 : : *
9 : : *
10 : : * IDENTIFICATION
11 : : * src/backend/executor/nodeHashjoin.c
12 : : *
13 : : * HASH JOIN
14 : : *
15 : : * This is based on the "hybrid hash join" algorithm described shortly in the
16 : : * following page
17 : : *
18 : : * https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
19 : : *
20 : : * and in detail in the referenced paper:
21 : : *
22 : : * "An Adaptive Hash Join Algorithm for Multiuser Environments"
23 : : * Hansjörg Zeller; Jim Gray (1990). Proceedings of the 16th VLDB conference.
24 : : * Brisbane: 186–197.
25 : : *
26 : : * If the inner side tuples of a hash join do not fit in memory, the hash join
27 : : * can be executed in multiple batches.
28 : : *
29 : : * If the statistics on the inner side relation are accurate, planner chooses a
30 : : * multi-batch strategy and estimates the number of batches.
31 : : *
32 : : * The query executor measures the real size of the hashtable and increases the
33 : : * number of batches if the hashtable grows too large.
34 : : *
35 : : * The number of batches is always a power of two, so an increase in the number
36 : : * of batches doubles it.
37 : : *
38 : : * Serial hash join measures batch size lazily -- waiting until it is loading a
39 : : * batch to determine if it will fit in memory. While inserting tuples into the
40 : : * hashtable, serial hash join will, if that tuple were to exceed work_mem,
41 : : * dump out the hashtable and reassign them either to other batch files or the
42 : : * current batch resident in the hashtable.
43 : : *
44 : : * Parallel hash join, on the other hand, completes all changes to the number
45 : : * of batches during the build phase. If it increases the number of batches, it
46 : : * dumps out all the tuples from all batches and reassigns them to entirely new
47 : : * batch files. Then it checks every batch to ensure it will fit in the space
48 : : * budget for the query.
49 : : *
50 : : * In both parallel and serial hash join, the executor currently makes a best
51 : : * effort. If a particular batch will not fit in memory, it tries doubling the
52 : : * number of batches. If after a batch increase, there is a batch which
53 : : * retained all or none of its tuples, the executor disables growth in the
54 : : * number of batches globally. After growth is disabled, all batches that would
55 : : * have previously triggered an increase in the number of batches instead
56 : : * exceed the space allowed.
57 : : *
58 : : * PARALLELISM
59 : : *
60 : : * Hash joins can participate in parallel query execution in several ways. A
61 : : * parallel-oblivious hash join is one where the node is unaware that it is
62 : : * part of a parallel plan. In this case, a copy of the inner plan is used to
63 : : * build a copy of the hash table in every backend, and the outer plan could
64 : : * either be built from a partial or complete path, so that the results of the
65 : : * hash join are correspondingly either partial or complete. A parallel-aware
66 : : * hash join is one that behaves differently, coordinating work between
67 : : * backends, and appears as Parallel Hash Join in EXPLAIN output. A Parallel
68 : : * Hash Join always appears with a Parallel Hash node.
69 : : *
70 : : * Parallel-aware hash joins use the same per-backend state machine to track
71 : : * progress through the hash join algorithm as parallel-oblivious hash joins.
72 : : * In a parallel-aware hash join, there is also a shared state machine that
73 : : * co-operating backends use to synchronize their local state machines and
74 : : * program counters. The shared state machine is managed with a Barrier IPC
75 : : * primitive. When all attached participants arrive at a barrier, the phase
76 : : * advances and all waiting participants are released.
77 : : *
78 : : * When a participant begins working on a parallel hash join, it must first
79 : : * figure out how much progress has already been made, because participants
80 : : * don't wait for each other to begin. For this reason there are switch
81 : : * statements at key points in the code where we have to synchronize our local
82 : : * state machine with the phase, and then jump to the correct part of the
83 : : * algorithm so that we can get started.
84 : : *
85 : : * One barrier called build_barrier is used to coordinate the hashing phases.
86 : : * The phase is represented by an integer which begins at zero and increments
87 : : * one by one, but in the code it is referred to by symbolic names as follows.
88 : : * An asterisk indicates a phase that is performed by a single arbitrarily
89 : : * chosen process.
90 : : *
91 : : * PHJ_BUILD_ELECT -- initial state
92 : : * PHJ_BUILD_ALLOCATE* -- one sets up the batches and table 0
93 : : * PHJ_BUILD_HASH_INNER -- all hash the inner rel
94 : : * PHJ_BUILD_HASH_OUTER -- (multi-batch only) all hash the outer
95 : : * PHJ_BUILD_RUN -- building done, probing can begin
96 : : * PHJ_BUILD_FREE* -- all work complete, one frees batches
97 : : *
98 : : * While in the phase PHJ_BUILD_HASH_INNER a separate pair of barriers may
99 : : * be used repeatedly as required to coordinate expansions in the number of
100 : : * batches or buckets. Their phases are as follows:
101 : : *
102 : : * PHJ_GROW_BATCHES_ELECT -- initial state
103 : : * PHJ_GROW_BATCHES_REALLOCATE* -- one allocates new batches
104 : : * PHJ_GROW_BATCHES_REPARTITION -- all repartition
105 : : * PHJ_GROW_BATCHES_DECIDE* -- one detects skew and cleans up
106 : : * PHJ_GROW_BATCHES_FINISH -- finished one growth cycle
107 : : *
108 : : * PHJ_GROW_BUCKETS_ELECT -- initial state
109 : : * PHJ_GROW_BUCKETS_REALLOCATE* -- one allocates new buckets
110 : : * PHJ_GROW_BUCKETS_REINSERT -- all insert tuples
111 : : *
112 : : * If the planner got the number of batches and buckets right, those won't be
113 : : * necessary, but on the other hand we might finish up needing to expand the
114 : : * buckets or batches multiple times while hashing the inner relation to stay
115 : : * within our memory budget and load factor target. For that reason it's a
116 : : * separate pair of barriers using circular phases.
117 : : *
118 : : * The PHJ_BUILD_HASH_OUTER phase is required only for multi-batch joins,
119 : : * because we need to divide the outer relation into batches up front in order
120 : : * to be able to process batches entirely independently. In contrast, the
121 : : * parallel-oblivious algorithm simply throws tuples 'forward' to 'later'
122 : : * batches whenever it encounters them while scanning and probing, which it
123 : : * can do because it processes batches in serial order.
124 : : *
125 : : * Once PHJ_BUILD_RUN is reached, backends then split up and process
126 : : * different batches, or gang up and work together on probing batches if there
127 : : * aren't enough to go around. For each batch there is a separate barrier
128 : : * with the following phases:
129 : : *
130 : : * PHJ_BATCH_ELECT -- initial state
131 : : * PHJ_BATCH_ALLOCATE* -- one allocates buckets
132 : : * PHJ_BATCH_LOAD -- all load the hash table from disk
133 : : * PHJ_BATCH_PROBE -- all probe
134 : : * PHJ_BATCH_SCAN* -- one does right/right-anti/full unmatched scan
135 : : * PHJ_BATCH_FREE* -- one frees memory
136 : : *
137 : : * Batch 0 is a special case, because it starts out in phase
138 : : * PHJ_BATCH_PROBE; populating batch 0's hash table is done during
139 : : * PHJ_BUILD_HASH_INNER so we can skip loading.
140 : : *
141 : : * Initially we try to plan for a single-batch hash join using the combined
142 : : * hash_mem of all participants to create a large shared hash table. If that
143 : : * turns out either at planning or execution time to be impossible then we
144 : : * fall back to regular hash_mem sized hash tables.
145 : : *
146 : : * To avoid deadlocks, we never wait for any barrier unless it is known that
147 : : * all other backends attached to it are actively executing the node or have
148 : : * finished. Practically, that means that we never emit a tuple while attached
149 : : * to a barrier, unless the barrier has reached a phase that means that no
150 : : * process will wait on it again. We emit tuples while attached to the build
151 : : * barrier in phase PHJ_BUILD_RUN, and to a per-batch barrier in phase
152 : : * PHJ_BATCH_PROBE. These are advanced to PHJ_BUILD_FREE and PHJ_BATCH_SCAN
153 : : * respectively without waiting, using BarrierArriveAndDetach() and
154 : : * BarrierArriveAndDetachExceptLast() respectively. The last to detach
155 : : * receives a different return value so that it knows that it's safe to
156 : : * clean up. Any straggler process that attaches after that phase is reached
157 : : * will see that it's too late to participate or access the relevant shared
158 : : * memory objects.
159 : : *
160 : : *-------------------------------------------------------------------------
161 : : */
162 : :
163 : : #include "postgres.h"
164 : :
165 : : #include "access/htup_details.h"
166 : : #include "access/parallel.h"
167 : : #include "executor/executor.h"
168 : : #include "executor/hashjoin.h"
169 : : #include "executor/instrument.h"
170 : : #include "executor/nodeHash.h"
171 : : #include "executor/nodeHashjoin.h"
172 : : #include "miscadmin.h"
173 : : #include "utils/lsyscache.h"
174 : : #include "utils/sharedtuplestore.h"
175 : : #include "utils/tuplestore.h"
176 : : #include "utils/wait_event.h"
177 : :
178 : :
179 : : /*
180 : : * States of the ExecHashJoin state machine
181 : : */
182 : : #define HJ_BUILD_HASHTABLE 1
183 : : #define HJ_NEED_NEW_OUTER 2
184 : : #define HJ_SCAN_BUCKET 3
185 : : #define HJ_FILL_OUTER_TUPLE 4
186 : : #define HJ_FILL_INNER_TUPLES 5
187 : : #define HJ_FILL_OUTER_NULL_TUPLES 6
188 : : #define HJ_FILL_INNER_NULL_TUPLES 7
189 : : #define HJ_NEED_NEW_BATCH 8
190 : :
191 : : /* Returns true if doing null-fill on outer relation */
192 : : #define HJ_FILL_OUTER(hjstate) ((hjstate)->hj_NullInnerTupleSlot != NULL)
193 : : /* Returns true if doing null-fill on inner relation */
194 : : #define HJ_FILL_INNER(hjstate) ((hjstate)->hj_NullOuterTupleSlot != NULL)
195 : :
196 : : static TupleTableSlot *ExecHashJoinOuterGetTuple(PlanState *outerNode,
197 : : HashJoinState *hjstate,
198 : : uint32 *hashvalue,
199 : : bool *is_null);
200 : : static TupleTableSlot *ExecParallelHashJoinOuterGetTuple(PlanState *outerNode,
201 : : HashJoinState *hjstate,
202 : : uint32 *hashvalue,
203 : : bool *is_null);
204 : : static TupleTableSlot *ExecHashJoinGetSavedTuple(HashJoinState *hjstate,
205 : : BufFile *file,
206 : : uint32 *hashvalue,
207 : : TupleTableSlot *tupleSlot);
208 : : static bool ExecHashJoinNewBatch(HashJoinState *hjstate);
209 : : static bool ExecParallelHashJoinNewBatch(HashJoinState *hjstate);
210 : : static void ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate);
211 : :
212 : :
213 : : /* ----------------------------------------------------------------
214 : : * ExecHashJoinImpl
215 : : *
216 : : * This function implements the Hybrid Hashjoin algorithm. It is marked
217 : : * with an always-inline attribute so that ExecHashJoin() and
218 : : * ExecParallelHashJoin() can inline it. Compilers that respect the
219 : : * attribute should create versions specialized for parallel == true and
220 : : * parallel == false with unnecessary branches removed.
221 : : *
222 : : * Note: the relation we build hash table on is the "inner"
223 : : * the other one is "outer".
224 : : * ----------------------------------------------------------------
225 : : */
226 : : static pg_always_inline TupleTableSlot *
227 : 8458701 : ExecHashJoinImpl(PlanState *pstate, bool parallel)
228 : : {
229 : 8458701 : HashJoinState *node = castNode(HashJoinState, pstate);
230 : : PlanState *outerNode;
231 : : HashState *hashNode;
232 : : ExprState *joinqual;
233 : : ExprState *otherqual;
234 : : ExprContext *econtext;
235 : : HashJoinTable hashtable;
236 : : TupleTableSlot *outerTupleSlot;
237 : : uint32 hashvalue;
238 : : bool is_null;
239 : : int batchno;
240 : : ParallelHashJoinState *parallel_state;
241 : :
242 : : /*
243 : : * get information from HashJoin node
244 : : */
245 : 8458701 : joinqual = node->js.joinqual;
246 : 8458701 : otherqual = node->js.ps.qual;
247 : 8458701 : hashNode = (HashState *) innerPlanState(node);
248 : 8458701 : outerNode = outerPlanState(node);
249 : 8458701 : hashtable = node->hj_HashTable;
250 : 8458701 : econtext = node->js.ps.ps_ExprContext;
251 : 8458701 : parallel_state = hashNode->parallel_state;
252 : :
253 : : /*
254 : : * Reset per-tuple memory context to free any expression evaluation
255 : : * storage allocated in the previous tuple cycle.
256 : : */
257 : 8458701 : ResetExprContext(econtext);
258 : :
259 : : /*
260 : : * run the hash join state machine
261 : : */
262 : : for (;;)
263 : : {
264 : : /*
265 : : * It's possible to iterate this loop many times before returning a
266 : : * tuple, in some pathological cases such as needing to move much of
267 : : * the current batch to a later batch. So let's check for interrupts
268 : : * each time through.
269 : : */
270 [ + + ]: 30947821 : CHECK_FOR_INTERRUPTS();
271 : :
272 [ + + + + : 30947821 : switch (node->hj_JoinState)
+ + + +
- ]
273 : : {
274 : 24672 : case HJ_BUILD_HASHTABLE:
275 : :
276 : : /*
277 : : * First time through: build hash table for inner relation.
278 : : */
279 : : Assert(hashtable == NULL);
280 : :
281 : : /*
282 : : * If the outer relation is completely empty, and it's not
283 : : * right/right-anti/full join, we can quit without building
284 : : * the hash table. However, for an inner join it is only a
285 : : * win to check this when the outer relation's startup cost is
286 : : * less than the projected cost of building the hash table.
287 : : * Otherwise it's best to build the hash table first and see
288 : : * if the inner relation is empty. (When it's a left join, we
289 : : * should always make this check, since we aren't going to be
290 : : * able to skip the join on the strength of an empty inner
291 : : * relation anyway.)
292 : : *
293 : : * If we are rescanning the join, we make use of information
294 : : * gained on the previous scan: don't bother to try the
295 : : * prefetch if the previous scan found the outer relation
296 : : * nonempty. This is not 100% reliable since with new
297 : : * parameters the outer relation might yield different
298 : : * results, but it's a good heuristic.
299 : : *
300 : : * The only way to make the check is to try to fetch a tuple
301 : : * from the outer plan node. If we succeed, we have to stash
302 : : * it away for later consumption by ExecHashJoinOuterGetTuple.
303 : : */
304 [ + + ]: 24672 : if (HJ_FILL_INNER(node))
305 : : {
306 : : /* no chance to not build the hash table */
307 : 4608 : node->hj_FirstOuterTupleSlot = NULL;
308 : : }
309 [ + + ]: 20064 : else if (parallel)
310 : : {
311 : : /*
312 : : * The empty-outer optimization is not implemented for
313 : : * shared hash tables, because no one participant can
314 : : * determine that there are no outer tuples, and it's not
315 : : * yet clear that it's worth the synchronization overhead
316 : : * of reaching consensus to figure that out. So we have
317 : : * to build the hash table.
318 : : */
319 : 228 : node->hj_FirstOuterTupleSlot = NULL;
320 : : }
321 [ + + ]: 19836 : else if (HJ_FILL_OUTER(node) ||
322 [ + + ]: 15627 : (outerNode->plan->startup_cost < hashNode->ps.plan->total_cost &&
323 [ + + ]: 13780 : !node->hj_OuterNotEmpty))
324 : : {
325 : 16718 : node->hj_FirstOuterTupleSlot = ExecProcNode(outerNode);
326 [ + + + + ]: 16718 : if (TupIsNull(node->hj_FirstOuterTupleSlot))
327 : : {
328 : 3666 : node->hj_OuterNotEmpty = false;
329 : 3666 : return NULL;
330 : : }
331 : : else
332 : 13052 : node->hj_OuterNotEmpty = true;
333 : : }
334 : : else
335 : 3118 : node->hj_FirstOuterTupleSlot = NULL;
336 : :
337 : : /*
338 : : * Create the hash table. If using Parallel Hash, then
339 : : * whoever gets here first will create the hash table and any
340 : : * later arrivals will merely attach to it.
341 : : */
342 : 21006 : hashtable = ExecHashTableCreate(hashNode);
343 : 21006 : node->hj_HashTable = hashtable;
344 : :
345 : : /*
346 : : * Execute the Hash node, to build the hash table. If using
347 : : * Parallel Hash, then we'll try to help hashing unless we
348 : : * arrived too late.
349 : : */
350 : 21006 : hashNode->hashtable = hashtable;
351 : 21006 : (void) MultiExecProcNode((PlanState *) hashNode);
352 : :
353 : : /*
354 : : * If the inner relation is completely empty, and we're not
355 : : * doing a left outer join, we can quit without scanning the
356 : : * outer relation. (If the inner relation contains only
357 : : * null-keyed tuples that we need to emit, we'll fall through
358 : : * and do the outer-relation scan. In principle we could go
359 : : * emit those tuples then quit, but it would complicate the
360 : : * state machine logic. The case seems rare enough to not be
361 : : * worth optimizing.)
362 : : */
363 [ + + ]: 21006 : if (hashtable->totalTuples == 0 &&
364 [ + + ]: 3246 : hashNode->null_tuple_store == NULL &&
365 [ + + ]: 3218 : !HJ_FILL_OUTER(node))
366 : : {
367 [ - + ]: 3022 : if (parallel)
368 : : {
369 : : /*
370 : : * Advance the build barrier to PHJ_BUILD_RUN before
371 : : * proceeding so we can negotiate resource cleanup.
372 : : */
373 : 0 : Barrier *build_barrier = ¶llel_state->build_barrier;
374 : :
375 [ # # ]: 0 : while (BarrierPhase(build_barrier) < PHJ_BUILD_RUN)
376 : 0 : BarrierArriveAndWait(build_barrier, 0);
377 : : }
378 : 3022 : return NULL;
379 : : }
380 : :
381 : : /*
382 : : * need to remember whether nbatch has increased since we
383 : : * began scanning the outer relation
384 : : */
385 : 17984 : hashtable->nbatch_outstart = hashtable->nbatch;
386 : :
387 : : /*
388 : : * Reset OuterNotEmpty for scan. (It's OK if we fetched a
389 : : * tuple above, because ExecHashJoinOuterGetTuple will
390 : : * immediately set it again.)
391 : : */
392 : 17984 : node->hj_OuterNotEmpty = false;
393 : :
394 [ + + ]: 17984 : if (parallel)
395 : 276 : {
396 : : Barrier *build_barrier;
397 : :
398 : 276 : build_barrier = ¶llel_state->build_barrier;
399 : : Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASH_OUTER ||
400 : : BarrierPhase(build_barrier) == PHJ_BUILD_RUN ||
401 : : BarrierPhase(build_barrier) == PHJ_BUILD_FREE);
402 [ + + ]: 276 : if (BarrierPhase(build_barrier) == PHJ_BUILD_HASH_OUTER)
403 : : {
404 : : /*
405 : : * If multi-batch, we need to hash the outer relation
406 : : * up front.
407 : : */
408 [ + + ]: 217 : if (hashtable->nbatch > 1)
409 : 120 : ExecParallelHashJoinPartitionOuter(node);
410 : 217 : BarrierArriveAndWait(build_barrier,
411 : : WAIT_EVENT_HASH_BUILD_HASH_OUTER);
412 : : Assert(BarrierPhase(build_barrier) == PHJ_BUILD_RUN);
413 : : }
414 : :
415 : : /*
416 : : * Each backend should now select a batch to work on.
417 : : * However, if we've already collected some null-keyed
418 : : * tuples, dump them first. (That is critical when we
419 : : * arrive late enough that no more batches are available;
420 : : * otherwise we'd fail to dump those tuples at all.)
421 : : */
422 : 276 : hashtable->curbatch = -1;
423 : :
424 [ + + ]: 276 : if (node->hj_NullOuterTupleStore)
425 : 4 : node->hj_JoinState = HJ_FILL_OUTER_NULL_TUPLES;
426 [ + + ]: 272 : else if (hashNode->null_tuple_store)
427 : 10 : node->hj_JoinState = HJ_FILL_INNER_NULL_TUPLES;
428 : : else
429 : 262 : node->hj_JoinState = HJ_NEED_NEW_BATCH;
430 : :
431 : 276 : continue;
432 : : }
433 : : else
434 : 17708 : node->hj_JoinState = HJ_NEED_NEW_OUTER;
435 : :
436 : : pg_fallthrough;
437 : :
438 : 15356974 : case HJ_NEED_NEW_OUTER:
439 : :
440 : : /*
441 : : * We don't have an outer tuple, try to get the next one
442 : : */
443 [ + + ]: 15356974 : if (parallel)
444 : : outerTupleSlot =
445 : 1444671 : ExecParallelHashJoinOuterGetTuple(outerNode, node,
446 : : &hashvalue,
447 : : &is_null);
448 : : else
449 : : outerTupleSlot =
450 : 13912303 : ExecHashJoinOuterGetTuple(outerNode, node,
451 : : &hashvalue,
452 : : &is_null);
453 : :
454 [ + + - + ]: 15356974 : if (TupIsNull(outerTupleSlot))
455 : : {
456 : : /* end of batch, or maybe whole join */
457 [ + + ]: 20185 : if (HJ_FILL_INNER(node))
458 : : {
459 : : /* set up to scan for unmatched inner tuples */
460 [ + + ]: 4363 : if (parallel)
461 : : {
462 : : /*
463 : : * Only one process is currently allowed to handle
464 : : * each batch's unmatched tuples, in a parallel
465 : : * join. However, each process must deal with any
466 : : * null-keyed tuples it found.
467 : : */
468 [ + + ]: 69 : if (ExecParallelPrepHashTableForUnmatched(node))
469 : 44 : node->hj_JoinState = HJ_FILL_INNER_TUPLES;
470 [ - + ]: 25 : else if (node->hj_NullOuterTupleStore)
471 : 0 : node->hj_JoinState = HJ_FILL_OUTER_NULL_TUPLES;
472 [ + + ]: 25 : else if (hashNode->null_tuple_store)
473 : 10 : node->hj_JoinState = HJ_FILL_INNER_NULL_TUPLES;
474 : : else
475 : 15 : node->hj_JoinState = HJ_NEED_NEW_BATCH;
476 : : }
477 : : else
478 : : {
479 : 4294 : ExecPrepHashTableForUnmatched(node);
480 : 4294 : node->hj_JoinState = HJ_FILL_INNER_TUPLES;
481 : : }
482 : : }
483 : : else
484 : : {
485 : : /* might have outer null-keyed tuples to fill */
486 : : Assert(hashNode->null_tuple_store == NULL);
487 [ - + ]: 15822 : if (node->hj_NullOuterTupleStore)
488 : 0 : node->hj_JoinState = HJ_FILL_OUTER_NULL_TUPLES;
489 : : else
490 : 15822 : node->hj_JoinState = HJ_NEED_NEW_BATCH;
491 : : }
492 : 20185 : continue;
493 : : }
494 : :
495 : 15336789 : econtext->ecxt_outertuple = outerTupleSlot;
496 : 15336789 : node->hj_MatchedOuter = false;
497 : :
498 : : /*
499 : : * If the tuple can't match anything, skip all the tedium and
500 : : * go directly to emitting it in null-extended form.
501 : : */
502 [ + + ]: 15336789 : if (is_null)
503 : : {
504 : 129 : node->hj_JoinState = HJ_FILL_OUTER_TUPLE;
505 : 129 : continue;
506 : : }
507 : :
508 : : /*
509 : : * Find the corresponding bucket for this tuple in the main
510 : : * hash table or skew hash table.
511 : : */
512 : 15336660 : node->hj_CurHashValue = hashvalue;
513 : 15336660 : ExecHashGetBucketAndBatch(hashtable, hashvalue,
514 : : &node->hj_CurBucketNo, &batchno);
515 : 15336660 : node->hj_CurSkewBucketNo = ExecHashGetSkewBucket(hashtable,
516 : : hashvalue);
517 : 15336660 : node->hj_CurTuple = NULL;
518 : :
519 : : /*
520 : : * The tuple might not belong to the current batch (where
521 : : * "current batch" includes the skew buckets if any).
522 : : */
523 [ + + ]: 15336660 : if (batchno != hashtable->curbatch &&
524 [ + + ]: 959656 : node->hj_CurSkewBucketNo == INVALID_SKEW_BUCKET_NO)
525 : 958856 : {
526 : : bool shouldFree;
527 : 958856 : MinimalTuple mintuple = ExecFetchSlotMinimalTuple(outerTupleSlot,
528 : : &shouldFree);
529 : :
530 : : /*
531 : : * Need to postpone this outer tuple to a later batch.
532 : : * Save it in the corresponding outer-batch file.
533 : : */
534 : : Assert(parallel_state == NULL);
535 : : Assert(batchno > hashtable->curbatch);
536 : 958856 : ExecHashJoinSaveTuple(mintuple, hashvalue,
537 : 958856 : &hashtable->outerBatchFile[batchno],
538 : : hashtable);
539 : :
540 [ + - ]: 958856 : if (shouldFree)
541 : 958856 : heap_free_minimal_tuple(mintuple);
542 : :
543 : : /* Loop around, staying in HJ_NEED_NEW_OUTER state */
544 : 958856 : continue;
545 : : }
546 : :
547 : : /* OK, let's scan the bucket for matches */
548 : 14377804 : node->hj_JoinState = HJ_SCAN_BUCKET;
549 : :
550 : : pg_fallthrough;
551 : :
552 : 19112424 : case HJ_SCAN_BUCKET:
553 : :
554 : : /*
555 : : * Scan the selected hash bucket for matches to current outer
556 : : */
557 [ + + ]: 19112424 : if (parallel)
558 : : {
559 [ + + ]: 2804080 : if (!ExecParallelScanHashBucket(node, econtext))
560 : : {
561 : : /* out of matches; check for possible outer-join fill */
562 : 1444028 : node->hj_JoinState = HJ_FILL_OUTER_TUPLE;
563 : 1444028 : continue;
564 : : }
565 : : }
566 : : else
567 : : {
568 [ + + ]: 16308344 : if (!ExecScanHashBucket(node, econtext))
569 : : {
570 : : /* out of matches; check for possible outer-join fill */
571 : 9088995 : node->hj_JoinState = HJ_FILL_OUTER_TUPLE;
572 : 9088995 : continue;
573 : : }
574 : : }
575 : :
576 : : /*
577 : : * In a right-semijoin, we only need the first match for each
578 : : * inner tuple.
579 : : */
580 [ + + + + ]: 8581285 : if (node->js.jointype == JOIN_RIGHT_SEMI &&
581 : 1884 : HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(node->hj_CurTuple)))
582 : 808 : continue;
583 : :
584 : : /*
585 : : * We've got a match, but still need to test non-hashed quals.
586 : : * ExecScanHashBucket already set up all the state needed to
587 : : * call ExecQual.
588 : : *
589 : : * If we pass the qual, then save state for next call and have
590 : : * ExecProject form the projection, store it in the tuple
591 : : * table, and return the slot.
592 : : *
593 : : * Only the joinquals determine tuple match status, but all
594 : : * quals must pass to actually return the tuple.
595 : : */
596 [ + + + + ]: 8578593 : if (joinqual == NULL || ExecQual(joinqual, econtext))
597 : : {
598 : 8469673 : node->hj_MatchedOuter = true;
599 : :
600 : : /*
601 : : * This is really only needed if HJ_FILL_INNER(node) or if
602 : : * we are in a right-semijoin, but we'll avoid the branch
603 : : * and just set it always.
604 : : */
605 [ + + ]: 8469673 : if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(node->hj_CurTuple)))
606 : 3977542 : HeapTupleHeaderSetMatch(HJTUPLE_MINTUPLE(node->hj_CurTuple));
607 : :
608 : : /* In an antijoin, we never return a matched tuple */
609 [ + + ]: 8469673 : if (node->js.jointype == JOIN_ANTI)
610 : : {
611 : 849530 : node->hj_JoinState = HJ_NEED_NEW_OUTER;
612 : 849530 : continue;
613 : : }
614 : :
615 : : /*
616 : : * If we only need to consider the first matching inner
617 : : * tuple, then advance to next outer tuple after we've
618 : : * processed this one.
619 : : */
620 [ + + ]: 7620143 : if (node->js.single_match)
621 : 2995172 : node->hj_JoinState = HJ_NEED_NEW_OUTER;
622 : :
623 : : /*
624 : : * In a right-antijoin, we never return a matched tuple.
625 : : * If it's not an inner_unique join, we need to stay on
626 : : * the current outer tuple to continue scanning the inner
627 : : * side for matches.
628 : : */
629 [ + + ]: 7620143 : if (node->js.jointype == JOIN_RIGHT_ANTI)
630 : 49889 : continue;
631 : :
632 [ + + + + ]: 7570254 : if (otherqual == NULL || ExecQual(otherqual, econtext))
633 : 7440890 : return ExecProject(node->js.ps.ps_ProjInfo);
634 : : else
635 [ - + ]: 129364 : InstrCountFiltered2(node, 1);
636 : : }
637 : : else
638 [ - + ]: 108920 : InstrCountFiltered1(node, 1);
639 : 238284 : break;
640 : :
641 : 10533152 : case HJ_FILL_OUTER_TUPLE:
642 : :
643 : : /*
644 : : * The current outer tuple has run out of matches, so check
645 : : * whether to emit a dummy outer-join tuple. Whether we emit
646 : : * one or not, the next state is NEED_NEW_OUTER.
647 : : */
648 : 10533152 : node->hj_JoinState = HJ_NEED_NEW_OUTER;
649 : :
650 [ + + ]: 10533152 : if (!node->hj_MatchedOuter &&
651 [ + + ]: 7314538 : HJ_FILL_OUTER(node))
652 : : {
653 : : /*
654 : : * Generate a fake join tuple with nulls for the inner
655 : : * tuple, and return it if it passes the non-join quals.
656 : : */
657 : 1400868 : econtext->ecxt_innertuple = node->hj_NullInnerTupleSlot;
658 : :
659 [ + + + + ]: 1400868 : if (otherqual == NULL || ExecQual(otherqual, econtext))
660 : 705739 : return ExecProject(node->js.ps.ps_ProjInfo);
661 : : else
662 [ - + ]: 695129 : InstrCountFiltered2(node, 1);
663 : : }
664 : 9827413 : break;
665 : :
666 : 295491 : case HJ_FILL_INNER_TUPLES:
667 : :
668 : : /*
669 : : * We have finished a batch, but we are doing
670 : : * right/right-anti/full join, so any unmatched inner tuples
671 : : * in the hashtable have to be emitted before we continue to
672 : : * the next batch.
673 : : */
674 [ + + + + ]: 510934 : if (!(parallel ? ExecParallelScanHashTableForUnmatched(node, econtext)
675 : 215443 : : ExecScanHashTableForUnmatched(node, econtext)))
676 : : {
677 : : /* no more unmatched tuples, but maybe there are nulls */
678 [ - + ]: 4334 : if (node->hj_NullOuterTupleStore)
679 : 0 : node->hj_JoinState = HJ_FILL_OUTER_NULL_TUPLES;
680 [ + + ]: 4334 : else if (hashNode->null_tuple_store)
681 : 73 : node->hj_JoinState = HJ_FILL_INNER_NULL_TUPLES;
682 : : else
683 : 4261 : node->hj_JoinState = HJ_NEED_NEW_BATCH;
684 : 4334 : continue;
685 : : }
686 : :
687 : : /*
688 : : * Generate a fake join tuple with nulls for the outer tuple,
689 : : * and return it if it passes the non-join quals.
690 : : */
691 : 291157 : econtext->ecxt_outertuple = node->hj_NullOuterTupleSlot;
692 : :
693 [ + + + + ]: 291157 : if (otherqual == NULL || ExecQual(otherqual, econtext))
694 : 286292 : return ExecProject(node->js.ps.ps_ProjInfo);
695 : : else
696 [ - + ]: 4865 : InstrCountFiltered2(node, 1);
697 : 4865 : break;
698 : :
699 : 8 : case HJ_FILL_OUTER_NULL_TUPLES:
700 : :
701 : : /*
702 : : * We have finished a batch, but we are doing left/full join,
703 : : * so any null-keyed outer tuples have to be emitted before we
704 : : * continue to the next batch.
705 : : *
706 : : * (We could delay this till the end of the join, but there
707 : : * seems little percentage in that.)
708 : : *
709 : : * We have to use tuplestore_gettupleslot_force because
710 : : * hj_OuterTupleSlot may not be able to store a MinimalTuple.
711 : : */
712 [ + + ]: 16 : while (tuplestore_gettupleslot_force(node->hj_NullOuterTupleStore,
713 : : true, false,
714 : : node->hj_OuterTupleSlot))
715 : : {
716 : : /*
717 : : * Generate a fake join tuple with nulls for the inner
718 : : * tuple, and return it if it passes the non-join quals.
719 : : */
720 : 4 : econtext->ecxt_outertuple = node->hj_OuterTupleSlot;
721 : 4 : econtext->ecxt_innertuple = node->hj_NullInnerTupleSlot;
722 : :
723 [ - + - - ]: 4 : if (otherqual == NULL || ExecQual(otherqual, econtext))
724 : 4 : return ExecProject(node->js.ps.ps_ProjInfo);
725 : : else
726 [ # # ]: 0 : InstrCountFiltered2(node, 1);
727 : :
728 : 0 : ResetExprContext(econtext);
729 : :
730 : : /* allow this loop to be cancellable */
731 [ # # ]: 0 : CHECK_FOR_INTERRUPTS();
732 : : }
733 : :
734 : : /* We don't need the tuplestore any more, so discard it. */
735 : 4 : tuplestore_end(node->hj_NullOuterTupleStore);
736 : 4 : node->hj_NullOuterTupleStore = NULL;
737 : :
738 : : /* Fill inner tuples too if it's a full join, else advance. */
739 [ + + ]: 4 : if (hashNode->null_tuple_store)
740 : 2 : node->hj_JoinState = HJ_FILL_INNER_NULL_TUPLES;
741 : : else
742 : 2 : node->hj_JoinState = HJ_NEED_NEW_BATCH;
743 : 4 : break;
744 : :
745 : 159 : case HJ_FILL_INNER_NULL_TUPLES:
746 : :
747 : : /*
748 : : * We have finished a batch, but we are doing
749 : : * right/right-anti/full join, so any null-keyed inner tuples
750 : : * have to be emitted before we continue to the next batch.
751 : : *
752 : : * (We could delay this till the end of the join, but there
753 : : * seems little percentage in that.)
754 : : */
755 [ + + ]: 322 : while (tuplestore_gettupleslot(hashNode->null_tuple_store,
756 : : true, false,
757 : : node->hj_HashTupleSlot))
758 : : {
759 : : /*
760 : : * Generate a fake join tuple with nulls for the outer
761 : : * tuple, and return it if it passes the non-join quals.
762 : : */
763 : 72 : econtext->ecxt_outertuple = node->hj_NullOuterTupleSlot;
764 : 72 : econtext->ecxt_innertuple = node->hj_HashTupleSlot;
765 : :
766 [ + + + + ]: 72 : if (otherqual == NULL || ExecQual(otherqual, econtext))
767 : 68 : return ExecProject(node->js.ps.ps_ProjInfo);
768 : : else
769 [ - + ]: 4 : InstrCountFiltered2(node, 1);
770 : :
771 : 4 : ResetExprContext(econtext);
772 : :
773 : : /* allow this loop to be cancellable */
774 [ - + ]: 4 : CHECK_FOR_INTERRUPTS();
775 : : }
776 : :
777 : : /*
778 : : * Ideally we'd discard the tuplestore now, but we can't
779 : : * because we might need it for rescans.
780 : : */
781 : :
782 : : /* Now we can advance to the next batch. */
783 : 91 : node->hj_JoinState = HJ_NEED_NEW_BATCH;
784 : 91 : break;
785 : :
786 : 20453 : case HJ_NEED_NEW_BATCH:
787 : :
788 : : /*
789 : : * Try to advance to next batch. Done if there are no more.
790 : : */
791 [ + + ]: 20453 : if (parallel)
792 : : {
793 [ + + ]: 911 : if (!ExecParallelHashJoinNewBatch(node))
794 : 276 : return NULL; /* end of parallel-aware join */
795 : : }
796 : : else
797 : : {
798 [ + + ]: 19542 : if (!ExecHashJoinNewBatch(node))
799 : 18744 : return NULL; /* end of parallel-oblivious join */
800 : : }
801 : 1433 : node->hj_JoinState = HJ_NEED_NEW_OUTER;
802 : 1433 : break;
803 : :
804 : 0 : default:
805 [ # # ]: 0 : elog(ERROR, "unrecognized hashjoin state: %d",
806 : : (int) node->hj_JoinState);
807 : : }
808 : : }
809 : : }
810 : :
811 : : /* ----------------------------------------------------------------
812 : : * ExecHashJoin
813 : : *
814 : : * Parallel-oblivious version.
815 : : * ----------------------------------------------------------------
816 : : */
817 : : static TupleTableSlot * /* return: a tuple or NULL */
818 : 6938333 : ExecHashJoin(PlanState *pstate)
819 : : {
820 : : /*
821 : : * On sufficiently smart compilers this should be inlined with the
822 : : * parallel-aware branches removed.
823 : : */
824 : 6938333 : return ExecHashJoinImpl(pstate, false);
825 : : }
826 : :
827 : : /* ----------------------------------------------------------------
828 : : * ExecParallelHashJoin
829 : : *
830 : : * Parallel-aware version.
831 : : * ----------------------------------------------------------------
832 : : */
833 : : static TupleTableSlot * /* return: a tuple or NULL */
834 : 1520368 : ExecParallelHashJoin(PlanState *pstate)
835 : : {
836 : : /*
837 : : * On sufficiently smart compilers this should be inlined with the
838 : : * parallel-oblivious branches removed.
839 : : */
840 : 1520368 : return ExecHashJoinImpl(pstate, true);
841 : : }
842 : :
843 : : /* ----------------------------------------------------------------
844 : : * ExecInitHashJoin
845 : : *
846 : : * Init routine for HashJoin node.
847 : : * ----------------------------------------------------------------
848 : : */
849 : : HashJoinState *
850 : 29174 : ExecInitHashJoin(HashJoin *node, EState *estate, int eflags)
851 : : {
852 : : HashJoinState *hjstate;
853 : : Plan *outerNode;
854 : : Hash *hashNode;
855 : : TupleDesc outerDesc,
856 : : innerDesc;
857 : : const TupleTableSlotOps *ops;
858 : :
859 : : /* check for unsupported flags */
860 : : Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
861 : :
862 : : /*
863 : : * create state structure
864 : : */
865 : 29174 : hjstate = makeNode(HashJoinState);
866 : 29174 : hjstate->js.ps.plan = (Plan *) node;
867 : 29174 : hjstate->js.ps.state = estate;
868 : :
869 : : /*
870 : : * See ExecHashJoinInitializeDSM() and ExecHashJoinInitializeWorker()
871 : : * where this function may be replaced with a parallel version, if we
872 : : * managed to launch a parallel query.
873 : : */
874 : 29174 : hjstate->js.ps.ExecProcNode = ExecHashJoin;
875 : 29174 : hjstate->js.jointype = node->join.jointype;
876 : :
877 : : /*
878 : : * Miscellaneous initialization
879 : : *
880 : : * create expression context for node
881 : : */
882 : 29174 : ExecAssignExprContext(estate, &hjstate->js.ps);
883 : :
884 : : /*
885 : : * initialize child nodes
886 : : *
887 : : * Note: we could suppress the REWIND flag for the inner input, which
888 : : * would amount to betting that the hash will be a single batch. Not
889 : : * clear if this would be a win or not.
890 : : */
891 : 29174 : outerNode = outerPlan(node);
892 : 29174 : hashNode = (Hash *) innerPlan(node);
893 : :
894 : 29174 : outerPlanState(hjstate) = ExecInitNode(outerNode, estate, eflags);
895 : 29174 : outerDesc = ExecGetResultType(outerPlanState(hjstate));
896 : 29174 : innerPlanState(hjstate) = ExecInitNode((Plan *) hashNode, estate, eflags);
897 : 29174 : innerDesc = ExecGetResultType(innerPlanState(hjstate));
898 : :
899 : : /*
900 : : * Initialize result slot, type and projection.
901 : : */
902 : 29174 : ExecInitResultTupleSlotTL(&hjstate->js.ps, &TTSOpsVirtual);
903 : 29174 : ExecAssignProjectionInfo(&hjstate->js.ps, NULL);
904 : :
905 : : /*
906 : : * tuple table initialization
907 : : */
908 : 29174 : ops = ExecGetResultSlotOps(outerPlanState(hjstate), NULL);
909 : 29174 : hjstate->hj_OuterTupleSlot = ExecInitExtraTupleSlot(estate, outerDesc,
910 : : ops);
911 : :
912 : : /*
913 : : * detect whether we need only consider the first matching inner tuple
914 : : */
915 [ + + ]: 42389 : hjstate->js.single_match = (node->join.inner_unique ||
916 [ + + ]: 13215 : node->join.jointype == JOIN_SEMI);
917 : :
918 : : /* set up null tuples for outer joins, if needed */
919 [ + + + + : 29174 : switch (node->join.jointype)
- ]
920 : : {
921 : 18992 : case JOIN_INNER:
922 : : case JOIN_SEMI:
923 : : case JOIN_RIGHT_SEMI:
924 : 18992 : break;
925 : 4815 : case JOIN_LEFT:
926 : : case JOIN_ANTI:
927 : 4815 : hjstate->hj_NullInnerTupleSlot =
928 : 4815 : ExecInitNullTupleSlot(estate, innerDesc, &TTSOpsVirtual);
929 : 4815 : break;
930 : 4602 : case JOIN_RIGHT:
931 : : case JOIN_RIGHT_ANTI:
932 : 4602 : hjstate->hj_NullOuterTupleSlot =
933 : 4602 : ExecInitNullTupleSlot(estate, outerDesc, &TTSOpsVirtual);
934 : 4602 : break;
935 : 765 : case JOIN_FULL:
936 : 765 : hjstate->hj_NullOuterTupleSlot =
937 : 765 : ExecInitNullTupleSlot(estate, outerDesc, &TTSOpsVirtual);
938 : 765 : hjstate->hj_NullInnerTupleSlot =
939 : 765 : ExecInitNullTupleSlot(estate, innerDesc, &TTSOpsVirtual);
940 : 765 : break;
941 : 0 : default:
942 [ # # ]: 0 : elog(ERROR, "unrecognized join type: %d",
943 : : (int) node->join.jointype);
944 : : }
945 : :
946 : : /*
947 : : * now for some voodoo. our temporary tuple slot is actually the result
948 : : * tuple slot of the Hash node (which is our inner plan). we can do this
949 : : * because Hash nodes don't return tuples via ExecProcNode() -- instead
950 : : * the hash join node uses ExecScanHashBucket() to get at the contents of
951 : : * the hash table. -cim 6/9/91
952 : : */
953 : : {
954 : 29174 : HashState *hashstate = (HashState *) innerPlanState(hjstate);
955 : 29174 : Hash *hash = (Hash *) hashstate->ps.plan;
956 : 29174 : TupleTableSlot *slot = hashstate->ps.ps_ResultTupleSlot;
957 : : Oid *outer_hashfuncid;
958 : : Oid *inner_hashfuncid;
959 : : bool *hash_strict;
960 : : ListCell *lc;
961 : : int nkeys;
962 : :
963 : :
964 : 29174 : hjstate->hj_HashTupleSlot = slot;
965 : :
966 : : /*
967 : : * Build ExprStates to obtain hash values for either side of the join.
968 : : * Note: must build the ExprStates before ExecHashTableCreate() so we
969 : : * properly attribute any SubPlans that exist in the hash expressions
970 : : * to the correct PlanState.
971 : : */
972 : 29174 : nkeys = list_length(node->hashoperators);
973 : :
974 : 29174 : outer_hashfuncid = palloc_array(Oid, nkeys);
975 : 29174 : inner_hashfuncid = palloc_array(Oid, nkeys);
976 : 29174 : hash_strict = palloc_array(bool, nkeys);
977 : :
978 : : /*
979 : : * Determine the hash function for each side of the join for the given
980 : : * join operator, and detect whether the join operator is strict.
981 : : */
982 [ + - + + : 60781 : foreach(lc, node->hashoperators)
+ + ]
983 : : {
984 : 31607 : Oid hashop = lfirst_oid(lc);
985 : 31607 : int i = foreach_current_index(lc);
986 : :
987 [ - + ]: 31607 : if (!get_op_hash_functions(hashop,
988 : 31607 : &outer_hashfuncid[i],
989 : 31607 : &inner_hashfuncid[i]))
990 [ # # ]: 0 : elog(ERROR,
991 : : "could not find hash function for hash operator %u",
992 : : hashop);
993 : 31607 : hash_strict[i] = op_strict(hashop);
994 : : }
995 : :
996 : : /*
997 : : * Build an ExprState to generate the hash value for the expressions
998 : : * on the outer side of the join.
999 : : */
1000 : 29174 : hjstate->hj_OuterHash =
1001 : 29174 : ExecBuildHash32Expr(hjstate->js.ps.ps_ResultTupleDesc,
1002 : : hjstate->js.ps.resultops,
1003 : : outer_hashfuncid,
1004 : 29174 : node->hashcollations,
1005 : 29174 : node->hashkeys,
1006 : : hash_strict,
1007 : : &hjstate->js.ps,
1008 : : 0);
1009 : :
1010 : : /* As above, but for the inner side of the join */
1011 : 29174 : hashstate->hash_expr =
1012 : 29174 : ExecBuildHash32Expr(hashstate->ps.ps_ResultTupleDesc,
1013 : : hashstate->ps.resultops,
1014 : : inner_hashfuncid,
1015 : 29174 : node->hashcollations,
1016 : 29174 : hash->hashkeys,
1017 : : hash_strict,
1018 : : &hashstate->ps,
1019 : : 0);
1020 : :
1021 : : /* Remember whether we need to save tuples with null join keys */
1022 : 29174 : hjstate->hj_KeepNullTuples = HJ_FILL_OUTER(hjstate);
1023 : 29174 : hashstate->keep_null_tuples = HJ_FILL_INNER(hjstate);
1024 : :
1025 : : /*
1026 : : * Set up the skew table hash function while we have a record of the
1027 : : * first key's hash function Oid.
1028 : : */
1029 [ + + ]: 29174 : if (OidIsValid(hash->skewTable))
1030 : : {
1031 : 22173 : hashstate->skew_hashfunction = palloc0_object(FmgrInfo);
1032 : 22173 : hashstate->skew_collation = linitial_oid(node->hashcollations);
1033 : 22173 : fmgr_info(outer_hashfuncid[0], hashstate->skew_hashfunction);
1034 : : }
1035 : :
1036 : : /* no need to keep these */
1037 : 29174 : pfree(outer_hashfuncid);
1038 : 29174 : pfree(inner_hashfuncid);
1039 : 29174 : pfree(hash_strict);
1040 : : }
1041 : :
1042 : : /*
1043 : : * initialize child expressions
1044 : : */
1045 : 29174 : hjstate->js.ps.qual =
1046 : 29174 : ExecInitQual(node->join.plan.qual, (PlanState *) hjstate);
1047 : 29174 : hjstate->js.joinqual =
1048 : 29174 : ExecInitQual(node->join.joinqual, (PlanState *) hjstate);
1049 : 29174 : hjstate->hashclauses =
1050 : 29174 : ExecInitQual(node->hashclauses, (PlanState *) hjstate);
1051 : :
1052 : : /*
1053 : : * initialize hash-specific info
1054 : : */
1055 : 29174 : hjstate->hj_HashTable = NULL;
1056 : 29174 : hjstate->hj_NullOuterTupleStore = NULL;
1057 : 29174 : hjstate->hj_FirstOuterTupleSlot = NULL;
1058 : :
1059 : 29174 : hjstate->hj_CurHashValue = 0;
1060 : 29174 : hjstate->hj_CurBucketNo = 0;
1061 : 29174 : hjstate->hj_CurSkewBucketNo = INVALID_SKEW_BUCKET_NO;
1062 : 29174 : hjstate->hj_CurTuple = NULL;
1063 : :
1064 : 29174 : hjstate->hj_JoinState = HJ_BUILD_HASHTABLE;
1065 : 29174 : hjstate->hj_MatchedOuter = false;
1066 : 29174 : hjstate->hj_OuterNotEmpty = false;
1067 : :
1068 : 29174 : return hjstate;
1069 : : }
1070 : :
1071 : : /* ----------------------------------------------------------------
1072 : : * ExecEndHashJoin
1073 : : *
1074 : : * clean up routine for HashJoin node
1075 : : * ----------------------------------------------------------------
1076 : : */
1077 : : void
1078 : 29102 : ExecEndHashJoin(HashJoinState *node)
1079 : : {
1080 : 29102 : HashState *hashNode = castNode(HashState, innerPlanState(node));
1081 : :
1082 : : /*
1083 : : * Free tuple stores if we made them (must do this before
1084 : : * ExecHashTableDestroy deletes hashCxt)
1085 : : */
1086 [ - + ]: 29102 : if (node->hj_NullOuterTupleStore)
1087 : : {
1088 : 0 : tuplestore_end(node->hj_NullOuterTupleStore);
1089 : 0 : node->hj_NullOuterTupleStore = NULL;
1090 : : }
1091 [ + + ]: 29102 : if (hashNode->null_tuple_store)
1092 : : {
1093 : 64 : tuplestore_end(hashNode->null_tuple_store);
1094 : 64 : hashNode->null_tuple_store = NULL;
1095 : : }
1096 : :
1097 : : /*
1098 : : * Free hash table
1099 : : */
1100 [ + + ]: 29102 : if (node->hj_HashTable)
1101 : : {
1102 : 19420 : ExecHashTableDestroy(node->hj_HashTable);
1103 : 19420 : node->hj_HashTable = NULL;
1104 : : }
1105 : :
1106 : : /*
1107 : : * clean up subtrees
1108 : : */
1109 : 29102 : ExecEndNode(outerPlanState(node));
1110 : 29102 : ExecEndNode(innerPlanState(node));
1111 : 29102 : }
1112 : :
1113 : : /*
1114 : : * ExecHashJoinOuterGetTuple
1115 : : *
1116 : : * get the next outer tuple for a parallel oblivious hashjoin: either by
1117 : : * executing the outer plan node in the first pass, or from the temp
1118 : : * files for the hashjoin batches.
1119 : : *
1120 : : * Returns a null slot if no more outer tuples (within the current batch).
1121 : : *
1122 : : * On success, the tuple's hash value is stored at *hashvalue --- this is
1123 : : * either originally computed, or re-read from the temp file. Also, we
1124 : : * set *is_null to true if the tuple cannot have any matches due to
1125 : : * containing a null join key (in which case the hashvalue is meaningless).
1126 : : */
1127 : : static TupleTableSlot *
1128 : 13912303 : ExecHashJoinOuterGetTuple(PlanState *outerNode,
1129 : : HashJoinState *hjstate,
1130 : : uint32 *hashvalue,
1131 : : bool *is_null)
1132 : : {
1133 : 13912303 : HashJoinTable hashtable = hjstate->hj_HashTable;
1134 : 13912303 : int curbatch = hashtable->curbatch;
1135 : : TupleTableSlot *slot;
1136 : :
1137 [ + + ]: 13912303 : if (curbatch == 0) /* if it is the first pass */
1138 : : {
1139 : : /*
1140 : : * Check to see if first outer tuple was already fetched by
1141 : : * ExecHashJoin() and not used yet.
1142 : : */
1143 : 12952649 : slot = hjstate->hj_FirstOuterTupleSlot;
1144 [ + + + - ]: 12952649 : if (!TupIsNull(slot))
1145 : 11171 : hjstate->hj_FirstOuterTupleSlot = NULL;
1146 : : else
1147 : 12941478 : slot = ExecProcNode(outerNode);
1148 : :
1149 [ + + + + ]: 12953268 : while (!TupIsNull(slot))
1150 : : {
1151 : : /*
1152 : : * We have to compute the tuple's hash value.
1153 : : */
1154 : 12934516 : ExprContext *econtext = hjstate->js.ps.ps_ExprContext;
1155 : :
1156 : 12934516 : econtext->ecxt_outertuple = slot;
1157 : :
1158 : 12934516 : ResetExprContext(econtext);
1159 : :
1160 : 12934516 : *hashvalue = DatumGetUInt32(ExecEvalExprSwitchContext(hjstate->hj_OuterHash,
1161 : : econtext,
1162 : : is_null));
1163 : :
1164 [ + + ]: 12934516 : if (!*is_null)
1165 : : {
1166 : : /* normal case with a non-null join key */
1167 : : /* remember outer relation is not empty for possible rescan */
1168 : 12933776 : hjstate->hj_OuterNotEmpty = true;
1169 : :
1170 : 12933776 : return slot;
1171 : : }
1172 [ + + ]: 740 : else if (hjstate->hj_KeepNullTuples)
1173 : : {
1174 : : /* null join key, but we must return tuple to be emitted */
1175 : 121 : return slot;
1176 : : }
1177 : :
1178 : : /*
1179 : : * That tuple couldn't match because of a NULL, so discard it and
1180 : : * continue with the next one.
1181 : : */
1182 : 619 : slot = ExecProcNode(outerNode);
1183 : : }
1184 : : }
1185 [ + - ]: 959654 : else if (curbatch < hashtable->nbatch)
1186 : : {
1187 : 959654 : BufFile *file = hashtable->outerBatchFile[curbatch];
1188 : :
1189 : : /*
1190 : : * In outer-join cases, we could get here even though the batch file
1191 : : * is empty.
1192 : : */
1193 [ - + ]: 959654 : if (file == NULL)
1194 : 0 : return NULL;
1195 : :
1196 : 959654 : slot = ExecHashJoinGetSavedTuple(hjstate,
1197 : : file,
1198 : : hashvalue,
1199 : : hjstate->hj_OuterTupleSlot);
1200 [ + + + - ]: 959654 : if (!TupIsNull(slot))
1201 : : {
1202 : 958856 : *is_null = false; /* else it'd not be in the batch file */
1203 : 958856 : return slot;
1204 : : }
1205 : : }
1206 : :
1207 : : /* End of this batch */
1208 : 19550 : return NULL;
1209 : : }
1210 : :
1211 : : /*
1212 : : * ExecHashJoinOuterGetTuple variant for the parallel case.
1213 : : */
1214 : : static TupleTableSlot *
1215 : 1444671 : ExecParallelHashJoinOuterGetTuple(PlanState *outerNode,
1216 : : HashJoinState *hjstate,
1217 : : uint32 *hashvalue,
1218 : : bool *is_null)
1219 : : {
1220 : 1444671 : HashJoinTable hashtable = hjstate->hj_HashTable;
1221 : 1444671 : int curbatch = hashtable->curbatch;
1222 : : TupleTableSlot *slot;
1223 : :
1224 : : /*
1225 : : * In the Parallel Hash case we only run the outer plan directly for
1226 : : * single-batch hash joins. Otherwise we have to go to batch files, even
1227 : : * for batch 0.
1228 : : */
1229 [ + + + + ]: 1444671 : if (curbatch == 0 && hashtable->nbatch == 1)
1230 : : {
1231 : 644126 : slot = ExecProcNode(outerNode);
1232 : :
1233 [ + + + - ]: 644138 : while (!TupIsNull(slot))
1234 : : {
1235 : 644024 : ExprContext *econtext = hjstate->js.ps.ps_ExprContext;
1236 : :
1237 : 644024 : econtext->ecxt_outertuple = slot;
1238 : :
1239 : 644024 : ResetExprContext(econtext);
1240 : :
1241 : 644024 : *hashvalue = DatumGetUInt32(ExecEvalExprSwitchContext(hjstate->hj_OuterHash,
1242 : : econtext,
1243 : : is_null));
1244 : :
1245 [ + + ]: 644024 : if (!*is_null)
1246 : : {
1247 : : /* normal case with a non-null join key */
1248 : 644004 : return slot;
1249 : : }
1250 [ + + ]: 20 : else if (hjstate->hj_KeepNullTuples)
1251 : : {
1252 : : /* null join key, but we must return tuple to be emitted */
1253 : 8 : return slot;
1254 : : }
1255 : :
1256 : : /*
1257 : : * That tuple couldn't match because of a NULL, so discard it and
1258 : : * continue with the next one.
1259 : : */
1260 : 12 : slot = ExecProcNode(outerNode);
1261 : : }
1262 : : }
1263 [ + - ]: 800545 : else if (curbatch < hashtable->nbatch)
1264 : : {
1265 : : MinimalTuple tuple;
1266 : :
1267 : 800545 : tuple = sts_parallel_scan_next(hashtable->batches[curbatch].outer_tuples,
1268 : : hashvalue);
1269 [ + + ]: 800545 : if (tuple != NULL)
1270 : : {
1271 : 800024 : ExecForceStoreMinimalTuple(tuple,
1272 : : hjstate->hj_OuterTupleSlot,
1273 : : false);
1274 : 800024 : *is_null = false; /* else it'd not be in the batch file */
1275 : 800024 : slot = hjstate->hj_OuterTupleSlot;
1276 : 800024 : return slot;
1277 : : }
1278 : : else
1279 : 521 : ExecClearTuple(hjstate->hj_OuterTupleSlot);
1280 : : }
1281 : :
1282 : : /* End of this batch */
1283 : 635 : hashtable->batches[curbatch].outer_eof = true;
1284 : :
1285 : 635 : return NULL;
1286 : : }
1287 : :
1288 : : /*
1289 : : * ExecHashJoinNewBatch
1290 : : * switch to a new hashjoin batch
1291 : : *
1292 : : * Returns true if successful, false if there are no more batches.
1293 : : */
1294 : : static bool
1295 : 19542 : ExecHashJoinNewBatch(HashJoinState *hjstate)
1296 : : {
1297 : 19542 : HashJoinTable hashtable = hjstate->hj_HashTable;
1298 : : int nbatch;
1299 : : int curbatch;
1300 : : BufFile *innerFile;
1301 : : TupleTableSlot *slot;
1302 : : uint32 hashvalue;
1303 : :
1304 : 19542 : nbatch = hashtable->nbatch;
1305 : 19542 : curbatch = hashtable->curbatch;
1306 : :
1307 [ + + ]: 19542 : if (curbatch > 0)
1308 : : {
1309 : : /*
1310 : : * We no longer need the previous outer batch file; close it right
1311 : : * away to free disk space.
1312 : : */
1313 [ + - ]: 798 : if (hashtable->outerBatchFile[curbatch])
1314 : 798 : BufFileClose(hashtable->outerBatchFile[curbatch]);
1315 : 798 : hashtable->outerBatchFile[curbatch] = NULL;
1316 : : }
1317 : : else /* we just finished the first batch */
1318 : : {
1319 : : /*
1320 : : * Reset some of the skew optimization state variables, since we no
1321 : : * longer need to consider skew tuples after the first batch. The
1322 : : * memory context reset we are about to do will release the skew
1323 : : * hashtable itself.
1324 : : */
1325 : 18744 : hashtable->skewEnabled = false;
1326 : 18744 : hashtable->skewBucket = NULL;
1327 : 18744 : hashtable->skewBucketNums = NULL;
1328 : 18744 : hashtable->nSkewBuckets = 0;
1329 : 18744 : hashtable->spaceUsedSkew = 0;
1330 : : }
1331 : :
1332 : : /*
1333 : : * We can always skip over any batches that are completely empty on both
1334 : : * sides. We can sometimes skip over batches that are empty on only one
1335 : : * side, but there are exceptions:
1336 : : *
1337 : : * 1. In a left/full outer join, we have to process outer batches even if
1338 : : * the inner batch is empty. Similarly, in a right/right-anti/full outer
1339 : : * join, we have to process inner batches even if the outer batch is
1340 : : * empty.
1341 : : *
1342 : : * 2. If we have increased nbatch since the initial estimate, we have to
1343 : : * scan inner batches since they might contain tuples that need to be
1344 : : * reassigned to later inner batches.
1345 : : *
1346 : : * 3. Similarly, if we have increased nbatch since starting the outer
1347 : : * scan, we have to rescan outer batches in case they contain tuples that
1348 : : * need to be reassigned.
1349 : : */
1350 : 19542 : curbatch++;
1351 [ + + ]: 19542 : while (curbatch < nbatch &&
1352 [ - + ]: 798 : (hashtable->outerBatchFile[curbatch] == NULL ||
1353 [ - + ]: 798 : hashtable->innerBatchFile[curbatch] == NULL))
1354 : : {
1355 [ # # ]: 0 : if (hashtable->outerBatchFile[curbatch] &&
1356 [ # # ]: 0 : HJ_FILL_OUTER(hjstate))
1357 : 0 : break; /* must process due to rule 1 */
1358 [ # # ]: 0 : if (hashtable->innerBatchFile[curbatch] &&
1359 [ # # ]: 0 : HJ_FILL_INNER(hjstate))
1360 : 0 : break; /* must process due to rule 1 */
1361 [ # # ]: 0 : if (hashtable->innerBatchFile[curbatch] &&
1362 [ # # ]: 0 : nbatch != hashtable->nbatch_original)
1363 : 0 : break; /* must process due to rule 2 */
1364 [ # # ]: 0 : if (hashtable->outerBatchFile[curbatch] &&
1365 [ # # ]: 0 : nbatch != hashtable->nbatch_outstart)
1366 : 0 : break; /* must process due to rule 3 */
1367 : : /* We can ignore this batch. */
1368 : : /* Release associated temp files right away. */
1369 [ # # ]: 0 : if (hashtable->innerBatchFile[curbatch])
1370 : 0 : BufFileClose(hashtable->innerBatchFile[curbatch]);
1371 : 0 : hashtable->innerBatchFile[curbatch] = NULL;
1372 [ # # ]: 0 : if (hashtable->outerBatchFile[curbatch])
1373 : 0 : BufFileClose(hashtable->outerBatchFile[curbatch]);
1374 : 0 : hashtable->outerBatchFile[curbatch] = NULL;
1375 : 0 : curbatch++;
1376 : : }
1377 : :
1378 [ + + ]: 19542 : if (curbatch >= nbatch)
1379 : 18744 : return false; /* no more batches */
1380 : :
1381 : 798 : hashtable->curbatch = curbatch;
1382 : :
1383 : : /*
1384 : : * Reload the hash table with the new inner batch (which could be empty)
1385 : : */
1386 : 798 : ExecHashTableReset(hashtable);
1387 : :
1388 : 798 : innerFile = hashtable->innerBatchFile[curbatch];
1389 : :
1390 [ + - ]: 798 : if (innerFile != NULL)
1391 : : {
1392 [ - + ]: 798 : if (BufFileSeek(innerFile, 0, 0, SEEK_SET))
1393 [ # # ]: 0 : ereport(ERROR,
1394 : : (errcode_for_file_access(),
1395 : : errmsg("could not rewind hash-join temporary file")));
1396 : :
1397 [ + + ]: 2269764 : while ((slot = ExecHashJoinGetSavedTuple(hjstate,
1398 : : innerFile,
1399 : : &hashvalue,
1400 : : hjstate->hj_HashTupleSlot)))
1401 : : {
1402 : : /*
1403 : : * NOTE: some tuples may be sent to future batches. Also, it is
1404 : : * possible for hashtable->nbatch to be increased here!
1405 : : */
1406 : 2268966 : ExecHashTableInsert(hashtable, slot, hashvalue);
1407 : : }
1408 : :
1409 : : /*
1410 : : * after we build the hash table, the inner batch file is no longer
1411 : : * needed
1412 : : */
1413 : 798 : BufFileClose(innerFile);
1414 : 798 : hashtable->innerBatchFile[curbatch] = NULL;
1415 : : }
1416 : :
1417 : : /*
1418 : : * Rewind outer batch file (if present), so that we can start reading it.
1419 : : */
1420 [ + - ]: 798 : if (hashtable->outerBatchFile[curbatch] != NULL)
1421 : : {
1422 [ - + ]: 798 : if (BufFileSeek(hashtable->outerBatchFile[curbatch], 0, 0, SEEK_SET))
1423 [ # # ]: 0 : ereport(ERROR,
1424 : : (errcode_for_file_access(),
1425 : : errmsg("could not rewind hash-join temporary file")));
1426 : : }
1427 : :
1428 : 798 : return true;
1429 : : }
1430 : :
1431 : : /*
1432 : : * Choose a batch to work on, and attach to it. Returns true if successful,
1433 : : * false if there are no more batches.
1434 : : */
1435 : : static bool
1436 : 911 : ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
1437 : : {
1438 : 911 : HashJoinTable hashtable = hjstate->hj_HashTable;
1439 : : int start_batchno;
1440 : : int batchno;
1441 : :
1442 : : /*
1443 : : * If we are a very slow worker, MultiExecParallelHash could have observed
1444 : : * build_barrier phase PHJ_BUILD_FREE and not bothered to set up batch
1445 : : * accessors. In that case we must be done.
1446 : : */
1447 [ + + ]: 911 : if (hashtable->batches == NULL)
1448 : 2 : return false;
1449 : :
1450 : : /*
1451 : : * If we were already attached to a batch, remember not to bother checking
1452 : : * it again, and detach from it (possibly freeing the hash table if we are
1453 : : * last to detach).
1454 : : */
1455 [ + + ]: 909 : if (hashtable->curbatch >= 0)
1456 : : {
1457 : 610 : hashtable->batches[hashtable->curbatch].done = true;
1458 : 610 : ExecHashTableDetachBatch(hashtable);
1459 : : }
1460 : :
1461 : : /*
1462 : : * Search for a batch that isn't done. We use an atomic counter to start
1463 : : * our search at a different batch in every participant when there are
1464 : : * more batches than participants.
1465 : : */
1466 : 909 : batchno = start_batchno =
1467 : 909 : pg_atomic_fetch_add_u32(&hashtable->parallel_state->distributor, 1) %
1468 : 909 : hashtable->nbatch;
1469 : : do
1470 : : {
1471 : : uint32 hashvalue;
1472 : : MinimalTuple tuple;
1473 : : TupleTableSlot *slot;
1474 : :
1475 [ + + ]: 2220 : if (!hashtable->batches[batchno].done)
1476 : : {
1477 : : SharedTuplestoreAccessor *inner_tuples;
1478 : 1190 : Barrier *batch_barrier =
1479 : 1190 : &hashtable->batches[batchno].shared->batch_barrier;
1480 : :
1481 [ + - + + : 1190 : switch (BarrierAttach(batch_barrier))
+ + - ]
1482 : : {
1483 : 388 : case PHJ_BATCH_ELECT:
1484 : :
1485 : : /* One backend allocates the hash table. */
1486 [ + - ]: 388 : if (BarrierArriveAndWait(batch_barrier,
1487 : : WAIT_EVENT_HASH_BATCH_ELECT))
1488 : 388 : ExecParallelHashTableAlloc(hashtable, batchno);
1489 : : pg_fallthrough;
1490 : :
1491 : : case PHJ_BATCH_ALLOCATE:
1492 : : /* Wait for allocation to complete. */
1493 : 388 : BarrierArriveAndWait(batch_barrier,
1494 : : WAIT_EVENT_HASH_BATCH_ALLOCATE);
1495 : : pg_fallthrough;
1496 : :
1497 : 403 : case PHJ_BATCH_LOAD:
1498 : : /* Start (or join in) loading tuples. */
1499 : 403 : ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
1500 : 403 : inner_tuples = hashtable->batches[batchno].inner_tuples;
1501 : 403 : sts_begin_parallel_scan(inner_tuples);
1502 [ + + ]: 721515 : while ((tuple = sts_parallel_scan_next(inner_tuples,
1503 : : &hashvalue)))
1504 : : {
1505 : 721112 : ExecForceStoreMinimalTuple(tuple,
1506 : : hjstate->hj_HashTupleSlot,
1507 : : false);
1508 : 721112 : slot = hjstate->hj_HashTupleSlot;
1509 : 721112 : ExecParallelHashTableInsertCurrentBatch(hashtable, slot,
1510 : : hashvalue);
1511 : : }
1512 : 403 : sts_end_parallel_scan(inner_tuples);
1513 : 403 : BarrierArriveAndWait(batch_barrier,
1514 : : WAIT_EVENT_HASH_BATCH_LOAD);
1515 : : pg_fallthrough;
1516 : :
1517 : 635 : case PHJ_BATCH_PROBE:
1518 : :
1519 : : /*
1520 : : * This batch is ready to probe. Return control to
1521 : : * caller. We stay attached to batch_barrier so that the
1522 : : * hash table stays alive until everyone's finished
1523 : : * probing it, but no participant is allowed to wait at
1524 : : * this barrier again (or else a deadlock could occur).
1525 : : * All attached participants must eventually detach from
1526 : : * the barrier and one worker must advance the phase so
1527 : : * that the final phase is reached.
1528 : : */
1529 : 635 : ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
1530 : 635 : sts_begin_parallel_scan(hashtable->batches[batchno].outer_tuples);
1531 : :
1532 : 635 : return true;
1533 : 3 : case PHJ_BATCH_SCAN:
1534 : :
1535 : : /*
1536 : : * In principle, we could help scan for unmatched tuples,
1537 : : * since that phase is already underway (the thing we
1538 : : * can't do under current deadlock-avoidance rules is wait
1539 : : * for others to arrive at PHJ_BATCH_SCAN, because
1540 : : * PHJ_BATCH_PROBE emits tuples, but in this case we just
1541 : : * got here without waiting). That is not yet done. For
1542 : : * now, we just detach and go around again. We have to
1543 : : * use ExecHashTableDetachBatch() because there's a small
1544 : : * chance we'll be the last to detach, and then we're
1545 : : * responsible for freeing memory.
1546 : : */
1547 : 3 : ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
1548 : 3 : hashtable->batches[batchno].done = true;
1549 : 3 : ExecHashTableDetachBatch(hashtable);
1550 : 3 : break;
1551 : :
1552 : 552 : case PHJ_BATCH_FREE:
1553 : :
1554 : : /*
1555 : : * Already done. Detach and go around again (if any
1556 : : * remain).
1557 : : */
1558 : 552 : BarrierDetach(batch_barrier);
1559 : 552 : hashtable->batches[batchno].done = true;
1560 : 552 : hashtable->curbatch = -1;
1561 : 552 : break;
1562 : :
1563 : 0 : default:
1564 [ # # ]: 0 : elog(ERROR, "unexpected batch phase %d",
1565 : : BarrierPhase(batch_barrier));
1566 : : }
1567 : : }
1568 : 1585 : batchno = (batchno + 1) % hashtable->nbatch;
1569 [ + + ]: 1585 : } while (batchno != start_batchno);
1570 : :
1571 : 274 : return false;
1572 : : }
1573 : :
1574 : : /*
1575 : : * ExecHashJoinSaveTuple
1576 : : * save a tuple to a batch file.
1577 : : *
1578 : : * The data recorded in the file for each tuple is its hash value,
1579 : : * then the tuple in MinimalTuple format.
1580 : : *
1581 : : * fileptr points to a batch file in one of the hashtable arrays.
1582 : : *
1583 : : * The batch files (and their buffers) are allocated in the spill context
1584 : : * created for the hashtable.
1585 : : */
1586 : : void
1587 : 3227822 : ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
1588 : : BufFile **fileptr, HashJoinTable hashtable)
1589 : : {
1590 : 3227822 : BufFile *file = *fileptr;
1591 : :
1592 : : /*
1593 : : * The batch file is lazily created. If this is the first tuple written to
1594 : : * this batch, the batch file is created and its buffer is allocated in
1595 : : * the spillCxt context, NOT in the batchCxt.
1596 : : *
1597 : : * During the build phase, buffered files are created for inner batches.
1598 : : * Each batch's buffered file is closed (and its buffer freed) after the
1599 : : * batch is loaded into memory during the outer side scan. Therefore, it
1600 : : * is necessary to allocate the batch file buffer in a memory context
1601 : : * which outlives the batch itself.
1602 : : *
1603 : : * Also, we use spillCxt instead of hashCxt for a better accounting of the
1604 : : * spilling memory consumption.
1605 : : */
1606 [ + + ]: 3227822 : if (file == NULL)
1607 : : {
1608 : 1596 : MemoryContext oldctx = MemoryContextSwitchTo(hashtable->spillCxt);
1609 : :
1610 : 1596 : file = BufFileCreateTemp(false);
1611 : 1596 : *fileptr = file;
1612 : :
1613 : 1596 : MemoryContextSwitchTo(oldctx);
1614 : : }
1615 : :
1616 : 3227822 : BufFileWrite(file, &hashvalue, sizeof(uint32));
1617 : 3227822 : BufFileWrite(file, tuple, tuple->t_len);
1618 : 3227822 : }
1619 : :
1620 : : /*
1621 : : * ExecHashJoinGetSavedTuple
1622 : : * read the next tuple from a batch file. Return NULL if no more.
1623 : : *
1624 : : * On success, *hashvalue is set to the tuple's hash value, and the tuple
1625 : : * itself is stored in the given slot.
1626 : : */
1627 : : static TupleTableSlot *
1628 : 3229418 : ExecHashJoinGetSavedTuple(HashJoinState *hjstate,
1629 : : BufFile *file,
1630 : : uint32 *hashvalue,
1631 : : TupleTableSlot *tupleSlot)
1632 : : {
1633 : : uint32 header[2];
1634 : : size_t nread;
1635 : : MinimalTuple tuple;
1636 : :
1637 : : /*
1638 : : * We check for interrupts here because this is typically taken as an
1639 : : * alternative code path to an ExecProcNode() call, which would include
1640 : : * such a check.
1641 : : */
1642 [ + + ]: 3229418 : CHECK_FOR_INTERRUPTS();
1643 : :
1644 : : /*
1645 : : * Since both the hash value and the MinimalTuple length word are uint32,
1646 : : * we can read them both in one BufFileRead() call without any type
1647 : : * cheating.
1648 : : */
1649 : 3229418 : nread = BufFileReadMaybeEOF(file, header, sizeof(header), true);
1650 [ + + ]: 3229418 : if (nread == 0) /* end of file */
1651 : : {
1652 : 1596 : ExecClearTuple(tupleSlot);
1653 : 1596 : return NULL;
1654 : : }
1655 : 3227822 : *hashvalue = header[0];
1656 : 3227822 : tuple = (MinimalTuple) palloc(header[1]);
1657 : 3227822 : tuple->t_len = header[1];
1658 : 3227822 : BufFileReadExact(file,
1659 : : (char *) tuple + sizeof(uint32),
1660 : 3227822 : header[1] - sizeof(uint32));
1661 : 3227822 : ExecForceStoreMinimalTuple(tuple, tupleSlot, true);
1662 : 3227822 : return tupleSlot;
1663 : : }
1664 : :
1665 : :
1666 : : void
1667 : 3184 : ExecReScanHashJoin(HashJoinState *node)
1668 : : {
1669 : 3184 : PlanState *outerPlan = outerPlanState(node);
1670 : 3184 : PlanState *innerPlan = innerPlanState(node);
1671 : :
1672 : : /*
1673 : : * We're always going to rescan the outer rel, so drop the associated
1674 : : * null-keys tuplestore; we'll rebuild it during the rescan. (Must do
1675 : : * this before ExecHashTableDestroy deletes hashCxt.)
1676 : : */
1677 [ - + ]: 3184 : if (node->hj_NullOuterTupleStore)
1678 : : {
1679 : 0 : tuplestore_end(node->hj_NullOuterTupleStore);
1680 : 0 : node->hj_NullOuterTupleStore = NULL;
1681 : : }
1682 : :
1683 : : /*
1684 : : * In a multi-batch join, we currently have to do rescans the hard way,
1685 : : * primarily because batch temp files may have already been released. But
1686 : : * if it's a single-batch join, and there is no parameter change for the
1687 : : * inner subnode, then we can just re-use the existing hash table without
1688 : : * rebuilding it.
1689 : : */
1690 [ + + ]: 3184 : if (node->hj_HashTable != NULL)
1691 : : {
1692 : 2728 : HashState *hashNode = castNode(HashState, innerPlan);
1693 : :
1694 : : Assert(hashNode->hashtable == node->hj_HashTable);
1695 : :
1696 [ + - ]: 2728 : if (node->hj_HashTable->nbatch == 1 &&
1697 [ + + ]: 2728 : innerPlan->chgParam == NULL)
1698 : : {
1699 : : /*
1700 : : * Okay to reuse the hash table; needn't rescan inner, either.
1701 : : *
1702 : : * However, if it's a right/right-anti/right-semi/full join, we'd
1703 : : * better reset the inner-tuple match flags contained in the
1704 : : * table.
1705 : : */
1706 [ + + + + ]: 1213 : if (HJ_FILL_INNER(node) || node->js.jointype == JOIN_RIGHT_SEMI)
1707 : 44 : ExecHashTableResetMatchFlags(node->hj_HashTable);
1708 : :
1709 : : /*
1710 : : * Also, we need to reset our state about the emptiness of the
1711 : : * outer relation, so that the new scan of the outer will update
1712 : : * it correctly if it turns out to be empty this time. (There's no
1713 : : * harm in clearing it now because ExecHashJoin won't need the
1714 : : * info. In the other cases, where the hash table doesn't exist
1715 : : * or we are destroying it, we leave this state alone because
1716 : : * ExecHashJoin will need it the first time through.)
1717 : : */
1718 : 1213 : node->hj_OuterNotEmpty = false;
1719 : :
1720 : : /*
1721 : : * Also, rewind inner null-key tuplestore so that we can return
1722 : : * those tuples again.
1723 : : */
1724 [ + + ]: 1213 : if (hashNode->null_tuple_store)
1725 : 4 : tuplestore_rescan(hashNode->null_tuple_store);
1726 : :
1727 : : /* ExecHashJoin can skip the BUILD_HASHTABLE step */
1728 : 1213 : node->hj_JoinState = HJ_NEED_NEW_OUTER;
1729 : : }
1730 : : else
1731 : : {
1732 : : /* must destroy and rebuild hash table */
1733 : :
1734 : : /* accumulate stats from old hash table, if wanted */
1735 : : /* (this should match ExecShutdownHash) */
1736 [ - + - - ]: 1515 : if (hashNode->ps.instrument && !hashNode->hinstrument)
1737 : 0 : hashNode->hinstrument = palloc0_object(HashInstrumentation);
1738 [ - + ]: 1515 : if (hashNode->hinstrument)
1739 : 0 : ExecHashAccumInstrumentation(hashNode->hinstrument,
1740 : : hashNode->hashtable);
1741 : :
1742 : : /* free inner null-key tuplestore before ExecHashTableDestroy */
1743 [ - + ]: 1515 : if (hashNode->null_tuple_store)
1744 : : {
1745 : 0 : tuplestore_end(hashNode->null_tuple_store);
1746 : 0 : hashNode->null_tuple_store = NULL;
1747 : : }
1748 : :
1749 : : /* for safety, be sure to clear child plan node's pointer too */
1750 : 1515 : hashNode->hashtable = NULL;
1751 : :
1752 : 1515 : ExecHashTableDestroy(node->hj_HashTable);
1753 : 1515 : node->hj_HashTable = NULL;
1754 : 1515 : node->hj_JoinState = HJ_BUILD_HASHTABLE;
1755 : :
1756 : : /*
1757 : : * if chgParam of subnode is not null then plan will be re-scanned
1758 : : * by first ExecProcNode.
1759 : : */
1760 [ - + ]: 1515 : if (innerPlan->chgParam == NULL)
1761 : 0 : ExecReScan(innerPlan);
1762 : : }
1763 : : }
1764 : :
1765 : : /* Always reset intra-tuple state */
1766 : 3184 : node->hj_CurHashValue = 0;
1767 : 3184 : node->hj_CurBucketNo = 0;
1768 : 3184 : node->hj_CurSkewBucketNo = INVALID_SKEW_BUCKET_NO;
1769 : 3184 : node->hj_CurTuple = NULL;
1770 : :
1771 : 3184 : node->hj_MatchedOuter = false;
1772 : 3184 : node->hj_FirstOuterTupleSlot = NULL;
1773 : :
1774 : : /*
1775 : : * if chgParam of subnode is not null then plan will be re-scanned by
1776 : : * first ExecProcNode.
1777 : : */
1778 [ + + ]: 3184 : if (outerPlan->chgParam == NULL)
1779 : 2173 : ExecReScan(outerPlan);
1780 : 3184 : }
1781 : :
1782 : : void
1783 : 26055 : ExecShutdownHashJoin(HashJoinState *node)
1784 : : {
1785 [ + + ]: 26055 : if (node->hj_HashTable)
1786 : : {
1787 : : /*
1788 : : * Detach from shared state before DSM memory goes away. This makes
1789 : : * sure that we don't have any pointers into DSM memory by the time
1790 : : * ExecEndHashJoin runs.
1791 : : */
1792 : 19411 : ExecHashTableDetachBatch(node->hj_HashTable);
1793 : 19411 : ExecHashTableDetach(node->hj_HashTable);
1794 : : }
1795 : 26055 : }
1796 : :
1797 : : static void
1798 : 120 : ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate)
1799 : : {
1800 : 120 : PlanState *outerState = outerPlanState(hjstate);
1801 : 120 : ExprContext *econtext = hjstate->js.ps.ps_ExprContext;
1802 : 120 : HashJoinTable hashtable = hjstate->hj_HashTable;
1803 : : TupleTableSlot *slot;
1804 : : int i;
1805 : :
1806 : : Assert(hjstate->hj_FirstOuterTupleSlot == NULL);
1807 : :
1808 : : /* Execute outer plan, writing all tuples to shared tuplestores. */
1809 : : for (;;)
1810 : 800052 : {
1811 : : bool isnull;
1812 : : uint32 hashvalue;
1813 : :
1814 : 800172 : slot = ExecProcNode(outerState);
1815 [ + + + + ]: 800172 : if (TupIsNull(slot))
1816 : : break;
1817 : 800052 : econtext->ecxt_outertuple = slot;
1818 : :
1819 : 800052 : ResetExprContext(econtext);
1820 : :
1821 : 800052 : hashvalue = DatumGetUInt32(ExecEvalExprSwitchContext(hjstate->hj_OuterHash,
1822 : : econtext,
1823 : : &isnull));
1824 : :
1825 [ + + ]: 800052 : if (!isnull)
1826 : : {
1827 : : /* normal case with a non-null join key */
1828 : : int batchno;
1829 : : int bucketno;
1830 : : bool shouldFree;
1831 : 800024 : MinimalTuple mintup = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1832 : :
1833 : 800024 : ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
1834 : : &batchno);
1835 : 800024 : sts_puttuple(hashtable->batches[batchno].outer_tuples,
1836 : : &hashvalue, mintup);
1837 : :
1838 [ + - ]: 800024 : if (shouldFree)
1839 : 800024 : heap_free_minimal_tuple(mintup);
1840 : : }
1841 [ + + ]: 28 : else if (hjstate->hj_KeepNullTuples)
1842 : : {
1843 : : /* null join key, but we must save tuple to be emitted later */
1844 [ + - ]: 4 : if (hjstate->hj_NullOuterTupleStore == NULL)
1845 : 4 : hjstate->hj_NullOuterTupleStore = ExecHashBuildNullTupleStore(hashtable);
1846 : 4 : tuplestore_puttupleslot(hjstate->hj_NullOuterTupleStore, slot);
1847 : : }
1848 : : /* else we can just discard the tuple immediately */
1849 : :
1850 [ - + ]: 800052 : CHECK_FOR_INTERRUPTS();
1851 : : }
1852 : :
1853 : : /* Make sure all outer partitions are readable by any backend. */
1854 [ + + ]: 1016 : for (i = 0; i < hashtable->nbatch; ++i)
1855 : 896 : sts_end_write(hashtable->batches[i].outer_tuples);
1856 : 120 : }
1857 : :
1858 : : void
1859 : 84 : ExecHashJoinEstimate(HashJoinState *state, ParallelContext *pcxt)
1860 : : {
1861 : 84 : shm_toc_estimate_chunk(&pcxt->estimator, sizeof(ParallelHashJoinState));
1862 : 84 : shm_toc_estimate_keys(&pcxt->estimator, 1);
1863 : 84 : }
1864 : :
1865 : : void
1866 : 84 : ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
1867 : : {
1868 : 84 : int plan_node_id = state->js.ps.plan->plan_node_id;
1869 : : HashState *hashNode;
1870 : : ParallelHashJoinState *pstate;
1871 : :
1872 : : /*
1873 : : * Disable shared hash table mode if we failed to create a real DSM
1874 : : * segment, because that means that we don't have a DSA area to work with.
1875 : : */
1876 [ - + ]: 84 : if (pcxt->seg == NULL)
1877 : 0 : return;
1878 : :
1879 : 84 : ExecSetExecProcNode(&state->js.ps, ExecParallelHashJoin);
1880 : :
1881 : : /*
1882 : : * Set up the state needed to coordinate access to the shared hash
1883 : : * table(s), using the plan node ID as the toc key.
1884 : : */
1885 : 84 : pstate = shm_toc_allocate(pcxt->toc, sizeof(ParallelHashJoinState));
1886 : 84 : shm_toc_insert(pcxt->toc, plan_node_id, pstate);
1887 : :
1888 : : /*
1889 : : * Set up the shared hash join state with no batches initially.
1890 : : * ExecHashTableCreate() will prepare at least one later and set nbatch
1891 : : * and space_allowed.
1892 : : */
1893 : 84 : pstate->nbatch = 0;
1894 : 84 : pstate->space_allowed = 0;
1895 : 84 : pstate->batches = InvalidDsaPointer;
1896 : 84 : pstate->old_batches = InvalidDsaPointer;
1897 : 84 : pstate->nbuckets = 0;
1898 : 84 : pstate->growth = PHJ_GROWTH_OK;
1899 : 84 : pstate->chunk_work_queue = InvalidDsaPointer;
1900 : 84 : pg_atomic_init_u32(&pstate->distributor, 0);
1901 : 84 : pstate->nparticipants = pcxt->nworkers + 1;
1902 : 84 : pstate->total_tuples = 0;
1903 : 84 : LWLockInitialize(&pstate->lock,
1904 : : LWTRANCHE_PARALLEL_HASH_JOIN);
1905 : 84 : BarrierInit(&pstate->build_barrier, 0);
1906 : 84 : BarrierInit(&pstate->grow_batches_barrier, 0);
1907 : 84 : BarrierInit(&pstate->grow_buckets_barrier, 0);
1908 : :
1909 : : /* Set up the space we'll use for shared temporary files. */
1910 : 84 : SharedFileSetInit(&pstate->fileset, pcxt->seg);
1911 : :
1912 : : /* Initialize the shared state in the hash node. */
1913 : 84 : hashNode = (HashState *) innerPlanState(state);
1914 : 84 : hashNode->parallel_state = pstate;
1915 : : }
1916 : :
1917 : : /* ----------------------------------------------------------------
1918 : : * ExecHashJoinReInitializeDSM
1919 : : *
1920 : : * Reset shared state before beginning a fresh scan.
1921 : : * ----------------------------------------------------------------
1922 : : */
1923 : : void
1924 : 32 : ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
1925 : : {
1926 : 32 : int plan_node_id = state->js.ps.plan->plan_node_id;
1927 : : ParallelHashJoinState *pstate;
1928 : : HashState *hashNode;
1929 : :
1930 : : /* Nothing to do if we failed to create a DSM segment. */
1931 [ - + ]: 32 : if (pcxt->seg == NULL)
1932 : 0 : return;
1933 : :
1934 : 32 : pstate = shm_toc_lookup(pcxt->toc, plan_node_id, false);
1935 : :
1936 : : /*
1937 : : * It would be possible to reuse the shared hash table in single-batch
1938 : : * cases by resetting and then fast-forwarding build_barrier to
1939 : : * PHJ_BUILD_FREE and batch 0's batch_barrier to PHJ_BATCH_PROBE, but
1940 : : * currently shared hash tables are already freed by now (by the last
1941 : : * participant to detach from the batch). We could consider keeping it
1942 : : * around for single-batch joins. We'd also need to adjust
1943 : : * finalize_plan() so that it doesn't record a dummy dependency for
1944 : : * Parallel Hash nodes, preventing the rescan optimization. For now we
1945 : : * don't try.
1946 : : */
1947 : :
1948 : : /* Detach, freeing any remaining shared memory. */
1949 [ - + ]: 32 : if (state->hj_HashTable != NULL)
1950 : : {
1951 : 0 : ExecHashTableDetachBatch(state->hj_HashTable);
1952 : 0 : ExecHashTableDetach(state->hj_HashTable);
1953 : : }
1954 : :
1955 : : /* Clear any shared batch files. */
1956 : 32 : SharedFileSetDeleteAll(&pstate->fileset);
1957 : :
1958 : : /* We'd better clear our local null-key tuplestores, too. */
1959 [ - + ]: 32 : if (state->hj_NullOuterTupleStore)
1960 : : {
1961 : 0 : tuplestore_end(state->hj_NullOuterTupleStore);
1962 : 0 : state->hj_NullOuterTupleStore = NULL;
1963 : : }
1964 : 32 : hashNode = (HashState *) innerPlanState(state);
1965 [ - + ]: 32 : if (hashNode->null_tuple_store)
1966 : : {
1967 : 0 : tuplestore_end(hashNode->null_tuple_store);
1968 : 0 : hashNode->null_tuple_store = NULL;
1969 : : }
1970 : :
1971 : :
1972 : : /* Reset build_barrier to PHJ_BUILD_ELECT so we can go around again. */
1973 : 32 : BarrierInit(&pstate->build_barrier, 0);
1974 : : }
1975 : :
1976 : : void
1977 : 212 : ExecHashJoinInitializeWorker(HashJoinState *state,
1978 : : ParallelWorkerContext *pwcxt)
1979 : : {
1980 : : HashState *hashNode;
1981 : 212 : int plan_node_id = state->js.ps.plan->plan_node_id;
1982 : : ParallelHashJoinState *pstate =
1983 : 212 : shm_toc_lookup(pwcxt->toc, plan_node_id, false);
1984 : :
1985 : : /* Attach to the space for shared temporary files. */
1986 : 212 : SharedFileSetAttach(&pstate->fileset, pwcxt->seg);
1987 : :
1988 : : /* Attach to the shared state in the hash node. */
1989 : 212 : hashNode = (HashState *) innerPlanState(state);
1990 : 212 : hashNode->parallel_state = pstate;
1991 : :
1992 : 212 : ExecSetExecProcNode(&state->js.ps, ExecParallelHashJoin);
1993 : 212 : }
|
