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