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