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