Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * nodeHash.c
4 : : * Routines to hash relations for hashjoin
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/nodeHash.c
12 : : *
13 : : * See note on parallelism in nodeHashjoin.c.
14 : : *
15 : : *-------------------------------------------------------------------------
16 : : */
17 : : /*
18 : : * INTERFACE ROUTINES
19 : : * MultiExecHash - generate an in-memory hash table of the relation
20 : : * ExecInitHash - initialize node and subnodes
21 : : * ExecEndHash - shutdown node and subnodes
22 : : */
23 : :
24 : : #include "postgres.h"
25 : :
26 : : #include <math.h>
27 : : #include <limits.h>
28 : :
29 : : #include "access/htup_details.h"
30 : : #include "access/parallel.h"
31 : : #include "catalog/pg_statistic.h"
32 : : #include "commands/tablespace.h"
33 : : #include "executor/executor.h"
34 : : #include "executor/hashjoin.h"
35 : : #include "executor/instrument.h"
36 : : #include "executor/nodeHash.h"
37 : : #include "executor/nodeHashjoin.h"
38 : : #include "miscadmin.h"
39 : : #include "port/pg_bitutils.h"
40 : : #include "utils/lsyscache.h"
41 : : #include "utils/memutils.h"
42 : : #include "utils/syscache.h"
43 : : #include "utils/tuplestore.h"
44 : : #include "utils/wait_event.h"
45 : :
46 : : static void ExecHashIncreaseNumBatches(HashJoinTable hashtable);
47 : : static void ExecHashIncreaseNumBuckets(HashJoinTable hashtable);
48 : : static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable);
49 : : static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable);
50 : : static void ExecHashBuildSkewHash(HashState *hashstate,
51 : : HashJoinTable hashtable, Hash *node,
52 : : int mcvsToUse);
53 : : static void ExecHashSkewTableInsert(HashJoinTable hashtable,
54 : : TupleTableSlot *slot,
55 : : uint32 hashvalue,
56 : : int bucketNumber);
57 : : static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable);
58 : :
59 : : static void *dense_alloc(HashJoinTable hashtable, Size size);
60 : : static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable,
61 : : size_t size,
62 : : dsa_pointer *shared);
63 : : static void MultiExecPrivateHash(HashState *node);
64 : : static void MultiExecParallelHash(HashState *node);
65 : : static inline HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable hashtable,
66 : : int bucketno);
67 : : static inline HashJoinTuple ExecParallelHashNextTuple(HashJoinTable hashtable,
68 : : HashJoinTuple tuple);
69 : : static inline void ExecParallelHashPushTuple(dsa_pointer_atomic *head,
70 : : HashJoinTuple tuple,
71 : : dsa_pointer tuple_shared);
72 : : static void ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch);
73 : : static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable);
74 : : static void ExecParallelHashRepartitionFirst(HashJoinTable hashtable);
75 : : static void ExecParallelHashRepartitionRest(HashJoinTable hashtable);
76 : : static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable hashtable,
77 : : dsa_pointer *shared);
78 : : static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable,
79 : : int batchno,
80 : : size_t size);
81 : : static void ExecParallelHashMergeCounters(HashJoinTable hashtable);
82 : : static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable);
83 : :
84 : :
85 : : /* ----------------------------------------------------------------
86 : : * ExecHash
87 : : *
88 : : * stub for pro forma compliance
89 : : * ----------------------------------------------------------------
90 : : */
91 : : static TupleTableSlot *
92 : 0 : ExecHash(PlanState *pstate)
93 : : {
94 [ # # ]: 0 : elog(ERROR, "Hash node does not support ExecProcNode call convention");
95 : : return NULL;
96 : : }
97 : :
98 : : /* ----------------------------------------------------------------
99 : : * MultiExecHash
100 : : *
101 : : * build hash table for hashjoin, doing partitioning if more
102 : : * than one batch is required.
103 : : * ----------------------------------------------------------------
104 : : */
105 : : Node *
106 : 20968 : MultiExecHash(HashState *node)
107 : : {
108 : : /* must provide our own instrumentation support */
109 [ + + ]: 20968 : if (node->ps.instrument)
110 : 227 : InstrStartNode(node->ps.instrument);
111 : :
112 [ + + ]: 20968 : if (node->parallel_state != NULL)
113 : 276 : MultiExecParallelHash(node);
114 : : else
115 : 20692 : MultiExecPrivateHash(node);
116 : :
117 : : /* must provide our own instrumentation support */
118 [ + + ]: 20968 : if (node->ps.instrument)
119 : 227 : InstrStopNode(node->ps.instrument, node->hashtable->reportTuples);
120 : :
121 : : /*
122 : : * We do not return the hash table directly because it's not a subtype of
123 : : * Node, and so would violate the MultiExecProcNode API. Instead, our
124 : : * parent Hashjoin node is expected to know how to fish it out of our node
125 : : * state. Ugly but not really worth cleaning up, since Hashjoin knows
126 : : * quite a bit more about Hash besides that.
127 : : */
128 : 20968 : return NULL;
129 : : }
130 : :
131 : : /* ----------------------------------------------------------------
132 : : * MultiExecPrivateHash
133 : : *
134 : : * parallel-oblivious version, building a backend-private
135 : : * hash table and (if necessary) batch files.
136 : : * ----------------------------------------------------------------
137 : : */
138 : : static void
139 : 20692 : MultiExecPrivateHash(HashState *node)
140 : : {
141 : : PlanState *outerNode;
142 : : HashJoinTable hashtable;
143 : : TupleTableSlot *slot;
144 : : ExprContext *econtext;
145 : 20692 : double nullTuples = 0;
146 : :
147 : : /*
148 : : * get state info from node
149 : : */
150 : 20692 : outerNode = outerPlanState(node);
151 : 20692 : hashtable = node->hashtable;
152 : :
153 : : /*
154 : : * set expression context
155 : : */
156 : 20692 : econtext = node->ps.ps_ExprContext;
157 : :
158 : : /*
159 : : * Get all tuples from the node below the Hash node and insert the
160 : : * potentially-matchable ones into the hash table (or temp files). Tuples
161 : : * that can't possibly match because they have null join keys are dumped
162 : : * into a separate tuplestore, or just summarily discarded if we don't
163 : : * need to emit them with null-extension.
164 : : */
165 : : for (;;)
166 : 6674969 : {
167 : : bool isnull;
168 : : Datum hashdatum;
169 : :
170 : 6695661 : slot = ExecProcNode(outerNode);
171 [ + + + + ]: 6695661 : if (TupIsNull(slot))
172 : : break;
173 : : /* We have to compute the hash value */
174 : 6674969 : econtext->ecxt_outertuple = slot;
175 : :
176 : 6674969 : ResetExprContext(econtext);
177 : :
178 : 6674969 : hashdatum = ExecEvalExprSwitchContext(node->hash_expr, econtext,
179 : : &isnull);
180 : :
181 [ + + ]: 6674969 : if (!isnull)
182 : : {
183 : : /* normal case with a non-null join key */
184 : 6674790 : uint32 hashvalue = DatumGetUInt32(hashdatum);
185 : : int bucketNumber;
186 : :
187 : 6674790 : bucketNumber = ExecHashGetSkewBucket(hashtable, hashvalue);
188 [ + + ]: 6674790 : if (bucketNumber != INVALID_SKEW_BUCKET_NO)
189 : : {
190 : : /* It's a skew tuple, so put it into that hash table */
191 : 392 : ExecHashSkewTableInsert(hashtable, slot, hashvalue,
192 : : bucketNumber);
193 : : }
194 : : else
195 : : {
196 : : /* Not subject to skew optimization, so insert normally */
197 : 6674398 : ExecHashTableInsert(hashtable, slot, hashvalue);
198 : : }
199 : 6674790 : hashtable->totalTuples += 1;
200 : : }
201 [ + + ]: 179 : else if (node->keep_null_tuples)
202 : : {
203 : : /* null join key, but we must save tuple to be emitted later */
204 [ + - ]: 56 : if (node->null_tuple_store == NULL)
205 : 56 : node->null_tuple_store = ExecHashBuildNullTupleStore(hashtable);
206 : 56 : tuplestore_puttupleslot(node->null_tuple_store, slot);
207 : 56 : nullTuples += 1;
208 : : }
209 : : /* else we can discard the tuple immediately */
210 : : }
211 : :
212 : : /* resize the hash table if needed (NTUP_PER_BUCKET exceeded) */
213 [ + + ]: 20692 : if (hashtable->nbuckets != hashtable->nbuckets_optimal)
214 : 112 : ExecHashIncreaseNumBuckets(hashtable);
215 : :
216 : : /* Account for the buckets in spaceUsed (reported in EXPLAIN ANALYZE) */
217 : 20692 : hashtable->spaceUsed += hashtable->nbuckets * sizeof(HashJoinTuple);
218 [ + + ]: 20692 : if (hashtable->spaceUsed > hashtable->spacePeak)
219 : 20661 : hashtable->spacePeak = hashtable->spaceUsed;
220 : :
221 : : /* Report total number of tuples output (but not those discarded) */
222 : 20692 : hashtable->reportTuples = hashtable->totalTuples + nullTuples;
223 : 20692 : }
224 : :
225 : : /* ----------------------------------------------------------------
226 : : * MultiExecParallelHash
227 : : *
228 : : * parallel-aware version, building a shared hash table and
229 : : * (if necessary) batch files using the combined effort of
230 : : * a set of co-operating backends.
231 : : * ----------------------------------------------------------------
232 : : */
233 : : static void
234 : 276 : MultiExecParallelHash(HashState *node)
235 : : {
236 : : ParallelHashJoinState *pstate;
237 : : PlanState *outerNode;
238 : : HashJoinTable hashtable;
239 : : TupleTableSlot *slot;
240 : : ExprContext *econtext;
241 : : Barrier *build_barrier;
242 : : int i;
243 : :
244 : : /*
245 : : * get state info from node
246 : : */
247 : 276 : outerNode = outerPlanState(node);
248 : 276 : hashtable = node->hashtable;
249 : :
250 : : /*
251 : : * set expression context
252 : : */
253 : 276 : econtext = node->ps.ps_ExprContext;
254 : :
255 : : /*
256 : : * Synchronize the parallel hash table build. At this stage we know that
257 : : * the shared hash table has been or is being set up by
258 : : * ExecHashTableCreate(), but we don't know if our peers have returned
259 : : * from there or are here in MultiExecParallelHash(), and if so how far
260 : : * through they are. To find out, we check the build_barrier phase then
261 : : * and jump to the right step in the build algorithm.
262 : : */
263 : 276 : pstate = hashtable->parallel_state;
264 : 276 : build_barrier = &pstate->build_barrier;
265 : : Assert(BarrierPhase(build_barrier) >= PHJ_BUILD_ALLOCATE);
266 [ + + + ]: 276 : switch (BarrierPhase(build_barrier))
267 : : {
268 : 121 : case PHJ_BUILD_ALLOCATE:
269 : :
270 : : /*
271 : : * Either I just allocated the initial hash table in
272 : : * ExecHashTableCreate(), or someone else is doing that. Either
273 : : * way, wait for everyone to arrive here so we can proceed.
274 : : */
275 : 121 : BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ALLOCATE);
276 : : pg_fallthrough;
277 : :
278 : 211 : case PHJ_BUILD_HASH_INNER:
279 : :
280 : : /*
281 : : * It's time to begin hashing, or if we just arrived here then
282 : : * hashing is already underway, so join in that effort. While
283 : : * hashing we have to be prepared to help increase the number of
284 : : * batches or buckets at any time, and if we arrived here when
285 : : * that was already underway we'll have to help complete that work
286 : : * immediately so that it's safe to access batches and buckets
287 : : * below.
288 : : */
289 [ + + ]: 211 : if (PHJ_GROW_BATCHES_PHASE(BarrierAttach(&pstate->grow_batches_barrier)) !=
290 : : PHJ_GROW_BATCHES_ELECT)
291 : 2 : ExecParallelHashIncreaseNumBatches(hashtable);
292 [ - + ]: 211 : if (PHJ_GROW_BUCKETS_PHASE(BarrierAttach(&pstate->grow_buckets_barrier)) !=
293 : : PHJ_GROW_BUCKETS_ELECT)
294 : 0 : ExecParallelHashIncreaseNumBuckets(hashtable);
295 : 211 : ExecParallelHashEnsureBatchAccessors(hashtable);
296 : 211 : ExecParallelHashTableSetCurrentBatch(hashtable, 0);
297 : : for (;;)
298 : 1440160 : {
299 : : bool isnull;
300 : : uint32 hashvalue;
301 : :
302 : 1440371 : slot = ExecProcNode(outerNode);
303 [ + + + + ]: 1440371 : if (TupIsNull(slot))
304 : : break;
305 : 1440160 : econtext->ecxt_outertuple = slot;
306 : :
307 : 1440160 : ResetExprContext(econtext);
308 : :
309 : 1440160 : hashvalue = DatumGetUInt32(ExecEvalExprSwitchContext(node->hash_expr,
310 : : econtext,
311 : : &isnull));
312 : :
313 [ + + ]: 1440160 : if (!isnull)
314 : : {
315 : : /* normal case with a non-null join key */
316 : 1440128 : ExecParallelHashTableInsert(hashtable, slot, hashvalue);
317 : 1440128 : hashtable->reportTuples++;
318 : : }
319 [ + + ]: 32 : else if (node->keep_null_tuples)
320 : : {
321 : : /* null join key, but save tuple to be emitted later */
322 [ + - ]: 12 : if (node->null_tuple_store == NULL)
323 : 12 : node->null_tuple_store = ExecHashBuildNullTupleStore(hashtable);
324 : 12 : tuplestore_puttupleslot(node->null_tuple_store, slot);
325 : 12 : hashtable->reportTuples++;
326 : : }
327 : : /* else we can discard the tuple immediately */
328 : : }
329 : :
330 : : /*
331 : : * Make sure that any tuples we wrote to disk are visible to
332 : : * others before anyone tries to load them.
333 : : */
334 [ + + ]: 1160 : for (i = 0; i < hashtable->nbatch; ++i)
335 : 949 : sts_end_write(hashtable->batches[i].inner_tuples);
336 : :
337 : : /*
338 : : * Update shared counters. We need an accurate total tuple count
339 : : * to control the empty table optimization.
340 : : */
341 : 211 : ExecParallelHashMergeCounters(hashtable);
342 : :
343 : 211 : BarrierDetach(&pstate->grow_buckets_barrier);
344 : 211 : BarrierDetach(&pstate->grow_batches_barrier);
345 : :
346 : : /*
347 : : * Wait for everyone to finish building and flushing files and
348 : : * counters.
349 : : */
350 [ + + ]: 211 : if (BarrierArriveAndWait(build_barrier,
351 : : WAIT_EVENT_HASH_BUILD_HASH_INNER))
352 : : {
353 : : /*
354 : : * Elect one backend to disable any further growth. Batches
355 : : * are now fixed. While building them we made sure they'd fit
356 : : * in our memory budget when we load them back in later (or we
357 : : * tried to do that and gave up because we detected extreme
358 : : * skew).
359 : : */
360 : 116 : pstate->growth = PHJ_GROWTH_DISABLED;
361 : : }
362 : : }
363 : :
364 : : /*
365 : : * We're not yet attached to a batch. We all agree on the dimensions and
366 : : * number of inner tuples. (In parallel mode, totalTuples isn't used in
367 : : * this module, but we must report it for nodeHashjoin.c's empty-table
368 : : * optimization.)
369 : : */
370 : 276 : hashtable->curbatch = -1;
371 : 276 : hashtable->nbuckets = pstate->nbuckets;
372 : 276 : hashtable->log2_nbuckets = pg_ceil_log2_32(hashtable->nbuckets);
373 : 276 : hashtable->totalTuples = pstate->total_tuples;
374 : :
375 : : /*
376 : : * Unless we're completely done and the batch state has been freed, make
377 : : * sure we have accessors.
378 : : */
379 [ + + ]: 276 : if (BarrierPhase(build_barrier) < PHJ_BUILD_FREE)
380 : 275 : ExecParallelHashEnsureBatchAccessors(hashtable);
381 : :
382 : : /*
383 : : * The next synchronization point is in ExecHashJoin's HJ_BUILD_HASHTABLE
384 : : * case, which will bring the build phase to PHJ_BUILD_RUN (if it isn't
385 : : * there already).
386 : : */
387 : : Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASH_OUTER ||
388 : : BarrierPhase(build_barrier) == PHJ_BUILD_RUN ||
389 : : BarrierPhase(build_barrier) == PHJ_BUILD_FREE);
390 : 276 : }
391 : :
392 : : /* ----------------------------------------------------------------
393 : : * ExecInitHash
394 : : *
395 : : * Init routine for Hash node
396 : : * ----------------------------------------------------------------
397 : : */
398 : : HashState *
399 : 29616 : ExecInitHash(Hash *node, EState *estate, int eflags)
400 : : {
401 : : HashState *hashstate;
402 : :
403 : : /* check for unsupported flags */
404 : : Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
405 : :
406 : : /*
407 : : * create state structure
408 : : */
409 : 29616 : hashstate = makeNode(HashState);
410 : 29616 : hashstate->ps.plan = (Plan *) node;
411 : 29616 : hashstate->ps.state = estate;
412 : 29616 : hashstate->ps.ExecProcNode = ExecHash;
413 : : /* delay building hashtable until ExecHashTableCreate() in executor run */
414 : 29616 : hashstate->hashtable = NULL;
415 : :
416 : : /*
417 : : * Miscellaneous initialization
418 : : *
419 : : * create expression context for node
420 : : */
421 : 29616 : ExecAssignExprContext(estate, &hashstate->ps);
422 : :
423 : : /*
424 : : * initialize child nodes
425 : : */
426 : 29616 : outerPlanState(hashstate) = ExecInitNode(outerPlan(node), estate, eflags);
427 : :
428 : : /*
429 : : * initialize our result slot and type. No need to build projection
430 : : * because this node doesn't do projections.
431 : : */
432 : 29616 : ExecInitResultTupleSlotTL(&hashstate->ps, &TTSOpsMinimalTuple);
433 : 29616 : hashstate->ps.ps_ProjInfo = NULL;
434 : :
435 : : Assert(node->plan.qual == NIL);
436 : :
437 : : /* these fields will be filled by ExecInitHashJoin() */
438 : 29616 : hashstate->hash_expr = NULL;
439 : 29616 : hashstate->null_tuple_store = NULL;
440 : 29616 : hashstate->keep_null_tuples = false;
441 : :
442 : 29616 : return hashstate;
443 : : }
444 : :
445 : : /* ---------------------------------------------------------------
446 : : * ExecEndHash
447 : : *
448 : : * clean up routine for Hash node
449 : : * ----------------------------------------------------------------
450 : : */
451 : : void
452 : 29544 : ExecEndHash(HashState *node)
453 : : {
454 : : PlanState *outerPlan;
455 : :
456 : : /*
457 : : * shut down the subplan
458 : : */
459 : 29544 : outerPlan = outerPlanState(node);
460 : 29544 : ExecEndNode(outerPlan);
461 : 29544 : }
462 : :
463 : :
464 : : /* ----------------------------------------------------------------
465 : : * ExecHashTableCreate
466 : : *
467 : : * create an empty hashtable data structure for hashjoin.
468 : : * ----------------------------------------------------------------
469 : : */
470 : : HashJoinTable
471 : 20968 : ExecHashTableCreate(HashState *state)
472 : : {
473 : : Hash *node;
474 : : HashJoinTable hashtable;
475 : : Plan *outerNode;
476 : : size_t space_allowed;
477 : : int nbuckets;
478 : : int nbatch;
479 : : double rows;
480 : : int num_skew_mcvs;
481 : : int log2_nbuckets;
482 : : MemoryContext oldcxt;
483 : :
484 : : /*
485 : : * Get information about the size of the relation to be hashed (it's the
486 : : * "outer" subtree of this node, but the inner relation of the hashjoin).
487 : : * Compute the appropriate size of the hash table.
488 : : */
489 : 20968 : node = (Hash *) state->ps.plan;
490 : 20968 : outerNode = outerPlan(node);
491 : :
492 : : /*
493 : : * If this is shared hash table with a partial plan, then we can't use
494 : : * outerNode->plan_rows to estimate its size. We need an estimate of the
495 : : * total number of rows across all copies of the partial plan.
496 : : */
497 [ + + ]: 20968 : rows = node->plan.parallel_aware ? node->rows_total : outerNode->plan_rows;
498 : :
499 : 20692 : ExecChooseHashTableSize(rows, outerNode->plan_width,
500 : 20968 : OidIsValid(node->skewTable),
501 : 20968 : state->parallel_state != NULL,
502 [ + + ]: 20968 : state->parallel_state != NULL ?
503 : 276 : state->parallel_state->nparticipants - 1 : 0,
504 : : &space_allowed,
505 : : &nbuckets, &nbatch, &num_skew_mcvs);
506 : :
507 : : /* nbuckets must be a power of 2 */
508 : 20968 : log2_nbuckets = pg_ceil_log2_32(nbuckets);
509 : : Assert(nbuckets == (1 << log2_nbuckets));
510 : :
511 : : /*
512 : : * Initialize the hash table control block.
513 : : *
514 : : * The hashtable control block is just palloc'd from the executor's
515 : : * per-query memory context. Everything else should be kept inside the
516 : : * subsidiary hashCxt, batchCxt or spillCxt.
517 : : */
518 : 20968 : hashtable = palloc_object(HashJoinTableData);
519 : 20968 : hashtable->nbuckets = nbuckets;
520 : 20968 : hashtable->nbuckets_original = nbuckets;
521 : 20968 : hashtable->nbuckets_optimal = nbuckets;
522 : 20968 : hashtable->log2_nbuckets = log2_nbuckets;
523 : 20968 : hashtable->log2_nbuckets_optimal = log2_nbuckets;
524 : 20968 : hashtable->buckets.unshared = NULL;
525 : 20968 : hashtable->skewEnabled = false;
526 : 20968 : hashtable->skewBucket = NULL;
527 : 20968 : hashtable->skewBucketLen = 0;
528 : 20968 : hashtable->nSkewBuckets = 0;
529 : 20968 : hashtable->skewBucketNums = NULL;
530 : 20968 : hashtable->nbatch = nbatch;
531 : 20968 : hashtable->curbatch = 0;
532 : 20968 : hashtable->nbatch_original = nbatch;
533 : 20968 : hashtable->nbatch_outstart = nbatch;
534 : 20968 : hashtable->growEnabled = true;
535 : 20968 : hashtable->totalTuples = 0;
536 : 20968 : hashtable->reportTuples = 0;
537 : 20968 : hashtable->skewTuples = 0;
538 : 20968 : hashtable->innerBatchFile = NULL;
539 : 20968 : hashtable->outerBatchFile = NULL;
540 : 20968 : hashtable->spaceUsed = 0;
541 : 20968 : hashtable->spacePeak = 0;
542 : 20968 : hashtable->spaceAllowed = space_allowed;
543 : 20968 : hashtable->spaceUsedSkew = 0;
544 : 20968 : hashtable->spaceAllowedSkew =
545 : 20968 : hashtable->spaceAllowed * SKEW_HASH_MEM_PERCENT / 100;
546 : 20968 : hashtable->chunks = NULL;
547 : 20968 : hashtable->current_chunk = NULL;
548 : 20968 : hashtable->parallel_state = state->parallel_state;
549 : 20968 : hashtable->area = state->ps.state->es_query_dsa;
550 : 20968 : hashtable->batches = NULL;
551 : :
552 : : #ifdef HJDEBUG
553 : : printf("Hashjoin %p: initial nbatch = %d, nbuckets = %d\n",
554 : : hashtable, nbatch, nbuckets);
555 : : #endif
556 : :
557 : : /*
558 : : * Create temporary memory contexts in which to keep the hashtable working
559 : : * storage. See notes in executor/hashjoin.h.
560 : : */
561 : 20968 : hashtable->hashCxt = AllocSetContextCreate(CurrentMemoryContext,
562 : : "HashTableContext",
563 : : ALLOCSET_DEFAULT_SIZES);
564 : :
565 : 20968 : hashtable->batchCxt = AllocSetContextCreate(hashtable->hashCxt,
566 : : "HashBatchContext",
567 : : ALLOCSET_DEFAULT_SIZES);
568 : :
569 : 20968 : hashtable->spillCxt = AllocSetContextCreate(hashtable->hashCxt,
570 : : "HashSpillContext",
571 : : ALLOCSET_DEFAULT_SIZES);
572 : :
573 : : /* Allocate data that will live for the life of the hashjoin */
574 : :
575 : 20968 : oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
576 : :
577 [ + + + + ]: 20968 : if (nbatch > 1 && hashtable->parallel_state == NULL)
578 : : {
579 : : MemoryContext oldctx;
580 : :
581 : : /*
582 : : * allocate and initialize the file arrays in hashCxt (not needed for
583 : : * parallel case which uses shared tuplestores instead of raw files)
584 : : */
585 : 82 : oldctx = MemoryContextSwitchTo(hashtable->spillCxt);
586 : :
587 : 82 : hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch);
588 : 82 : hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch);
589 : :
590 : 82 : MemoryContextSwitchTo(oldctx);
591 : :
592 : : /* The files will not be opened until needed... */
593 : : /* ... but make sure we have temp tablespaces established for them */
594 : 82 : PrepareTempTablespaces();
595 : : }
596 : :
597 : 20968 : MemoryContextSwitchTo(oldcxt);
598 : :
599 [ + + ]: 20968 : if (hashtable->parallel_state)
600 : : {
601 : 276 : ParallelHashJoinState *pstate = hashtable->parallel_state;
602 : : Barrier *build_barrier;
603 : :
604 : : /*
605 : : * Attach to the build barrier. The corresponding detach operation is
606 : : * in ExecHashTableDetach. Note that we won't attach to the
607 : : * batch_barrier for batch 0 yet. We'll attach later and start it out
608 : : * in PHJ_BATCH_PROBE phase, because batch 0 is allocated up front and
609 : : * then loaded while hashing (the standard hybrid hash join
610 : : * algorithm), and we'll coordinate that using build_barrier.
611 : : */
612 : 276 : build_barrier = &pstate->build_barrier;
613 : 276 : BarrierAttach(build_barrier);
614 : :
615 : : /*
616 : : * So far we have no idea whether there are any other participants,
617 : : * and if so, what phase they are working on. The only thing we care
618 : : * about at this point is whether someone has already created the
619 : : * SharedHashJoinBatch objects and the hash table for batch 0. One
620 : : * backend will be elected to do that now if necessary.
621 : : */
622 [ + + + - ]: 392 : if (BarrierPhase(build_barrier) == PHJ_BUILD_ELECT &&
623 : 116 : BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ELECT))
624 : : {
625 : 116 : pstate->nbatch = nbatch;
626 : 116 : pstate->space_allowed = space_allowed;
627 : 116 : pstate->growth = PHJ_GROWTH_OK;
628 : :
629 : : /* Set up the shared state for coordinating batches. */
630 : 116 : ExecParallelHashJoinSetUpBatches(hashtable, nbatch);
631 : :
632 : : /*
633 : : * Allocate batch 0's hash table up front so we can load it
634 : : * directly while hashing.
635 : : */
636 : 116 : pstate->nbuckets = nbuckets;
637 : 116 : ExecParallelHashTableAlloc(hashtable, 0);
638 : : }
639 : :
640 : : /*
641 : : * The next Parallel Hash synchronization point is in
642 : : * MultiExecParallelHash(), which will progress it all the way to
643 : : * PHJ_BUILD_RUN. The caller must not return control from this
644 : : * executor node between now and then.
645 : : */
646 : : }
647 : : else
648 : : {
649 : : /*
650 : : * Prepare context for the first-scan space allocations; allocate the
651 : : * hashbucket array therein, and set each bucket "empty".
652 : : */
653 : 20692 : MemoryContextSwitchTo(hashtable->batchCxt);
654 : :
655 : 20692 : hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets);
656 : :
657 : : /*
658 : : * Set up for skew optimization, if possible and there's a need for
659 : : * more than one batch. (In a one-batch join, there's no point in
660 : : * it.)
661 : : */
662 [ + + ]: 20692 : if (nbatch > 1)
663 : 82 : ExecHashBuildSkewHash(state, hashtable, node, num_skew_mcvs);
664 : :
665 : 20692 : MemoryContextSwitchTo(oldcxt);
666 : : }
667 : :
668 : 20968 : return hashtable;
669 : : }
670 : :
671 : :
672 : : /*
673 : : * Compute appropriate size for hashtable given the estimated size of the
674 : : * relation to be hashed (number of rows and average row width).
675 : : *
676 : : * This is exported so that the planner's costsize.c can use it.
677 : : */
678 : :
679 : : /* Target bucket loading (tuples per bucket) */
680 : : #define NTUP_PER_BUCKET 1
681 : :
682 : : void
683 : 642035 : ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
684 : : bool try_combined_hash_mem,
685 : : int parallel_workers,
686 : : size_t *space_allowed,
687 : : int *numbuckets,
688 : : int *numbatches,
689 : : int *num_skew_mcvs)
690 : : {
691 : : int tupsize;
692 : : double inner_rel_bytes;
693 : : size_t hash_table_bytes;
694 : : size_t bucket_bytes;
695 : : size_t max_pointers;
696 : 642035 : int nbatch = 1;
697 : : int nbuckets;
698 : : double dbuckets;
699 : :
700 : : /* Force a plausible relation size if no info */
701 [ + + ]: 642035 : if (ntuples <= 0.0)
702 : 119 : ntuples = 1000.0;
703 : :
704 : : /*
705 : : * Estimate tupsize based on footprint of tuple in hashtable... note this
706 : : * does not allow for any palloc overhead. The manipulations of spaceUsed
707 : : * don't count palloc overhead either.
708 : : */
709 : 642035 : tupsize = HJTUPLE_OVERHEAD +
710 : 642035 : MAXALIGN(SizeofMinimalTupleHeader) +
711 : 642035 : MAXALIGN(tupwidth);
712 : 642035 : inner_rel_bytes = ntuples * tupsize;
713 : :
714 : : /*
715 : : * Compute in-memory hashtable size limit from GUCs.
716 : : */
717 : 642035 : hash_table_bytes = get_hash_memory_limit();
718 : :
719 : : /*
720 : : * Parallel Hash tries to use the combined hash_mem of all workers to
721 : : * avoid the need to batch. If that won't work, it falls back to hash_mem
722 : : * per worker and tries to process batches in parallel.
723 : : */
724 [ + + ]: 642035 : if (try_combined_hash_mem)
725 : : {
726 : : /* Careful, this could overflow size_t */
727 : : double newlimit;
728 : :
729 : 58218 : newlimit = (double) hash_table_bytes * (double) (parallel_workers + 1);
730 [ + - ]: 58218 : newlimit = Min(newlimit, (double) SIZE_MAX);
731 : 58218 : hash_table_bytes = (size_t) newlimit;
732 : : }
733 : :
734 : 642035 : *space_allowed = hash_table_bytes;
735 : :
736 : : /*
737 : : * If skew optimization is possible, estimate the number of skew buckets
738 : : * that will fit in the memory allowed, and decrement the assumed space
739 : : * available for the main hash table accordingly.
740 : : *
741 : : * We make the optimistic assumption that each skew bucket will contain
742 : : * one inner-relation tuple. If that turns out to be low, we will recover
743 : : * at runtime by reducing the number of skew buckets.
744 : : *
745 : : * hashtable->skewBucket will have up to 8 times as many HashSkewBucket
746 : : * pointers as the number of MCVs we allow, since ExecHashBuildSkewHash
747 : : * will round up to the next power of 2 and then multiply by 4 to reduce
748 : : * collisions.
749 : : */
750 [ + + ]: 642035 : if (useskew)
751 : : {
752 : : size_t bytes_per_mcv;
753 : : size_t skew_mcvs;
754 : :
755 : : /*----------
756 : : * Compute number of MCVs we could hold in hash_table_bytes
757 : : *
758 : : * Divisor is:
759 : : * size of a hash tuple +
760 : : * worst-case size of skewBucket[] per MCV +
761 : : * size of skewBucketNums[] entry +
762 : : * size of skew bucket struct itself
763 : : *----------
764 : : */
765 : 638289 : bytes_per_mcv = tupsize +
766 : : (8 * sizeof(HashSkewBucket *)) +
767 : 638289 : sizeof(int) +
768 : : SKEW_BUCKET_OVERHEAD;
769 : 638289 : skew_mcvs = hash_table_bytes / bytes_per_mcv;
770 : :
771 : : /*
772 : : * Now scale by SKEW_HASH_MEM_PERCENT (we do it in this order so as
773 : : * not to worry about size_t overflow in the multiplication)
774 : : */
775 : 638289 : skew_mcvs = (skew_mcvs * SKEW_HASH_MEM_PERCENT) / 100;
776 : :
777 : : /* Now clamp to integer range */
778 : 638289 : skew_mcvs = Min(skew_mcvs, INT_MAX);
779 : :
780 : 638289 : *num_skew_mcvs = (int) skew_mcvs;
781 : :
782 : : /* Reduce hash_table_bytes by the amount needed for the skew table */
783 [ + - ]: 638289 : if (skew_mcvs > 0)
784 : 638289 : hash_table_bytes -= skew_mcvs * bytes_per_mcv;
785 : : }
786 : : else
787 : 3746 : *num_skew_mcvs = 0;
788 : :
789 : : /*
790 : : * Set nbuckets to achieve an average bucket load of NTUP_PER_BUCKET when
791 : : * memory is filled, assuming a single batch; but limit the value so that
792 : : * the pointer arrays we'll try to allocate do not exceed hash_table_bytes
793 : : * nor MaxAllocSize.
794 : : *
795 : : * Note that both nbuckets and nbatch must be powers of 2 to make
796 : : * ExecHashGetBucketAndBatch fast.
797 : : */
798 : 642035 : max_pointers = hash_table_bytes / sizeof(HashJoinTuple);
799 : 642035 : max_pointers = Min(max_pointers, MaxAllocSize / sizeof(HashJoinTuple));
800 : : /* If max_pointers isn't a power of 2, must round it down to one */
801 : 642035 : max_pointers = pg_prevpower2_size_t(max_pointers);
802 : :
803 : : /* Also ensure we avoid integer overflow in nbatch and nbuckets */
804 : : /* (this step is redundant given the current value of MaxAllocSize) */
805 : 642035 : max_pointers = Min(max_pointers, INT_MAX / 2 + 1);
806 : :
807 : 642035 : dbuckets = ceil(ntuples / NTUP_PER_BUCKET);
808 [ + + ]: 642035 : dbuckets = Min(dbuckets, max_pointers);
809 : 642035 : nbuckets = (int) dbuckets;
810 : : /* don't let nbuckets be really small, though ... */
811 : 642035 : nbuckets = Max(nbuckets, 1024);
812 : : /* ... and force it to be a power of 2. */
813 : 642035 : nbuckets = pg_nextpower2_32(nbuckets);
814 : :
815 : : /*
816 : : * If there's not enough space to store the projected number of tuples and
817 : : * the required bucket headers, we will need multiple batches.
818 : : */
819 : 642035 : bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
820 [ + + ]: 642035 : if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
821 : : {
822 : : /* We'll need multiple batches */
823 : : size_t sbuckets;
824 : : double dbatch;
825 : : int minbatch;
826 : : size_t bucket_size;
827 : :
828 : : /*
829 : : * If Parallel Hash with combined hash_mem would still need multiple
830 : : * batches, we'll have to fall back to regular hash_mem budget.
831 : : */
832 [ + + ]: 3394 : if (try_combined_hash_mem)
833 : : {
834 : 172 : ExecChooseHashTableSize(ntuples, tupwidth, useskew,
835 : : false, parallel_workers,
836 : : space_allowed,
837 : : numbuckets,
838 : : numbatches,
839 : : num_skew_mcvs);
840 : 172 : return;
841 : : }
842 : :
843 : : /*
844 : : * Estimate the number of buckets we'll want to have when hash_mem is
845 : : * entirely full. Each bucket will contain a bucket pointer plus
846 : : * NTUP_PER_BUCKET tuples, whose projected size already includes
847 : : * overhead for the hash code, pointer to the next tuple, etc.
848 : : */
849 : 3222 : bucket_size = (tupsize * NTUP_PER_BUCKET + sizeof(HashJoinTuple));
850 [ - + ]: 3222 : if (hash_table_bytes <= bucket_size)
851 : 0 : sbuckets = 1; /* avoid pg_nextpower2_size_t(0) */
852 : : else
853 : 3222 : sbuckets = pg_nextpower2_size_t(hash_table_bytes / bucket_size);
854 : 3222 : sbuckets = Min(sbuckets, max_pointers);
855 : 3222 : nbuckets = (int) sbuckets;
856 : 3222 : nbuckets = pg_nextpower2_32(nbuckets);
857 : 3222 : bucket_bytes = nbuckets * sizeof(HashJoinTuple);
858 : :
859 : : /*
860 : : * Buckets are simple pointers to hashjoin tuples, while tupsize
861 : : * includes the pointer, hash code, and MinimalTupleData. So buckets
862 : : * should never really exceed 25% of hash_mem (even for
863 : : * NTUP_PER_BUCKET=1); except maybe for hash_mem values that are not
864 : : * 2^N bytes, where we might get more because of doubling. So let's
865 : : * look for 50% here.
866 : : */
867 : : Assert(bucket_bytes <= hash_table_bytes / 2);
868 : :
869 : : /* Calculate required number of batches. */
870 : 3222 : dbatch = ceil(inner_rel_bytes / (hash_table_bytes - bucket_bytes));
871 [ + - ]: 3222 : dbatch = Min(dbatch, max_pointers);
872 : 3222 : minbatch = (int) dbatch;
873 : 3222 : nbatch = pg_nextpower2_32(Max(2, minbatch));
874 : : }
875 : :
876 : : /*
877 : : * Optimize the total amount of memory consumed by the hash node.
878 : : *
879 : : * The nbatch calculation above focuses on the in-memory hash table,
880 : : * assuming no per-batch overhead. But each batch may have two files, each
881 : : * with a BLCKSZ buffer. For large nbatch values these buffers may use
882 : : * significantly more memory than the hash table.
883 : : *
884 : : * The total memory usage may be expressed by this formula:
885 : : *
886 : : * (inner_rel_bytes / nbatch) + (2 * nbatch * BLCKSZ)
887 : : *
888 : : * where (inner_rel_bytes / nbatch) is the size of the in-memory hash
889 : : * table and (2 * nbatch * BLCKSZ) is the amount of memory used by file
890 : : * buffers.
891 : : *
892 : : * The nbatch calculation however ignores the second part. And for very
893 : : * large inner_rel_bytes, there may be no nbatch that keeps total memory
894 : : * usage under the budget (work_mem * hash_mem_multiplier). To deal with
895 : : * that, we will adjust nbatch to minimize total memory consumption across
896 : : * both the hashtable and file buffers.
897 : : *
898 : : * As we increase the size of the hashtable, the number of batches
899 : : * decreases, and the total memory usage follows a U-shaped curve. We find
900 : : * the minimum nbatch by "walking back" -- checking if halving nbatch
901 : : * would lower the total memory usage. We stop when it no longer helps.
902 : : *
903 : : * We only reduce the number of batches. Adding batches reduces memory
904 : : * usage only when most of the memory is used by the hash table, with
905 : : * total memory usage within the limit or not far from it. We don't want
906 : : * to start batching when not needed, even if that would reduce memory
907 : : * usage.
908 : : *
909 : : * While growing the hashtable, we also adjust the number of buckets to
910 : : * maintain a load factor of NTUP_PER_BUCKET while squeezing tuples back
911 : : * from batches into the hashtable.
912 : : *
913 : : * Note that we can only change nbuckets during initial hashtable sizing.
914 : : * Once we start building the hash, nbuckets is fixed (we may still grow
915 : : * the hash table).
916 : : *
917 : : * We double several parameters (space_allowed, nbuckets, num_skew_mcvs),
918 : : * which introduces a risk of overflow. We avoid this by exiting the loop.
919 : : * We could do something smarter (e.g. capping nbuckets and continue), but
920 : : * the complexity is not worth it. Such cases are extremely rare, and this
921 : : * is a best-effort attempt to reduce memory usage.
922 : : */
923 [ + + ]: 642471 : while (nbatch > 1)
924 : : {
925 : : /* Check that buckets won't overflow MaxAllocSize */
926 [ - + ]: 3830 : if (nbuckets > (MaxAllocSize / sizeof(HashJoinTuple) / 2))
927 : 0 : break;
928 : :
929 : : /* num_skew_mcvs should be less than nbuckets */
930 : : Assert((*num_skew_mcvs) < (INT_MAX / 2));
931 : :
932 : : /*
933 : : * Check that space_allowed won't overflow SIZE_MAX.
934 : : *
935 : : * We don't use hash_table_bytes here, because it does not include the
936 : : * skew buckets. And we want to limit the overall memory limit.
937 : : */
938 [ - + ]: 3830 : if ((*space_allowed) > (SIZE_MAX / 2))
939 : 0 : break;
940 : :
941 : : /*
942 : : * Will halving the number of batches and doubling the size of the
943 : : * hashtable reduce overall memory usage?
944 : : *
945 : : * This is the same as (S = space_allowed):
946 : : *
947 : : * (S + 2 * nbatch * BLCKSZ) < (S * 2 + nbatch * BLCKSZ)
948 : : *
949 : : * but avoiding intermediate overflow.
950 : : */
951 [ + + ]: 3830 : if (nbatch < (*space_allowed) / BLCKSZ)
952 : 3222 : break;
953 : :
954 : : /*
955 : : * MaxAllocSize is sufficiently small that we are not worried about
956 : : * overflowing nbuckets.
957 : : */
958 : 608 : nbuckets *= 2;
959 : :
960 : 608 : *num_skew_mcvs = (*num_skew_mcvs) * 2;
961 : 608 : *space_allowed = (*space_allowed) * 2;
962 : :
963 : 608 : nbatch /= 2;
964 : : }
965 : :
966 : : Assert(nbuckets > 0);
967 : : Assert(nbatch > 0);
968 : :
969 : 641863 : *numbuckets = nbuckets;
970 : 641863 : *numbatches = nbatch;
971 : : }
972 : :
973 : :
974 : : /* ----------------------------------------------------------------
975 : : * ExecHashTableDestroy
976 : : *
977 : : * destroy a hash table
978 : : * ----------------------------------------------------------------
979 : : */
980 : : void
981 : 20897 : ExecHashTableDestroy(HashJoinTable hashtable)
982 : : {
983 : : int i;
984 : :
985 : : /*
986 : : * Make sure all the temp files are closed. We skip batch 0, since it
987 : : * can't have any temp files (and the arrays might not even exist if
988 : : * nbatch is only 1). Parallel hash joins don't use these files.
989 : : */
990 [ + + ]: 20897 : if (hashtable->innerBatchFile != NULL)
991 : : {
992 [ + + ]: 942 : for (i = 1; i < hashtable->nbatch; i++)
993 : : {
994 [ - + ]: 799 : if (hashtable->innerBatchFile[i])
995 : 0 : BufFileClose(hashtable->innerBatchFile[i]);
996 [ - + ]: 799 : if (hashtable->outerBatchFile[i])
997 : 0 : BufFileClose(hashtable->outerBatchFile[i]);
998 : : }
999 : : }
1000 : :
1001 : : /* Release working memory (batchCxt is a child, so it goes away too) */
1002 : 20897 : MemoryContextDelete(hashtable->hashCxt);
1003 : :
1004 : : /* And drop the control block */
1005 : 20897 : pfree(hashtable);
1006 : 20897 : }
1007 : :
1008 : : /*
1009 : : * Consider adjusting the allowed hash table size, depending on the number
1010 : : * of batches, to minimize the overall memory usage (for both the hashtable
1011 : : * and batch files).
1012 : : *
1013 : : * We're adjusting the size of the hash table, not the (optimal) number of
1014 : : * buckets. We can't change that once we start building the hash, due to how
1015 : : * ExecHashGetBucketAndBatch calculates batchno/bucketno from the hash. This
1016 : : * means the load factor may not be optimal, but we're in damage control so
1017 : : * we accept slower lookups. It's still much better than batch explosion.
1018 : : *
1019 : : * Returns true if we chose to increase the batch size (and thus we don't
1020 : : * need to add batches), and false if we should increase nbatch.
1021 : : */
1022 : : static bool
1023 : 125 : ExecHashIncreaseBatchSize(HashJoinTable hashtable)
1024 : : {
1025 : : /*
1026 : : * How much additional memory would doubling nbatch use? Each batch may
1027 : : * require two buffered files (inner/outer), with a BLCKSZ buffer.
1028 : : */
1029 : 125 : size_t batchSpace = (hashtable->nbatch * 2 * (size_t) BLCKSZ);
1030 : :
1031 : : /*
1032 : : * Compare the new space needed for doubling nbatch and for enlarging the
1033 : : * in-memory hash table. If doubling the hash table needs less memory,
1034 : : * just do that. Otherwise, continue with doubling the nbatch.
1035 : : *
1036 : : * We're either doubling spaceAllowed or batchSpace, so which of those
1037 : : * increases the memory usage the least is the same as comparing the
1038 : : * values directly.
1039 : : */
1040 [ - + ]: 125 : if (hashtable->spaceAllowed <= batchSpace)
1041 : : {
1042 : 0 : hashtable->spaceAllowed *= 2;
1043 : 0 : return true;
1044 : : }
1045 : :
1046 : 125 : return false;
1047 : : }
1048 : :
1049 : : /*
1050 : : * ExecHashIncreaseNumBatches
1051 : : * increase the original number of batches in order to reduce
1052 : : * current memory consumption
1053 : : */
1054 : : static void
1055 : 535495 : ExecHashIncreaseNumBatches(HashJoinTable hashtable)
1056 : : {
1057 : 535495 : int oldnbatch = hashtable->nbatch;
1058 : 535495 : int curbatch = hashtable->curbatch;
1059 : : int nbatch;
1060 : : long ninmemory;
1061 : : long nfreed;
1062 : : HashMemoryChunk oldchunks;
1063 : :
1064 : : /* do nothing if we've decided to shut off growth */
1065 [ + + ]: 535495 : if (!hashtable->growEnabled)
1066 : 535370 : return;
1067 : :
1068 : : /* safety check to avoid overflow */
1069 [ - + ]: 125 : if (oldnbatch > Min(INT_MAX / 2, MaxAllocSize / (sizeof(void *) * 2)))
1070 : 0 : return;
1071 : :
1072 : : /* consider increasing size of the in-memory hash table instead */
1073 [ - + ]: 125 : if (ExecHashIncreaseBatchSize(hashtable))
1074 : 0 : return;
1075 : :
1076 : 125 : nbatch = oldnbatch * 2;
1077 : : Assert(nbatch > 1);
1078 : :
1079 : : #ifdef HJDEBUG
1080 : : printf("Hashjoin %p: increasing nbatch to %d because space = %zu\n",
1081 : : hashtable, nbatch, hashtable->spaceUsed);
1082 : : #endif
1083 : :
1084 [ + + ]: 125 : if (hashtable->innerBatchFile == NULL)
1085 : : {
1086 : 61 : MemoryContext oldcxt = MemoryContextSwitchTo(hashtable->spillCxt);
1087 : :
1088 : : /* we had no file arrays before */
1089 : 61 : hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch);
1090 : 61 : hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch);
1091 : :
1092 : 61 : MemoryContextSwitchTo(oldcxt);
1093 : :
1094 : : /* time to establish the temp tablespaces, too */
1095 : 61 : PrepareTempTablespaces();
1096 : : }
1097 : : else
1098 : : {
1099 : : /* enlarge arrays and zero out added entries */
1100 : 64 : hashtable->innerBatchFile = repalloc0_array(hashtable->innerBatchFile, BufFile *, oldnbatch, nbatch);
1101 : 64 : hashtable->outerBatchFile = repalloc0_array(hashtable->outerBatchFile, BufFile *, oldnbatch, nbatch);
1102 : : }
1103 : :
1104 : 125 : hashtable->nbatch = nbatch;
1105 : :
1106 : : /*
1107 : : * Scan through the existing hash table entries and dump out any that are
1108 : : * no longer of the current batch.
1109 : : */
1110 : 125 : ninmemory = nfreed = 0;
1111 : :
1112 : : /* If know we need to resize nbuckets, we can do it while rebatching. */
1113 [ + + ]: 125 : if (hashtable->nbuckets_optimal != hashtable->nbuckets)
1114 : : {
1115 : : /* we never decrease the number of buckets */
1116 : : Assert(hashtable->nbuckets_optimal > hashtable->nbuckets);
1117 : :
1118 : 61 : hashtable->nbuckets = hashtable->nbuckets_optimal;
1119 : 61 : hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
1120 : :
1121 : 61 : hashtable->buckets.unshared =
1122 : 61 : repalloc_array(hashtable->buckets.unshared,
1123 : : HashJoinTuple, hashtable->nbuckets);
1124 : : }
1125 : :
1126 : : /*
1127 : : * We will scan through the chunks directly, so that we can reset the
1128 : : * buckets now and not have to keep track which tuples in the buckets have
1129 : : * already been processed. We will free the old chunks as we go.
1130 : : */
1131 : 125 : memset(hashtable->buckets.unshared, 0,
1132 : 125 : sizeof(HashJoinTuple) * hashtable->nbuckets);
1133 : 125 : oldchunks = hashtable->chunks;
1134 : 125 : hashtable->chunks = NULL;
1135 : :
1136 : : /* so, let's scan through the old chunks, and all tuples in each chunk */
1137 [ + + ]: 625 : while (oldchunks != NULL)
1138 : : {
1139 : 500 : HashMemoryChunk nextchunk = oldchunks->next.unshared;
1140 : :
1141 : : /* position within the buffer (up to oldchunks->used) */
1142 : 500 : size_t idx = 0;
1143 : :
1144 : : /* process all tuples stored in this chunk (and then free it) */
1145 [ + + ]: 341587 : while (idx < oldchunks->used)
1146 : : {
1147 : 341087 : HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(oldchunks) + idx);
1148 : 341087 : MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
1149 : 341087 : int hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len);
1150 : : int bucketno;
1151 : : int batchno;
1152 : :
1153 : 341087 : ninmemory++;
1154 : 341087 : ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1155 : : &bucketno, &batchno);
1156 : :
1157 [ + + ]: 341087 : if (batchno == curbatch)
1158 : : {
1159 : : /* keep tuple in memory - copy it into the new chunk */
1160 : : HashJoinTuple copyTuple;
1161 : :
1162 : 127136 : copyTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
1163 : 127136 : memcpy(copyTuple, hashTuple, hashTupleSize);
1164 : :
1165 : : /* and add it back to the appropriate bucket */
1166 : 127136 : copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1167 : 127136 : hashtable->buckets.unshared[bucketno] = copyTuple;
1168 : : }
1169 : : else
1170 : : {
1171 : : /* dump it out */
1172 : : Assert(batchno > curbatch);
1173 : 213951 : ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple),
1174 : : hashTuple->hashvalue,
1175 : 213951 : &hashtable->innerBatchFile[batchno],
1176 : : hashtable);
1177 : :
1178 : 213951 : hashtable->spaceUsed -= hashTupleSize;
1179 : 213951 : nfreed++;
1180 : : }
1181 : :
1182 : : /* next tuple in this chunk */
1183 : 341087 : idx += MAXALIGN(hashTupleSize);
1184 : :
1185 : : /* allow this loop to be cancellable */
1186 [ - + ]: 341087 : CHECK_FOR_INTERRUPTS();
1187 : : }
1188 : :
1189 : : /* we're done with this chunk - free it and proceed to the next one */
1190 : 500 : pfree(oldchunks);
1191 : 500 : oldchunks = nextchunk;
1192 : : }
1193 : :
1194 : : #ifdef HJDEBUG
1195 : : printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n",
1196 : : hashtable, nfreed, ninmemory, hashtable->spaceUsed);
1197 : : #endif
1198 : :
1199 : : /*
1200 : : * If we dumped out either all or none of the tuples in the table, disable
1201 : : * further expansion of nbatch. This situation implies that we have
1202 : : * enough tuples of identical hashvalues to overflow spaceAllowed.
1203 : : * Increasing nbatch will not fix it since there's no way to subdivide the
1204 : : * group any more finely. We have to just gut it out and hope the server
1205 : : * has enough RAM.
1206 : : */
1207 [ + - + + ]: 125 : if (nfreed == 0 || nfreed == ninmemory)
1208 : : {
1209 : 31 : hashtable->growEnabled = false;
1210 : : #ifdef HJDEBUG
1211 : : printf("Hashjoin %p: disabling further increase of nbatch\n",
1212 : : hashtable);
1213 : : #endif
1214 : : }
1215 : : }
1216 : :
1217 : : /*
1218 : : * ExecParallelHashIncreaseNumBatches
1219 : : * Every participant attached to grow_batches_barrier must run this
1220 : : * function when it observes growth == PHJ_GROWTH_NEED_MORE_BATCHES.
1221 : : */
1222 : : static void
1223 : 41 : ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
1224 : : {
1225 : 41 : ParallelHashJoinState *pstate = hashtable->parallel_state;
1226 : :
1227 : : Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASH_INNER);
1228 : :
1229 : : /*
1230 : : * It's unlikely, but we need to be prepared for new participants to show
1231 : : * up while we're in the middle of this operation so we need to switch on
1232 : : * barrier phase here.
1233 : : */
1234 [ + - + - : 41 : switch (PHJ_GROW_BATCHES_PHASE(BarrierPhase(&pstate->grow_batches_barrier)))
- - ]
1235 : : {
1236 : 39 : case PHJ_GROW_BATCHES_ELECT:
1237 : :
1238 : : /*
1239 : : * Elect one participant to prepare to grow the number of batches.
1240 : : * This involves reallocating or resetting the buckets of batch 0
1241 : : * in preparation for all participants to begin repartitioning the
1242 : : * tuples.
1243 : : */
1244 [ + + ]: 39 : if (BarrierArriveAndWait(&pstate->grow_batches_barrier,
1245 : : WAIT_EVENT_HASH_GROW_BATCHES_ELECT))
1246 : : {
1247 : : dsa_pointer_atomic *buckets;
1248 : : ParallelHashJoinBatch *old_batch0;
1249 : : int new_nbatch;
1250 : : int i;
1251 : :
1252 : : /* Move the old batch out of the way. */
1253 : 32 : old_batch0 = hashtable->batches[0].shared;
1254 : 32 : pstate->old_batches = pstate->batches;
1255 : 32 : pstate->old_nbatch = hashtable->nbatch;
1256 : 32 : pstate->batches = InvalidDsaPointer;
1257 : :
1258 : : /* Free this backend's old accessors. */
1259 : 32 : ExecParallelHashCloseBatchAccessors(hashtable);
1260 : :
1261 : : /* Figure out how many batches to use. */
1262 [ + + ]: 32 : if (hashtable->nbatch == 1)
1263 : : {
1264 : : /*
1265 : : * We are going from single-batch to multi-batch. We need
1266 : : * to switch from one large combined memory budget to the
1267 : : * regular hash_mem budget.
1268 : : */
1269 : 24 : pstate->space_allowed = get_hash_memory_limit();
1270 : :
1271 : : /*
1272 : : * The combined hash_mem of all participants wasn't
1273 : : * enough. Therefore one batch per participant would be
1274 : : * approximately equivalent and would probably also be
1275 : : * insufficient. So try two batches per participant,
1276 : : * rounded up to a power of two.
1277 : : */
1278 : 24 : new_nbatch = pg_nextpower2_32(pstate->nparticipants * 2);
1279 : : }
1280 : : else
1281 : : {
1282 : : /*
1283 : : * We were already multi-batched. Try doubling the number
1284 : : * of batches.
1285 : : */
1286 : 8 : new_nbatch = hashtable->nbatch * 2;
1287 : : }
1288 : :
1289 : : /* Allocate new larger generation of batches. */
1290 : : Assert(hashtable->nbatch == pstate->nbatch);
1291 : 32 : ExecParallelHashJoinSetUpBatches(hashtable, new_nbatch);
1292 : : Assert(hashtable->nbatch == pstate->nbatch);
1293 : :
1294 : : /* Replace or recycle batch 0's bucket array. */
1295 [ + + ]: 32 : if (pstate->old_nbatch == 1)
1296 : : {
1297 : : double dtuples;
1298 : : double dbuckets;
1299 : : int new_nbuckets;
1300 : : uint32 max_buckets;
1301 : :
1302 : : /*
1303 : : * We probably also need a smaller bucket array. How many
1304 : : * tuples do we expect per batch, assuming we have only
1305 : : * half of them so far? Normally we don't need to change
1306 : : * the bucket array's size, because the size of each batch
1307 : : * stays the same as we add more batches, but in this
1308 : : * special case we move from a large batch to many smaller
1309 : : * batches and it would be wasteful to keep the large
1310 : : * array.
1311 : : */
1312 : 24 : dtuples = (old_batch0->ntuples * 2.0) / new_nbatch;
1313 : :
1314 : : /*
1315 : : * We need to calculate the maximum number of buckets to
1316 : : * stay within the MaxAllocSize boundary. Round the
1317 : : * maximum number to the previous power of 2 given that
1318 : : * later we round the number to the next power of 2.
1319 : : */
1320 : 24 : max_buckets = pg_prevpower2_32((uint32)
1321 : : (MaxAllocSize / sizeof(dsa_pointer_atomic)));
1322 : 24 : dbuckets = ceil(dtuples / NTUP_PER_BUCKET);
1323 [ + - ]: 24 : dbuckets = Min(dbuckets, max_buckets);
1324 : 24 : new_nbuckets = (int) dbuckets;
1325 : 24 : new_nbuckets = Max(new_nbuckets, 1024);
1326 : 24 : new_nbuckets = pg_nextpower2_32(new_nbuckets);
1327 : 24 : dsa_free(hashtable->area, old_batch0->buckets);
1328 : 48 : hashtable->batches[0].shared->buckets =
1329 : 24 : dsa_allocate(hashtable->area,
1330 : : sizeof(dsa_pointer_atomic) * new_nbuckets);
1331 : : buckets = (dsa_pointer_atomic *)
1332 : 24 : dsa_get_address(hashtable->area,
1333 : 24 : hashtable->batches[0].shared->buckets);
1334 [ + + ]: 73752 : for (i = 0; i < new_nbuckets; ++i)
1335 : 73728 : dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
1336 : 24 : pstate->nbuckets = new_nbuckets;
1337 : : }
1338 : : else
1339 : : {
1340 : : /* Recycle the existing bucket array. */
1341 : 8 : hashtable->batches[0].shared->buckets = old_batch0->buckets;
1342 : : buckets = (dsa_pointer_atomic *)
1343 : 8 : dsa_get_address(hashtable->area, old_batch0->buckets);
1344 [ + + ]: 32776 : for (i = 0; i < hashtable->nbuckets; ++i)
1345 : 32768 : dsa_pointer_atomic_write(&buckets[i], InvalidDsaPointer);
1346 : : }
1347 : :
1348 : : /* Move all chunks to the work queue for parallel processing. */
1349 : 32 : pstate->chunk_work_queue = old_batch0->chunks;
1350 : :
1351 : : /* Disable further growth temporarily while we're growing. */
1352 : 32 : pstate->growth = PHJ_GROWTH_DISABLED;
1353 : : }
1354 : : else
1355 : : {
1356 : : /* All other participants just flush their tuples to disk. */
1357 : 7 : ExecParallelHashCloseBatchAccessors(hashtable);
1358 : : }
1359 : : pg_fallthrough;
1360 : :
1361 : : case PHJ_GROW_BATCHES_REALLOCATE:
1362 : : /* Wait for the above to be finished. */
1363 : 39 : BarrierArriveAndWait(&pstate->grow_batches_barrier,
1364 : : WAIT_EVENT_HASH_GROW_BATCHES_REALLOCATE);
1365 : : pg_fallthrough;
1366 : :
1367 : 41 : case PHJ_GROW_BATCHES_REPARTITION:
1368 : : /* Make sure that we have the current dimensions and buckets. */
1369 : 41 : ExecParallelHashEnsureBatchAccessors(hashtable);
1370 : 41 : ExecParallelHashTableSetCurrentBatch(hashtable, 0);
1371 : : /* Then partition, flush counters. */
1372 : 41 : ExecParallelHashRepartitionFirst(hashtable);
1373 : 41 : ExecParallelHashRepartitionRest(hashtable);
1374 : 41 : ExecParallelHashMergeCounters(hashtable);
1375 : : /* Wait for the above to be finished. */
1376 : 41 : BarrierArriveAndWait(&pstate->grow_batches_barrier,
1377 : : WAIT_EVENT_HASH_GROW_BATCHES_REPARTITION);
1378 : : pg_fallthrough;
1379 : :
1380 : 41 : case PHJ_GROW_BATCHES_DECIDE:
1381 : :
1382 : : /*
1383 : : * Elect one participant to clean up and decide whether further
1384 : : * repartitioning is needed, or should be disabled because it's
1385 : : * not helping.
1386 : : */
1387 [ + + ]: 41 : if (BarrierArriveAndWait(&pstate->grow_batches_barrier,
1388 : : WAIT_EVENT_HASH_GROW_BATCHES_DECIDE))
1389 : : {
1390 : : ParallelHashJoinBatch *old_batches;
1391 : 32 : bool space_exhausted = false;
1392 : 32 : bool extreme_skew_detected = false;
1393 : :
1394 : : /* Make sure that we have the current dimensions and buckets. */
1395 : 32 : ExecParallelHashEnsureBatchAccessors(hashtable);
1396 : 32 : ExecParallelHashTableSetCurrentBatch(hashtable, 0);
1397 : :
1398 : 32 : old_batches = dsa_get_address(hashtable->area, pstate->old_batches);
1399 : :
1400 : : /* Are any of the new generation of batches exhausted? */
1401 [ + + ]: 224 : for (int i = 0; i < hashtable->nbatch; ++i)
1402 : : {
1403 : : ParallelHashJoinBatch *batch;
1404 : : ParallelHashJoinBatch *old_batch;
1405 : : int parent;
1406 : :
1407 : 192 : batch = hashtable->batches[i].shared;
1408 [ + - ]: 192 : if (batch->space_exhausted ||
1409 [ + + ]: 192 : batch->estimated_size > pstate->space_allowed)
1410 : 16 : space_exhausted = true;
1411 : :
1412 : 192 : parent = i % pstate->old_nbatch;
1413 : 192 : old_batch = NthParallelHashJoinBatch(old_batches, parent);
1414 [ + + ]: 192 : if (old_batch->space_exhausted ||
1415 [ - + ]: 48 : batch->estimated_size > pstate->space_allowed)
1416 : : {
1417 : : /*
1418 : : * Did this batch receive ALL of the tuples from its
1419 : : * parent batch? That would indicate that further
1420 : : * repartitioning isn't going to help (the hash values
1421 : : * are probably all the same).
1422 : : */
1423 [ + + ]: 144 : if (batch->ntuples == hashtable->batches[parent].shared->old_ntuples)
1424 : 16 : extreme_skew_detected = true;
1425 : : }
1426 : : }
1427 : :
1428 : : /* Don't keep growing if it's not helping or we'd overflow. */
1429 [ + + - + ]: 32 : if (extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2)
1430 : 16 : pstate->growth = PHJ_GROWTH_DISABLED;
1431 [ - + ]: 16 : else if (space_exhausted)
1432 : 0 : pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
1433 : : else
1434 : 16 : pstate->growth = PHJ_GROWTH_OK;
1435 : :
1436 : : /* Free the old batches in shared memory. */
1437 : 32 : dsa_free(hashtable->area, pstate->old_batches);
1438 : 32 : pstate->old_batches = InvalidDsaPointer;
1439 : : }
1440 : : pg_fallthrough;
1441 : :
1442 : : case PHJ_GROW_BATCHES_FINISH:
1443 : : /* Wait for the above to complete. */
1444 : 41 : BarrierArriveAndWait(&pstate->grow_batches_barrier,
1445 : : WAIT_EVENT_HASH_GROW_BATCHES_FINISH);
1446 : : }
1447 : 41 : }
1448 : :
1449 : : /*
1450 : : * Repartition the tuples currently loaded into memory for inner batch 0
1451 : : * because the number of batches has been increased. Some tuples are retained
1452 : : * in memory and some are written out to a later batch.
1453 : : */
1454 : : static void
1455 : 41 : ExecParallelHashRepartitionFirst(HashJoinTable hashtable)
1456 : : {
1457 : : dsa_pointer chunk_shared;
1458 : : HashMemoryChunk chunk;
1459 : :
1460 : : Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);
1461 : :
1462 [ + + ]: 275 : while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared)))
1463 : : {
1464 : 193 : size_t idx = 0;
1465 : :
1466 : : /* Repartition all tuples in this chunk. */
1467 [ + + ]: 148159 : while (idx < chunk->used)
1468 : : {
1469 : 147966 : HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1470 : 147966 : MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
1471 : : HashJoinTuple copyTuple;
1472 : : dsa_pointer shared;
1473 : : int bucketno;
1474 : : int batchno;
1475 : :
1476 : 147966 : ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1477 : : &bucketno, &batchno);
1478 : :
1479 : : Assert(batchno < hashtable->nbatch);
1480 [ + + ]: 147966 : if (batchno == 0)
1481 : : {
1482 : : /* It still belongs in batch 0. Copy to a new chunk. */
1483 : : copyTuple =
1484 : 34075 : ExecParallelHashTupleAlloc(hashtable,
1485 : 34075 : HJTUPLE_OVERHEAD + tuple->t_len,
1486 : : &shared);
1487 : 34075 : copyTuple->hashvalue = hashTuple->hashvalue;
1488 : 34075 : memcpy(HJTUPLE_MINTUPLE(copyTuple), tuple, tuple->t_len);
1489 : 34075 : ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1490 : : copyTuple, shared);
1491 : : }
1492 : : else
1493 : : {
1494 : 113891 : size_t tuple_size =
1495 : 113891 : MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1496 : :
1497 : : /* It belongs in a later batch. */
1498 : 113891 : hashtable->batches[batchno].estimated_size += tuple_size;
1499 : 113891 : sts_puttuple(hashtable->batches[batchno].inner_tuples,
1500 : 113891 : &hashTuple->hashvalue, tuple);
1501 : : }
1502 : :
1503 : : /* Count this tuple. */
1504 : 147966 : ++hashtable->batches[0].old_ntuples;
1505 : 147966 : ++hashtable->batches[batchno].ntuples;
1506 : :
1507 : 147966 : idx += MAXALIGN(HJTUPLE_OVERHEAD +
1508 : : HJTUPLE_MINTUPLE(hashTuple)->t_len);
1509 : : }
1510 : :
1511 : : /* Free this chunk. */
1512 : 193 : dsa_free(hashtable->area, chunk_shared);
1513 : :
1514 [ - + ]: 193 : CHECK_FOR_INTERRUPTS();
1515 : : }
1516 : 41 : }
1517 : :
1518 : : /*
1519 : : * Help repartition inner batches 1..n.
1520 : : */
1521 : : static void
1522 : 41 : ExecParallelHashRepartitionRest(HashJoinTable hashtable)
1523 : : {
1524 : 41 : ParallelHashJoinState *pstate = hashtable->parallel_state;
1525 : 41 : int old_nbatch = pstate->old_nbatch;
1526 : : SharedTuplestoreAccessor **old_inner_tuples;
1527 : : ParallelHashJoinBatch *old_batches;
1528 : : int i;
1529 : :
1530 : : /* Get our hands on the previous generation of batches. */
1531 : : old_batches = (ParallelHashJoinBatch *)
1532 : 41 : dsa_get_address(hashtable->area, pstate->old_batches);
1533 : 41 : old_inner_tuples = palloc0_array(SharedTuplestoreAccessor *, old_nbatch);
1534 [ + + ]: 86 : for (i = 1; i < old_nbatch; ++i)
1535 : : {
1536 : 45 : ParallelHashJoinBatch *shared =
1537 : 45 : NthParallelHashJoinBatch(old_batches, i);
1538 : :
1539 : 45 : old_inner_tuples[i] = sts_attach(ParallelHashJoinBatchInner(shared),
1540 : : ParallelWorkerNumber + 1,
1541 : : &pstate->fileset);
1542 : : }
1543 : :
1544 : : /* Join in the effort to repartition them. */
1545 [ + + ]: 86 : for (i = 1; i < old_nbatch; ++i)
1546 : : {
1547 : : MinimalTuple tuple;
1548 : : uint32 hashvalue;
1549 : :
1550 : : /* Scan one partition from the previous generation. */
1551 : 45 : sts_begin_parallel_scan(old_inner_tuples[i]);
1552 [ + + ]: 107693 : while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue)))
1553 : : {
1554 : 107648 : size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1555 : : int bucketno;
1556 : : int batchno;
1557 : :
1558 : : /* Decide which partition it goes to in the new generation. */
1559 : 107648 : ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
1560 : : &batchno);
1561 : :
1562 : 107648 : hashtable->batches[batchno].estimated_size += tuple_size;
1563 : 107648 : ++hashtable->batches[batchno].ntuples;
1564 : 107648 : ++hashtable->batches[i].old_ntuples;
1565 : :
1566 : : /* Store the tuple its new batch. */
1567 : 107648 : sts_puttuple(hashtable->batches[batchno].inner_tuples,
1568 : : &hashvalue, tuple);
1569 : :
1570 [ - + ]: 107648 : CHECK_FOR_INTERRUPTS();
1571 : : }
1572 : 45 : sts_end_parallel_scan(old_inner_tuples[i]);
1573 : : }
1574 : :
1575 : 41 : pfree(old_inner_tuples);
1576 : 41 : }
1577 : :
1578 : : /*
1579 : : * Transfer the backend-local per-batch counters to the shared totals.
1580 : : */
1581 : : static void
1582 : 252 : ExecParallelHashMergeCounters(HashJoinTable hashtable)
1583 : : {
1584 : 252 : ParallelHashJoinState *pstate = hashtable->parallel_state;
1585 : : int i;
1586 : :
1587 : 252 : LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
1588 : 252 : pstate->total_tuples = 0;
1589 [ + + ]: 1457 : for (i = 0; i < hashtable->nbatch; ++i)
1590 : : {
1591 : 1205 : ParallelHashJoinBatchAccessor *batch = &hashtable->batches[i];
1592 : :
1593 : 1205 : batch->shared->size += batch->size;
1594 : 1205 : batch->shared->estimated_size += batch->estimated_size;
1595 : 1205 : batch->shared->ntuples += batch->ntuples;
1596 : 1205 : batch->shared->old_ntuples += batch->old_ntuples;
1597 : 1205 : batch->size = 0;
1598 : 1205 : batch->estimated_size = 0;
1599 : 1205 : batch->ntuples = 0;
1600 : 1205 : batch->old_ntuples = 0;
1601 : 1205 : pstate->total_tuples += batch->shared->ntuples;
1602 : : }
1603 : 252 : LWLockRelease(&pstate->lock);
1604 : 252 : }
1605 : :
1606 : : /*
1607 : : * ExecHashIncreaseNumBuckets
1608 : : * increase the original number of buckets in order to reduce
1609 : : * number of tuples per bucket
1610 : : */
1611 : : static void
1612 : 112 : ExecHashIncreaseNumBuckets(HashJoinTable hashtable)
1613 : : {
1614 : : HashMemoryChunk chunk;
1615 : :
1616 : : /* do nothing if not an increase (it's called increase for a reason) */
1617 [ - + ]: 112 : if (hashtable->nbuckets >= hashtable->nbuckets_optimal)
1618 : 0 : return;
1619 : :
1620 : : #ifdef HJDEBUG
1621 : : printf("Hashjoin %p: increasing nbuckets %d => %d\n",
1622 : : hashtable, hashtable->nbuckets, hashtable->nbuckets_optimal);
1623 : : #endif
1624 : :
1625 : 112 : hashtable->nbuckets = hashtable->nbuckets_optimal;
1626 : 112 : hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
1627 : :
1628 : : Assert(hashtable->nbuckets > 1);
1629 : : Assert(hashtable->nbuckets <= (INT_MAX / 2));
1630 : : Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets));
1631 : :
1632 : : /*
1633 : : * Just reallocate the proper number of buckets - we don't need to walk
1634 : : * through them - we can walk the dense-allocated chunks (just like in
1635 : : * ExecHashIncreaseNumBatches, but without all the copying into new
1636 : : * chunks)
1637 : : */
1638 : 112 : hashtable->buckets.unshared =
1639 : 112 : repalloc_array(hashtable->buckets.unshared,
1640 : : HashJoinTuple, hashtable->nbuckets);
1641 : :
1642 : 112 : memset(hashtable->buckets.unshared, 0,
1643 : 112 : hashtable->nbuckets * sizeof(HashJoinTuple));
1644 : :
1645 : : /* scan through all tuples in all chunks to rebuild the hash table */
1646 [ + + ]: 2190 : for (chunk = hashtable->chunks; chunk != NULL; chunk = chunk->next.unshared)
1647 : : {
1648 : : /* process all tuples stored in this chunk */
1649 : 2078 : size_t idx = 0;
1650 : :
1651 [ + + ]: 403873 : while (idx < chunk->used)
1652 : : {
1653 : 401795 : HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1654 : : int bucketno;
1655 : : int batchno;
1656 : :
1657 : 401795 : ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1658 : : &bucketno, &batchno);
1659 : :
1660 : : /* add the tuple to the proper bucket */
1661 : 401795 : hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1662 : 401795 : hashtable->buckets.unshared[bucketno] = hashTuple;
1663 : :
1664 : : /* advance index past the tuple */
1665 : 401795 : idx += MAXALIGN(HJTUPLE_OVERHEAD +
1666 : : HJTUPLE_MINTUPLE(hashTuple)->t_len);
1667 : : }
1668 : :
1669 : : /* allow this loop to be cancellable */
1670 [ - + ]: 2078 : CHECK_FOR_INTERRUPTS();
1671 : : }
1672 : : }
1673 : :
1674 : : static void
1675 : 48 : ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
1676 : : {
1677 : 48 : ParallelHashJoinState *pstate = hashtable->parallel_state;
1678 : : int i;
1679 : : HashMemoryChunk chunk;
1680 : : dsa_pointer chunk_s;
1681 : :
1682 : : Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASH_INNER);
1683 : :
1684 : : /*
1685 : : * It's unlikely, but we need to be prepared for new participants to show
1686 : : * up while we're in the middle of this operation so we need to switch on
1687 : : * barrier phase here.
1688 : : */
1689 [ + - - - ]: 48 : switch (PHJ_GROW_BUCKETS_PHASE(BarrierPhase(&pstate->grow_buckets_barrier)))
1690 : : {
1691 : 48 : case PHJ_GROW_BUCKETS_ELECT:
1692 : : /* Elect one participant to prepare to increase nbuckets. */
1693 [ + - ]: 48 : if (BarrierArriveAndWait(&pstate->grow_buckets_barrier,
1694 : : WAIT_EVENT_HASH_GROW_BUCKETS_ELECT))
1695 : : {
1696 : : size_t size;
1697 : : dsa_pointer_atomic *buckets;
1698 : :
1699 : : /* Double the size of the bucket array. */
1700 : 48 : pstate->nbuckets *= 2;
1701 : 48 : size = pstate->nbuckets * sizeof(dsa_pointer_atomic);
1702 : 48 : hashtable->batches[0].shared->size += size / 2;
1703 : 48 : dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
1704 : 96 : hashtable->batches[0].shared->buckets =
1705 : 48 : dsa_allocate(hashtable->area, size);
1706 : : buckets = (dsa_pointer_atomic *)
1707 : 48 : dsa_get_address(hashtable->area,
1708 : 48 : hashtable->batches[0].shared->buckets);
1709 [ + + ]: 229424 : for (i = 0; i < pstate->nbuckets; ++i)
1710 : 229376 : dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
1711 : :
1712 : : /* Put the chunk list onto the work queue. */
1713 : 48 : pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;
1714 : :
1715 : : /* Clear the flag. */
1716 : 48 : pstate->growth = PHJ_GROWTH_OK;
1717 : : }
1718 : : pg_fallthrough;
1719 : :
1720 : : case PHJ_GROW_BUCKETS_REALLOCATE:
1721 : : /* Wait for the above to complete. */
1722 : 48 : BarrierArriveAndWait(&pstate->grow_buckets_barrier,
1723 : : WAIT_EVENT_HASH_GROW_BUCKETS_REALLOCATE);
1724 : : pg_fallthrough;
1725 : :
1726 : 48 : case PHJ_GROW_BUCKETS_REINSERT:
1727 : : /* Reinsert all tuples into the hash table. */
1728 : 48 : ExecParallelHashEnsureBatchAccessors(hashtable);
1729 : 48 : ExecParallelHashTableSetCurrentBatch(hashtable, 0);
1730 [ + + ]: 272 : while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_s)))
1731 : : {
1732 : 176 : size_t idx = 0;
1733 : :
1734 [ + + ]: 144144 : while (idx < chunk->used)
1735 : : {
1736 : 143968 : HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1737 : 143968 : dsa_pointer shared = chunk_s + HASH_CHUNK_HEADER_SIZE + idx;
1738 : : int bucketno;
1739 : : int batchno;
1740 : :
1741 : 143968 : ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1742 : : &bucketno, &batchno);
1743 : : Assert(batchno == 0);
1744 : :
1745 : : /* add the tuple to the proper bucket */
1746 : 143968 : ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1747 : : hashTuple, shared);
1748 : :
1749 : : /* advance index past the tuple */
1750 : 143968 : idx += MAXALIGN(HJTUPLE_OVERHEAD +
1751 : : HJTUPLE_MINTUPLE(hashTuple)->t_len);
1752 : : }
1753 : :
1754 : : /* allow this loop to be cancellable */
1755 [ - + ]: 176 : CHECK_FOR_INTERRUPTS();
1756 : : }
1757 : 48 : BarrierArriveAndWait(&pstate->grow_buckets_barrier,
1758 : : WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT);
1759 : : }
1760 : 48 : }
1761 : :
1762 : : /*
1763 : : * ExecHashTableInsert
1764 : : * insert a tuple into the hash table depending on the hash value
1765 : : * it may just go to a temp file for later batches
1766 : : *
1767 : : * Note: the passed TupleTableSlot may contain a regular, minimal, or virtual
1768 : : * tuple; the minimal case in particular is certain to happen while reloading
1769 : : * tuples from batch files. We could save some cycles in the regular-tuple
1770 : : * case by not forcing the slot contents into minimal form; not clear if it's
1771 : : * worth the messiness required.
1772 : : */
1773 : : void
1774 : 8944908 : ExecHashTableInsert(HashJoinTable hashtable,
1775 : : TupleTableSlot *slot,
1776 : : uint32 hashvalue)
1777 : : {
1778 : : bool shouldFree;
1779 : 8944908 : MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1780 : : int bucketno;
1781 : : int batchno;
1782 : :
1783 : 8944908 : ExecHashGetBucketAndBatch(hashtable, hashvalue,
1784 : : &bucketno, &batchno);
1785 : :
1786 : : /*
1787 : : * decide whether to put the tuple in the hash table or a temp file
1788 : : */
1789 [ + + ]: 8944908 : if (batchno == hashtable->curbatch)
1790 : : {
1791 : : /*
1792 : : * put the tuple in hash table
1793 : : */
1794 : : HashJoinTuple hashTuple;
1795 : : int hashTupleSize;
1796 : :
1797 : : /* Create the HashJoinTuple */
1798 : 6888489 : hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
1799 : 6888489 : hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
1800 : :
1801 : 6888489 : hashTuple->hashvalue = hashvalue;
1802 : 6888489 : memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1803 : :
1804 : : /*
1805 : : * We always reset the tuple-matched flag on insertion. This is okay
1806 : : * even when reloading a tuple from a batch file, since the tuple
1807 : : * could not possibly have been matched to an outer tuple before it
1808 : : * went into the batch file.
1809 : : */
1810 : 6888489 : HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
1811 : :
1812 : : /* Push it onto the front of the bucket's list */
1813 : 6888489 : hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1814 : 6888489 : hashtable->buckets.unshared[bucketno] = hashTuple;
1815 : :
1816 : : /*
1817 : : * Increase the (optimal) number of buckets if we just exceeded the
1818 : : * NTUP_PER_BUCKET threshold, but only when there's still a single
1819 : : * batch. Note that totalTuples - skewTuples is a reliable indicator
1820 : : * of the hash table's size only as long as there's just one batch.
1821 : : */
1822 [ + + ]: 6888489 : if (hashtable->nbatch == 1 &&
1823 : 4501381 : (hashtable->totalTuples - hashtable->skewTuples) >
1824 [ + + ]: 4501381 : (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
1825 : : {
1826 : : /* Guard against integer overflow and alloc size overflow */
1827 [ + - ]: 234 : if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
1828 [ + - ]: 234 : hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
1829 : : {
1830 : 234 : hashtable->nbuckets_optimal *= 2;
1831 : 234 : hashtable->log2_nbuckets_optimal += 1;
1832 : : }
1833 : : }
1834 : :
1835 : : /* Account for space used, and back off if we've used too much */
1836 : 6888489 : hashtable->spaceUsed += hashTupleSize;
1837 [ + + ]: 6888489 : if (hashtable->spaceUsed > hashtable->spacePeak)
1838 : 5182118 : hashtable->spacePeak = hashtable->spaceUsed;
1839 : 6888489 : if (hashtable->spaceUsed +
1840 : 6888489 : hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
1841 [ + + ]: 6888489 : > hashtable->spaceAllowed)
1842 : 535495 : ExecHashIncreaseNumBatches(hashtable);
1843 : : }
1844 : : else
1845 : : {
1846 : : /*
1847 : : * put the tuple into a temp file for later batches
1848 : : */
1849 : : Assert(batchno > hashtable->curbatch);
1850 : 2056419 : ExecHashJoinSaveTuple(tuple,
1851 : : hashvalue,
1852 : 2056419 : &hashtable->innerBatchFile[batchno],
1853 : : hashtable);
1854 : : }
1855 : :
1856 [ + + ]: 8944908 : if (shouldFree)
1857 : 6628927 : heap_free_minimal_tuple(tuple);
1858 : 8944908 : }
1859 : :
1860 : : /*
1861 : : * ExecParallelHashTableInsert
1862 : : * insert a tuple into a shared hash table or shared batch tuplestore
1863 : : */
1864 : : void
1865 : 1440128 : ExecParallelHashTableInsert(HashJoinTable hashtable,
1866 : : TupleTableSlot *slot,
1867 : : uint32 hashvalue)
1868 : : {
1869 : : bool shouldFree;
1870 : 1440128 : MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1871 : : dsa_pointer shared;
1872 : : int bucketno;
1873 : : int batchno;
1874 : :
1875 : 167 : retry:
1876 : 1440295 : ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1877 : :
1878 [ + + ]: 1440295 : if (batchno == 0)
1879 : : {
1880 : : HashJoinTuple hashTuple;
1881 : :
1882 : : /* Try to load it into memory. */
1883 : : Assert(BarrierPhase(&hashtable->parallel_state->build_barrier) ==
1884 : : PHJ_BUILD_HASH_INNER);
1885 : 833056 : hashTuple = ExecParallelHashTupleAlloc(hashtable,
1886 : 833056 : HJTUPLE_OVERHEAD + tuple->t_len,
1887 : : &shared);
1888 [ + + ]: 833056 : if (hashTuple == NULL)
1889 : 149 : goto retry;
1890 : :
1891 : : /* Store the hash value in the HashJoinTuple header. */
1892 : 832907 : hashTuple->hashvalue = hashvalue;
1893 : 832907 : memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1894 : 832907 : HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
1895 : :
1896 : : /* Push it onto the front of the bucket's list */
1897 : 832907 : ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1898 : : hashTuple, shared);
1899 : : }
1900 : : else
1901 : : {
1902 : 607239 : size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1903 : :
1904 : : Assert(batchno > 0);
1905 : :
1906 : : /* Try to preallocate space in the batch if necessary. */
1907 [ + + ]: 607239 : if (hashtable->batches[batchno].preallocated < tuple_size)
1908 : : {
1909 [ + + ]: 1153 : if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size))
1910 : 18 : goto retry;
1911 : : }
1912 : :
1913 : : Assert(hashtable->batches[batchno].preallocated >= tuple_size);
1914 : 607221 : hashtable->batches[batchno].preallocated -= tuple_size;
1915 : 607221 : sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue,
1916 : : tuple);
1917 : : }
1918 : 1440128 : ++hashtable->batches[batchno].ntuples;
1919 : :
1920 [ + - ]: 1440128 : if (shouldFree)
1921 : 1440128 : heap_free_minimal_tuple(tuple);
1922 : 1440128 : }
1923 : :
1924 : : /*
1925 : : * Insert a tuple into the current hash table. Unlike
1926 : : * ExecParallelHashTableInsert, this version is not prepared to send the tuple
1927 : : * to other batches or to run out of memory, and should only be called with
1928 : : * tuples that belong in the current batch once growth has been disabled.
1929 : : */
1930 : : void
1931 : 721112 : ExecParallelHashTableInsertCurrentBatch(HashJoinTable hashtable,
1932 : : TupleTableSlot *slot,
1933 : : uint32 hashvalue)
1934 : : {
1935 : : bool shouldFree;
1936 : 721112 : MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1937 : : HashJoinTuple hashTuple;
1938 : : dsa_pointer shared;
1939 : : int batchno;
1940 : : int bucketno;
1941 : :
1942 : 721112 : ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1943 : : Assert(batchno == hashtable->curbatch);
1944 : 721112 : hashTuple = ExecParallelHashTupleAlloc(hashtable,
1945 : 721112 : HJTUPLE_OVERHEAD + tuple->t_len,
1946 : : &shared);
1947 : 721112 : hashTuple->hashvalue = hashvalue;
1948 : 721112 : memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1949 : 721112 : HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
1950 : 721112 : ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1951 : : hashTuple, shared);
1952 : :
1953 [ - + ]: 721112 : if (shouldFree)
1954 : 0 : heap_free_minimal_tuple(tuple);
1955 : 721112 : }
1956 : :
1957 : :
1958 : : /*
1959 : : * ExecHashGetBucketAndBatch
1960 : : * Determine the bucket number and batch number for a hash value
1961 : : *
1962 : : * Note: on-the-fly increases of nbatch must not change the bucket number
1963 : : * for a given hash code (since we don't move tuples to different hash
1964 : : * chains), and must only cause the batch number to remain the same or
1965 : : * increase. Our algorithm is
1966 : : * bucketno = hashvalue MOD nbuckets
1967 : : * batchno = ROR(hashvalue, log2_nbuckets) MOD nbatch
1968 : : * where nbuckets and nbatch are both expected to be powers of 2, so we can
1969 : : * do the computations by shifting and masking. (This assumes that all hash
1970 : : * functions are good about randomizing all their output bits, else we are
1971 : : * likely to have very skewed bucket or batch occupancy.)
1972 : : *
1973 : : * nbuckets and log2_nbuckets may change while nbatch == 1 because of dynamic
1974 : : * bucket count growth. Once we start batching, the value is fixed and does
1975 : : * not change over the course of the join (making it possible to compute batch
1976 : : * number the way we do here).
1977 : : *
1978 : : * nbatch is always a power of 2; we increase it only by doubling it. This
1979 : : * effectively adds one more bit to the top of the batchno. In very large
1980 : : * joins, we might run out of bits to add, so we do this by rotating the hash
1981 : : * value. This causes batchno to steal bits from bucketno when the number of
1982 : : * virtual buckets exceeds 2^32. It's better to have longer bucket chains
1983 : : * than to lose the ability to divide batches.
1984 : : */
1985 : : void
1986 : 27265472 : ExecHashGetBucketAndBatch(HashJoinTable hashtable,
1987 : : uint32 hashvalue,
1988 : : int *bucketno,
1989 : : int *batchno)
1990 : : {
1991 : 27265472 : uint32 nbuckets = (uint32) hashtable->nbuckets;
1992 : 27265472 : uint32 nbatch = (uint32) hashtable->nbatch;
1993 : :
1994 [ + + ]: 27265472 : if (nbatch > 1)
1995 : : {
1996 : 10248276 : *bucketno = hashvalue & (nbuckets - 1);
1997 : 10248276 : *batchno = pg_rotate_right32(hashvalue,
1998 : 10248276 : hashtable->log2_nbuckets) & (nbatch - 1);
1999 : : }
2000 : : else
2001 : : {
2002 : 17017196 : *bucketno = hashvalue & (nbuckets - 1);
2003 : 17017196 : *batchno = 0;
2004 : : }
2005 : 27265472 : }
2006 : :
2007 : : /*
2008 : : * ExecScanHashBucket
2009 : : * scan a hash bucket for matches to the current outer tuple
2010 : : *
2011 : : * The current outer tuple must be stored in econtext->ecxt_outertuple.
2012 : : *
2013 : : * On success, the inner tuple is stored into hjstate->hj_CurTuple and
2014 : : * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot
2015 : : * for the latter.
2016 : : */
2017 : : bool
2018 : 15424465 : ExecScanHashBucket(HashJoinState *hjstate,
2019 : : ExprContext *econtext)
2020 : : {
2021 : 15424465 : ExprState *hjclauses = hjstate->hashclauses;
2022 : 15424465 : HashJoinTable hashtable = hjstate->hj_HashTable;
2023 : 15424465 : HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2024 : 15424465 : uint32 hashvalue = hjstate->hj_CurHashValue;
2025 : :
2026 : : /*
2027 : : * hj_CurTuple is the address of the tuple last returned from the current
2028 : : * bucket, or NULL if it's time to start scanning a new bucket.
2029 : : *
2030 : : * If the tuple hashed to a skew bucket then scan the skew bucket
2031 : : * otherwise scan the standard hashtable bucket.
2032 : : */
2033 [ + + ]: 15424465 : if (hashTuple != NULL)
2034 : 3610748 : hashTuple = hashTuple->next.unshared;
2035 [ + + ]: 11813717 : else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
2036 : 1600 : hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
2037 : : else
2038 : 11812117 : hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
2039 : :
2040 [ + + ]: 18453258 : while (hashTuple != NULL)
2041 : : {
2042 [ + + ]: 10342128 : if (hashTuple->hashvalue == hashvalue)
2043 : : {
2044 : : TupleTableSlot *inntuple;
2045 : :
2046 : : /* insert hashtable's tuple into exec slot so ExecQual sees it */
2047 : 7313347 : inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2048 : : hjstate->hj_HashTupleSlot,
2049 : : false); /* do not pfree */
2050 : 7313347 : econtext->ecxt_innertuple = inntuple;
2051 : :
2052 [ + + ]: 7313347 : if (ExecQualAndReset(hjclauses, econtext))
2053 : : {
2054 : 7313335 : hjstate->hj_CurTuple = hashTuple;
2055 : 7313335 : return true;
2056 : : }
2057 : : }
2058 : :
2059 : 3028793 : hashTuple = hashTuple->next.unshared;
2060 : : }
2061 : :
2062 : : /*
2063 : : * no match
2064 : : */
2065 : 8111130 : return false;
2066 : : }
2067 : :
2068 : : /*
2069 : : * ExecParallelScanHashBucket
2070 : : * scan a hash bucket for matches to the current outer tuple
2071 : : *
2072 : : * The current outer tuple must be stored in econtext->ecxt_outertuple.
2073 : : *
2074 : : * On success, the inner tuple is stored into hjstate->hj_CurTuple and
2075 : : * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot
2076 : : * for the latter.
2077 : : */
2078 : : bool
2079 : 2804080 : ExecParallelScanHashBucket(HashJoinState *hjstate,
2080 : : ExprContext *econtext)
2081 : : {
2082 : 2804080 : ExprState *hjclauses = hjstate->hashclauses;
2083 : 2804080 : HashJoinTable hashtable = hjstate->hj_HashTable;
2084 : 2804080 : HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2085 : 2804080 : uint32 hashvalue = hjstate->hj_CurHashValue;
2086 : :
2087 : : /*
2088 : : * hj_CurTuple is the address of the tuple last returned from the current
2089 : : * bucket, or NULL if it's time to start scanning a new bucket.
2090 : : */
2091 [ + + ]: 2804080 : if (hashTuple != NULL)
2092 : 1360052 : hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2093 : : else
2094 : 1444028 : hashTuple = ExecParallelHashFirstTuple(hashtable,
2095 : : hjstate->hj_CurBucketNo);
2096 : :
2097 [ + + ]: 3665828 : while (hashTuple != NULL)
2098 : : {
2099 [ + + ]: 2221800 : if (hashTuple->hashvalue == hashvalue)
2100 : : {
2101 : : TupleTableSlot *inntuple;
2102 : :
2103 : : /* insert hashtable's tuple into exec slot so ExecQual sees it */
2104 : 1360052 : inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2105 : : hjstate->hj_HashTupleSlot,
2106 : : false); /* do not pfree */
2107 : 1360052 : econtext->ecxt_innertuple = inntuple;
2108 : :
2109 [ + - ]: 1360052 : if (ExecQualAndReset(hjclauses, econtext))
2110 : : {
2111 : 1360052 : hjstate->hj_CurTuple = hashTuple;
2112 : 1360052 : return true;
2113 : : }
2114 : : }
2115 : :
2116 : 861748 : hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2117 : : }
2118 : :
2119 : : /*
2120 : : * no match
2121 : : */
2122 : 1444028 : return false;
2123 : : }
2124 : :
2125 : : /*
2126 : : * ExecPrepHashTableForUnmatched
2127 : : * set up for a series of ExecScanHashTableForUnmatched calls
2128 : : */
2129 : : void
2130 : 4201 : ExecPrepHashTableForUnmatched(HashJoinState *hjstate)
2131 : : {
2132 : : /*----------
2133 : : * During this scan we use the HashJoinState fields as follows:
2134 : : *
2135 : : * hj_CurBucketNo: next regular bucket to scan
2136 : : * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
2137 : : * hj_CurTuple: last tuple returned, or NULL to start next bucket
2138 : : *----------
2139 : : */
2140 : 4201 : hjstate->hj_CurBucketNo = 0;
2141 : 4201 : hjstate->hj_CurSkewBucketNo = 0;
2142 : 4201 : hjstate->hj_CurTuple = NULL;
2143 : 4201 : }
2144 : :
2145 : : /*
2146 : : * Decide if this process is allowed to run the unmatched scan. If so, the
2147 : : * batch barrier is advanced to PHJ_BATCH_SCAN and true is returned.
2148 : : * Otherwise the batch is detached and false is returned.
2149 : : */
2150 : : bool
2151 : 70 : ExecParallelPrepHashTableForUnmatched(HashJoinState *hjstate)
2152 : : {
2153 : 70 : HashJoinTable hashtable = hjstate->hj_HashTable;
2154 : 70 : int curbatch = hashtable->curbatch;
2155 : 70 : ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
2156 : :
2157 : : Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE);
2158 : :
2159 : : /*
2160 : : * It would not be deadlock-free to wait on the batch barrier, because it
2161 : : * is in PHJ_BATCH_PROBE phase, and thus processes attached to it have
2162 : : * already emitted tuples. Therefore, we'll hold a wait-free election:
2163 : : * only one process can continue to the next phase, and all others detach
2164 : : * from this batch. They can still go any work on other batches, if there
2165 : : * are any.
2166 : : */
2167 [ + + ]: 70 : if (!BarrierArriveAndDetachExceptLast(&batch->batch_barrier))
2168 : : {
2169 : : /* This process considers the batch to be done. */
2170 : 26 : hashtable->batches[hashtable->curbatch].done = true;
2171 : :
2172 : : /* Make sure any temporary files are closed. */
2173 : 26 : sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
2174 : 26 : sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
2175 : :
2176 : : /*
2177 : : * Track largest batch we've seen, which would normally happen in
2178 : : * ExecHashTableDetachBatch().
2179 : : */
2180 : 26 : hashtable->spacePeak =
2181 : 26 : Max(hashtable->spacePeak,
2182 : : batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
2183 : 26 : hashtable->curbatch = -1;
2184 : 26 : return false;
2185 : : }
2186 : :
2187 : : /* Now we are alone with this batch. */
2188 : : Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_SCAN);
2189 : :
2190 : : /*
2191 : : * Has another process decided to give up early and command all processes
2192 : : * to skip the unmatched scan?
2193 : : */
2194 [ - + ]: 44 : if (batch->skip_unmatched)
2195 : : {
2196 : 0 : hashtable->batches[hashtable->curbatch].done = true;
2197 : 0 : ExecHashTableDetachBatch(hashtable);
2198 : 0 : return false;
2199 : : }
2200 : :
2201 : : /* Now prepare the process local state, just as for non-parallel join. */
2202 : 44 : ExecPrepHashTableForUnmatched(hjstate);
2203 : :
2204 : 44 : return true;
2205 : : }
2206 : :
2207 : : /*
2208 : : * ExecScanHashTableForUnmatched
2209 : : * scan the hash table for unmatched inner tuples
2210 : : *
2211 : : * On success, the inner tuple is stored into hjstate->hj_CurTuple and
2212 : : * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot
2213 : : * for the latter.
2214 : : */
2215 : : bool
2216 : 272089 : ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
2217 : : {
2218 : 272089 : HashJoinTable hashtable = hjstate->hj_HashTable;
2219 : 272089 : HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2220 : :
2221 : : for (;;)
2222 : : {
2223 : : /*
2224 : : * hj_CurTuple is the address of the tuple last returned from the
2225 : : * current bucket, or NULL if it's time to start scanning a new
2226 : : * bucket.
2227 : : */
2228 [ + + ]: 5285233 : if (hashTuple != NULL)
2229 : 267932 : hashTuple = hashTuple->next.unshared;
2230 [ + + ]: 5017301 : else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
2231 : : {
2232 : 5013148 : hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
2233 : 5013148 : hjstate->hj_CurBucketNo++;
2234 : : }
2235 [ - + ]: 4153 : else if (hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
2236 : : {
2237 : 0 : int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
2238 : :
2239 : 0 : hashTuple = hashtable->skewBucket[j]->tuples;
2240 : 0 : hjstate->hj_CurSkewBucketNo++;
2241 : : }
2242 : : else
2243 : 4153 : break; /* finished all buckets */
2244 : :
2245 [ + + ]: 5556123 : while (hashTuple != NULL)
2246 : : {
2247 [ + + ]: 542979 : if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
2248 : : {
2249 : : TupleTableSlot *inntuple;
2250 : :
2251 : : /* insert hashtable's tuple into exec slot */
2252 : 267936 : inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2253 : : hjstate->hj_HashTupleSlot,
2254 : : false); /* do not pfree */
2255 : 267936 : econtext->ecxt_innertuple = inntuple;
2256 : :
2257 : : /*
2258 : : * Reset temp memory each time; although this function doesn't
2259 : : * do any qual eval, the caller will, so let's keep it
2260 : : * parallel to ExecScanHashBucket.
2261 : : */
2262 : 267936 : ResetExprContext(econtext);
2263 : :
2264 : 267936 : hjstate->hj_CurTuple = hashTuple;
2265 : 267936 : return true;
2266 : : }
2267 : :
2268 : 275043 : hashTuple = hashTuple->next.unshared;
2269 : : }
2270 : :
2271 : : /* allow this loop to be cancellable */
2272 [ - + ]: 5013144 : CHECK_FOR_INTERRUPTS();
2273 : : }
2274 : :
2275 : : /*
2276 : : * no more unmatched tuples
2277 : : */
2278 : 4153 : return false;
2279 : : }
2280 : :
2281 : : /*
2282 : : * ExecParallelScanHashTableForUnmatched
2283 : : * scan the hash table for unmatched inner tuples, in parallel join
2284 : : *
2285 : : * On success, the inner tuple is stored into hjstate->hj_CurTuple and
2286 : : * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot
2287 : : * for the latter.
2288 : : */
2289 : : bool
2290 : 80048 : ExecParallelScanHashTableForUnmatched(HashJoinState *hjstate,
2291 : : ExprContext *econtext)
2292 : : {
2293 : 80048 : HashJoinTable hashtable = hjstate->hj_HashTable;
2294 : 80048 : HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2295 : :
2296 : : for (;;)
2297 : : {
2298 : : /*
2299 : : * hj_CurTuple is the address of the tuple last returned from the
2300 : : * current bucket, or NULL if it's time to start scanning a new
2301 : : * bucket.
2302 : : */
2303 [ + + ]: 489648 : if (hashTuple != NULL)
2304 : 80004 : hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2305 [ + + ]: 409644 : else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
2306 : 409600 : hashTuple = ExecParallelHashFirstTuple(hashtable,
2307 : 409600 : hjstate->hj_CurBucketNo++);
2308 : : else
2309 : 44 : break; /* finished all buckets */
2310 : :
2311 [ + + ]: 649604 : while (hashTuple != NULL)
2312 : : {
2313 [ + + ]: 240004 : if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
2314 : : {
2315 : : TupleTableSlot *inntuple;
2316 : :
2317 : : /* insert hashtable's tuple into exec slot */
2318 : 80004 : inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2319 : : hjstate->hj_HashTupleSlot,
2320 : : false); /* do not pfree */
2321 : 80004 : econtext->ecxt_innertuple = inntuple;
2322 : :
2323 : : /*
2324 : : * Reset temp memory each time; although this function doesn't
2325 : : * do any qual eval, the caller will, so let's keep it
2326 : : * parallel to ExecScanHashBucket.
2327 : : */
2328 : 80004 : ResetExprContext(econtext);
2329 : :
2330 : 80004 : hjstate->hj_CurTuple = hashTuple;
2331 : 80004 : return true;
2332 : : }
2333 : :
2334 : 160000 : hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2335 : : }
2336 : :
2337 : : /* allow this loop to be cancellable */
2338 [ + + ]: 409600 : CHECK_FOR_INTERRUPTS();
2339 : : }
2340 : :
2341 : : /*
2342 : : * no more unmatched tuples
2343 : : */
2344 : 44 : return false;
2345 : : }
2346 : :
2347 : : /*
2348 : : * ExecHashTableReset
2349 : : *
2350 : : * reset hash table header for new batch
2351 : : */
2352 : : void
2353 : 799 : ExecHashTableReset(HashJoinTable hashtable)
2354 : : {
2355 : : MemoryContext oldcxt;
2356 : 799 : int nbuckets = hashtable->nbuckets;
2357 : :
2358 : : /*
2359 : : * Release all the hash buckets and tuples acquired in the prior pass, and
2360 : : * reinitialize the context for a new pass.
2361 : : */
2362 : 799 : MemoryContextReset(hashtable->batchCxt);
2363 : 799 : oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
2364 : :
2365 : : /* Reallocate and reinitialize the hash bucket headers. */
2366 : 799 : hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets);
2367 : :
2368 : 799 : hashtable->spaceUsed = 0;
2369 : :
2370 : 799 : MemoryContextSwitchTo(oldcxt);
2371 : :
2372 : : /* Forget the chunks (the memory was freed by the context reset above). */
2373 : 799 : hashtable->chunks = NULL;
2374 : 799 : }
2375 : :
2376 : : /*
2377 : : * ExecHashTableResetMatchFlags
2378 : : * Clear all the HeapTupleHeaderHasMatch flags in the table
2379 : : */
2380 : : void
2381 : 44 : ExecHashTableResetMatchFlags(HashJoinTable hashtable)
2382 : : {
2383 : : HashJoinTuple tuple;
2384 : : int i;
2385 : :
2386 : : /* Reset all flags in the main table ... */
2387 [ + + ]: 45100 : for (i = 0; i < hashtable->nbuckets; i++)
2388 : : {
2389 [ + + ]: 45236 : for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
2390 : 180 : tuple = tuple->next.unshared)
2391 : 180 : HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
2392 : : }
2393 : :
2394 : : /* ... and the same for the skew buckets, if any */
2395 [ - + ]: 44 : for (i = 0; i < hashtable->nSkewBuckets; i++)
2396 : : {
2397 : 0 : int j = hashtable->skewBucketNums[i];
2398 : 0 : HashSkewBucket *skewBucket = hashtable->skewBucket[j];
2399 : :
2400 [ # # ]: 0 : for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
2401 : 0 : HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
2402 : : }
2403 : 44 : }
2404 : :
2405 : :
2406 : : void
2407 : 1091 : ExecReScanHash(HashState *node)
2408 : : {
2409 : 1091 : PlanState *outerPlan = outerPlanState(node);
2410 : :
2411 : : /*
2412 : : * if chgParam of subnode is not null then plan will be re-scanned by
2413 : : * first ExecProcNode.
2414 : : */
2415 [ + + ]: 1091 : if (outerPlan->chgParam == NULL)
2416 : 20 : ExecReScan(outerPlan);
2417 : 1091 : }
2418 : :
2419 : :
2420 : : /*
2421 : : * ExecHashBuildSkewHash
2422 : : *
2423 : : * Set up for skew optimization if we can identify the most common values
2424 : : * (MCVs) of the outer relation's join key. We make a skew hash bucket
2425 : : * for the hash value of each MCV, up to the number of slots allowed
2426 : : * based on available memory.
2427 : : */
2428 : : static void
2429 : 82 : ExecHashBuildSkewHash(HashState *hashstate, HashJoinTable hashtable,
2430 : : Hash *node, int mcvsToUse)
2431 : : {
2432 : : HeapTupleData *statsTuple;
2433 : : AttStatsSlot sslot;
2434 : :
2435 : : /* Do nothing if planner didn't identify the outer relation's join key */
2436 [ - + ]: 82 : if (!OidIsValid(node->skewTable))
2437 : 0 : return;
2438 : : /* Also, do nothing if we don't have room for at least one skew bucket */
2439 [ - + ]: 82 : if (mcvsToUse <= 0)
2440 : 0 : return;
2441 : :
2442 : : /*
2443 : : * Try to find the MCV statistics for the outer relation's join key.
2444 : : */
2445 : 82 : statsTuple = SearchSysCache3(STATRELATTINH,
2446 : : ObjectIdGetDatum(node->skewTable),
2447 : 82 : Int16GetDatum(node->skewColumn),
2448 : 82 : BoolGetDatum(node->skewInherit));
2449 [ - + ]: 82 : if (!HeapTupleIsValid(statsTuple))
2450 : 0 : return;
2451 : :
2452 [ + + ]: 82 : if (get_attstatsslot(&sslot, statsTuple,
2453 : : STATISTIC_KIND_MCV, InvalidOid,
2454 : : ATTSTATSSLOT_VALUES | ATTSTATSSLOT_NUMBERS))
2455 : : {
2456 : : double frac;
2457 : : int nbuckets;
2458 : : int i;
2459 : :
2460 [ - + ]: 4 : if (mcvsToUse > sslot.nvalues)
2461 : 0 : mcvsToUse = sslot.nvalues;
2462 : :
2463 : : /*
2464 : : * Calculate the expected fraction of outer relation that will
2465 : : * participate in the skew optimization. If this isn't at least
2466 : : * SKEW_MIN_OUTER_FRACTION, don't use skew optimization.
2467 : : */
2468 : 4 : frac = 0;
2469 [ + + ]: 88 : for (i = 0; i < mcvsToUse; i++)
2470 : 84 : frac += sslot.numbers[i];
2471 [ - + ]: 4 : if (frac < SKEW_MIN_OUTER_FRACTION)
2472 : : {
2473 : 0 : free_attstatsslot(&sslot);
2474 : 0 : ReleaseSysCache(statsTuple);
2475 : 0 : return;
2476 : : }
2477 : :
2478 : : /*
2479 : : * Okay, set up the skew hashtable.
2480 : : *
2481 : : * skewBucket[] is an open addressing hashtable with a power of 2 size
2482 : : * that is greater than the number of MCV values. (This ensures there
2483 : : * will be at least one null entry, so searches will always
2484 : : * terminate.)
2485 : : *
2486 : : * Note: this code could fail if mcvsToUse exceeds INT_MAX/8 or
2487 : : * MaxAllocSize/sizeof(void *)/8, but that is not currently possible
2488 : : * since we limit pg_statistic entries to much less than that.
2489 : : */
2490 : 4 : nbuckets = pg_nextpower2_32(mcvsToUse + 1);
2491 : : /* use two more bits just to help avoid collisions */
2492 : 4 : nbuckets <<= 2;
2493 : :
2494 : 4 : hashtable->skewEnabled = true;
2495 : 4 : hashtable->skewBucketLen = nbuckets;
2496 : :
2497 : : /*
2498 : : * We allocate the bucket memory in the hashtable's batch context. It
2499 : : * is only needed during the first batch, and this ensures it will be
2500 : : * automatically removed once the first batch is done.
2501 : : */
2502 : 4 : hashtable->skewBucket = (HashSkewBucket **)
2503 : 4 : MemoryContextAllocZero(hashtable->batchCxt,
2504 : : nbuckets * sizeof(HashSkewBucket *));
2505 : 4 : hashtable->skewBucketNums = (int *)
2506 : 4 : MemoryContextAllocZero(hashtable->batchCxt,
2507 : : mcvsToUse * sizeof(int));
2508 : :
2509 : 4 : hashtable->spaceUsed += nbuckets * sizeof(HashSkewBucket *)
2510 : 4 : + mcvsToUse * sizeof(int);
2511 : 4 : hashtable->spaceUsedSkew += nbuckets * sizeof(HashSkewBucket *)
2512 : 4 : + mcvsToUse * sizeof(int);
2513 [ + - ]: 4 : if (hashtable->spaceUsed > hashtable->spacePeak)
2514 : 4 : hashtable->spacePeak = hashtable->spaceUsed;
2515 : :
2516 : : /*
2517 : : * Create a skew bucket for each MCV hash value.
2518 : : *
2519 : : * Note: it is very important that we create the buckets in order of
2520 : : * decreasing MCV frequency. If we have to remove some buckets, they
2521 : : * must be removed in reverse order of creation (see notes in
2522 : : * ExecHashRemoveNextSkewBucket) and we want the least common MCVs to
2523 : : * be removed first.
2524 : : */
2525 : :
2526 [ + + ]: 88 : for (i = 0; i < mcvsToUse; i++)
2527 : : {
2528 : : uint32 hashvalue;
2529 : : int bucket;
2530 : :
2531 : 84 : hashvalue = DatumGetUInt32(FunctionCall1Coll(hashstate->skew_hashfunction,
2532 : : hashstate->skew_collation,
2533 : 84 : sslot.values[i]));
2534 : :
2535 : : /*
2536 : : * While we have not hit a hole in the hashtable and have not hit
2537 : : * the desired bucket, we have collided with some previous hash
2538 : : * value, so try the next bucket location. NB: this code must
2539 : : * match ExecHashGetSkewBucket.
2540 : : */
2541 : 84 : bucket = hashvalue & (nbuckets - 1);
2542 [ - + ]: 84 : while (hashtable->skewBucket[bucket] != NULL &&
2543 [ # # ]: 0 : hashtable->skewBucket[bucket]->hashvalue != hashvalue)
2544 : 0 : bucket = (bucket + 1) & (nbuckets - 1);
2545 : :
2546 : : /*
2547 : : * If we found an existing bucket with the same hashvalue, leave
2548 : : * it alone. It's okay for two MCVs to share a hashvalue.
2549 : : */
2550 [ - + ]: 84 : if (hashtable->skewBucket[bucket] != NULL)
2551 : 0 : continue;
2552 : :
2553 : : /* Okay, create a new skew bucket for this hashvalue. */
2554 : 168 : hashtable->skewBucket[bucket] = (HashSkewBucket *)
2555 : 84 : MemoryContextAlloc(hashtable->batchCxt,
2556 : : sizeof(HashSkewBucket));
2557 : 84 : hashtable->skewBucket[bucket]->hashvalue = hashvalue;
2558 : 84 : hashtable->skewBucket[bucket]->tuples = NULL;
2559 : 84 : hashtable->skewBucketNums[hashtable->nSkewBuckets] = bucket;
2560 : 84 : hashtable->nSkewBuckets++;
2561 : 84 : hashtable->spaceUsed += SKEW_BUCKET_OVERHEAD;
2562 : 84 : hashtable->spaceUsedSkew += SKEW_BUCKET_OVERHEAD;
2563 [ + - ]: 84 : if (hashtable->spaceUsed > hashtable->spacePeak)
2564 : 84 : hashtable->spacePeak = hashtable->spaceUsed;
2565 : : }
2566 : :
2567 : 4 : free_attstatsslot(&sslot);
2568 : : }
2569 : :
2570 : 82 : ReleaseSysCache(statsTuple);
2571 : : }
2572 : :
2573 : : /*
2574 : : * ExecHashGetSkewBucket
2575 : : *
2576 : : * Returns the index of the skew bucket for this hashvalue,
2577 : : * or INVALID_SKEW_BUCKET_NO if the hashvalue is not
2578 : : * associated with any active skew bucket.
2579 : : */
2580 : : int
2581 : 20891391 : ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue)
2582 : : {
2583 : : int bucket;
2584 : :
2585 : : /*
2586 : : * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
2587 : : * particular, this happens after the initial batch is done).
2588 : : */
2589 [ + + ]: 20891391 : if (!hashtable->skewEnabled)
2590 : 20811391 : return INVALID_SKEW_BUCKET_NO;
2591 : :
2592 : : /*
2593 : : * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
2594 : : */
2595 : 80000 : bucket = hashvalue & (hashtable->skewBucketLen - 1);
2596 : :
2597 : : /*
2598 : : * While we have not hit a hole in the hashtable and have not hit the
2599 : : * desired bucket, we have collided with some other hash value, so try the
2600 : : * next bucket location.
2601 : : */
2602 [ + + ]: 85220 : while (hashtable->skewBucket[bucket] != NULL &&
2603 [ + + ]: 7212 : hashtable->skewBucket[bucket]->hashvalue != hashvalue)
2604 : 5220 : bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
2605 : :
2606 : : /*
2607 : : * Found the desired bucket?
2608 : : */
2609 [ + + ]: 80000 : if (hashtable->skewBucket[bucket] != NULL)
2610 : 1992 : return bucket;
2611 : :
2612 : : /*
2613 : : * There must not be any hashtable entry for this hash value.
2614 : : */
2615 : 78008 : return INVALID_SKEW_BUCKET_NO;
2616 : : }
2617 : :
2618 : : /*
2619 : : * ExecHashSkewTableInsert
2620 : : *
2621 : : * Insert a tuple into the skew hashtable.
2622 : : *
2623 : : * This should generally match up with the current-batch case in
2624 : : * ExecHashTableInsert.
2625 : : */
2626 : : static void
2627 : 392 : ExecHashSkewTableInsert(HashJoinTable hashtable,
2628 : : TupleTableSlot *slot,
2629 : : uint32 hashvalue,
2630 : : int bucketNumber)
2631 : : {
2632 : : bool shouldFree;
2633 : 392 : MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
2634 : : HashJoinTuple hashTuple;
2635 : : int hashTupleSize;
2636 : :
2637 : : /* Create the HashJoinTuple */
2638 : 392 : hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
2639 : 392 : hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
2640 : : hashTupleSize);
2641 : 392 : hashTuple->hashvalue = hashvalue;
2642 : 392 : memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
2643 : 392 : HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
2644 : :
2645 : : /* Push it onto the front of the skew bucket's list */
2646 : 392 : hashTuple->next.unshared = hashtable->skewBucket[bucketNumber]->tuples;
2647 : 392 : hashtable->skewBucket[bucketNumber]->tuples = hashTuple;
2648 : : Assert(hashTuple != hashTuple->next.unshared);
2649 : :
2650 : : /* Account for space used, and back off if we've used too much */
2651 : 392 : hashtable->skewTuples += 1;
2652 : 392 : hashtable->spaceUsed += hashTupleSize;
2653 : 392 : hashtable->spaceUsedSkew += hashTupleSize;
2654 [ + + ]: 392 : if (hashtable->spaceUsed > hashtable->spacePeak)
2655 : 288 : hashtable->spacePeak = hashtable->spaceUsed;
2656 [ + + ]: 460 : while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
2657 : 68 : ExecHashRemoveNextSkewBucket(hashtable);
2658 : :
2659 : : /* Check we are not over the total spaceAllowed, either */
2660 [ - + ]: 392 : if (hashtable->spaceUsed > hashtable->spaceAllowed)
2661 : 0 : ExecHashIncreaseNumBatches(hashtable);
2662 : :
2663 [ + - ]: 392 : if (shouldFree)
2664 : 392 : heap_free_minimal_tuple(tuple);
2665 : 392 : }
2666 : :
2667 : : /*
2668 : : * ExecHashRemoveNextSkewBucket
2669 : : *
2670 : : * Remove the least valuable skew bucket by pushing its tuples into
2671 : : * the main hash table.
2672 : : */
2673 : : static void
2674 : 68 : ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
2675 : : {
2676 : : int bucketToRemove;
2677 : : HashSkewBucket *bucket;
2678 : : uint32 hashvalue;
2679 : : int bucketno;
2680 : : int batchno;
2681 : : HashJoinTuple hashTuple;
2682 : :
2683 : : /* Locate the bucket to remove */
2684 : 68 : bucketToRemove = hashtable->skewBucketNums[hashtable->nSkewBuckets - 1];
2685 : 68 : bucket = hashtable->skewBucket[bucketToRemove];
2686 : :
2687 : : /*
2688 : : * Calculate which bucket and batch the tuples belong to in the main
2689 : : * hashtable. They all have the same hash value, so it's the same for all
2690 : : * of them. Also note that it's not possible for nbatch to increase while
2691 : : * we are processing the tuples.
2692 : : */
2693 : 68 : hashvalue = bucket->hashvalue;
2694 : 68 : ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
2695 : :
2696 : : /* Process all tuples in the bucket */
2697 : 68 : hashTuple = bucket->tuples;
2698 [ + + ]: 300 : while (hashTuple != NULL)
2699 : : {
2700 : 232 : HashJoinTuple nextHashTuple = hashTuple->next.unshared;
2701 : : MinimalTuple tuple;
2702 : : Size tupleSize;
2703 : :
2704 : : /*
2705 : : * This code must agree with ExecHashTableInsert. We do not use
2706 : : * ExecHashTableInsert directly as ExecHashTableInsert expects a
2707 : : * TupleTableSlot while we already have HashJoinTuples.
2708 : : */
2709 : 232 : tuple = HJTUPLE_MINTUPLE(hashTuple);
2710 : 232 : tupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
2711 : :
2712 : : /* Decide whether to put the tuple in the hash table or a temp file */
2713 [ + + ]: 232 : if (batchno == hashtable->curbatch)
2714 : : {
2715 : : /* Move the tuple to the main hash table */
2716 : : HashJoinTuple copyTuple;
2717 : :
2718 : : /*
2719 : : * We must copy the tuple into the dense storage, else it will not
2720 : : * be found by, eg, ExecHashIncreaseNumBatches.
2721 : : */
2722 : 92 : copyTuple = (HashJoinTuple) dense_alloc(hashtable, tupleSize);
2723 : 92 : memcpy(copyTuple, hashTuple, tupleSize);
2724 : 92 : pfree(hashTuple);
2725 : :
2726 : 92 : copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
2727 : 92 : hashtable->buckets.unshared[bucketno] = copyTuple;
2728 : :
2729 : : /* We have reduced skew space, but overall space doesn't change */
2730 : 92 : hashtable->spaceUsedSkew -= tupleSize;
2731 : : }
2732 : : else
2733 : : {
2734 : : /* Put the tuple into a temp file for later batches */
2735 : : Assert(batchno > hashtable->curbatch);
2736 : 140 : ExecHashJoinSaveTuple(tuple, hashvalue,
2737 : 140 : &hashtable->innerBatchFile[batchno],
2738 : : hashtable);
2739 : 140 : pfree(hashTuple);
2740 : 140 : hashtable->spaceUsed -= tupleSize;
2741 : 140 : hashtable->spaceUsedSkew -= tupleSize;
2742 : : }
2743 : :
2744 : : /*
2745 : : * We must reduce skewTuples, but totalTuples doesn't change since it
2746 : : * counts both main and skew tuples.
2747 : : */
2748 : 232 : hashtable->skewTuples -= 1;
2749 : :
2750 : 232 : hashTuple = nextHashTuple;
2751 : :
2752 : : /* allow this loop to be cancellable */
2753 [ - + ]: 232 : CHECK_FOR_INTERRUPTS();
2754 : : }
2755 : :
2756 : : /*
2757 : : * Free the bucket struct itself and reset the hashtable entry to NULL.
2758 : : *
2759 : : * NOTE: this is not nearly as simple as it looks on the surface, because
2760 : : * of the possibility of collisions in the hashtable. Suppose that hash
2761 : : * values A and B collide at a particular hashtable entry, and that A was
2762 : : * entered first so B gets shifted to a different table entry. If we were
2763 : : * to remove A first then ExecHashGetSkewBucket would mistakenly start
2764 : : * reporting that B is not in the hashtable, because it would hit the NULL
2765 : : * before finding B. However, we always remove entries in the reverse
2766 : : * order of creation, so this failure cannot happen.
2767 : : */
2768 : 68 : hashtable->skewBucket[bucketToRemove] = NULL;
2769 : 68 : hashtable->nSkewBuckets--;
2770 : 68 : pfree(bucket);
2771 : 68 : hashtable->spaceUsed -= SKEW_BUCKET_OVERHEAD;
2772 : 68 : hashtable->spaceUsedSkew -= SKEW_BUCKET_OVERHEAD;
2773 : :
2774 : : /*
2775 : : * If we have removed all skew buckets then give up on skew optimization.
2776 : : * Release the arrays since they aren't useful any more.
2777 : : */
2778 [ - + ]: 68 : if (hashtable->nSkewBuckets == 0)
2779 : : {
2780 : 0 : hashtable->skewEnabled = false;
2781 : 0 : pfree(hashtable->skewBucket);
2782 : 0 : pfree(hashtable->skewBucketNums);
2783 : 0 : hashtable->skewBucket = NULL;
2784 : 0 : hashtable->skewBucketNums = NULL;
2785 : 0 : hashtable->spaceUsed -= hashtable->spaceUsedSkew;
2786 : 0 : hashtable->spaceUsedSkew = 0;
2787 : : }
2788 : 68 : }
2789 : :
2790 : : /*
2791 : : * Build a tuplestore suitable for holding null-keyed input tuples.
2792 : : * (This function doesn't care whether it's for outer or inner tuples.)
2793 : : *
2794 : : * Note that in a parallel hash join, each worker has its own tuplestore(s)
2795 : : * for these. There's no need to interact with other workers to decide
2796 : : * what to do with them. So they're always in private storage.
2797 : : */
2798 : : Tuplestorestate *
2799 : 155 : ExecHashBuildNullTupleStore(HashJoinTable hashtable)
2800 : : {
2801 : : Tuplestorestate *tstore;
2802 : : MemoryContext oldcxt;
2803 : :
2804 : : /*
2805 : : * We keep the tuplestore in the hashCxt to ensure it won't go away too
2806 : : * soon. Size it at work_mem/16 so that it doesn't bloat the node's space
2807 : : * consumption too much.
2808 : : */
2809 : 155 : oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
2810 : 155 : tstore = tuplestore_begin_heap(false, false, work_mem / 16);
2811 : 155 : MemoryContextSwitchTo(oldcxt);
2812 : 155 : return tstore;
2813 : : }
2814 : :
2815 : : /*
2816 : : * Reserve space in the DSM segment for instrumentation data.
2817 : : */
2818 : : void
2819 : 208 : ExecHashEstimate(HashState *node, ParallelContext *pcxt)
2820 : : {
2821 : : size_t size;
2822 : :
2823 : : /* don't need this if not instrumenting or no workers */
2824 [ + + - + ]: 208 : if (!node->ps.instrument || pcxt->nworkers == 0)
2825 : 152 : return;
2826 : :
2827 : 56 : size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
2828 : 56 : size = add_size(size, offsetof(SharedHashInfo, hinstrument));
2829 : 56 : shm_toc_estimate_chunk(&pcxt->estimator, size);
2830 : 56 : shm_toc_estimate_keys(&pcxt->estimator, 1);
2831 : : }
2832 : :
2833 : : /*
2834 : : * Set up a space in the DSM for all workers to record instrumentation data
2835 : : * about their hash table.
2836 : : */
2837 : : void
2838 : 208 : ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt)
2839 : : {
2840 : : size_t size;
2841 : :
2842 : : /* don't need this if not instrumenting or no workers */
2843 [ + + - + ]: 208 : if (!node->ps.instrument || pcxt->nworkers == 0)
2844 : 152 : return;
2845 : :
2846 : 56 : size = offsetof(SharedHashInfo, hinstrument) +
2847 : 56 : pcxt->nworkers * sizeof(HashInstrumentation);
2848 : 56 : node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
2849 : :
2850 : : /* Each per-worker area must start out as zeroes. */
2851 : 56 : memset(node->shared_info, 0, size);
2852 : :
2853 : 56 : node->shared_info->num_workers = pcxt->nworkers;
2854 : 56 : shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id,
2855 : 56 : node->shared_info);
2856 : : }
2857 : :
2858 : : /*
2859 : : * Locate the DSM space for hash table instrumentation data that we'll write
2860 : : * to at shutdown time.
2861 : : */
2862 : : void
2863 : 524 : ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt)
2864 : : {
2865 : : SharedHashInfo *shared_info;
2866 : :
2867 : : /* don't need this if not instrumenting */
2868 [ + + ]: 524 : if (!node->ps.instrument)
2869 : 356 : return;
2870 : :
2871 : : /*
2872 : : * Find our entry in the shared area, and set up a pointer to it so that
2873 : : * we'll accumulate stats there when shutting down or rebuilding the hash
2874 : : * table.
2875 : : */
2876 : : shared_info = (SharedHashInfo *)
2877 : 168 : shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
2878 : 168 : node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
2879 : : }
2880 : :
2881 : : /*
2882 : : * Collect EXPLAIN stats if needed, saving them into DSM memory if
2883 : : * ExecHashInitializeWorker was called, or local storage if not. In the
2884 : : * parallel case, this must be done in ExecShutdownHash() rather than
2885 : : * ExecEndHash() because the latter runs after we've detached from the DSM
2886 : : * segment.
2887 : : */
2888 : : void
2889 : 26697 : ExecShutdownHash(HashState *node)
2890 : : {
2891 : : /* Allocate save space if EXPLAIN'ing and we didn't do so already */
2892 [ + + + + ]: 26697 : if (node->ps.instrument && !node->hinstrument)
2893 : 79 : node->hinstrument = palloc0_object(HashInstrumentation);
2894 : : /* Now accumulate data for the current (final) hash table */
2895 [ + + + + ]: 26697 : if (node->hinstrument && node->hashtable)
2896 : 225 : ExecHashAccumInstrumentation(node->hinstrument, node->hashtable);
2897 : 26697 : }
2898 : :
2899 : : /*
2900 : : * Retrieve instrumentation data from workers before the DSM segment is
2901 : : * detached, so that EXPLAIN can access it.
2902 : : */
2903 : : void
2904 : 56 : ExecHashRetrieveInstrumentation(HashState *node)
2905 : : {
2906 : 56 : SharedHashInfo *shared_info = node->shared_info;
2907 : : size_t size;
2908 : :
2909 [ - + ]: 56 : if (shared_info == NULL)
2910 : 0 : return;
2911 : :
2912 : : /* Replace node->shared_info with a copy in backend-local memory. */
2913 : 56 : size = offsetof(SharedHashInfo, hinstrument) +
2914 : 56 : shared_info->num_workers * sizeof(HashInstrumentation);
2915 : 56 : node->shared_info = palloc(size);
2916 : 56 : memcpy(node->shared_info, shared_info, size);
2917 : : }
2918 : :
2919 : : /*
2920 : : * Accumulate instrumentation data from 'hashtable' into an
2921 : : * initially-zeroed HashInstrumentation struct.
2922 : : *
2923 : : * This is used to merge information across successive hash table instances
2924 : : * within a single plan node. We take the maximum values of each interesting
2925 : : * number. The largest nbuckets and largest nbatch values might have occurred
2926 : : * in different instances, so there's some risk of confusion from reporting
2927 : : * unrelated numbers; but there's a bigger risk of misdiagnosing a performance
2928 : : * issue if we don't report the largest values. Similarly, we want to report
2929 : : * the largest spacePeak regardless of whether it happened in the same
2930 : : * instance as the largest nbuckets or nbatch. All the instances should have
2931 : : * the same nbuckets_original and nbatch_original; but there's little value
2932 : : * in depending on that here, so handle them the same way.
2933 : : */
2934 : : void
2935 : 225 : ExecHashAccumInstrumentation(HashInstrumentation *instrument,
2936 : : HashJoinTable hashtable)
2937 : : {
2938 : 225 : instrument->nbuckets = Max(instrument->nbuckets,
2939 : : hashtable->nbuckets);
2940 : 225 : instrument->nbuckets_original = Max(instrument->nbuckets_original,
2941 : : hashtable->nbuckets_original);
2942 : 225 : instrument->nbatch = Max(instrument->nbatch,
2943 : : hashtable->nbatch);
2944 : 225 : instrument->nbatch_original = Max(instrument->nbatch_original,
2945 : : hashtable->nbatch_original);
2946 : 225 : instrument->space_peak = Max(instrument->space_peak,
2947 : : hashtable->spacePeak);
2948 : 225 : }
2949 : :
2950 : : /*
2951 : : * Allocate 'size' bytes from the currently active HashMemoryChunk
2952 : : */
2953 : : static void *
2954 : 7015717 : dense_alloc(HashJoinTable hashtable, Size size)
2955 : : {
2956 : : HashMemoryChunk newChunk;
2957 : : char *ptr;
2958 : :
2959 : : /* just in case the size is not already aligned properly */
2960 : 7015717 : size = MAXALIGN(size);
2961 : :
2962 : : /*
2963 : : * If tuple size is larger than threshold, allocate a separate chunk.
2964 : : */
2965 [ - + ]: 7015717 : if (size > HASH_CHUNK_THRESHOLD)
2966 : : {
2967 : : /* allocate new chunk and put it at the beginning of the list */
2968 : 0 : newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
2969 : : HASH_CHUNK_HEADER_SIZE + size);
2970 : 0 : newChunk->maxlen = size;
2971 : 0 : newChunk->used = size;
2972 : 0 : newChunk->ntuples = 1;
2973 : :
2974 : : /*
2975 : : * Add this chunk to the list after the first existing chunk, so that
2976 : : * we don't lose the remaining space in the "current" chunk.
2977 : : */
2978 [ # # ]: 0 : if (hashtable->chunks != NULL)
2979 : : {
2980 : 0 : newChunk->next = hashtable->chunks->next;
2981 : 0 : hashtable->chunks->next.unshared = newChunk;
2982 : : }
2983 : : else
2984 : : {
2985 : 0 : newChunk->next.unshared = hashtable->chunks;
2986 : 0 : hashtable->chunks = newChunk;
2987 : : }
2988 : :
2989 : 0 : return HASH_CHUNK_DATA(newChunk);
2990 : : }
2991 : :
2992 : : /*
2993 : : * See if we have enough space for it in the current chunk (if any). If
2994 : : * not, allocate a fresh chunk.
2995 : : */
2996 [ + + ]: 7015717 : if ((hashtable->chunks == NULL) ||
2997 [ + + ]: 6997854 : (hashtable->chunks->maxlen - hashtable->chunks->used) < size)
2998 : : {
2999 : : /* allocate new chunk and put it at the beginning of the list */
3000 : 28698 : newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
3001 : : HASH_CHUNK_HEADER_SIZE + HASH_CHUNK_SIZE);
3002 : :
3003 : 28698 : newChunk->maxlen = HASH_CHUNK_SIZE;
3004 : 28698 : newChunk->used = size;
3005 : 28698 : newChunk->ntuples = 1;
3006 : :
3007 : 28698 : newChunk->next.unshared = hashtable->chunks;
3008 : 28698 : hashtable->chunks = newChunk;
3009 : :
3010 : 28698 : return HASH_CHUNK_DATA(newChunk);
3011 : : }
3012 : :
3013 : : /* There is enough space in the current chunk, let's add the tuple */
3014 : 6987019 : ptr = HASH_CHUNK_DATA(hashtable->chunks) + hashtable->chunks->used;
3015 : 6987019 : hashtable->chunks->used += size;
3016 : 6987019 : hashtable->chunks->ntuples += 1;
3017 : :
3018 : : /* return pointer to the start of the tuple memory */
3019 : 6987019 : return ptr;
3020 : : }
3021 : :
3022 : : /*
3023 : : * Allocate space for a tuple in shared dense storage. This is equivalent to
3024 : : * dense_alloc but for Parallel Hash using shared memory.
3025 : : *
3026 : : * While loading a tuple into shared memory, we might run out of memory and
3027 : : * decide to repartition, or determine that the load factor is too high and
3028 : : * decide to expand the bucket array, or discover that another participant has
3029 : : * commanded us to help do that. Return NULL if number of buckets or batches
3030 : : * has changed, indicating that the caller must retry (considering the
3031 : : * possibility that the tuple no longer belongs in the same batch).
3032 : : */
3033 : : static HashJoinTuple
3034 : 1588243 : ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size,
3035 : : dsa_pointer *shared)
3036 : : {
3037 : 1588243 : ParallelHashJoinState *pstate = hashtable->parallel_state;
3038 : : dsa_pointer chunk_shared;
3039 : : HashMemoryChunk chunk;
3040 : : Size chunk_size;
3041 : : HashJoinTuple result;
3042 : 1588243 : int curbatch = hashtable->curbatch;
3043 : :
3044 : 1588243 : size = MAXALIGN(size);
3045 : :
3046 : : /*
3047 : : * Fast path: if there is enough space in this backend's current chunk,
3048 : : * then we can allocate without any locking.
3049 : : */
3050 : 1588243 : chunk = hashtable->current_chunk;
3051 [ + + + - ]: 1588243 : if (chunk != NULL &&
3052 : 1587594 : size <= HASH_CHUNK_THRESHOLD &&
3053 [ + + ]: 1587594 : chunk->maxlen - chunk->used >= size)
3054 : : {
3055 : :
3056 : 1585795 : chunk_shared = hashtable->current_chunk_shared;
3057 : : Assert(chunk == dsa_get_address(hashtable->area, chunk_shared));
3058 : 1585795 : *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE + chunk->used;
3059 : 1585795 : result = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + chunk->used);
3060 : 1585795 : chunk->used += size;
3061 : :
3062 : : Assert(chunk->used <= chunk->maxlen);
3063 : : Assert(result == dsa_get_address(hashtable->area, *shared));
3064 : :
3065 : 1585795 : return result;
3066 : : }
3067 : :
3068 : : /* Slow path: try to allocate a new chunk. */
3069 : 2448 : LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3070 : :
3071 : : /*
3072 : : * Check if we need to help increase the number of buckets or batches.
3073 : : */
3074 [ + + ]: 2448 : if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
3075 [ + + ]: 2420 : pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3076 : : {
3077 : 76 : ParallelHashGrowth growth = pstate->growth;
3078 : :
3079 : 76 : hashtable->current_chunk = NULL;
3080 : 76 : LWLockRelease(&pstate->lock);
3081 : :
3082 : : /* Another participant has commanded us to help grow. */
3083 [ + + ]: 76 : if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
3084 : 28 : ExecParallelHashIncreaseNumBatches(hashtable);
3085 [ + - ]: 48 : else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3086 : 48 : ExecParallelHashIncreaseNumBuckets(hashtable);
3087 : :
3088 : : /* The caller must retry. */
3089 : 76 : return NULL;
3090 : : }
3091 : :
3092 : : /* Oversized tuples get their own chunk. */
3093 [ + + ]: 2372 : if (size > HASH_CHUNK_THRESHOLD)
3094 : 32 : chunk_size = size + HASH_CHUNK_HEADER_SIZE;
3095 : : else
3096 : 2340 : chunk_size = HASH_CHUNK_SIZE;
3097 : :
3098 : : /* Check if it's time to grow batches or buckets. */
3099 [ + + ]: 2372 : if (pstate->growth != PHJ_GROWTH_DISABLED)
3100 : : {
3101 : : Assert(curbatch == 0);
3102 : : Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASH_INNER);
3103 : :
3104 : : /*
3105 : : * Check if our space limit would be exceeded. To avoid choking on
3106 : : * very large tuples or very low hash_mem setting, we'll always allow
3107 : : * each backend to allocate at least one chunk.
3108 : : */
3109 [ + + ]: 1205 : if (hashtable->batches[0].at_least_one_chunk &&
3110 : 935 : hashtable->batches[0].shared->size +
3111 [ + + ]: 935 : chunk_size > pstate->space_allowed)
3112 : : {
3113 : 25 : pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
3114 : 25 : hashtable->batches[0].shared->space_exhausted = true;
3115 : 25 : LWLockRelease(&pstate->lock);
3116 : :
3117 : 25 : return NULL;
3118 : : }
3119 : :
3120 : : /* Check if our load factor limit would be exceeded. */
3121 [ + + ]: 1180 : if (hashtable->nbatch == 1)
3122 : : {
3123 : 1030 : hashtable->batches[0].shared->ntuples += hashtable->batches[0].ntuples;
3124 : 1030 : hashtable->batches[0].ntuples = 0;
3125 : : /* Guard against integer overflow and alloc size overflow */
3126 : 1030 : if (hashtable->batches[0].shared->ntuples + 1 >
3127 [ + + ]: 1030 : hashtable->nbuckets * NTUP_PER_BUCKET &&
3128 [ + - ]: 48 : hashtable->nbuckets < (INT_MAX / 2) &&
3129 [ + - ]: 48 : hashtable->nbuckets * 2 <=
3130 : : MaxAllocSize / sizeof(dsa_pointer_atomic))
3131 : : {
3132 : 48 : pstate->growth = PHJ_GROWTH_NEED_MORE_BUCKETS;
3133 : 48 : LWLockRelease(&pstate->lock);
3134 : :
3135 : 48 : return NULL;
3136 : : }
3137 : : }
3138 : : }
3139 : :
3140 : : /* We are cleared to allocate a new chunk. */
3141 : 2299 : chunk_shared = dsa_allocate(hashtable->area, chunk_size);
3142 : 2299 : hashtable->batches[curbatch].shared->size += chunk_size;
3143 : 2299 : hashtable->batches[curbatch].at_least_one_chunk = true;
3144 : :
3145 : : /* Set up the chunk. */
3146 : 2299 : chunk = (HashMemoryChunk) dsa_get_address(hashtable->area, chunk_shared);
3147 : 2299 : *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE;
3148 : 2299 : chunk->maxlen = chunk_size - HASH_CHUNK_HEADER_SIZE;
3149 : 2299 : chunk->used = size;
3150 : :
3151 : : /*
3152 : : * Push it onto the list of chunks, so that it can be found if we need to
3153 : : * increase the number of buckets or batches (batch 0 only) and later for
3154 : : * freeing the memory (all batches).
3155 : : */
3156 : 2299 : chunk->next.shared = hashtable->batches[curbatch].shared->chunks;
3157 : 2299 : hashtable->batches[curbatch].shared->chunks = chunk_shared;
3158 : :
3159 [ + + ]: 2299 : if (size <= HASH_CHUNK_THRESHOLD)
3160 : : {
3161 : : /*
3162 : : * Make this the current chunk so that we can use the fast path to
3163 : : * fill the rest of it up in future calls.
3164 : : */
3165 : 2275 : hashtable->current_chunk = chunk;
3166 : 2275 : hashtable->current_chunk_shared = chunk_shared;
3167 : : }
3168 : 2299 : LWLockRelease(&pstate->lock);
3169 : :
3170 : : Assert(HASH_CHUNK_DATA(chunk) == dsa_get_address(hashtable->area, *shared));
3171 : 2299 : result = (HashJoinTuple) HASH_CHUNK_DATA(chunk);
3172 : :
3173 : 2299 : return result;
3174 : : }
3175 : :
3176 : : /*
3177 : : * One backend needs to set up the shared batch state including tuplestores.
3178 : : * Other backends will ensure they have correctly configured accessors by
3179 : : * called ExecParallelHashEnsureBatchAccessors().
3180 : : */
3181 : : static void
3182 : 148 : ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
3183 : : {
3184 : 148 : ParallelHashJoinState *pstate = hashtable->parallel_state;
3185 : : ParallelHashJoinBatch *batches;
3186 : : MemoryContext oldcxt;
3187 : : int i;
3188 : :
3189 : : Assert(hashtable->batches == NULL);
3190 : :
3191 : : /* Allocate space. */
3192 : 148 : pstate->batches =
3193 : 148 : dsa_allocate0(hashtable->area,
3194 : : EstimateParallelHashJoinBatch(hashtable) * nbatch);
3195 : 148 : pstate->nbatch = nbatch;
3196 : 148 : batches = dsa_get_address(hashtable->area, pstate->batches);
3197 : :
3198 : : /*
3199 : : * Use hash join spill memory context to allocate accessors, including
3200 : : * buffers for the temporary files.
3201 : : */
3202 : 148 : oldcxt = MemoryContextSwitchTo(hashtable->spillCxt);
3203 : :
3204 : : /* Allocate this backend's accessor array. */
3205 : 148 : hashtable->nbatch = nbatch;
3206 : 148 : hashtable->batches =
3207 : 148 : palloc0_array(ParallelHashJoinBatchAccessor, hashtable->nbatch);
3208 : :
3209 : : /* Set up the shared state, tuplestores and backend-local accessors. */
3210 [ + + ]: 708 : for (i = 0; i < hashtable->nbatch; ++i)
3211 : : {
3212 : 560 : ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
3213 : 560 : ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
3214 : : char name[MAXPGPATH];
3215 : :
3216 : : /*
3217 : : * All members of shared were zero-initialized. We just need to set
3218 : : * up the Barrier.
3219 : : */
3220 : 560 : BarrierInit(&shared->batch_barrier, 0);
3221 [ + + ]: 560 : if (i == 0)
3222 : : {
3223 : : /* Batch 0 doesn't need to be loaded. */
3224 : 148 : BarrierAttach(&shared->batch_barrier);
3225 [ + + ]: 592 : while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBE)
3226 : 444 : BarrierArriveAndWait(&shared->batch_barrier, 0);
3227 : 148 : BarrierDetach(&shared->batch_barrier);
3228 : : }
3229 : :
3230 : : /* Initialize accessor state. All members were zero-initialized. */
3231 : 560 : accessor->shared = shared;
3232 : :
3233 : : /* Initialize the shared tuplestores. */
3234 : 560 : snprintf(name, sizeof(name), "i%dof%d", i, hashtable->nbatch);
3235 : 560 : accessor->inner_tuples =
3236 : 560 : sts_initialize(ParallelHashJoinBatchInner(shared),
3237 : : pstate->nparticipants,
3238 : : ParallelWorkerNumber + 1,
3239 : : sizeof(uint32),
3240 : : SHARED_TUPLESTORE_SINGLE_PASS,
3241 : : &pstate->fileset,
3242 : : name);
3243 : 560 : snprintf(name, sizeof(name), "o%dof%d", i, hashtable->nbatch);
3244 : 560 : accessor->outer_tuples =
3245 : 560 : sts_initialize(ParallelHashJoinBatchOuter(shared,
3246 : : pstate->nparticipants),
3247 : : pstate->nparticipants,
3248 : : ParallelWorkerNumber + 1,
3249 : : sizeof(uint32),
3250 : : SHARED_TUPLESTORE_SINGLE_PASS,
3251 : : &pstate->fileset,
3252 : : name);
3253 : : }
3254 : :
3255 : 148 : MemoryContextSwitchTo(oldcxt);
3256 : 148 : }
3257 : :
3258 : : /*
3259 : : * Free the current set of ParallelHashJoinBatchAccessor objects.
3260 : : */
3261 : : static void
3262 : 39 : ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
3263 : : {
3264 : : int i;
3265 : :
3266 [ + + ]: 120 : for (i = 0; i < hashtable->nbatch; ++i)
3267 : : {
3268 : : /* Make sure no files are left open. */
3269 : 81 : sts_end_write(hashtable->batches[i].inner_tuples);
3270 : 81 : sts_end_write(hashtable->batches[i].outer_tuples);
3271 : 81 : sts_end_parallel_scan(hashtable->batches[i].inner_tuples);
3272 : 81 : sts_end_parallel_scan(hashtable->batches[i].outer_tuples);
3273 : : }
3274 : 39 : pfree(hashtable->batches);
3275 : 39 : hashtable->batches = NULL;
3276 : 39 : }
3277 : :
3278 : : /*
3279 : : * Make sure this backend has up-to-date accessors for the current set of
3280 : : * batches.
3281 : : */
3282 : : static void
3283 : 607 : ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
3284 : : {
3285 : 607 : ParallelHashJoinState *pstate = hashtable->parallel_state;
3286 : : ParallelHashJoinBatch *batches;
3287 : : MemoryContext oldcxt;
3288 : : int i;
3289 : :
3290 [ + + ]: 607 : if (hashtable->batches != NULL)
3291 : : {
3292 [ + - ]: 441 : if (hashtable->nbatch == pstate->nbatch)
3293 : 441 : return;
3294 : 0 : ExecParallelHashCloseBatchAccessors(hashtable);
3295 : : }
3296 : :
3297 : : /*
3298 : : * We should never see a state where the batch-tracking array is freed,
3299 : : * because we should have given up sooner if we join when the build
3300 : : * barrier has reached the PHJ_BUILD_FREE phase.
3301 : : */
3302 : : Assert(DsaPointerIsValid(pstate->batches));
3303 : :
3304 : : /*
3305 : : * Use hash join spill memory context to allocate accessors, including
3306 : : * buffers for the temporary files.
3307 : : */
3308 : 166 : oldcxt = MemoryContextSwitchTo(hashtable->spillCxt);
3309 : :
3310 : : /* Allocate this backend's accessor array. */
3311 : 166 : hashtable->nbatch = pstate->nbatch;
3312 : 166 : hashtable->batches =
3313 : 166 : palloc0_array(ParallelHashJoinBatchAccessor, hashtable->nbatch);
3314 : :
3315 : : /* Find the base of the pseudo-array of ParallelHashJoinBatch objects. */
3316 : : batches = (ParallelHashJoinBatch *)
3317 : 166 : dsa_get_address(hashtable->area, pstate->batches);
3318 : :
3319 : : /* Set up the accessor array and attach to the tuplestores. */
3320 [ + + ]: 871 : for (i = 0; i < hashtable->nbatch; ++i)
3321 : : {
3322 : 705 : ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
3323 : 705 : ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
3324 : :
3325 : 705 : accessor->shared = shared;
3326 : 705 : accessor->preallocated = 0;
3327 : 705 : accessor->done = false;
3328 : 705 : accessor->outer_eof = false;
3329 : 705 : accessor->inner_tuples =
3330 : 705 : sts_attach(ParallelHashJoinBatchInner(shared),
3331 : : ParallelWorkerNumber + 1,
3332 : : &pstate->fileset);
3333 : 705 : accessor->outer_tuples =
3334 : 705 : sts_attach(ParallelHashJoinBatchOuter(shared,
3335 : : pstate->nparticipants),
3336 : : ParallelWorkerNumber + 1,
3337 : : &pstate->fileset);
3338 : : }
3339 : :
3340 : 166 : MemoryContextSwitchTo(oldcxt);
3341 : : }
3342 : :
3343 : : /*
3344 : : * Allocate an empty shared memory hash table for a given batch.
3345 : : */
3346 : : void
3347 : 504 : ExecParallelHashTableAlloc(HashJoinTable hashtable, int batchno)
3348 : : {
3349 : 504 : ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
3350 : : dsa_pointer_atomic *buckets;
3351 : 504 : int nbuckets = hashtable->parallel_state->nbuckets;
3352 : : int i;
3353 : :
3354 : 504 : batch->buckets =
3355 : 504 : dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
3356 : : buckets = (dsa_pointer_atomic *)
3357 : 504 : dsa_get_address(hashtable->area, batch->buckets);
3358 [ + + ]: 2662904 : for (i = 0; i < nbuckets; ++i)
3359 : 2662400 : dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
3360 : 504 : }
3361 : :
3362 : : /*
3363 : : * If we are currently attached to a shared hash join batch, detach. If we
3364 : : * are last to detach, clean up.
3365 : : */
3366 : : void
3367 : 20580 : ExecHashTableDetachBatch(HashJoinTable hashtable)
3368 : : {
3369 [ + + ]: 20580 : if (hashtable->parallel_state != NULL &&
3370 [ + + ]: 887 : hashtable->curbatch >= 0)
3371 : : {
3372 : 611 : int curbatch = hashtable->curbatch;
3373 : 611 : ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
3374 : 611 : bool attached = true;
3375 : :
3376 : : /* Make sure any temporary files are closed. */
3377 : 611 : sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
3378 : 611 : sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
3379 : :
3380 : : /* After attaching we always get at least to PHJ_BATCH_PROBE. */
3381 : : Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE ||
3382 : : BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_SCAN);
3383 : :
3384 : : /*
3385 : : * If we're abandoning the PHJ_BATCH_PROBE phase early without having
3386 : : * reached the end of it, it means the plan doesn't want any more
3387 : : * tuples, and it is happy to abandon any tuples buffered in this
3388 : : * process's subplans. For correctness, we can't allow any process to
3389 : : * execute the PHJ_BATCH_SCAN phase, because we will never have the
3390 : : * complete set of match bits. Therefore we skip emitting unmatched
3391 : : * tuples in all backends (if this is a full/right join), as if those
3392 : : * tuples were all due to be emitted by this process and it has
3393 : : * abandoned them too.
3394 : : */
3395 [ + + ]: 611 : if (BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE &&
3396 [ - + ]: 566 : !hashtable->batches[curbatch].outer_eof)
3397 : : {
3398 : : /*
3399 : : * This flag may be written to by multiple backends during
3400 : : * PHJ_BATCH_PROBE phase, but will only be read in PHJ_BATCH_SCAN
3401 : : * phase so requires no extra locking.
3402 : : */
3403 : 0 : batch->skip_unmatched = true;
3404 : : }
3405 : :
3406 : : /*
3407 : : * Even if we aren't doing a full/right outer join, we'll step through
3408 : : * the PHJ_BATCH_SCAN phase just to maintain the invariant that
3409 : : * freeing happens in PHJ_BATCH_FREE, but that'll be wait-free.
3410 : : */
3411 [ + + ]: 611 : if (BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE)
3412 : 566 : attached = BarrierArriveAndDetachExceptLast(&batch->batch_barrier);
3413 [ + + + + ]: 611 : if (attached && BarrierArriveAndDetach(&batch->batch_barrier))
3414 : : {
3415 : : /*
3416 : : * We are not longer attached to the batch barrier, but we're the
3417 : : * process that was chosen to free resources and it's safe to
3418 : : * assert the current phase. The ParallelHashJoinBatch can't go
3419 : : * away underneath us while we are attached to the build barrier,
3420 : : * making this access safe.
3421 : : */
3422 : : Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_FREE);
3423 : :
3424 : : /* Free shared chunks and buckets. */
3425 [ + + ]: 2610 : while (DsaPointerIsValid(batch->chunks))
3426 : : {
3427 : : HashMemoryChunk chunk =
3428 : 2106 : dsa_get_address(hashtable->area, batch->chunks);
3429 : 2106 : dsa_pointer next = chunk->next.shared;
3430 : :
3431 : 2106 : dsa_free(hashtable->area, batch->chunks);
3432 : 2106 : batch->chunks = next;
3433 : : }
3434 [ + - ]: 504 : if (DsaPointerIsValid(batch->buckets))
3435 : : {
3436 : 504 : dsa_free(hashtable->area, batch->buckets);
3437 : 504 : batch->buckets = InvalidDsaPointer;
3438 : : }
3439 : : }
3440 : :
3441 : : /*
3442 : : * Track the largest batch we've been attached to. Though each
3443 : : * backend might see a different subset of batches, explain.c will
3444 : : * scan the results from all backends to find the largest value.
3445 : : */
3446 : 611 : hashtable->spacePeak =
3447 : 611 : Max(hashtable->spacePeak,
3448 : : batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
3449 : :
3450 : : /* Remember that we are not attached to a batch. */
3451 : 611 : hashtable->curbatch = -1;
3452 : : }
3453 : 20580 : }
3454 : :
3455 : : /*
3456 : : * Detach from all shared resources. If we are last to detach, clean up.
3457 : : */
3458 : : void
3459 : 19969 : ExecHashTableDetach(HashJoinTable hashtable)
3460 : : {
3461 : 19969 : ParallelHashJoinState *pstate = hashtable->parallel_state;
3462 : :
3463 : : /*
3464 : : * If we're involved in a parallel query, we must either have gotten all
3465 : : * the way to PHJ_BUILD_RUN, or joined too late and be in PHJ_BUILD_FREE.
3466 : : */
3467 : : Assert(!pstate ||
3468 : : BarrierPhase(&pstate->build_barrier) >= PHJ_BUILD_RUN);
3469 : :
3470 [ + + + + ]: 19969 : if (pstate && BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_RUN)
3471 : : {
3472 : : int i;
3473 : :
3474 : : /* Make sure any temporary files are closed. */
3475 [ + - ]: 275 : if (hashtable->batches)
3476 : : {
3477 [ + + ]: 1459 : for (i = 0; i < hashtable->nbatch; ++i)
3478 : : {
3479 : 1184 : sts_end_write(hashtable->batches[i].inner_tuples);
3480 : 1184 : sts_end_write(hashtable->batches[i].outer_tuples);
3481 : 1184 : sts_end_parallel_scan(hashtable->batches[i].inner_tuples);
3482 : 1184 : sts_end_parallel_scan(hashtable->batches[i].outer_tuples);
3483 : : }
3484 : : }
3485 : :
3486 : : /* If we're last to detach, clean up shared memory. */
3487 [ + + ]: 275 : if (BarrierArriveAndDetach(&pstate->build_barrier))
3488 : : {
3489 : : /*
3490 : : * Late joining processes will see this state and give up
3491 : : * immediately.
3492 : : */
3493 : : Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_FREE);
3494 : :
3495 [ + - ]: 116 : if (DsaPointerIsValid(pstate->batches))
3496 : : {
3497 : 116 : dsa_free(hashtable->area, pstate->batches);
3498 : 116 : pstate->batches = InvalidDsaPointer;
3499 : : }
3500 : : }
3501 : : }
3502 : 19969 : hashtable->parallel_state = NULL;
3503 : 19969 : }
3504 : :
3505 : : /*
3506 : : * Get the first tuple in a given bucket identified by number.
3507 : : */
3508 : : static inline HashJoinTuple
3509 : 1853628 : ExecParallelHashFirstTuple(HashJoinTable hashtable, int bucketno)
3510 : : {
3511 : : HashJoinTuple tuple;
3512 : : dsa_pointer p;
3513 : :
3514 : : Assert(hashtable->parallel_state);
3515 : 1853628 : p = dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
3516 : 1853628 : tuple = (HashJoinTuple) dsa_get_address(hashtable->area, p);
3517 : :
3518 : 1853628 : return tuple;
3519 : : }
3520 : :
3521 : : /*
3522 : : * Get the next tuple in the same bucket as 'tuple'.
3523 : : */
3524 : : static inline HashJoinTuple
3525 : 2461804 : ExecParallelHashNextTuple(HashJoinTable hashtable, HashJoinTuple tuple)
3526 : : {
3527 : : HashJoinTuple next;
3528 : :
3529 : : Assert(hashtable->parallel_state);
3530 : 2461804 : next = (HashJoinTuple) dsa_get_address(hashtable->area, tuple->next.shared);
3531 : :
3532 : 2461804 : return next;
3533 : : }
3534 : :
3535 : : /*
3536 : : * Insert a tuple at the front of a chain of tuples in DSA memory atomically.
3537 : : */
3538 : : static inline void
3539 : 1732062 : ExecParallelHashPushTuple(dsa_pointer_atomic *head,
3540 : : HashJoinTuple tuple,
3541 : : dsa_pointer tuple_shared)
3542 : : {
3543 : : for (;;)
3544 : : {
3545 : 1741933 : tuple->next.shared = dsa_pointer_atomic_read(head);
3546 [ + + ]: 1741933 : if (dsa_pointer_atomic_compare_exchange(head,
3547 : 1741933 : &tuple->next.shared,
3548 : : tuple_shared))
3549 : 1732062 : break;
3550 : : }
3551 : 1732062 : }
3552 : :
3553 : : /*
3554 : : * Prepare to work on a given batch.
3555 : : */
3556 : : void
3557 : 1377 : ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno)
3558 : : {
3559 : : Assert(hashtable->batches[batchno].shared->buckets != InvalidDsaPointer);
3560 : :
3561 : 1377 : hashtable->curbatch = batchno;
3562 : 1377 : hashtable->buckets.shared = (dsa_pointer_atomic *)
3563 : 1377 : dsa_get_address(hashtable->area,
3564 : 1377 : hashtable->batches[batchno].shared->buckets);
3565 : 1377 : hashtable->nbuckets = hashtable->parallel_state->nbuckets;
3566 : 1377 : hashtable->log2_nbuckets = pg_ceil_log2_32(hashtable->nbuckets);
3567 : 1377 : hashtable->current_chunk = NULL;
3568 : 1377 : hashtable->current_chunk_shared = InvalidDsaPointer;
3569 : 1377 : hashtable->batches[batchno].at_least_one_chunk = false;
3570 : 1377 : }
3571 : :
3572 : : /*
3573 : : * Take the next available chunk from the queue of chunks being worked on in
3574 : : * parallel. Return NULL if there are none left. Otherwise return a pointer
3575 : : * to the chunk, and set *shared to the DSA pointer to the chunk.
3576 : : */
3577 : : static HashMemoryChunk
3578 : 458 : ExecParallelHashPopChunkQueue(HashJoinTable hashtable, dsa_pointer *shared)
3579 : : {
3580 : 458 : ParallelHashJoinState *pstate = hashtable->parallel_state;
3581 : : HashMemoryChunk chunk;
3582 : :
3583 : 458 : LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3584 [ + + ]: 458 : if (DsaPointerIsValid(pstate->chunk_work_queue))
3585 : : {
3586 : 369 : *shared = pstate->chunk_work_queue;
3587 : : chunk = (HashMemoryChunk)
3588 : 369 : dsa_get_address(hashtable->area, *shared);
3589 : 369 : pstate->chunk_work_queue = chunk->next.shared;
3590 : : }
3591 : : else
3592 : 89 : chunk = NULL;
3593 : 458 : LWLockRelease(&pstate->lock);
3594 : :
3595 : 458 : return chunk;
3596 : : }
3597 : :
3598 : : /*
3599 : : * Increase the space preallocated in this backend for a given inner batch by
3600 : : * at least a given amount. This allows us to track whether a given batch
3601 : : * would fit in memory when loaded back in. Also increase the number of
3602 : : * batches or buckets if required.
3603 : : *
3604 : : * This maintains a running estimation of how much space will be taken when we
3605 : : * load the batch back into memory by simulating the way chunks will be handed
3606 : : * out to workers. It's not perfectly accurate because the tuples will be
3607 : : * packed into memory chunks differently by ExecParallelHashTupleAlloc(), but
3608 : : * it should be pretty close. It tends to overestimate by a fraction of a
3609 : : * chunk per worker since all workers gang up to preallocate during hashing,
3610 : : * but workers tend to reload batches alone if there are enough to go around,
3611 : : * leaving fewer partially filled chunks. This effect is bounded by
3612 : : * nparticipants.
3613 : : *
3614 : : * Return false if the number of batches or buckets has changed, and the
3615 : : * caller should reconsider which batch a given tuple now belongs in and call
3616 : : * again.
3617 : : */
3618 : : static bool
3619 : 1153 : ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size)
3620 : : {
3621 : 1153 : ParallelHashJoinState *pstate = hashtable->parallel_state;
3622 : 1153 : ParallelHashJoinBatchAccessor *batch = &hashtable->batches[batchno];
3623 : 1153 : size_t want = Max(size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);
3624 : :
3625 : : Assert(batchno > 0);
3626 : : Assert(batchno < hashtable->nbatch);
3627 : : Assert(size == MAXALIGN(size));
3628 : :
3629 : 1153 : LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3630 : :
3631 : : /* Has another participant commanded us to help grow? */
3632 [ + + ]: 1153 : if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
3633 [ - + ]: 1142 : pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3634 : : {
3635 : 11 : ParallelHashGrowth growth = pstate->growth;
3636 : :
3637 : 11 : LWLockRelease(&pstate->lock);
3638 [ + - ]: 11 : if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
3639 : 11 : ExecParallelHashIncreaseNumBatches(hashtable);
3640 [ # # ]: 0 : else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3641 : 0 : ExecParallelHashIncreaseNumBuckets(hashtable);
3642 : :
3643 : 11 : return false;
3644 : : }
3645 : :
3646 [ + + ]: 1142 : if (pstate->growth != PHJ_GROWTH_DISABLED &&
3647 [ + + ]: 989 : batch->at_least_one_chunk &&
3648 : 317 : (batch->shared->estimated_size + want + HASH_CHUNK_HEADER_SIZE
3649 [ + + ]: 317 : > pstate->space_allowed))
3650 : : {
3651 : : /*
3652 : : * We have determined that this batch would exceed the space budget if
3653 : : * loaded into memory. Command all participants to help repartition.
3654 : : */
3655 : 7 : batch->shared->space_exhausted = true;
3656 : 7 : pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
3657 : 7 : LWLockRelease(&pstate->lock);
3658 : :
3659 : 7 : return false;
3660 : : }
3661 : :
3662 : 1135 : batch->at_least_one_chunk = true;
3663 : 1135 : batch->shared->estimated_size += want + HASH_CHUNK_HEADER_SIZE;
3664 : 1135 : batch->preallocated = want;
3665 : 1135 : LWLockRelease(&pstate->lock);
3666 : :
3667 : 1135 : return true;
3668 : : }
3669 : :
3670 : : /*
3671 : : * Calculate the limit on how much memory can be used by Hash and similar
3672 : : * plan types. This is work_mem times hash_mem_multiplier, and is
3673 : : * expressed in bytes.
3674 : : *
3675 : : * Exported for use by the planner, as well as other hash-like executor
3676 : : * nodes. This is a rather random place for this, but there is no better
3677 : : * place.
3678 : : */
3679 : : size_t
3680 : 1230491 : get_hash_memory_limit(void)
3681 : : {
3682 : : double mem_limit;
3683 : :
3684 : : /* Do initial calculation in double arithmetic */
3685 : 1230491 : mem_limit = (double) work_mem * hash_mem_multiplier * 1024.0;
3686 : :
3687 : : /* Clamp in case it doesn't fit in size_t */
3688 [ + - ]: 1230491 : mem_limit = Min(mem_limit, (double) SIZE_MAX);
3689 : :
3690 : 1230491 : return (size_t) mem_limit;
3691 : : }
|
