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