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