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