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