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