PostgreSQL Source Code  git master
nodeHash.c File Reference
#include "postgres.h"
#include <math.h>
#include <limits.h>
#include "access/htup_details.h"
#include "access/parallel.h"
#include "catalog/pg_statistic.h"
#include "commands/tablespace.h"
#include "executor/execdebug.h"
#include "executor/hashjoin.h"
#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "port/atomics.h"
#include "utils/dynahash.h"
#include "utils/memutils.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"
Include dependency graph for nodeHash.c:

Go to the source code of this file.

Macros

#define NTUP_PER_BUCKET   1
 

Functions

static void ExecHashIncreaseNumBatches (HashJoinTable hashtable)
 
static void ExecHashIncreaseNumBuckets (HashJoinTable hashtable)
 
static void ExecParallelHashIncreaseNumBatches (HashJoinTable hashtable)
 
static void ExecParallelHashIncreaseNumBuckets (HashJoinTable hashtable)
 
static void ExecHashBuildSkewHash (HashJoinTable hashtable, Hash *node, int mcvsToUse)
 
static void ExecHashSkewTableInsert (HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue, int bucketNumber)
 
static void ExecHashRemoveNextSkewBucket (HashJoinTable hashtable)
 
static void * dense_alloc (HashJoinTable hashtable, Size size)
 
static HashJoinTuple ExecParallelHashTupleAlloc (HashJoinTable hashtable, size_t size, dsa_pointer *shared)
 
static void MultiExecPrivateHash (HashState *node)
 
static void MultiExecParallelHash (HashState *node)
 
static HashJoinTuple ExecParallelHashFirstTuple (HashJoinTable table, int bucketno)
 
static HashJoinTuple ExecParallelHashNextTuple (HashJoinTable table, HashJoinTuple tuple)
 
static void ExecParallelHashPushTuple (dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
 
static void ExecParallelHashJoinSetUpBatches (HashJoinTable hashtable, int nbatch)
 
static void ExecParallelHashEnsureBatchAccessors (HashJoinTable hashtable)
 
static void ExecParallelHashRepartitionFirst (HashJoinTable hashtable)
 
static void ExecParallelHashRepartitionRest (HashJoinTable hashtable)
 
static HashMemoryChunk ExecParallelHashPopChunkQueue (HashJoinTable table, dsa_pointer *shared)
 
static bool ExecParallelHashTuplePrealloc (HashJoinTable hashtable, int batchno, size_t size)
 
static void ExecParallelHashMergeCounters (HashJoinTable hashtable)
 
static void ExecParallelHashCloseBatchAccessors (HashJoinTable hashtable)
 
static TupleTableSlotExecHash (PlanState *pstate)
 
NodeMultiExecHash (HashState *node)
 
HashStateExecInitHash (Hash *node, EState *estate, int eflags)
 
void ExecEndHash (HashState *node)
 
HashJoinTable ExecHashTableCreate (HashState *state, List *hashOperators, bool keepNulls)
 
void ExecChooseHashTableSize (double ntuples, int tupwidth, bool useskew, bool try_combined_work_mem, int parallel_workers, size_t *space_allowed, int *numbuckets, int *numbatches, int *num_skew_mcvs)
 
void ExecHashTableDestroy (HashJoinTable hashtable)
 
void ExecHashTableInsert (HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue)
 
void ExecParallelHashTableInsert (HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue)
 
void ExecParallelHashTableInsertCurrentBatch (HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue)
 
bool ExecHashGetHashValue (HashJoinTable hashtable, ExprContext *econtext, List *hashkeys, bool outer_tuple, bool keep_nulls, uint32 *hashvalue)
 
void ExecHashGetBucketAndBatch (HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
 
bool ExecScanHashBucket (HashJoinState *hjstate, ExprContext *econtext)
 
bool ExecParallelScanHashBucket (HashJoinState *hjstate, ExprContext *econtext)
 
void ExecPrepHashTableForUnmatched (HashJoinState *hjstate)
 
bool ExecScanHashTableForUnmatched (HashJoinState *hjstate, ExprContext *econtext)
 
void ExecHashTableReset (HashJoinTable hashtable)
 
void ExecHashTableResetMatchFlags (HashJoinTable hashtable)
 
void ExecReScanHash (HashState *node)
 
int ExecHashGetSkewBucket (HashJoinTable hashtable, uint32 hashvalue)
 
void ExecHashEstimate (HashState *node, ParallelContext *pcxt)
 
void ExecHashInitializeDSM (HashState *node, ParallelContext *pcxt)
 
void ExecHashInitializeWorker (HashState *node, ParallelWorkerContext *pwcxt)
 
void ExecShutdownHash (HashState *node)
 
void ExecHashRetrieveInstrumentation (HashState *node)
 
void ExecHashGetInstrumentation (HashInstrumentation *instrument, HashJoinTable hashtable)
 
void ExecParallelHashTableAlloc (HashJoinTable hashtable, int batchno)
 
void ExecHashTableDetachBatch (HashJoinTable hashtable)
 
void ExecHashTableDetach (HashJoinTable hashtable)
 
void ExecParallelHashTableSetCurrentBatch (HashJoinTable hashtable, int batchno)
 

Macro Definition Documentation

◆ NTUP_PER_BUCKET

#define NTUP_PER_BUCKET   1

Function Documentation

◆ dense_alloc()

static void * dense_alloc ( HashJoinTable  hashtable,
Size  size 
)
static

Definition at line 2649 of file nodeHash.c.

References HashJoinTableData::batchCxt, HashJoinTableData::chunks, HASH_CHUNK_DATA, HASH_CHUNK_HEADER_SIZE, HASH_CHUNK_SIZE, HASH_CHUNK_THRESHOLD, MAXALIGN, HashMemoryChunkData::maxlen, MemoryContextAlloc(), HashMemoryChunkData::next, HashMemoryChunkData::ntuples, HashMemoryChunkData::unshared, and HashMemoryChunkData::used.

Referenced by ExecHashIncreaseNumBatches(), ExecHashRemoveNextSkewBucket(), and ExecHashTableInsert().

2650 {
2651  HashMemoryChunk newChunk;
2652  char *ptr;
2653 
2654  /* just in case the size is not already aligned properly */
2655  size = MAXALIGN(size);
2656 
2657  /*
2658  * If tuple size is larger than threshold, allocate a separate chunk.
2659  */
2660  if (size > HASH_CHUNK_THRESHOLD)
2661  {
2662  /* allocate new chunk and put it at the beginning of the list */
2663  newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
2664  HASH_CHUNK_HEADER_SIZE + size);
2665  newChunk->maxlen = size;
2666  newChunk->used = size;
2667  newChunk->ntuples = 1;
2668 
2669  /*
2670  * Add this chunk to the list after the first existing chunk, so that
2671  * we don't lose the remaining space in the "current" chunk.
2672  */
2673  if (hashtable->chunks != NULL)
2674  {
2675  newChunk->next = hashtable->chunks->next;
2676  hashtable->chunks->next.unshared = newChunk;
2677  }
2678  else
2679  {
2680  newChunk->next.unshared = hashtable->chunks;
2681  hashtable->chunks = newChunk;
2682  }
2683 
2684  return HASH_CHUNK_DATA(newChunk);
2685  }
2686 
2687  /*
2688  * See if we have enough space for it in the current chunk (if any). If
2689  * not, allocate a fresh chunk.
2690  */
2691  if ((hashtable->chunks == NULL) ||
2692  (hashtable->chunks->maxlen - hashtable->chunks->used) < size)
2693  {
2694  /* allocate new chunk and put it at the beginning of the list */
2695  newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
2697 
2698  newChunk->maxlen = HASH_CHUNK_SIZE;
2699  newChunk->used = size;
2700  newChunk->ntuples = 1;
2701 
2702  newChunk->next.unshared = hashtable->chunks;
2703  hashtable->chunks = newChunk;
2704 
2705  return HASH_CHUNK_DATA(newChunk);
2706  }
2707 
2708  /* There is enough space in the current chunk, let's add the tuple */
2709  ptr = HASH_CHUNK_DATA(hashtable->chunks) + hashtable->chunks->used;
2710  hashtable->chunks->used += size;
2711  hashtable->chunks->ntuples += 1;
2712 
2713  /* return pointer to the start of the tuple memory */
2714  return ptr;
2715 }
#define HASH_CHUNK_SIZE
Definition: hashjoin.h:139
union HashMemoryChunkData::@96 next
#define HASH_CHUNK_THRESHOLD
Definition: hashjoin.h:143
struct HashMemoryChunkData * unshared
Definition: hashjoin.h:126
MemoryContext batchCxt
Definition: hashjoin.h:348
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define MAXALIGN(LEN)
Definition: c.h:641
HashMemoryChunk chunks
Definition: hashjoin.h:351
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:693
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141

◆ ExecChooseHashTableSize()

void ExecChooseHashTableSize ( double  ntuples,
int  tupwidth,
bool  useskew,
bool  try_combined_work_mem,
int  parallel_workers,
size_t *  space_allowed,
int *  numbuckets,
int *  numbatches,
int *  num_skew_mcvs 
)

Definition at line 661 of file nodeHash.c.

References Assert, ExecChooseHashTableSize(), HJTUPLE_OVERHEAD, Max, MAXALIGN, MaxAllocSize, Min, my_log2(), NTUP_PER_BUCKET, SizeofMinimalTupleHeader, SKEW_BUCKET_OVERHEAD, SKEW_WORK_MEM_PERCENT, and work_mem.

Referenced by ExecChooseHashTableSize(), ExecHashTableCreate(), and initial_cost_hashjoin().

668 {
669  int tupsize;
670  double inner_rel_bytes;
671  long bucket_bytes;
672  long hash_table_bytes;
673  long skew_table_bytes;
674  long max_pointers;
675  long mppow2;
676  int nbatch = 1;
677  int nbuckets;
678  double dbuckets;
679 
680  /* Force a plausible relation size if no info */
681  if (ntuples <= 0.0)
682  ntuples = 1000.0;
683 
684  /*
685  * Estimate tupsize based on footprint of tuple in hashtable... note this
686  * does not allow for any palloc overhead. The manipulations of spaceUsed
687  * don't count palloc overhead either.
688  */
689  tupsize = HJTUPLE_OVERHEAD +
691  MAXALIGN(tupwidth);
692  inner_rel_bytes = ntuples * tupsize;
693 
694  /*
695  * Target in-memory hashtable size is work_mem kilobytes.
696  */
697  hash_table_bytes = work_mem * 1024L;
698 
699  /*
700  * Parallel Hash tries to use the combined work_mem of all workers to
701  * avoid the need to batch. If that won't work, it falls back to work_mem
702  * per worker and tries to process batches in parallel.
703  */
704  if (try_combined_work_mem)
705  hash_table_bytes += hash_table_bytes * parallel_workers;
706 
707  *space_allowed = hash_table_bytes;
708 
709  /*
710  * If skew optimization is possible, estimate the number of skew buckets
711  * that will fit in the memory allowed, and decrement the assumed space
712  * available for the main hash table accordingly.
713  *
714  * We make the optimistic assumption that each skew bucket will contain
715  * one inner-relation tuple. If that turns out to be low, we will recover
716  * at runtime by reducing the number of skew buckets.
717  *
718  * hashtable->skewBucket will have up to 8 times as many HashSkewBucket
719  * pointers as the number of MCVs we allow, since ExecHashBuildSkewHash
720  * will round up to the next power of 2 and then multiply by 4 to reduce
721  * collisions.
722  */
723  if (useskew)
724  {
725  skew_table_bytes = hash_table_bytes * SKEW_WORK_MEM_PERCENT / 100;
726 
727  /*----------
728  * Divisor is:
729  * size of a hash tuple +
730  * worst-case size of skewBucket[] per MCV +
731  * size of skewBucketNums[] entry +
732  * size of skew bucket struct itself
733  *----------
734  */
735  *num_skew_mcvs = skew_table_bytes / (tupsize +
736  (8 * sizeof(HashSkewBucket *)) +
737  sizeof(int) +
739  if (*num_skew_mcvs > 0)
740  hash_table_bytes -= skew_table_bytes;
741  }
742  else
743  *num_skew_mcvs = 0;
744 
745  /*
746  * Set nbuckets to achieve an average bucket load of NTUP_PER_BUCKET when
747  * memory is filled, assuming a single batch; but limit the value so that
748  * the pointer arrays we'll try to allocate do not exceed work_mem nor
749  * MaxAllocSize.
750  *
751  * Note that both nbuckets and nbatch must be powers of 2 to make
752  * ExecHashGetBucketAndBatch fast.
753  */
754  max_pointers = *space_allowed / sizeof(HashJoinTuple);
755  max_pointers = Min(max_pointers, MaxAllocSize / sizeof(HashJoinTuple));
756  /* If max_pointers isn't a power of 2, must round it down to one */
757  mppow2 = 1L << my_log2(max_pointers);
758  if (max_pointers != mppow2)
759  max_pointers = mppow2 / 2;
760 
761  /* Also ensure we avoid integer overflow in nbatch and nbuckets */
762  /* (this step is redundant given the current value of MaxAllocSize) */
763  max_pointers = Min(max_pointers, INT_MAX / 2);
764 
765  dbuckets = ceil(ntuples / NTUP_PER_BUCKET);
766  dbuckets = Min(dbuckets, max_pointers);
767  nbuckets = (int) dbuckets;
768  /* don't let nbuckets be really small, though ... */
769  nbuckets = Max(nbuckets, 1024);
770  /* ... and force it to be a power of 2. */
771  nbuckets = 1 << my_log2(nbuckets);
772 
773  /*
774  * If there's not enough space to store the projected number of tuples and
775  * the required bucket headers, we will need multiple batches.
776  */
777  bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
778  if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
779  {
780  /* We'll need multiple batches */
781  long lbuckets;
782  double dbatch;
783  int minbatch;
784  long bucket_size;
785 
786  /*
787  * If Parallel Hash with combined work_mem would still need multiple
788  * batches, we'll have to fall back to regular work_mem budget.
789  */
790  if (try_combined_work_mem)
791  {
792  ExecChooseHashTableSize(ntuples, tupwidth, useskew,
793  false, parallel_workers,
794  space_allowed,
795  numbuckets,
796  numbatches,
797  num_skew_mcvs);
798  return;
799  }
800 
801  /*
802  * Estimate the number of buckets we'll want to have when work_mem is
803  * entirely full. Each bucket will contain a bucket pointer plus
804  * NTUP_PER_BUCKET tuples, whose projected size already includes
805  * overhead for the hash code, pointer to the next tuple, etc.
806  */
807  bucket_size = (tupsize * NTUP_PER_BUCKET + sizeof(HashJoinTuple));
808  lbuckets = 1L << my_log2(hash_table_bytes / bucket_size);
809  lbuckets = Min(lbuckets, max_pointers);
810  nbuckets = (int) lbuckets;
811  nbuckets = 1 << my_log2(nbuckets);
812  bucket_bytes = nbuckets * sizeof(HashJoinTuple);
813 
814  /*
815  * Buckets are simple pointers to hashjoin tuples, while tupsize
816  * includes the pointer, hash code, and MinimalTupleData. So buckets
817  * should never really exceed 25% of work_mem (even for
818  * NTUP_PER_BUCKET=1); except maybe for work_mem values that are not
819  * 2^N bytes, where we might get more because of doubling. So let's
820  * look for 50% here.
821  */
822  Assert(bucket_bytes <= hash_table_bytes / 2);
823 
824  /* Calculate required number of batches. */
825  dbatch = ceil(inner_rel_bytes / (hash_table_bytes - bucket_bytes));
826  dbatch = Min(dbatch, max_pointers);
827  minbatch = (int) dbatch;
828  nbatch = 2;
829  while (nbatch < minbatch)
830  nbatch <<= 1;
831  }
832 
833  Assert(nbuckets > 0);
834  Assert(nbatch > 0);
835 
836  *numbuckets = nbuckets;
837  *numbatches = nbatch;
838 }
#define SKEW_BUCKET_OVERHEAD
Definition: hashjoin.h:108
#define Min(x, y)
Definition: c.h:846
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
int my_log2(long num)
Definition: dynahash.c:1716
#define MaxAllocSize
Definition: memutils.h:40
#define SizeofMinimalTupleHeader
Definition: htup_details.h:655
#define NTUP_PER_BUCKET
Definition: nodeHash.c:658
int work_mem
Definition: globals.c:113
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
#define Max(x, y)
Definition: c.h:840
#define Assert(condition)
Definition: c.h:688
#define MAXALIGN(LEN)
Definition: c.h:641
#define SKEW_WORK_MEM_PERCENT
Definition: hashjoin.h:110
void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew, bool try_combined_work_mem, int parallel_workers, size_t *space_allowed, int *numbuckets, int *numbatches, int *num_skew_mcvs)
Definition: nodeHash.c:661

◆ ExecEndHash()

void ExecEndHash ( HashState node)

Definition at line 404 of file nodeHash.c.

References ExecEndNode(), ExecFreeExprContext(), outerPlan, outerPlanState, and HashState::ps.

Referenced by ExecEndNode().

405 {
407 
408  /*
409  * free exprcontext
410  */
411  ExecFreeExprContext(&node->ps);
412 
413  /*
414  * shut down the subplan
415  */
416  outerPlan = outerPlanState(node);
417  ExecEndNode(outerPlan);
418 }
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:539
void ExecFreeExprContext(PlanState *planstate)
Definition: execUtils.c:561
#define outerPlanState(node)
Definition: execnodes.h:914
PlanState ps
Definition: execnodes.h:2058
#define outerPlan(node)
Definition: plannodes.h:174

◆ ExecHash()

static TupleTableSlot* ExecHash ( PlanState pstate)
static

Definition at line 91 of file nodeHash.c.

References elog, and ERROR.

Referenced by ExecInitHash().

92 {
93  elog(ERROR, "Hash node does not support ExecProcNode call convention");
94  return NULL;
95 }
#define ERROR
Definition: elog.h:43
#define elog
Definition: elog.h:219

◆ ExecHashBuildSkewHash()

static void ExecHashBuildSkewHash ( HashJoinTable  hashtable,
Hash node,
int  mcvsToUse 
)
static

Definition at line 2188 of file nodeHash.c.

References ATTSTATSSLOT_NUMBERS, ATTSTATSSLOT_VALUES, HashJoinTableData::batchCxt, BoolGetDatum, DatumGetUInt32, free_attstatsslot(), FunctionCall1, get_attstatsslot(), HashSkewBucket::hashvalue, HeapTupleIsValid, i, Int16GetDatum, InvalidOid, MemoryContextAlloc(), MemoryContextAllocZero(), HashJoinTableData::nSkewBuckets, AttStatsSlot::numbers, AttStatsSlot::nvalues, ObjectIdGetDatum, OidIsValid, HashJoinTableData::outer_hashfunctions, ReleaseSysCache(), SearchSysCache3(), SKEW_BUCKET_OVERHEAD, SKEW_MIN_OUTER_FRACTION, HashJoinTableData::skewBucket, HashJoinTableData::skewBucketLen, HashJoinTableData::skewBucketNums, Hash::skewColumn, HashJoinTableData::skewEnabled, Hash::skewInherit, Hash::skewTable, HashJoinTableData::spacePeak, HashJoinTableData::spaceUsed, HashJoinTableData::spaceUsedSkew, STATISTIC_KIND_MCV, STATRELATTINH, HashSkewBucket::tuples, and AttStatsSlot::values.

Referenced by ExecHashTableCreate().

2189 {
2190  HeapTupleData *statsTuple;
2191  AttStatsSlot sslot;
2192 
2193  /* Do nothing if planner didn't identify the outer relation's join key */
2194  if (!OidIsValid(node->skewTable))
2195  return;
2196  /* Also, do nothing if we don't have room for at least one skew bucket */
2197  if (mcvsToUse <= 0)
2198  return;
2199 
2200  /*
2201  * Try to find the MCV statistics for the outer relation's join key.
2202  */
2203  statsTuple = SearchSysCache3(STATRELATTINH,
2204  ObjectIdGetDatum(node->skewTable),
2205  Int16GetDatum(node->skewColumn),
2206  BoolGetDatum(node->skewInherit));
2207  if (!HeapTupleIsValid(statsTuple))
2208  return;
2209 
2210  if (get_attstatsslot(&sslot, statsTuple,
2213  {
2214  double frac;
2215  int nbuckets;
2216  FmgrInfo *hashfunctions;
2217  int i;
2218 
2219  if (mcvsToUse > sslot.nvalues)
2220  mcvsToUse = sslot.nvalues;
2221 
2222  /*
2223  * Calculate the expected fraction of outer relation that will
2224  * participate in the skew optimization. If this isn't at least
2225  * SKEW_MIN_OUTER_FRACTION, don't use skew optimization.
2226  */
2227  frac = 0;
2228  for (i = 0; i < mcvsToUse; i++)
2229  frac += sslot.numbers[i];
2230  if (frac < SKEW_MIN_OUTER_FRACTION)
2231  {
2232  free_attstatsslot(&sslot);
2233  ReleaseSysCache(statsTuple);
2234  return;
2235  }
2236 
2237  /*
2238  * Okay, set up the skew hashtable.
2239  *
2240  * skewBucket[] is an open addressing hashtable with a power of 2 size
2241  * that is greater than the number of MCV values. (This ensures there
2242  * will be at least one null entry, so searches will always
2243  * terminate.)
2244  *
2245  * Note: this code could fail if mcvsToUse exceeds INT_MAX/8 or
2246  * MaxAllocSize/sizeof(void *)/8, but that is not currently possible
2247  * since we limit pg_statistic entries to much less than that.
2248  */
2249  nbuckets = 2;
2250  while (nbuckets <= mcvsToUse)
2251  nbuckets <<= 1;
2252  /* use two more bits just to help avoid collisions */
2253  nbuckets <<= 2;
2254 
2255  hashtable->skewEnabled = true;
2256  hashtable->skewBucketLen = nbuckets;
2257 
2258  /*
2259  * We allocate the bucket memory in the hashtable's batch context. It
2260  * is only needed during the first batch, and this ensures it will be
2261  * automatically removed once the first batch is done.
2262  */
2263  hashtable->skewBucket = (HashSkewBucket **)
2264  MemoryContextAllocZero(hashtable->batchCxt,
2265  nbuckets * sizeof(HashSkewBucket *));
2266  hashtable->skewBucketNums = (int *)
2267  MemoryContextAllocZero(hashtable->batchCxt,
2268  mcvsToUse * sizeof(int));
2269 
2270  hashtable->spaceUsed += nbuckets * sizeof(HashSkewBucket *)
2271  + mcvsToUse * sizeof(int);
2272  hashtable->spaceUsedSkew += nbuckets * sizeof(HashSkewBucket *)
2273  + mcvsToUse * sizeof(int);
2274  if (hashtable->spaceUsed > hashtable->spacePeak)
2275  hashtable->spacePeak = hashtable->spaceUsed;
2276 
2277  /*
2278  * Create a skew bucket for each MCV hash value.
2279  *
2280  * Note: it is very important that we create the buckets in order of
2281  * decreasing MCV frequency. If we have to remove some buckets, they
2282  * must be removed in reverse order of creation (see notes in
2283  * ExecHashRemoveNextSkewBucket) and we want the least common MCVs to
2284  * be removed first.
2285  */
2286  hashfunctions = hashtable->outer_hashfunctions;
2287 
2288  for (i = 0; i < mcvsToUse; i++)
2289  {
2290  uint32 hashvalue;
2291  int bucket;
2292 
2293  hashvalue = DatumGetUInt32(FunctionCall1(&hashfunctions[0],
2294  sslot.values[i]));
2295 
2296  /*
2297  * While we have not hit a hole in the hashtable and have not hit
2298  * the desired bucket, we have collided with some previous hash
2299  * value, so try the next bucket location. NB: this code must
2300  * match ExecHashGetSkewBucket.
2301  */
2302  bucket = hashvalue & (nbuckets - 1);
2303  while (hashtable->skewBucket[bucket] != NULL &&
2304  hashtable->skewBucket[bucket]->hashvalue != hashvalue)
2305  bucket = (bucket + 1) & (nbuckets - 1);
2306 
2307  /*
2308  * If we found an existing bucket with the same hashvalue, leave
2309  * it alone. It's okay for two MCVs to share a hashvalue.
2310  */
2311  if (hashtable->skewBucket[bucket] != NULL)
2312  continue;
2313 
2314  /* Okay, create a new skew bucket for this hashvalue. */
2315  hashtable->skewBucket[bucket] = (HashSkewBucket *)
2316  MemoryContextAlloc(hashtable->batchCxt,
2317  sizeof(HashSkewBucket));
2318  hashtable->skewBucket[bucket]->hashvalue = hashvalue;
2319  hashtable->skewBucket[bucket]->tuples = NULL;
2320  hashtable->skewBucketNums[hashtable->nSkewBuckets] = bucket;
2321  hashtable->nSkewBuckets++;
2322  hashtable->spaceUsed += SKEW_BUCKET_OVERHEAD;
2323  hashtable->spaceUsedSkew += SKEW_BUCKET_OVERHEAD;
2324  if (hashtable->spaceUsed > hashtable->spacePeak)
2325  hashtable->spacePeak = hashtable->spaceUsed;
2326  }
2327 
2328  free_attstatsslot(&sslot);
2329  }
2330 
2331  ReleaseSysCache(statsTuple);
2332 }
Oid skewTable
Definition: plannodes.h:886
#define DatumGetUInt32(X)
Definition: postgres.h:469
Definition: fmgr.h:56
#define SKEW_BUCKET_OVERHEAD
Definition: hashjoin.h:108
bool skewInherit
Definition: plannodes.h:888
#define SKEW_MIN_OUTER_FRACTION
Definition: hashjoin.h:111
#define ATTSTATSSLOT_VALUES
Definition: lsyscache.h:39
#define Int16GetDatum(X)
Definition: postgres.h:434
#define OidIsValid(objectId)
Definition: c.h:594
int * skewBucketNums
Definition: hashjoin.h:308
#define ATTSTATSSLOT_NUMBERS
Definition: lsyscache.h:40
#define ObjectIdGetDatum(X)
Definition: postgres.h:490
HeapTuple SearchSysCache3(int cacheId, Datum key1, Datum key2, Datum key3)
Definition: syscache.c:1134
float4 * numbers
Definition: lsyscache.h:52
AttrNumber skewColumn
Definition: plannodes.h:887
HashJoinTuple tuples
Definition: hashjoin.h:105
unsigned int uint32
Definition: c.h:314
#define STATISTIC_KIND_MCV
Definition: pg_statistic.h:202
MemoryContext batchCxt
Definition: hashjoin.h:348
FmgrInfo * outer_hashfunctions
Definition: hashjoin.h:337
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:1160
void * MemoryContextAllocZero(MemoryContext context, Size size)
Definition: mcxt.c:728
#define BoolGetDatum(X)
Definition: postgres.h:385
#define InvalidOid
Definition: postgres_ext.h:36
uint32 hashvalue
Definition: hashjoin.h:104
#define HeapTupleIsValid(tuple)
Definition: htup.h:77
bool get_attstatsslot(AttStatsSlot *sslot, HeapTuple statstuple, int reqkind, Oid reqop, int flags)
Definition: lsyscache.c:2913
Datum * values
Definition: lsyscache.h:49
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:693
int i
#define FunctionCall1(flinfo, arg1)
Definition: fmgr.h:603
void free_attstatsslot(AttStatsSlot *sslot)
Definition: lsyscache.c:3029

◆ ExecHashEstimate()

void ExecHashEstimate ( HashState node,
ParallelContext pcxt 
)

Definition at line 2543 of file nodeHash.c.

References add_size(), ParallelContext::estimator, PlanState::instrument, mul_size(), ParallelContext::nworkers, offsetof, HashState::ps, shm_toc_estimate_chunk, and shm_toc_estimate_keys.

Referenced by ExecParallelEstimate().

2544 {
2545  size_t size;
2546 
2547  /* don't need this if not instrumenting or no workers */
2548  if (!node->ps.instrument || pcxt->nworkers == 0)
2549  return;
2550 
2551  size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
2552  size = add_size(size, offsetof(SharedHashInfo, hinstrument));
2553  shm_toc_estimate_chunk(&pcxt->estimator, size);
2554  shm_toc_estimate_keys(&pcxt->estimator, 1);
2555 }
Instrumentation * instrument
Definition: execnodes.h:878
shm_toc_estimator estimator
Definition: parallel.h:41
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
PlanState ps
Definition: execnodes.h:2058
Size mul_size(Size s1, Size s2)
Definition: shmem.c:492
Size add_size(Size s1, Size s2)
Definition: shmem.c:475
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
#define offsetof(type, field)
Definition: c.h:611

◆ ExecHashGetBucketAndBatch()

void ExecHashGetBucketAndBatch ( HashJoinTable  hashtable,
uint32  hashvalue,
int *  bucketno,
int *  batchno 
)

Definition at line 1874 of file nodeHash.c.

References HashJoinTableData::log2_nbuckets, HashJoinTableData::nbatch, and HashJoinTableData::nbuckets.

Referenced by ExecHashIncreaseNumBatches(), ExecHashIncreaseNumBuckets(), ExecHashJoinImpl(), ExecHashRemoveNextSkewBucket(), ExecHashTableInsert(), ExecParallelHashIncreaseNumBuckets(), ExecParallelHashJoinPartitionOuter(), ExecParallelHashRepartitionFirst(), ExecParallelHashRepartitionRest(), ExecParallelHashTableInsert(), and ExecParallelHashTableInsertCurrentBatch().

1878 {
1879  uint32 nbuckets = (uint32) hashtable->nbuckets;
1880  uint32 nbatch = (uint32) hashtable->nbatch;
1881 
1882  if (nbatch > 1)
1883  {
1884  /* we can do MOD by masking, DIV by shifting */
1885  *bucketno = hashvalue & (nbuckets - 1);
1886  *batchno = (hashvalue >> hashtable->log2_nbuckets) & (nbatch - 1);
1887  }
1888  else
1889  {
1890  *bucketno = hashvalue & (nbuckets - 1);
1891  *batchno = 0;
1892  }
1893 }
unsigned int uint32
Definition: c.h:314

◆ ExecHashGetHashValue()

bool ExecHashGetHashValue ( HashJoinTable  hashtable,
ExprContext econtext,
List hashkeys,
bool  outer_tuple,
bool  keep_nulls,
uint32 hashvalue 
)

Definition at line 1770 of file nodeHash.c.

References DatumGetUInt32, ExprContext::ecxt_per_tuple_memory, ExecEvalExpr(), FunctionCall1, HashJoinTableData::hashStrict, i, HashJoinTableData::inner_hashfunctions, lfirst, MemoryContextSwitchTo(), HashJoinTableData::outer_hashfunctions, and ResetExprContext.

Referenced by ExecHashJoinOuterGetTuple(), ExecParallelHashJoinOuterGetTuple(), ExecParallelHashJoinPartitionOuter(), MultiExecParallelHash(), and MultiExecPrivateHash().

1776 {
1777  uint32 hashkey = 0;
1778  FmgrInfo *hashfunctions;
1779  ListCell *hk;
1780  int i = 0;
1781  MemoryContext oldContext;
1782 
1783  /*
1784  * We reset the eval context each time to reclaim any memory leaked in the
1785  * hashkey expressions.
1786  */
1787  ResetExprContext(econtext);
1788 
1789  oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
1790 
1791  if (outer_tuple)
1792  hashfunctions = hashtable->outer_hashfunctions;
1793  else
1794  hashfunctions = hashtable->inner_hashfunctions;
1795 
1796  foreach(hk, hashkeys)
1797  {
1798  ExprState *keyexpr = (ExprState *) lfirst(hk);
1799  Datum keyval;
1800  bool isNull;
1801 
1802  /* rotate hashkey left 1 bit at each step */
1803  hashkey = (hashkey << 1) | ((hashkey & 0x80000000) ? 1 : 0);
1804 
1805  /*
1806  * Get the join attribute value of the tuple
1807  */
1808  keyval = ExecEvalExpr(keyexpr, econtext, &isNull);
1809 
1810  /*
1811  * If the attribute is NULL, and the join operator is strict, then
1812  * this tuple cannot pass the join qual so we can reject it
1813  * immediately (unless we're scanning the outside of an outer join, in
1814  * which case we must not reject it). Otherwise we act like the
1815  * hashcode of NULL is zero (this will support operators that act like
1816  * IS NOT DISTINCT, though not any more-random behavior). We treat
1817  * the hash support function as strict even if the operator is not.
1818  *
1819  * Note: currently, all hashjoinable operators must be strict since
1820  * the hash index AM assumes that. However, it takes so little extra
1821  * code here to allow non-strict that we may as well do it.
1822  */
1823  if (isNull)
1824  {
1825  if (hashtable->hashStrict[i] && !keep_nulls)
1826  {
1827  MemoryContextSwitchTo(oldContext);
1828  return false; /* cannot match */
1829  }
1830  /* else, leave hashkey unmodified, equivalent to hashcode 0 */
1831  }
1832  else
1833  {
1834  /* Compute the hash function */
1835  uint32 hkey;
1836 
1837  hkey = DatumGetUInt32(FunctionCall1(&hashfunctions[i], keyval));
1838  hashkey ^= hkey;
1839  }
1840 
1841  i++;
1842  }
1843 
1844  MemoryContextSwitchTo(oldContext);
1845 
1846  *hashvalue = hashkey;
1847  return true;
1848 }
#define DatumGetUInt32(X)
Definition: postgres.h:469
Definition: fmgr.h:56
MemoryContext ecxt_per_tuple_memory
Definition: execnodes.h:217
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
FmgrInfo * inner_hashfunctions
Definition: hashjoin.h:338
static Datum ExecEvalExpr(ExprState *state, ExprContext *econtext, bool *isNull)
Definition: executor.h:282
unsigned int uint32
Definition: c.h:314
FmgrInfo * outer_hashfunctions
Definition: hashjoin.h:337
uintptr_t Datum
Definition: postgres.h:365
#define lfirst(lc)
Definition: pg_list.h:106
int i
#define FunctionCall1(flinfo, arg1)
Definition: fmgr.h:603
bool * hashStrict
Definition: hashjoin.h:339
#define ResetExprContext(econtext)
Definition: executor.h:484

◆ ExecHashGetInstrumentation()

void ExecHashGetInstrumentation ( HashInstrumentation instrument,
HashJoinTable  hashtable 
)

◆ ExecHashGetSkewBucket()

int ExecHashGetSkewBucket ( HashJoinTable  hashtable,
uint32  hashvalue 
)

Definition at line 2342 of file nodeHash.c.

References HashSkewBucket::hashvalue, INVALID_SKEW_BUCKET_NO, HashJoinTableData::skewBucket, HashJoinTableData::skewBucketLen, and HashJoinTableData::skewEnabled.

Referenced by ExecHashJoinImpl(), and MultiExecPrivateHash().

2343 {
2344  int bucket;
2345 
2346  /*
2347  * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
2348  * particular, this happens after the initial batch is done).
2349  */
2350  if (!hashtable->skewEnabled)
2351  return INVALID_SKEW_BUCKET_NO;
2352 
2353  /*
2354  * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
2355  */
2356  bucket = hashvalue & (hashtable->skewBucketLen - 1);
2357 
2358  /*
2359  * While we have not hit a hole in the hashtable and have not hit the
2360  * desired bucket, we have collided with some other hash value, so try the
2361  * next bucket location.
2362  */
2363  while (hashtable->skewBucket[bucket] != NULL &&
2364  hashtable->skewBucket[bucket]->hashvalue != hashvalue)
2365  bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
2366 
2367  /*
2368  * Found the desired bucket?
2369  */
2370  if (hashtable->skewBucket[bucket] != NULL)
2371  return bucket;
2372 
2373  /*
2374  * There must not be any hashtable entry for this hash value.
2375  */
2376  return INVALID_SKEW_BUCKET_NO;
2377 }
#define INVALID_SKEW_BUCKET_NO
Definition: hashjoin.h:109
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
uint32 hashvalue
Definition: hashjoin.h:104

◆ ExecHashIncreaseNumBatches()

static void ExecHashIncreaseNumBatches ( HashJoinTable  hashtable)
static

Definition at line 881 of file nodeHash.c.

References Assert, HashJoinTableData::buckets, CHECK_FOR_INTERRUPTS, HashJoinTableData::chunks, HashJoinTableData::curbatch, dense_alloc(), ExecHashGetBucketAndBatch(), ExecHashJoinSaveTuple(), HashJoinTableData::growEnabled, HASH_CHUNK_DATA, HashJoinTableData::hashCxt, HashJoinTupleData::hashvalue, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, idx(), HashJoinTableData::innerBatchFile, HashJoinTableData::log2_nbuckets, HashJoinTableData::log2_nbuckets_optimal, MAXALIGN, MaxAllocSize, MemoryContextSwitchTo(), MemSet, Min, HashJoinTableData::nbatch, HashJoinTableData::nbuckets, HashJoinTableData::nbuckets_optimal, HashJoinTupleData::next, HashMemoryChunkData::next, HashJoinTableData::outerBatchFile, palloc0(), pfree(), PrepareTempTablespaces(), repalloc(), HashJoinTableData::spaceUsed, MinimalTupleData::t_len, HashJoinTupleData::unshared, HashMemoryChunkData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashSkewTableInsert(), and ExecHashTableInsert().

882 {
883  int oldnbatch = hashtable->nbatch;
884  int curbatch = hashtable->curbatch;
885  int nbatch;
886  MemoryContext oldcxt;
887  long ninmemory;
888  long nfreed;
889  HashMemoryChunk oldchunks;
890 
891  /* do nothing if we've decided to shut off growth */
892  if (!hashtable->growEnabled)
893  return;
894 
895  /* safety check to avoid overflow */
896  if (oldnbatch > Min(INT_MAX / 2, MaxAllocSize / (sizeof(void *) * 2)))
897  return;
898 
899  nbatch = oldnbatch * 2;
900  Assert(nbatch > 1);
901 
902 #ifdef HJDEBUG
903  printf("Hashjoin %p: increasing nbatch to %d because space = %zu\n",
904  hashtable, nbatch, hashtable->spaceUsed);
905 #endif
906 
907  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
908 
909  if (hashtable->innerBatchFile == NULL)
910  {
911  /* we had no file arrays before */
912  hashtable->innerBatchFile = (BufFile **)
913  palloc0(nbatch * sizeof(BufFile *));
914  hashtable->outerBatchFile = (BufFile **)
915  palloc0(nbatch * sizeof(BufFile *));
916  /* time to establish the temp tablespaces, too */
918  }
919  else
920  {
921  /* enlarge arrays and zero out added entries */
922  hashtable->innerBatchFile = (BufFile **)
923  repalloc(hashtable->innerBatchFile, nbatch * sizeof(BufFile *));
924  hashtable->outerBatchFile = (BufFile **)
925  repalloc(hashtable->outerBatchFile, nbatch * sizeof(BufFile *));
926  MemSet(hashtable->innerBatchFile + oldnbatch, 0,
927  (nbatch - oldnbatch) * sizeof(BufFile *));
928  MemSet(hashtable->outerBatchFile + oldnbatch, 0,
929  (nbatch - oldnbatch) * sizeof(BufFile *));
930  }
931 
932  MemoryContextSwitchTo(oldcxt);
933 
934  hashtable->nbatch = nbatch;
935 
936  /*
937  * Scan through the existing hash table entries and dump out any that are
938  * no longer of the current batch.
939  */
940  ninmemory = nfreed = 0;
941 
942  /* If know we need to resize nbuckets, we can do it while rebatching. */
943  if (hashtable->nbuckets_optimal != hashtable->nbuckets)
944  {
945  /* we never decrease the number of buckets */
946  Assert(hashtable->nbuckets_optimal > hashtable->nbuckets);
947 
948  hashtable->nbuckets = hashtable->nbuckets_optimal;
949  hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
950 
951  hashtable->buckets.unshared =
952  repalloc(hashtable->buckets.unshared,
953  sizeof(HashJoinTuple) * hashtable->nbuckets);
954  }
955 
956  /*
957  * We will scan through the chunks directly, so that we can reset the
958  * buckets now and not have to keep track which tuples in the buckets have
959  * already been processed. We will free the old chunks as we go.
960  */
961  memset(hashtable->buckets.unshared, 0,
962  sizeof(HashJoinTuple) * hashtable->nbuckets);
963  oldchunks = hashtable->chunks;
964  hashtable->chunks = NULL;
965 
966  /* so, let's scan through the old chunks, and all tuples in each chunk */
967  while (oldchunks != NULL)
968  {
969  HashMemoryChunk nextchunk = oldchunks->next.unshared;
970 
971  /* position within the buffer (up to oldchunks->used) */
972  size_t idx = 0;
973 
974  /* process all tuples stored in this chunk (and then free it) */
975  while (idx < oldchunks->used)
976  {
977  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(oldchunks) + idx);
978  MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
979  int hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len);
980  int bucketno;
981  int batchno;
982 
983  ninmemory++;
984  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
985  &bucketno, &batchno);
986 
987  if (batchno == curbatch)
988  {
989  /* keep tuple in memory - copy it into the new chunk */
990  HashJoinTuple copyTuple;
991 
992  copyTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
993  memcpy(copyTuple, hashTuple, hashTupleSize);
994 
995  /* and add it back to the appropriate bucket */
996  copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
997  hashtable->buckets.unshared[bucketno] = copyTuple;
998  }
999  else
1000  {
1001  /* dump it out */
1002  Assert(batchno > curbatch);
1004  hashTuple->hashvalue,
1005  &hashtable->innerBatchFile[batchno]);
1006 
1007  hashtable->spaceUsed -= hashTupleSize;
1008  nfreed++;
1009  }
1010 
1011  /* next tuple in this chunk */
1012  idx += MAXALIGN(hashTupleSize);
1013 
1014  /* allow this loop to be cancellable */
1016  }
1017 
1018  /* we're done with this chunk - free it and proceed to the next one */
1019  pfree(oldchunks);
1020  oldchunks = nextchunk;
1021  }
1022 
1023 #ifdef HJDEBUG
1024  printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n",
1025  hashtable, nfreed, ninmemory, hashtable->spaceUsed);
1026 #endif
1027 
1028  /*
1029  * If we dumped out either all or none of the tuples in the table, disable
1030  * further expansion of nbatch. This situation implies that we have
1031  * enough tuples of identical hashvalues to overflow spaceAllowed.
1032  * Increasing nbatch will not fix it since there's no way to subdivide the
1033  * group any more finely. We have to just gut it out and hope the server
1034  * has enough RAM.
1035  */
1036  if (nfreed == 0 || nfreed == ninmemory)
1037  {
1038  hashtable->growEnabled = false;
1039 #ifdef HJDEBUG
1040  printf("Hashjoin %p: disabling further increase of nbatch\n",
1041  hashtable);
1042 #endif
1043  }
1044 }
int log2_nbuckets_optimal
Definition: hashjoin.h:291
union HashJoinTupleData::@95 next
#define Min(x, y)
Definition: c.h:846
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
#define MemSet(start, val, len)
Definition: c.h:897
Datum idx(PG_FUNCTION_ARGS)
Definition: _int_op.c:264
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
void pfree(void *pointer)
Definition: mcxt.c:936
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
void PrepareTempTablespaces(void)
Definition: tablespace.c:1287
union HashMemoryChunkData::@96 next
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
BufFile ** outerBatchFile
Definition: hashjoin.h:330
struct HashMemoryChunkData * unshared
Definition: hashjoin.h:126
#define MaxAllocSize
Definition: memutils.h:40
void * palloc0(Size size)
Definition: mcxt.c:864
static void * dense_alloc(HashJoinTable hashtable, Size size)
Definition: nodeHash.c:2649
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
BufFile ** innerBatchFile
Definition: hashjoin.h:329
#define MAXALIGN(LEN)
Definition: c.h:641
union HashJoinTableData::@97 buckets
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:949
HashMemoryChunk chunks
Definition: hashjoin.h:351
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
MemoryContext hashCxt
Definition: hashjoin.h:347
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, BufFile **fileptr)
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecHashIncreaseNumBuckets()

static void ExecHashIncreaseNumBuckets ( HashJoinTable  hashtable)
static

Definition at line 1426 of file nodeHash.c.

References Assert, HashJoinTableData::buckets, CHECK_FOR_INTERRUPTS, HashJoinTableData::chunks, ExecHashGetBucketAndBatch(), HASH_CHUNK_DATA, HashJoinTupleData::hashvalue, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, idx(), HashJoinTableData::log2_nbuckets, HashJoinTableData::log2_nbuckets_optimal, MAXALIGN, HashJoinTableData::nbuckets, HashJoinTableData::nbuckets_optimal, HashJoinTupleData::next, HashMemoryChunkData::next, repalloc(), HashJoinTupleData::unshared, HashMemoryChunkData::unshared, and HashJoinTableData::unshared.

Referenced by MultiExecPrivateHash().

1427 {
1428  HashMemoryChunk chunk;
1429 
1430  /* do nothing if not an increase (it's called increase for a reason) */
1431  if (hashtable->nbuckets >= hashtable->nbuckets_optimal)
1432  return;
1433 
1434 #ifdef HJDEBUG
1435  printf("Hashjoin %p: increasing nbuckets %d => %d\n",
1436  hashtable, hashtable->nbuckets, hashtable->nbuckets_optimal);
1437 #endif
1438 
1439  hashtable->nbuckets = hashtable->nbuckets_optimal;
1440  hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
1441 
1442  Assert(hashtable->nbuckets > 1);
1443  Assert(hashtable->nbuckets <= (INT_MAX / 2));
1444  Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets));
1445 
1446  /*
1447  * Just reallocate the proper number of buckets - we don't need to walk
1448  * through them - we can walk the dense-allocated chunks (just like in
1449  * ExecHashIncreaseNumBatches, but without all the copying into new
1450  * chunks)
1451  */
1452  hashtable->buckets.unshared =
1453  (HashJoinTuple *) repalloc(hashtable->buckets.unshared,
1454  hashtable->nbuckets * sizeof(HashJoinTuple));
1455 
1456  memset(hashtable->buckets.unshared, 0,
1457  hashtable->nbuckets * sizeof(HashJoinTuple));
1458 
1459  /* scan through all tuples in all chunks to rebuild the hash table */
1460  for (chunk = hashtable->chunks; chunk != NULL; chunk = chunk->next.unshared)
1461  {
1462  /* process all tuples stored in this chunk */
1463  size_t idx = 0;
1464 
1465  while (idx < chunk->used)
1466  {
1467  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1468  int bucketno;
1469  int batchno;
1470 
1471  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1472  &bucketno, &batchno);
1473 
1474  /* add the tuple to the proper bucket */
1475  hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1476  hashtable->buckets.unshared[bucketno] = hashTuple;
1477 
1478  /* advance index past the tuple */
1479  idx += MAXALIGN(HJTUPLE_OVERHEAD +
1480  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1481  }
1482 
1483  /* allow this loop to be cancellable */
1485  }
1486 }
int log2_nbuckets_optimal
Definition: hashjoin.h:291
union HashJoinTupleData::@95 next
Datum idx(PG_FUNCTION_ARGS)
Definition: _int_op.c:264
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
union HashMemoryChunkData::@96 next
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
struct HashMemoryChunkData * unshared
Definition: hashjoin.h:126
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
#define MAXALIGN(LEN)
Definition: c.h:641
union HashJoinTableData::@97 buckets
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:949
HashMemoryChunk chunks
Definition: hashjoin.h:351
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecHashInitializeDSM()

void ExecHashInitializeDSM ( HashState node,
ParallelContext pcxt 
)

Definition at line 2562 of file nodeHash.c.

References PlanState::instrument, SharedHashInfo::num_workers, ParallelContext::nworkers, offsetof, PlanState::plan, Plan::plan_node_id, HashState::ps, HashState::shared_info, shm_toc_allocate(), shm_toc_insert(), and ParallelContext::toc.

Referenced by ExecParallelInitializeDSM().

2563 {
2564  size_t size;
2565 
2566  /* don't need this if not instrumenting or no workers */
2567  if (!node->ps.instrument || pcxt->nworkers == 0)
2568  return;
2569 
2570  size = offsetof(SharedHashInfo, hinstrument) +
2571  pcxt->nworkers * sizeof(HashInstrumentation);
2572  node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
2573  memset(node->shared_info, 0, size);
2574  node->shared_info->num_workers = pcxt->nworkers;
2575  shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id,
2576  node->shared_info);
2577 }
Instrumentation * instrument
Definition: execnodes.h:878
struct HashInstrumentation HashInstrumentation
int plan_node_id
Definition: plannodes.h:143
SharedHashInfo * shared_info
Definition: execnodes.h:2063
PlanState ps
Definition: execnodes.h:2058
Plan * plan
Definition: execnodes.h:868
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
#define offsetof(type, field)
Definition: c.h:611
shm_toc * toc
Definition: parallel.h:44

◆ ExecHashInitializeWorker()

void ExecHashInitializeWorker ( HashState node,
ParallelWorkerContext pwcxt 
)

Definition at line 2584 of file nodeHash.c.

References SharedHashInfo::hinstrument, HashState::hinstrument, PlanState::instrument, ParallelWorkerNumber, PlanState::plan, Plan::plan_node_id, HashState::ps, shm_toc_lookup(), and ParallelWorkerContext::toc.

Referenced by ExecParallelInitializeWorker().

2585 {
2586  SharedHashInfo *shared_info;
2587 
2588  /* don't need this if not instrumenting */
2589  if (!node->ps.instrument)
2590  return;
2591 
2592  shared_info = (SharedHashInfo *)
2593  shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
2594  node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
2595 }
Instrumentation * instrument
Definition: execnodes.h:878
int plan_node_id
Definition: plannodes.h:143
int ParallelWorkerNumber
Definition: parallel.c:103
PlanState ps
Definition: execnodes.h:2058
HashInstrumentation * hinstrument
Definition: execnodes.h:2064
HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER]
Definition: execnodes.h:2049
Plan * plan
Definition: execnodes.h:868
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232

◆ ExecHashRemoveNextSkewBucket()

static void ExecHashRemoveNextSkewBucket ( HashJoinTable  hashtable)
static

Definition at line 2430 of file nodeHash.c.

References Assert, HashJoinTableData::buckets, CHECK_FOR_INTERRUPTS, HashJoinTableData::curbatch, dense_alloc(), ExecHashGetBucketAndBatch(), ExecHashJoinSaveTuple(), HashSkewBucket::hashvalue, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, HashJoinTableData::innerBatchFile, HashJoinTupleData::next, HashJoinTableData::nSkewBuckets, pfree(), SKEW_BUCKET_OVERHEAD, HashJoinTableData::skewBucket, HashJoinTableData::skewBucketNums, HashJoinTableData::skewEnabled, HashJoinTableData::spaceUsed, HashJoinTableData::spaceUsedSkew, MinimalTupleData::t_len, HashSkewBucket::tuples, HashJoinTupleData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashSkewTableInsert().

2431 {
2432  int bucketToRemove;
2433  HashSkewBucket *bucket;
2434  uint32 hashvalue;
2435  int bucketno;
2436  int batchno;
2437  HashJoinTuple hashTuple;
2438 
2439  /* Locate the bucket to remove */
2440  bucketToRemove = hashtable->skewBucketNums[hashtable->nSkewBuckets - 1];
2441  bucket = hashtable->skewBucket[bucketToRemove];
2442 
2443  /*
2444  * Calculate which bucket and batch the tuples belong to in the main
2445  * hashtable. They all have the same hash value, so it's the same for all
2446  * of them. Also note that it's not possible for nbatch to increase while
2447  * we are processing the tuples.
2448  */
2449  hashvalue = bucket->hashvalue;
2450  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
2451 
2452  /* Process all tuples in the bucket */
2453  hashTuple = bucket->tuples;
2454  while (hashTuple != NULL)
2455  {
2456  HashJoinTuple nextHashTuple = hashTuple->next.unshared;
2457  MinimalTuple tuple;
2458  Size tupleSize;
2459 
2460  /*
2461  * This code must agree with ExecHashTableInsert. We do not use
2462  * ExecHashTableInsert directly as ExecHashTableInsert expects a
2463  * TupleTableSlot while we already have HashJoinTuples.
2464  */
2465  tuple = HJTUPLE_MINTUPLE(hashTuple);
2466  tupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
2467 
2468  /* Decide whether to put the tuple in the hash table or a temp file */
2469  if (batchno == hashtable->curbatch)
2470  {
2471  /* Move the tuple to the main hash table */
2472  HashJoinTuple copyTuple;
2473 
2474  /*
2475  * We must copy the tuple into the dense storage, else it will not
2476  * be found by, eg, ExecHashIncreaseNumBatches.
2477  */
2478  copyTuple = (HashJoinTuple) dense_alloc(hashtable, tupleSize);
2479  memcpy(copyTuple, hashTuple, tupleSize);
2480  pfree(hashTuple);
2481 
2482  copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
2483  hashtable->buckets.unshared[bucketno] = copyTuple;
2484 
2485  /* We have reduced skew space, but overall space doesn't change */
2486  hashtable->spaceUsedSkew -= tupleSize;
2487  }
2488  else
2489  {
2490  /* Put the tuple into a temp file for later batches */
2491  Assert(batchno > hashtable->curbatch);
2492  ExecHashJoinSaveTuple(tuple, hashvalue,
2493  &hashtable->innerBatchFile[batchno]);
2494  pfree(hashTuple);
2495  hashtable->spaceUsed -= tupleSize;
2496  hashtable->spaceUsedSkew -= tupleSize;
2497  }
2498 
2499  hashTuple = nextHashTuple;
2500 
2501  /* allow this loop to be cancellable */
2503  }
2504 
2505  /*
2506  * Free the bucket struct itself and reset the hashtable entry to NULL.
2507  *
2508  * NOTE: this is not nearly as simple as it looks on the surface, because
2509  * of the possibility of collisions in the hashtable. Suppose that hash
2510  * values A and B collide at a particular hashtable entry, and that A was
2511  * entered first so B gets shifted to a different table entry. If we were
2512  * to remove A first then ExecHashGetSkewBucket would mistakenly start
2513  * reporting that B is not in the hashtable, because it would hit the NULL
2514  * before finding B. However, we always remove entries in the reverse
2515  * order of creation, so this failure cannot happen.
2516  */
2517  hashtable->skewBucket[bucketToRemove] = NULL;
2518  hashtable->nSkewBuckets--;
2519  pfree(bucket);
2520  hashtable->spaceUsed -= SKEW_BUCKET_OVERHEAD;
2521  hashtable->spaceUsedSkew -= SKEW_BUCKET_OVERHEAD;
2522 
2523  /*
2524  * If we have removed all skew buckets then give up on skew optimization.
2525  * Release the arrays since they aren't useful any more.
2526  */
2527  if (hashtable->nSkewBuckets == 0)
2528  {
2529  hashtable->skewEnabled = false;
2530  pfree(hashtable->skewBucket);
2531  pfree(hashtable->skewBucketNums);
2532  hashtable->skewBucket = NULL;
2533  hashtable->skewBucketNums = NULL;
2534  hashtable->spaceUsed -= hashtable->spaceUsedSkew;
2535  hashtable->spaceUsedSkew = 0;
2536  }
2537 }
#define SKEW_BUCKET_OVERHEAD
Definition: hashjoin.h:108
union HashJoinTupleData::@95 next
int * skewBucketNums
Definition: hashjoin.h:308
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
void pfree(void *pointer)
Definition: mcxt.c:936
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
HashJoinTuple tuples
Definition: hashjoin.h:105
unsigned int uint32
Definition: c.h:314
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
static void * dense_alloc(HashJoinTable hashtable, Size size)
Definition: nodeHash.c:2649
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
uint32 hashvalue
Definition: hashjoin.h:104
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
size_t Size
Definition: c.h:422
BufFile ** innerBatchFile
Definition: hashjoin.h:329
union HashJoinTableData::@97 buckets
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, BufFile **fileptr)

◆ ExecHashRetrieveInstrumentation()

void ExecHashRetrieveInstrumentation ( HashState node)

Definition at line 2615 of file nodeHash.c.

References SharedHashInfo::num_workers, offsetof, palloc(), and HashState::shared_info.

Referenced by ExecParallelRetrieveInstrumentation().

2616 {
2617  SharedHashInfo *shared_info = node->shared_info;
2618  size_t size;
2619 
2620  if (shared_info == NULL)
2621  return;
2622 
2623  /* Replace node->shared_info with a copy in backend-local memory. */
2624  size = offsetof(SharedHashInfo, hinstrument) +
2625  shared_info->num_workers * sizeof(HashInstrumentation);
2626  node->shared_info = palloc(size);
2627  memcpy(node->shared_info, shared_info, size);
2628 }
struct HashInstrumentation HashInstrumentation
SharedHashInfo * shared_info
Definition: execnodes.h:2063
void * palloc(Size size)
Definition: mcxt.c:835
#define offsetof(type, field)
Definition: c.h:611

◆ ExecHashSkewTableInsert()

static void ExecHashSkewTableInsert ( HashJoinTable  hashtable,
TupleTableSlot slot,
uint32  hashvalue,
int  bucketNumber 
)
static

Definition at line 2388 of file nodeHash.c.

References Assert, HashJoinTableData::batchCxt, ExecFetchSlotMinimalTuple(), ExecHashIncreaseNumBatches(), ExecHashRemoveNextSkewBucket(), HashJoinTupleData::hashvalue, HeapTupleHeaderClearMatch, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, MemoryContextAlloc(), HashJoinTupleData::next, HashJoinTableData::skewBucket, HashJoinTableData::spaceAllowed, HashJoinTableData::spaceAllowedSkew, HashJoinTableData::spacePeak, HashJoinTableData::spaceUsed, HashJoinTableData::spaceUsedSkew, MinimalTupleData::t_len, HashSkewBucket::tuples, and HashJoinTupleData::unshared.

Referenced by MultiExecPrivateHash().

2392 {
2394  HashJoinTuple hashTuple;
2395  int hashTupleSize;
2396 
2397  /* Create the HashJoinTuple */
2398  hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
2399  hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
2400  hashTupleSize);
2401  hashTuple->hashvalue = hashvalue;
2402  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
2404 
2405  /* Push it onto the front of the skew bucket's list */
2406  hashTuple->next.unshared = hashtable->skewBucket[bucketNumber]->tuples;
2407  hashtable->skewBucket[bucketNumber]->tuples = hashTuple;
2408  Assert(hashTuple != hashTuple->next.unshared);
2409 
2410  /* Account for space used, and back off if we've used too much */
2411  hashtable->spaceUsed += hashTupleSize;
2412  hashtable->spaceUsedSkew += hashTupleSize;
2413  if (hashtable->spaceUsed > hashtable->spacePeak)
2414  hashtable->spacePeak = hashtable->spaceUsed;
2415  while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
2416  ExecHashRemoveNextSkewBucket(hashtable);
2417 
2418  /* Check we are not over the total spaceAllowed, either */
2419  if (hashtable->spaceUsed > hashtable->spaceAllowed)
2420  ExecHashIncreaseNumBatches(hashtable);
2421 }
static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
Definition: nodeHash.c:2430
union HashJoinTupleData::@95 next
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot)
Definition: execTuples.c:688
static void ExecHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:881
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
Size spaceAllowedSkew
Definition: hashjoin.h:345
HashJoinTuple tuples
Definition: hashjoin.h:105
MemoryContext batchCxt
Definition: hashjoin.h:348
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:532
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:693
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecHashTableCreate()

HashJoinTable ExecHashTableCreate ( HashState state,
List hashOperators,
bool  keepNulls 
)

Definition at line 428 of file nodeHash.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, HashJoinTableData::area, Assert, BarrierArriveAndWait(), BarrierAttach(), BarrierPhase(), HashJoinTableData::batchCxt, HashJoinTableData::batches, HashJoinTableData::buckets, ParallelHashJoinState::build_barrier, HashJoinTableData::chunks, HashJoinTableData::curbatch, HashJoinTableData::current_chunk, CurrentMemoryContext, elog, ERROR, EState::es_query_dsa, ExecChooseHashTableSize(), ExecHashBuildSkewHash(), ExecParallelHashJoinSetUpBatches(), ExecParallelHashTableAlloc(), fmgr_info(), get_op_hash_functions(), HashJoinTableData::growEnabled, ParallelHashJoinState::growth, HashJoinTableData::hashCxt, HashJoinTableData::hashStrict, i, HashJoinTableData::inner_hashfunctions, HashJoinTableData::innerBatchFile, HashJoinTableData::keepNulls, lfirst_oid, list_length(), HashJoinTableData::log2_nbuckets, HashJoinTableData::log2_nbuckets_optimal, MemoryContextSwitchTo(), my_log2(), ParallelHashJoinState::nbatch, HashJoinTableData::nbatch, HashJoinTableData::nbatch_original, HashJoinTableData::nbatch_outstart, ParallelHashJoinState::nbuckets, HashJoinTableData::nbuckets, HashJoinTableData::nbuckets_optimal, HashJoinTableData::nbuckets_original, ParallelHashJoinState::nparticipants, HashJoinTableData::nSkewBuckets, OidIsValid, op_strict(), HashJoinTableData::outer_hashfunctions, HashJoinTableData::outerBatchFile, outerPlan, palloc(), palloc0(), Plan::parallel_aware, HashJoinTableData::parallel_state, HashState::parallel_state, HashJoinTableData::partialTuples, PHJ_BUILD_ELECTING, PHJ_GROWTH_OK, PlanState::plan, Hash::plan, Plan::plan_rows, Plan::plan_width, PrepareTempTablespaces(), HashState::ps, Hash::rows_total, SKEW_WORK_MEM_PERCENT, HashJoinTableData::skewBucket, HashJoinTableData::skewBucketLen, HashJoinTableData::skewBucketNums, HashJoinTableData::skewEnabled, Hash::skewTable, HashJoinTableData::skewTuples, ParallelHashJoinState::space_allowed, HashJoinTableData::spaceAllowed, HashJoinTableData::spaceAllowedSkew, HashJoinTableData::spacePeak, HashJoinTableData::spaceUsed, HashJoinTableData::spaceUsedSkew, PlanState::state, HashJoinTableData::totalTuples, HashJoinTableData::unshared, and WAIT_EVENT_HASH_BUILD_ELECTING.

Referenced by ExecHashJoinImpl().

429 {
430  Hash *node;
431  HashJoinTable hashtable;
432  Plan *outerNode;
433  size_t space_allowed;
434  int nbuckets;
435  int nbatch;
436  double rows;
437  int num_skew_mcvs;
438  int log2_nbuckets;
439  int nkeys;
440  int i;
441  ListCell *ho;
442  MemoryContext oldcxt;
443 
444  /*
445  * Get information about the size of the relation to be hashed (it's the
446  * "outer" subtree of this node, but the inner relation of the hashjoin).
447  * Compute the appropriate size of the hash table.
448  */
449  node = (Hash *) state->ps.plan;
450  outerNode = outerPlan(node);
451 
452  /*
453  * If this is shared hash table with a partial plan, then we can't use
454  * outerNode->plan_rows to estimate its size. We need an estimate of the
455  * total number of rows across all copies of the partial plan.
456  */
457  rows = node->plan.parallel_aware ? node->rows_total : outerNode->plan_rows;
458 
459  ExecChooseHashTableSize(rows, outerNode->plan_width,
460  OidIsValid(node->skewTable),
461  state->parallel_state != NULL,
462  state->parallel_state != NULL ?
463  state->parallel_state->nparticipants - 1 : 0,
464  &space_allowed,
465  &nbuckets, &nbatch, &num_skew_mcvs);
466 
467  /* nbuckets must be a power of 2 */
468  log2_nbuckets = my_log2(nbuckets);
469  Assert(nbuckets == (1 << log2_nbuckets));
470 
471  /*
472  * Initialize the hash table control block.
473  *
474  * The hashtable control block is just palloc'd from the executor's
475  * per-query memory context.
476  */
477  hashtable = (HashJoinTable) palloc(sizeof(HashJoinTableData));
478  hashtable->nbuckets = nbuckets;
479  hashtable->nbuckets_original = nbuckets;
480  hashtable->nbuckets_optimal = nbuckets;
481  hashtable->log2_nbuckets = log2_nbuckets;
482  hashtable->log2_nbuckets_optimal = log2_nbuckets;
483  hashtable->buckets.unshared = NULL;
484  hashtable->keepNulls = keepNulls;
485  hashtable->skewEnabled = false;
486  hashtable->skewBucket = NULL;
487  hashtable->skewBucketLen = 0;
488  hashtable->nSkewBuckets = 0;
489  hashtable->skewBucketNums = NULL;
490  hashtable->nbatch = nbatch;
491  hashtable->curbatch = 0;
492  hashtable->nbatch_original = nbatch;
493  hashtable->nbatch_outstart = nbatch;
494  hashtable->growEnabled = true;
495  hashtable->totalTuples = 0;
496  hashtable->partialTuples = 0;
497  hashtable->skewTuples = 0;
498  hashtable->innerBatchFile = NULL;
499  hashtable->outerBatchFile = NULL;
500  hashtable->spaceUsed = 0;
501  hashtable->spacePeak = 0;
502  hashtable->spaceAllowed = space_allowed;
503  hashtable->spaceUsedSkew = 0;
504  hashtable->spaceAllowedSkew =
505  hashtable->spaceAllowed * SKEW_WORK_MEM_PERCENT / 100;
506  hashtable->chunks = NULL;
507  hashtable->current_chunk = NULL;
508  hashtable->parallel_state = state->parallel_state;
509  hashtable->area = state->ps.state->es_query_dsa;
510  hashtable->batches = NULL;
511 
512 #ifdef HJDEBUG
513  printf("Hashjoin %p: initial nbatch = %d, nbuckets = %d\n",
514  hashtable, nbatch, nbuckets);
515 #endif
516 
517  /*
518  * Get info about the hash functions to be used for each hash key. Also
519  * remember whether the join operators are strict.
520  */
521  nkeys = list_length(hashOperators);
522  hashtable->outer_hashfunctions =
523  (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
524  hashtable->inner_hashfunctions =
525  (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
526  hashtable->hashStrict = (bool *) palloc(nkeys * sizeof(bool));
527  i = 0;
528  foreach(ho, hashOperators)
529  {
530  Oid hashop = lfirst_oid(ho);
531  Oid left_hashfn;
532  Oid right_hashfn;
533 
534  if (!get_op_hash_functions(hashop, &left_hashfn, &right_hashfn))
535  elog(ERROR, "could not find hash function for hash operator %u",
536  hashop);
537  fmgr_info(left_hashfn, &hashtable->outer_hashfunctions[i]);
538  fmgr_info(right_hashfn, &hashtable->inner_hashfunctions[i]);
539  hashtable->hashStrict[i] = op_strict(hashop);
540  i++;
541  }
542 
543  /*
544  * Create temporary memory contexts in which to keep the hashtable working
545  * storage. See notes in executor/hashjoin.h.
546  */
548  "HashTableContext",
550 
551  hashtable->batchCxt = AllocSetContextCreate(hashtable->hashCxt,
552  "HashBatchContext",
554 
555  /* Allocate data that will live for the life of the hashjoin */
556 
557  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
558 
559  if (nbatch > 1 && hashtable->parallel_state == NULL)
560  {
561  /*
562  * allocate and initialize the file arrays in hashCxt (not needed for
563  * parallel case which uses shared tuplestores instead of raw files)
564  */
565  hashtable->innerBatchFile = (BufFile **)
566  palloc0(nbatch * sizeof(BufFile *));
567  hashtable->outerBatchFile = (BufFile **)
568  palloc0(nbatch * sizeof(BufFile *));
569  /* The files will not be opened until needed... */
570  /* ... but make sure we have temp tablespaces established for them */
572  }
573 
574  MemoryContextSwitchTo(oldcxt);
575 
576  if (hashtable->parallel_state)
577  {
578  ParallelHashJoinState *pstate = hashtable->parallel_state;
579  Barrier *build_barrier;
580 
581  /*
582  * Attach to the build barrier. The corresponding detach operation is
583  * in ExecHashTableDetach. Note that we won't attach to the
584  * batch_barrier for batch 0 yet. We'll attach later and start it out
585  * in PHJ_BATCH_PROBING phase, because batch 0 is allocated up front
586  * and then loaded while hashing (the standard hybrid hash join
587  * algorithm), and we'll coordinate that using build_barrier.
588  */
589  build_barrier = &pstate->build_barrier;
590  BarrierAttach(build_barrier);
591 
592  /*
593  * So far we have no idea whether there are any other participants,
594  * and if so, what phase they are working on. The only thing we care
595  * about at this point is whether someone has already created the
596  * SharedHashJoinBatch objects and the hash table for batch 0. One
597  * backend will be elected to do that now if necessary.
598  */
599  if (BarrierPhase(build_barrier) == PHJ_BUILD_ELECTING &&
601  {
602  pstate->nbatch = nbatch;
603  pstate->space_allowed = space_allowed;
604  pstate->growth = PHJ_GROWTH_OK;
605 
606  /* Set up the shared state for coordinating batches. */
607  ExecParallelHashJoinSetUpBatches(hashtable, nbatch);
608 
609  /*
610  * Allocate batch 0's hash table up front so we can load it
611  * directly while hashing.
612  */
613  pstate->nbuckets = nbuckets;
614  ExecParallelHashTableAlloc(hashtable, 0);
615  }
616 
617  /*
618  * The next Parallel Hash synchronization point is in
619  * MultiExecParallelHash(), which will progress it all the way to
620  * PHJ_BUILD_DONE. The caller must not return control from this
621  * executor node between now and then.
622  */
623  }
624  else
625  {
626  /*
627  * Prepare context for the first-scan space allocations; allocate the
628  * hashbucket array therein, and set each bucket "empty".
629  */
630  MemoryContextSwitchTo(hashtable->batchCxt);
631 
632  hashtable->buckets.unshared = (HashJoinTuple *)
633  palloc0(nbuckets * sizeof(HashJoinTuple));
634 
635  /*
636  * Set up for skew optimization, if possible and there's a need for
637  * more than one batch. (In a one-batch join, there's no point in
638  * it.)
639  */
640  if (nbatch > 1)
641  ExecHashBuildSkewHash(hashtable, node, num_skew_mcvs);
642 
643  MemoryContextSwitchTo(oldcxt);
644  }
645 
646  return hashtable;
647 }
int log2_nbuckets_optimal
Definition: hashjoin.h:291
double rows_total
Definition: plannodes.h:890
Oid skewTable
Definition: plannodes.h:886
struct ParallelHashJoinState * parallel_state
Definition: execnodes.h:2067
double skewTuples
Definition: hashjoin.h:320
Definition: fmgr.h:56
struct dsa_area * es_query_dsa
Definition: execnodes.h:530
double plan_rows
Definition: plannodes.h:131
bool op_strict(Oid opno)
Definition: lsyscache.c:1266
bool get_op_hash_functions(Oid opno, RegProcedure *lhs_procno, RegProcedure *rhs_procno)
Definition: lsyscache.c:507
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
FmgrInfo * inner_hashfunctions
Definition: hashjoin.h:338
void ExecParallelHashTableAlloc(HashJoinTable hashtable, int batchno)
Definition: nodeHash.c:3035
EState * state
Definition: execnodes.h:870
unsigned int Oid
Definition: postgres_ext.h:31
#define OidIsValid(objectId)
Definition: c.h:594
static void ExecHashBuildSkewHash(HashJoinTable hashtable, Hash *node, int mcvsToUse)
Definition: nodeHash.c:2188
double partialTuples
Definition: hashjoin.h:319
dsa_area * area
Definition: hashjoin.h:355
int * skewBucketNums
Definition: hashjoin.h:308
#define ERROR
Definition: elog.h:43
void PrepareTempTablespaces(void)
Definition: tablespace.c:1287
void fmgr_info(Oid functionId, FmgrInfo *finfo)
Definition: fmgr.c:122
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:197
BufFile ** outerBatchFile
Definition: hashjoin.h:330
Size spaceAllowedSkew
Definition: hashjoin.h:345
bool parallel_aware
Definition: plannodes.h:137
PlanState ps
Definition: execnodes.h:2058
#define PHJ_BUILD_ELECTING
Definition: hashjoin.h:257
MemoryContext CurrentMemoryContext
Definition: mcxt.c:37
MemoryContext batchCxt
Definition: hashjoin.h:348
struct HashJoinTableData * HashJoinTable
Definition: execnodes.h:1720
int my_log2(long num)
Definition: dynahash.c:1716
#define outerPlan(node)
Definition: plannodes.h:174
FmgrInfo * outer_hashfunctions
Definition: hashjoin.h:337
#define AllocSetContextCreate(parent, name, allocparams)
Definition: memutils.h:165
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
int BarrierAttach(Barrier *barrier)
Definition: barrier.c:214
void * palloc0(Size size)
Definition: mcxt.c:864
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
Plan * plan
Definition: execnodes.h:868
double totalTuples
Definition: hashjoin.h:318
int plan_width
Definition: plannodes.h:132
#define Assert(condition)
Definition: c.h:688
ParallelHashGrowth growth
Definition: hashjoin.h:241
BufFile ** innerBatchFile
Definition: hashjoin.h:329
static int list_length(const List *l)
Definition: pg_list.h:89
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
union HashJoinTableData::@97 buckets
bool BarrierArriveAndWait(Barrier *barrier, uint32 wait_event_info)
Definition: barrier.c:125
HashMemoryChunk chunks
Definition: hashjoin.h:351
Plan plan
Definition: plannodes.h:885
void * palloc(Size size)
Definition: mcxt.c:835
static void ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
Definition: nodeHash.c:2874
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
HashMemoryChunk current_chunk
Definition: hashjoin.h:354
int i
bool * hashStrict
Definition: hashjoin.h:339
MemoryContext hashCxt
Definition: hashjoin.h:347
#define elog
Definition: elog.h:219
#define SKEW_WORK_MEM_PERCENT
Definition: hashjoin.h:110
void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew, bool try_combined_work_mem, int parallel_workers, size_t *space_allowed, int *numbuckets, int *numbatches, int *num_skew_mcvs)
Definition: nodeHash.c:661
#define lfirst_oid(lc)
Definition: pg_list.h:108

◆ ExecHashTableDestroy()

void ExecHashTableDestroy ( HashJoinTable  hashtable)

Definition at line 848 of file nodeHash.c.

References BufFileClose(), HashJoinTableData::hashCxt, i, HashJoinTableData::innerBatchFile, MemoryContextDelete(), HashJoinTableData::nbatch, HashJoinTableData::outerBatchFile, and pfree().

Referenced by ExecEndHashJoin(), and ExecReScanHashJoin().

849 {
850  int i;
851 
852  /*
853  * Make sure all the temp files are closed. We skip batch 0, since it
854  * can't have any temp files (and the arrays might not even exist if
855  * nbatch is only 1). Parallel hash joins don't use these files.
856  */
857  if (hashtable->innerBatchFile != NULL)
858  {
859  for (i = 1; i < hashtable->nbatch; i++)
860  {
861  if (hashtable->innerBatchFile[i])
862  BufFileClose(hashtable->innerBatchFile[i]);
863  if (hashtable->outerBatchFile[i])
864  BufFileClose(hashtable->outerBatchFile[i]);
865  }
866  }
867 
868  /* Release working memory (batchCxt is a child, so it goes away too) */
869  MemoryContextDelete(hashtable->hashCxt);
870 
871  /* And drop the control block */
872  pfree(hashtable);
873 }
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:198
void BufFileClose(BufFile *file)
Definition: buffile.c:397
void pfree(void *pointer)
Definition: mcxt.c:936
BufFile ** outerBatchFile
Definition: hashjoin.h:330
BufFile ** innerBatchFile
Definition: hashjoin.h:329
int i
MemoryContext hashCxt
Definition: hashjoin.h:347

◆ ExecHashTableDetach()

void ExecHashTableDetach ( HashJoinTable  hashtable)

Definition at line 3112 of file nodeHash.c.

References HashJoinTableData::area, BarrierDetach(), ParallelHashJoinState::batches, HashJoinTableData::batches, ParallelHashJoinState::build_barrier, dsa_free(), DsaPointerIsValid, i, ParallelHashJoinBatchAccessor::inner_tuples, InvalidDsaPointer, HashJoinTableData::nbatch, ParallelHashJoinBatchAccessor::outer_tuples, HashJoinTableData::parallel_state, sts_end_parallel_scan(), and sts_end_write().

Referenced by ExecHashJoinReInitializeDSM(), and ExecShutdownHashJoin().

3113 {
3114  if (hashtable->parallel_state)
3115  {
3116  ParallelHashJoinState *pstate = hashtable->parallel_state;
3117  int i;
3118 
3119  /* Make sure any temporary files are closed. */
3120  if (hashtable->batches)
3121  {
3122  for (i = 0; i < hashtable->nbatch; ++i)
3123  {
3124  sts_end_write(hashtable->batches[i].inner_tuples);
3125  sts_end_write(hashtable->batches[i].outer_tuples);
3128  }
3129  }
3130 
3131  /* If we're last to detach, clean up shared memory. */
3132  if (BarrierDetach(&pstate->build_barrier))
3133  {
3134  if (DsaPointerIsValid(pstate->batches))
3135  {
3136  dsa_free(hashtable->area, pstate->batches);
3137  pstate->batches = InvalidDsaPointer;
3138  }
3139  }
3140 
3141  hashtable->parallel_state = NULL;
3142  }
3143 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
#define InvalidDsaPointer
Definition: dsa.h:78
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_area * area
Definition: hashjoin.h:355
dsa_pointer batches
Definition: hashjoin.h:236
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
bool BarrierDetach(Barrier *barrier)
Definition: barrier.c:234
#define DsaPointerIsValid(x)
Definition: dsa.h:81
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:812
int i
void sts_end_write(SharedTuplestoreAccessor *accessor)

◆ ExecHashTableDetachBatch()

void ExecHashTableDetachBatch ( HashJoinTable  hashtable)

Definition at line 3055 of file nodeHash.c.

References HashJoinTableData::area, Assert, BarrierArriveAndDetach(), BarrierPhase(), ParallelHashJoinBatch::batch_barrier, HashJoinTableData::batches, ParallelHashJoinBatch::buckets, ParallelHashJoinBatch::chunks, HashJoinTableData::curbatch, dsa_free(), dsa_get_address(), DsaPointerIsValid, ParallelHashJoinBatchAccessor::inner_tuples, InvalidDsaPointer, Max, HashJoinTableData::nbuckets, HashMemoryChunkData::next, next, ParallelHashJoinBatchAccessor::outer_tuples, HashJoinTableData::parallel_state, PHJ_BATCH_DONE, HashMemoryChunkData::shared, ParallelHashJoinBatchAccessor::shared, ParallelHashJoinBatch::size, HashJoinTableData::spacePeak, and sts_end_parallel_scan().

Referenced by ExecHashJoinReInitializeDSM(), ExecParallelHashJoinNewBatch(), and ExecShutdownHashJoin().

3056 {
3057  if (hashtable->parallel_state != NULL &&
3058  hashtable->curbatch >= 0)
3059  {
3060  int curbatch = hashtable->curbatch;
3061  ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
3062 
3063  /* Make sure any temporary files are closed. */
3064  sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
3065  sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
3066 
3067  /* Detach from the batch we were last working on. */
3069  {
3070  /*
3071  * Technically we shouldn't access the barrier because we're no
3072  * longer attached, but since there is no way it's moving after
3073  * this point it seems safe to make the following assertion.
3074  */
3076 
3077  /* Free shared chunks and buckets. */
3078  while (DsaPointerIsValid(batch->chunks))
3079  {
3080  HashMemoryChunk chunk =
3081  dsa_get_address(hashtable->area, batch->chunks);
3082  dsa_pointer next = chunk->next.shared;
3083 
3084  dsa_free(hashtable->area, batch->chunks);
3085  batch->chunks = next;
3086  }
3087  if (DsaPointerIsValid(batch->buckets))
3088  {
3089  dsa_free(hashtable->area, batch->buckets);
3090  batch->buckets = InvalidDsaPointer;
3091  }
3092  }
3093 
3094  /*
3095  * Track the largest batch we've been attached to. Though each
3096  * backend might see a different subset of batches, explain.c will
3097  * scan the results from all backends to find the largest value.
3098  */
3099  hashtable->spacePeak =
3100  Max(hashtable->spacePeak,
3101  batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
3102 
3103  /* Remember that we are not attached to a batch. */
3104  hashtable->curbatch = -1;
3105  }
3106 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
#define PHJ_BATCH_DONE
Definition: hashjoin.h:268
static int32 next
Definition: blutils.c:210
#define InvalidDsaPointer
Definition: dsa.h:78
dsa_pointer chunks
Definition: hashjoin.h:156
uint64 dsa_pointer
Definition: dsa.h:62
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_area * area
Definition: hashjoin.h:355
dsa_pointer shared
Definition: hashjoin.h:127
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
union HashMemoryChunkData::@96 next
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
bool BarrierArriveAndDetach(Barrier *barrier)
Definition: barrier.c:203
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define Max(x, y)
Definition: c.h:840
#define Assert(condition)
Definition: c.h:688
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
#define DsaPointerIsValid(x)
Definition: dsa.h:81
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:812
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecHashTableInsert()

void ExecHashTableInsert ( HashJoinTable  hashtable,
TupleTableSlot slot,
uint32  hashvalue 
)

Definition at line 1588 of file nodeHash.c.

References Assert, HashJoinTableData::buckets, HashJoinTableData::curbatch, dense_alloc(), ExecFetchSlotMinimalTuple(), ExecHashGetBucketAndBatch(), ExecHashIncreaseNumBatches(), ExecHashJoinSaveTuple(), HashJoinTupleData::hashvalue, HeapTupleHeaderClearMatch, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, HashJoinTableData::innerBatchFile, HashJoinTableData::log2_nbuckets_optimal, MaxAllocSize, HashJoinTableData::nbatch, HashJoinTableData::nbuckets_optimal, HashJoinTupleData::next, NTUP_PER_BUCKET, HashJoinTableData::skewTuples, HashJoinTableData::spaceAllowed, HashJoinTableData::spacePeak, HashJoinTableData::spaceUsed, MinimalTupleData::t_len, HashJoinTableData::totalTuples, HashJoinTupleData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashJoinNewBatch(), and MultiExecPrivateHash().

1591 {
1593  int bucketno;
1594  int batchno;
1595 
1596  ExecHashGetBucketAndBatch(hashtable, hashvalue,
1597  &bucketno, &batchno);
1598 
1599  /*
1600  * decide whether to put the tuple in the hash table or a temp file
1601  */
1602  if (batchno == hashtable->curbatch)
1603  {
1604  /*
1605  * put the tuple in hash table
1606  */
1607  HashJoinTuple hashTuple;
1608  int hashTupleSize;
1609  double ntuples = (hashtable->totalTuples - hashtable->skewTuples);
1610 
1611  /* Create the HashJoinTuple */
1612  hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
1613  hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
1614 
1615  hashTuple->hashvalue = hashvalue;
1616  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1617 
1618  /*
1619  * We always reset the tuple-matched flag on insertion. This is okay
1620  * even when reloading a tuple from a batch file, since the tuple
1621  * could not possibly have been matched to an outer tuple before it
1622  * went into the batch file.
1623  */
1625 
1626  /* Push it onto the front of the bucket's list */
1627  hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1628  hashtable->buckets.unshared[bucketno] = hashTuple;
1629 
1630  /*
1631  * Increase the (optimal) number of buckets if we just exceeded the
1632  * NTUP_PER_BUCKET threshold, but only when there's still a single
1633  * batch.
1634  */
1635  if (hashtable->nbatch == 1 &&
1636  ntuples > (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
1637  {
1638  /* Guard against integer overflow and alloc size overflow */
1639  if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
1640  hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
1641  {
1642  hashtable->nbuckets_optimal *= 2;
1643  hashtable->log2_nbuckets_optimal += 1;
1644  }
1645  }
1646 
1647  /* Account for space used, and back off if we've used too much */
1648  hashtable->spaceUsed += hashTupleSize;
1649  if (hashtable->spaceUsed > hashtable->spacePeak)
1650  hashtable->spacePeak = hashtable->spaceUsed;
1651  if (hashtable->spaceUsed +
1652  hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
1653  > hashtable->spaceAllowed)
1654  ExecHashIncreaseNumBatches(hashtable);
1655  }
1656  else
1657  {
1658  /*
1659  * put the tuple into a temp file for later batches
1660  */
1661  Assert(batchno > hashtable->curbatch);
1662  ExecHashJoinSaveTuple(tuple,
1663  hashvalue,
1664  &hashtable->innerBatchFile[batchno]);
1665  }
1666 }
int log2_nbuckets_optimal
Definition: hashjoin.h:291
double skewTuples
Definition: hashjoin.h:320
union HashJoinTupleData::@95 next
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot)
Definition: execTuples.c:688
static void ExecHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:881
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
#define MaxAllocSize
Definition: memutils.h:40
static void * dense_alloc(HashJoinTable hashtable, Size size)
Definition: nodeHash.c:2649
#define NTUP_PER_BUCKET
Definition: nodeHash.c:658
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
double totalTuples
Definition: hashjoin.h:318
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
BufFile ** innerBatchFile
Definition: hashjoin.h:329
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:532
union HashJoinTableData::@97 buckets
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, BufFile **fileptr)
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecHashTableReset()

void ExecHashTableReset ( HashJoinTable  hashtable)

Definition at line 2113 of file nodeHash.c.

References HashJoinTableData::batchCxt, HashJoinTableData::buckets, HashJoinTableData::chunks, MemoryContextReset(), MemoryContextSwitchTo(), HashJoinTableData::nbuckets, palloc0(), HashJoinTableData::spaceUsed, and HashJoinTableData::unshared.

Referenced by ExecHashJoinNewBatch().

2114 {
2115  MemoryContext oldcxt;
2116  int nbuckets = hashtable->nbuckets;
2117 
2118  /*
2119  * Release all the hash buckets and tuples acquired in the prior pass, and
2120  * reinitialize the context for a new pass.
2121  */
2122  MemoryContextReset(hashtable->batchCxt);
2123  oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
2124 
2125  /* Reallocate and reinitialize the hash bucket headers. */
2126  hashtable->buckets.unshared = (HashJoinTuple *)
2127  palloc0(nbuckets * sizeof(HashJoinTuple));
2128 
2129  hashtable->spaceUsed = 0;
2130 
2131  MemoryContextSwitchTo(oldcxt);
2132 
2133  /* Forget the chunks (the memory was freed by the context reset above). */
2134  hashtable->chunks = NULL;
2135 }
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:134
MemoryContext batchCxt
Definition: hashjoin.h:348
void * palloc0(Size size)
Definition: mcxt.c:864
union HashJoinTableData::@97 buckets
HashMemoryChunk chunks
Definition: hashjoin.h:351
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297

◆ ExecHashTableResetMatchFlags()

void ExecHashTableResetMatchFlags ( HashJoinTable  hashtable)

Definition at line 2142 of file nodeHash.c.

References HashJoinTableData::buckets, HeapTupleHeaderClearMatch, HJTUPLE_MINTUPLE, i, HashJoinTableData::nbuckets, HashJoinTupleData::next, HashJoinTableData::nSkewBuckets, HashJoinTableData::skewBucket, HashJoinTableData::skewBucketNums, HashSkewBucket::tuples, HashJoinTupleData::unshared, and HashJoinTableData::unshared.

Referenced by ExecReScanHashJoin().

2143 {
2144  HashJoinTuple tuple;
2145  int i;
2146 
2147  /* Reset all flags in the main table ... */
2148  for (i = 0; i < hashtable->nbuckets; i++)
2149  {
2150  for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
2151  tuple = tuple->next.unshared)
2153  }
2154 
2155  /* ... and the same for the skew buckets, if any */
2156  for (i = 0; i < hashtable->nSkewBuckets; i++)
2157  {
2158  int j = hashtable->skewBucketNums[i];
2159  HashSkewBucket *skewBucket = hashtable->skewBucket[j];
2160 
2161  for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
2163  }
2164 }
union HashJoinTupleData::@95 next
int * skewBucketNums
Definition: hashjoin.h:308
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
HashJoinTuple tuples
Definition: hashjoin.h:105
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:532
union HashJoinTableData::@97 buckets
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
int i

◆ ExecInitHash()

HashState* ExecInitHash ( Hash node,
EState estate,
int  eflags 
)

Definition at line 352 of file nodeHash.c.

References Assert, EXEC_FLAG_BACKWARD, EXEC_FLAG_MARK, ExecAssignExprContext(), ExecHash(), ExecInitNode(), ExecInitQual(), ExecInitResultTupleSlotTL(), PlanState::ExecProcNode, HashState::hashkeys, HashState::hashtable, makeNode, NIL, outerPlan, outerPlanState, PlanState::plan, Hash::plan, HashState::ps, PlanState::ps_ProjInfo, Plan::qual, PlanState::qual, and PlanState::state.

Referenced by ExecInitNode().

353 {
354  HashState *hashstate;
355 
356  /* check for unsupported flags */
357  Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
358 
359  /*
360  * create state structure
361  */
362  hashstate = makeNode(HashState);
363  hashstate->ps.plan = (Plan *) node;
364  hashstate->ps.state = estate;
365  hashstate->ps.ExecProcNode = ExecHash;
366  hashstate->hashtable = NULL;
367  hashstate->hashkeys = NIL; /* will be set by parent HashJoin */
368 
369  /*
370  * Miscellaneous initialization
371  *
372  * create expression context for node
373  */
374  ExecAssignExprContext(estate, &hashstate->ps);
375 
376  /*
377  * initialize child nodes
378  */
379  outerPlanState(hashstate) = ExecInitNode(outerPlan(node), estate, eflags);
380 
381  /*
382  * initialize our result slot and type. No need to build projection
383  * because this node doesn't do projections.
384  */
385  ExecInitResultTupleSlotTL(estate, &hashstate->ps);
386  hashstate->ps.ps_ProjInfo = NULL;
387 
388  /*
389  * initialize child expressions
390  */
391  hashstate->ps.qual =
392  ExecInitQual(node->plan.qual, (PlanState *) hashstate);
393 
394  return hashstate;
395 }
#define NIL
Definition: pg_list.h:69
List * qual
Definition: plannodes.h:145
ProjectionInfo * ps_ProjInfo
Definition: execnodes.h:903
HashJoinTable hashtable
Definition: execnodes.h:2059
EState * state
Definition: execnodes.h:870
ExprState * ExecInitQual(List *qual, PlanState *parent)
Definition: execExpr.c:204
#define EXEC_FLAG_BACKWARD
Definition: executor.h:61
#define outerPlanState(node)
Definition: execnodes.h:914
List * hashkeys
Definition: execnodes.h:2060
PlanState ps
Definition: execnodes.h:2058
#define outerPlan(node)
Definition: plannodes.h:174
static TupleTableSlot * ExecHash(PlanState *pstate)
Definition: nodeHash.c:91
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:874
void ExecInitResultTupleSlotTL(EState *estate, PlanState *planstate)
Definition: execTuples.c:870
Plan * plan
Definition: execnodes.h:868
#define makeNode(_type_)
Definition: nodes.h:561
#define Assert(condition)
Definition: c.h:688
#define EXEC_FLAG_MARK
Definition: executor.h:62
void ExecAssignExprContext(EState *estate, PlanState *planstate)
Definition: execUtils.c:425
ExprState * qual
Definition: execnodes.h:886
Plan plan
Definition: plannodes.h:885
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:139

◆ ExecParallelHashCloseBatchAccessors()

static void ExecParallelHashCloseBatchAccessors ( HashJoinTable  hashtable)
static

Definition at line 2951 of file nodeHash.c.

References HashJoinTableData::batches, i, ParallelHashJoinBatchAccessor::inner_tuples, HashJoinTableData::nbatch, ParallelHashJoinBatchAccessor::outer_tuples, pfree(), sts_end_parallel_scan(), and sts_end_write().

Referenced by ExecParallelHashEnsureBatchAccessors(), and ExecParallelHashIncreaseNumBatches().

2952 {
2953  int i;
2954 
2955  for (i = 0; i < hashtable->nbatch; ++i)
2956  {
2957  /* Make sure no files are left open. */
2958  sts_end_write(hashtable->batches[i].inner_tuples);
2959  sts_end_write(hashtable->batches[i].outer_tuples);
2962  }
2963  pfree(hashtable->batches);
2964  hashtable->batches = NULL;
2965 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
void pfree(void *pointer)
Definition: mcxt.c:936
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
int i
void sts_end_write(SharedTuplestoreAccessor *accessor)

◆ ExecParallelHashEnsureBatchAccessors()

static void ExecParallelHashEnsureBatchAccessors ( HashJoinTable  hashtable)
static

Definition at line 2972 of file nodeHash.c.

References HashJoinTableData::area, ParallelHashJoinState::batches, HashJoinTableData::batches, ParallelHashJoinBatchAccessor::done, dsa_get_address(), DsaPointerIsValid, ExecParallelHashCloseBatchAccessors(), ParallelHashJoinState::fileset, HashJoinTableData::hashCxt, i, ParallelHashJoinBatchAccessor::inner_tuples, MemoryContextSwitchTo(), ParallelHashJoinState::nbatch, HashJoinTableData::nbatch, ParallelHashJoinState::nparticipants, NthParallelHashJoinBatch, ParallelHashJoinBatchAccessor::outer_tuples, palloc0(), HashJoinTableData::parallel_state, ParallelHashJoinBatchInner, ParallelHashJoinBatchOuter, ParallelWorkerNumber, ParallelHashJoinBatchAccessor::preallocated, ParallelHashJoinBatchAccessor::shared, and sts_attach().

Referenced by ExecParallelHashIncreaseNumBatches(), ExecParallelHashIncreaseNumBuckets(), and MultiExecParallelHash().

2973 {
2974  ParallelHashJoinState *pstate = hashtable->parallel_state;
2975  ParallelHashJoinBatch *batches;
2976  MemoryContext oldcxt;
2977  int i;
2978 
2979  if (hashtable->batches != NULL)
2980  {
2981  if (hashtable->nbatch == pstate->nbatch)
2982  return;
2984  }
2985 
2986  /*
2987  * It's possible for a backend to start up very late so that the whole
2988  * join is finished and the shm state for tracking batches has already
2989  * been freed by ExecHashTableDetach(). In that case we'll just leave
2990  * hashtable->batches as NULL so that ExecParallelHashJoinNewBatch() gives
2991  * up early.
2992  */
2993  if (!DsaPointerIsValid(pstate->batches))
2994  return;
2995 
2996  /* Use hash join memory context. */
2997  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
2998 
2999  /* Allocate this backend's accessor array. */
3000  hashtable->nbatch = pstate->nbatch;
3001  hashtable->batches = (ParallelHashJoinBatchAccessor *)
3002  palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
3003 
3004  /* Find the base of the pseudo-array of ParallelHashJoinBatch objects. */
3005  batches = (ParallelHashJoinBatch *)
3006  dsa_get_address(hashtable->area, pstate->batches);
3007 
3008  /* Set up the accessor array and attach to the tuplestores. */
3009  for (i = 0; i < hashtable->nbatch; ++i)
3010  {
3011  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
3012  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
3013 
3014  accessor->shared = shared;
3015  accessor->preallocated = 0;
3016  accessor->done = false;
3017  accessor->inner_tuples =
3020  &pstate->fileset);
3021  accessor->outer_tuples =
3023  pstate->nparticipants),
3025  &pstate->fileset);
3026  }
3027 
3028  MemoryContextSwitchTo(oldcxt);
3029 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2951
#define ParallelHashJoinBatchOuter(batch, nparticipants)
Definition: hashjoin.h:175
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
SharedFileSet fileset
Definition: hashjoin.h:253
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_area * area
Definition: hashjoin.h:355
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
dsa_pointer batches
Definition: hashjoin.h:236
#define ParallelHashJoinBatchInner(batch)
Definition: hashjoin.h:170
SharedTuplestoreAccessor * sts_attach(SharedTuplestore *sts, int my_participant_number, SharedFileSet *fileset)
int ParallelWorkerNumber
Definition: parallel.c:103
void * palloc0(Size size)
Definition: mcxt.c:864
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define NthParallelHashJoinBatch(base, n)
Definition: hashjoin.h:186
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
#define DsaPointerIsValid(x)
Definition: dsa.h:81
int i
MemoryContext hashCxt
Definition: hashjoin.h:347

◆ ExecParallelHashFirstTuple()

static HashJoinTuple ExecParallelHashFirstTuple ( HashJoinTable  table,
int  bucketno 
)
inlinestatic

Definition at line 3149 of file nodeHash.c.

References HashJoinTableData::area, Assert, HashJoinTableData::buckets, dsa_get_address(), dsa_pointer_atomic_read, HashJoinTableData::parallel_state, and HashJoinTableData::shared.

Referenced by ExecParallelScanHashBucket().

3150 {
3151  HashJoinTuple tuple;
3152  dsa_pointer p;
3153 
3154  Assert(hashtable->parallel_state);
3155  p = dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
3156  tuple = (HashJoinTuple) dsa_get_address(hashtable->area, p);
3157 
3158  return tuple;
3159 }
uint64 dsa_pointer
Definition: dsa.h:62
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
#define Assert(condition)
Definition: c.h:688
#define dsa_pointer_atomic_read
Definition: dsa.h:65

◆ ExecParallelHashIncreaseNumBatches()

static void ExecParallelHashIncreaseNumBatches ( HashJoinTable  hashtable)
static

Definition at line 1052 of file nodeHash.c.

References HashJoinTableData::area, Assert, BarrierArriveAndWait(), BarrierPhase(), ParallelHashJoinState::batches, HashJoinTableData::batches, ParallelHashJoinBatch::buckets, ParallelHashJoinState::build_barrier, ParallelHashJoinState::chunk_work_queue, ParallelHashJoinBatch::chunks, dsa_allocate, dsa_free(), dsa_get_address(), dsa_pointer_atomic_init, dsa_pointer_atomic_write, ParallelHashJoinBatch::estimated_size, ExecParallelHashCloseBatchAccessors(), ExecParallelHashEnsureBatchAccessors(), ExecParallelHashJoinSetUpBatches(), ExecParallelHashMergeCounters(), ExecParallelHashRepartitionFirst(), ExecParallelHashRepartitionRest(), ExecParallelHashTableSetCurrentBatch(), ParallelHashJoinState::grow_batches_barrier, ParallelHashJoinState::growth, i, InvalidDsaPointer, Max, MaxAllocSize, Min, my_log2(), ParallelHashJoinState::nbatch, HashJoinTableData::nbatch, ParallelHashJoinState::nbuckets, HashJoinTableData::nbuckets, ParallelHashJoinState::nparticipants, NTUP_PER_BUCKET, ParallelHashJoinBatch::ntuples, ParallelHashJoinState::old_batches, ParallelHashJoinState::old_nbatch, ParallelHashJoinBatch::old_ntuples, HashJoinTableData::parallel_state, PHJ_BUILD_HASHING_INNER, PHJ_GROW_BATCHES_ALLOCATING, PHJ_GROW_BATCHES_DECIDING, PHJ_GROW_BATCHES_ELECTING, PHJ_GROW_BATCHES_FINISHING, PHJ_GROW_BATCHES_PHASE, PHJ_GROW_BATCHES_REPARTITIONING, PHJ_GROWTH_DISABLED, PHJ_GROWTH_NEED_MORE_BATCHES, PHJ_GROWTH_OK, ParallelHashJoinBatchAccessor::shared, ParallelHashJoinState::space_allowed, ParallelHashJoinBatch::space_exhausted, WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATING, WAIT_EVENT_HASH_GROW_BATCHES_DECIDING, WAIT_EVENT_HASH_GROW_BATCHES_ELECTING, WAIT_EVENT_HASH_GROW_BATCHES_FINISHING, WAIT_EVENT_HASH_GROW_BATCHES_REPARTITIONING, and work_mem.

Referenced by ExecParallelHashTupleAlloc(), ExecParallelHashTuplePrealloc(), and MultiExecParallelHash().

1053 {
1054  ParallelHashJoinState *pstate = hashtable->parallel_state;
1055  int i;
1056 
1058 
1059  /*
1060  * It's unlikely, but we need to be prepared for new participants to show
1061  * up while we're in the middle of this operation so we need to switch on
1062  * barrier phase here.
1063  */
1065  {
1067 
1068  /*
1069  * Elect one participant to prepare to grow the number of batches.
1070  * This involves reallocating or resetting the buckets of batch 0
1071  * in preparation for all participants to begin repartitioning the
1072  * tuples.
1073  */
1076  {
1077  dsa_pointer_atomic *buckets;
1078  ParallelHashJoinBatch *old_batch0;
1079  int new_nbatch;
1080  int i;
1081 
1082  /* Move the old batch out of the way. */
1083  old_batch0 = hashtable->batches[0].shared;
1084  pstate->old_batches = pstate->batches;
1085  pstate->old_nbatch = hashtable->nbatch;
1086  pstate->batches = InvalidDsaPointer;
1087 
1088  /* Free this backend's old accessors. */
1090 
1091  /* Figure out how many batches to use. */
1092  if (hashtable->nbatch == 1)
1093  {
1094  /*
1095  * We are going from single-batch to multi-batch. We need
1096  * to switch from one large combined memory budget to the
1097  * regular work_mem budget.
1098  */
1099  pstate->space_allowed = work_mem * 1024L;
1100 
1101  /*
1102  * The combined work_mem of all participants wasn't
1103  * enough. Therefore one batch per participant would be
1104  * approximately equivalent and would probably also be
1105  * insufficient. So try two batches per particiant,
1106  * rounded up to a power of two.
1107  */
1108  new_nbatch = 1 << my_log2(pstate->nparticipants * 2);
1109  }
1110  else
1111  {
1112  /*
1113  * We were already multi-batched. Try doubling the number
1114  * of batches.
1115  */
1116  new_nbatch = hashtable->nbatch * 2;
1117  }
1118 
1119  /* Allocate new larger generation of batches. */
1120  Assert(hashtable->nbatch == pstate->nbatch);
1121  ExecParallelHashJoinSetUpBatches(hashtable, new_nbatch);
1122  Assert(hashtable->nbatch == pstate->nbatch);
1123 
1124  /* Replace or recycle batch 0's bucket array. */
1125  if (pstate->old_nbatch == 1)
1126  {
1127  double dtuples;
1128  double dbuckets;
1129  int new_nbuckets;
1130 
1131  /*
1132  * We probably also need a smaller bucket array. How many
1133  * tuples do we expect per batch, assuming we have only
1134  * half of them so far? Normally we don't need to change
1135  * the bucket array's size, because the size of each batch
1136  * stays the same as we add more batches, but in this
1137  * special case we move from a large batch to many smaller
1138  * batches and it would be wasteful to keep the large
1139  * array.
1140  */
1141  dtuples = (old_batch0->ntuples * 2.0) / new_nbatch;
1142  dbuckets = ceil(dtuples / NTUP_PER_BUCKET);
1143  dbuckets = Min(dbuckets,
1144  MaxAllocSize / sizeof(dsa_pointer_atomic));
1145  new_nbuckets = (int) dbuckets;
1146  new_nbuckets = Max(new_nbuckets, 1024);
1147  new_nbuckets = 1 << my_log2(new_nbuckets);
1148  dsa_free(hashtable->area, old_batch0->buckets);
1149  hashtable->batches[0].shared->buckets =
1150  dsa_allocate(hashtable->area,
1151  sizeof(dsa_pointer_atomic) * new_nbuckets);
1152  buckets = (dsa_pointer_atomic *)
1153  dsa_get_address(hashtable->area,
1154  hashtable->batches[0].shared->buckets);
1155  for (i = 0; i < new_nbuckets; ++i)
1157  pstate->nbuckets = new_nbuckets;
1158  }
1159  else
1160  {
1161  /* Recycle the existing bucket array. */
1162  hashtable->batches[0].shared->buckets = old_batch0->buckets;
1163  buckets = (dsa_pointer_atomic *)
1164  dsa_get_address(hashtable->area, old_batch0->buckets);
1165  for (i = 0; i < hashtable->nbuckets; ++i)
1167  }
1168 
1169  /* Move all chunks to the work queue for parallel processing. */
1170  pstate->chunk_work_queue = old_batch0->chunks;
1171 
1172  /* Disable further growth temporarily while we're growing. */
1173  pstate->growth = PHJ_GROWTH_DISABLED;
1174  }
1175  else
1176  {
1177  /* All other participants just flush their tuples to disk. */
1179  }
1180  /* Fall through. */
1181 
1183  /* Wait for the above to be finished. */
1186  /* Fall through. */
1187 
1189  /* Make sure that we have the current dimensions and buckets. */
1192  /* Then partition, flush counters. */
1195  ExecParallelHashMergeCounters(hashtable);
1196  /* Wait for the above to be finished. */
1199  /* Fall through. */
1200 
1202 
1203  /*
1204  * Elect one participant to clean up and decide whether further
1205  * repartitioning is needed, or should be disabled because it's
1206  * not helping.
1207  */
1210  {
1211  bool space_exhausted = false;
1212  bool extreme_skew_detected = false;
1213 
1214  /* Make sure that we have the current dimensions and buckets. */
1217 
1218  /* Are any of the new generation of batches exhausted? */
1219  for (i = 0; i < hashtable->nbatch; ++i)
1220  {
1221  ParallelHashJoinBatch *batch = hashtable->batches[i].shared;
1222 
1223  if (batch->space_exhausted ||
1224  batch->estimated_size > pstate->space_allowed)
1225  {
1226  int parent;
1227 
1228  space_exhausted = true;
1229 
1230  /*
1231  * Did this batch receive ALL of the tuples from its
1232  * parent batch? That would indicate that further
1233  * repartitioning isn't going to help (the hash values
1234  * are probably all the same).
1235  */
1236  parent = i % pstate->old_nbatch;
1237  if (batch->ntuples == hashtable->batches[parent].shared->old_ntuples)
1238  extreme_skew_detected = true;
1239  }
1240  }
1241 
1242  /* Don't keep growing if it's not helping or we'd overflow. */
1243  if (extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2)
1244  pstate->growth = PHJ_GROWTH_DISABLED;
1245  else if (space_exhausted)
1247  else
1248  pstate->growth = PHJ_GROWTH_OK;
1249 
1250  /* Free the old batches in shared memory. */
1251  dsa_free(hashtable->area, pstate->old_batches);
1252  pstate->old_batches = InvalidDsaPointer;
1253  }
1254  /* Fall through. */
1255 
1257  /* Wait for the above to complete. */
1260  }
1261 }
static void ExecParallelHashRepartitionRest(HashJoinTable hashtable)
Definition: nodeHash.c:1336
static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2951
dsa_pointer chunk_work_queue
Definition: hashjoin.h:242
#define PHJ_GROW_BATCHES_DECIDING
Definition: hashjoin.h:274
#define InvalidDsaPointer
Definition: dsa.h:78
void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno)
Definition: nodeHash.c:3197
dsa_pointer chunks
Definition: hashjoin.h:156
#define Min(x, y)
Definition: c.h:846
static void ExecParallelHashMergeCounters(HashJoinTable hashtable)
Definition: nodeHash.c:1396
static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2972
static void ExecParallelHashRepartitionFirst(HashJoinTable hashtable)
Definition: nodeHash.c:1269
dsa_area * area
Definition: hashjoin.h:355
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
dsa_pointer batches
Definition: hashjoin.h:236
#define PHJ_GROW_BATCHES_PHASE(n)
Definition: hashjoin.h:276
int my_log2(long num)
Definition: dynahash.c:1716
#define PHJ_GROW_BATCHES_ALLOCATING
Definition: hashjoin.h:272
#define MaxAllocSize
Definition: memutils.h:40
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
#define NTUP_PER_BUCKET
Definition: nodeHash.c:658
#define PHJ_GROW_BATCHES_REPARTITIONING
Definition: hashjoin.h:273
int work_mem
Definition: globals.c:113
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define Max(x, y)
Definition: c.h:840
#define dsa_pointer_atomic_write
Definition: dsa.h:66
#define Assert(condition)
Definition: c.h:688
ParallelHashGrowth growth
Definition: hashjoin.h:241
dsa_pointer old_batches
Definition: hashjoin.h:237
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
bool BarrierArriveAndWait(Barrier *barrier, uint32 wait_event_info)
Definition: barrier.c:125
#define dsa_pointer_atomic_init
Definition: dsa.h:64
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:812
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
static void ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
Definition: nodeHash.c:2874
int i
#define PHJ_GROW_BATCHES_FINISHING
Definition: hashjoin.h:275
Barrier grow_batches_barrier
Definition: hashjoin.h:249
#define PHJ_GROW_BATCHES_ELECTING
Definition: hashjoin.h:271
#define dsa_allocate(area, size)
Definition: dsa.h:84
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecParallelHashIncreaseNumBuckets()

static void ExecParallelHashIncreaseNumBuckets ( HashJoinTable  hashtable)
static

Definition at line 1489 of file nodeHash.c.

References HashJoinTableData::area, Assert, BarrierArriveAndWait(), BarrierPhase(), HashJoinTableData::batches, ParallelHashJoinBatch::buckets, HashJoinTableData::buckets, ParallelHashJoinState::build_barrier, CHECK_FOR_INTERRUPTS, ParallelHashJoinState::chunk_work_queue, ParallelHashJoinBatch::chunks, dsa_allocate, dsa_free(), dsa_get_address(), dsa_pointer_atomic_init, ExecHashGetBucketAndBatch(), ExecParallelHashEnsureBatchAccessors(), ExecParallelHashPopChunkQueue(), ExecParallelHashPushTuple(), ExecParallelHashTableSetCurrentBatch(), ParallelHashJoinState::grow_buckets_barrier, ParallelHashJoinState::growth, HASH_CHUNK_DATA, HASH_CHUNK_HEADER_SIZE, HashJoinTupleData::hashvalue, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, i, idx(), InvalidDsaPointer, MAXALIGN, ParallelHashJoinState::nbuckets, HashJoinTableData::parallel_state, PHJ_BUILD_HASHING_INNER, PHJ_GROW_BUCKETS_ALLOCATING, PHJ_GROW_BUCKETS_ELECTING, PHJ_GROW_BUCKETS_PHASE, PHJ_GROW_BUCKETS_REINSERTING, PHJ_GROWTH_OK, ParallelHashJoinBatchAccessor::shared, HashJoinTableData::shared, ParallelHashJoinBatch::size, WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATING, WAIT_EVENT_HASH_GROW_BUCKETS_ELECTING, and WAIT_EVENT_HASH_GROW_BUCKETS_REINSERTING.

Referenced by ExecParallelHashTupleAlloc(), ExecParallelHashTuplePrealloc(), and MultiExecParallelHash().

1490 {
1491  ParallelHashJoinState *pstate = hashtable->parallel_state;
1492  int i;
1493  HashMemoryChunk chunk;
1494  dsa_pointer chunk_s;
1495 
1497 
1498  /*
1499  * It's unlikely, but we need to be prepared for new participants to show
1500  * up while we're in the middle of this operation so we need to switch on
1501  * barrier phase here.
1502  */
1504  {
1506  /* Elect one participant to prepare to increase nbuckets. */
1509  {
1510  size_t size;
1511  dsa_pointer_atomic *buckets;
1512 
1513  /* Double the size of the bucket array. */
1514  pstate->nbuckets *= 2;
1515  size = pstate->nbuckets * sizeof(dsa_pointer_atomic);
1516  hashtable->batches[0].shared->size += size / 2;
1517  dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
1518  hashtable->batches[0].shared->buckets =
1519  dsa_allocate(hashtable->area, size);
1520  buckets = (dsa_pointer_atomic *)
1521  dsa_get_address(hashtable->area,
1522  hashtable->batches[0].shared->buckets);
1523  for (i = 0; i < pstate->nbuckets; ++i)
1525 
1526  /* Put the chunk list onto the work queue. */
1527  pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;
1528 
1529  /* Clear the flag. */
1530  pstate->growth = PHJ_GROWTH_OK;
1531  }
1532  /* Fall through. */
1533 
1535  /* Wait for the above to complete. */
1538  /* Fall through. */
1539 
1541  /* Reinsert all tuples into the hash table. */
1544  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_s)))
1545  {
1546  size_t idx = 0;
1547 
1548  while (idx < chunk->used)
1549  {
1550  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1551  dsa_pointer shared = chunk_s + HASH_CHUNK_HEADER_SIZE + idx;
1552  int bucketno;
1553  int batchno;
1554 
1555  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1556  &bucketno, &batchno);
1557  Assert(batchno == 0);
1558 
1559  /* add the tuple to the proper bucket */
1560  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1561  hashTuple, shared);
1562 
1563  /* advance index past the tuple */
1564  idx += MAXALIGN(HJTUPLE_OVERHEAD +
1565  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1566  }
1567 
1568  /* allow this loop to be cancellable */
1570  }
1573  }
1574 }
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
dsa_pointer chunk_work_queue
Definition: hashjoin.h:242
#define InvalidDsaPointer
Definition: dsa.h:78
void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno)
Definition: nodeHash.c:3197
dsa_pointer chunks
Definition: hashjoin.h:156
Datum idx(PG_FUNCTION_ARGS)
Definition: _int_op.c:264
static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2972
uint64 dsa_pointer
Definition: dsa.h:62
#define PHJ_GROW_BUCKETS_REINSERTING
Definition: hashjoin.h:281
dsa_area * area
Definition: hashjoin.h:355
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
#define PHJ_GROW_BUCKETS_ALLOCATING
Definition: hashjoin.h:280
Barrier grow_buckets_barrier
Definition: hashjoin.h:250
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table, dsa_pointer *shared)
Definition: nodeHash.c:3218
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
#define PHJ_GROW_BUCKETS_ELECTING
Definition: hashjoin.h:279
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define Assert(condition)
Definition: c.h:688
ParallelHashGrowth growth
Definition: hashjoin.h:241
#define PHJ_GROW_BUCKETS_PHASE(n)
Definition: hashjoin.h:282
pg_atomic_uint64 dsa_pointer_atomic
Definition: dsa.h:63
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
#define MAXALIGN(LEN)
Definition: c.h:641
union HashJoinTableData::@97 buckets
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
bool BarrierArriveAndWait(Barrier *barrier, uint32 wait_event_info)
Definition: barrier.c:125
#define dsa_pointer_atomic_init
Definition: dsa.h:64
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:812
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3179
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
int i
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
uint32 hashvalue
Definition: hashjoin.h:75
#define dsa_allocate(area, size)
Definition: dsa.h:84
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecParallelHashJoinSetUpBatches()

static void ExecParallelHashJoinSetUpBatches ( HashJoinTable  hashtable,
int  nbatch 
)
static

Definition at line 2874 of file nodeHash.c.

References HashJoinTableData::area, Assert, BarrierArriveAndWait(), BarrierAttach(), BarrierDetach(), BarrierInit(), BarrierPhase(), ParallelHashJoinBatch::batch_barrier, ParallelHashJoinState::batches, HashJoinTableData::batches, dsa_allocate0, dsa_get_address(), EstimateParallelHashJoinBatch, ParallelHashJoinState::fileset, HashJoinTableData::hashCxt, i, ParallelHashJoinBatchAccessor::inner_tuples, MAXPGPATH, MemoryContextSwitchTo(), name, ParallelHashJoinState::nbatch, HashJoinTableData::nbatch, ParallelHashJoinState::nparticipants, NthParallelHashJoinBatch, ParallelHashJoinBatchAccessor::outer_tuples, palloc0(), HashJoinTableData::parallel_state, ParallelHashJoinBatchInner, ParallelHashJoinBatchOuter, ParallelWorkerNumber, PHJ_BATCH_PROBING, ParallelHashJoinBatchAccessor::shared, SHARED_TUPLESTORE_SINGLE_PASS, snprintf(), and sts_initialize().

Referenced by ExecHashTableCreate(), and ExecParallelHashIncreaseNumBatches().

2875 {
2876  ParallelHashJoinState *pstate = hashtable->parallel_state;
2877  ParallelHashJoinBatch *batches;
2878  MemoryContext oldcxt;
2879  int i;
2880 
2881  Assert(hashtable->batches == NULL);
2882 
2883  /* Allocate space. */
2884  pstate->batches =
2885  dsa_allocate0(hashtable->area,
2886  EstimateParallelHashJoinBatch(hashtable) * nbatch);
2887  pstate->nbatch = nbatch;
2888  batches = dsa_get_address(hashtable->area, pstate->batches);
2889 
2890  /* Use hash join memory context. */
2891  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
2892 
2893  /* Allocate this backend's accessor array. */
2894  hashtable->nbatch = nbatch;
2895  hashtable->batches = (ParallelHashJoinBatchAccessor *)
2896  palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
2897 
2898  /* Set up the shared state, tuplestores and backend-local accessors. */
2899  for (i = 0; i < hashtable->nbatch; ++i)
2900  {
2901  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
2902  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
2903  char name[MAXPGPATH];
2904 
2905  /*
2906  * All members of shared were zero-initialized. We just need to set
2907  * up the Barrier.
2908  */
2909  BarrierInit(&shared->batch_barrier, 0);
2910  if (i == 0)
2911  {
2912  /* Batch 0 doesn't need to be loaded. */
2913  BarrierAttach(&shared->batch_barrier);
2914  while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING)
2915  BarrierArriveAndWait(&shared->batch_barrier, 0);
2916  BarrierDetach(&shared->batch_barrier);
2917  }
2918 
2919  /* Initialize accessor state. All members were zero-initialized. */
2920  accessor->shared = shared;
2921 
2922  /* Initialize the shared tuplestores. */
2923  snprintf(name, sizeof(name), "i%dof%d", i, hashtable->nbatch);
2924  accessor->inner_tuples =
2926  pstate->nparticipants,
2928  sizeof(uint32),
2930  &pstate->fileset,
2931  name);
2932  snprintf(name, sizeof(name), "o%dof%d", i, hashtable->nbatch);
2933  accessor->outer_tuples =
2935  pstate->nparticipants),
2936  pstate->nparticipants,
2938  sizeof(uint32),
2940  &pstate->fileset,
2941  name);
2942  }
2943 
2944  MemoryContextSwitchTo(oldcxt);
2945 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
SharedTuplestoreAccessor * sts_initialize(SharedTuplestore *sts, int participants, int my_participant_number, size_t meta_data_size, int flags, SharedFileSet *fileset, const char *name)
void BarrierInit(Barrier *barrier, int participants)
Definition: barrier.c:100
#define ParallelHashJoinBatchOuter(batch, nparticipants)
Definition: hashjoin.h:175
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
SharedFileSet fileset
Definition: hashjoin.h:253
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_area * area
Definition: hashjoin.h:355
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
dsa_pointer batches
Definition: hashjoin.h:236
#define ParallelHashJoinBatchInner(batch)
Definition: hashjoin.h:170
#define MAXPGPATH
#define dsa_allocate0(area, size)
Definition: dsa.h:88
int ParallelWorkerNumber
Definition: parallel.c:103
unsigned int uint32
Definition: c.h:314
#define SHARED_TUPLESTORE_SINGLE_PASS
int BarrierAttach(Barrier *barrier)
Definition: barrier.c:214
void * palloc0(Size size)
Definition: mcxt.c:864
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
#define PHJ_BATCH_PROBING
Definition: hashjoin.h:267
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define Assert(condition)
Definition: c.h:688
#define NthParallelHashJoinBatch(base, n)
Definition: hashjoin.h:186
bool BarrierDetach(Barrier *barrier)
Definition: barrier.c:234
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
const char * name
Definition: encode.c:521
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
bool BarrierArriveAndWait(Barrier *barrier, uint32 wait_event_info)
Definition: barrier.c:125
#define EstimateParallelHashJoinBatch(hashtable)
Definition: hashjoin.h:181
int i
MemoryContext hashCxt
Definition: hashjoin.h:347

◆ ExecParallelHashMergeCounters()

static void ExecParallelHashMergeCounters ( HashJoinTable  hashtable)
static

Definition at line 1396 of file nodeHash.c.

References HashJoinTableData::batches, ParallelHashJoinBatch::estimated_size, ParallelHashJoinBatchAccessor::estimated_size, i, ParallelHashJoinState::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), HashJoinTableData::nbatch, ParallelHashJoinBatch::ntuples, ParallelHashJoinBatchAccessor::ntuples, ParallelHashJoinBatch::old_ntuples, ParallelHashJoinBatchAccessor::old_ntuples, HashJoinTableData::parallel_state, ParallelHashJoinBatchAccessor::shared, ParallelHashJoinBatch::size, ParallelHashJoinBatchAccessor::size, and ParallelHashJoinState::total_tuples.

Referenced by ExecParallelHashIncreaseNumBatches(), and MultiExecParallelHash().

1397 {
1398  ParallelHashJoinState *pstate = hashtable->parallel_state;
1399  int i;
1400 
1401  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
1402  pstate->total_tuples = 0;
1403  for (i = 0; i < hashtable->nbatch; ++i)
1404  {
1405  ParallelHashJoinBatchAccessor *batch = &hashtable->batches[i];
1406 
1407  batch->shared->size += batch->size;
1408  batch->shared->estimated_size += batch->estimated_size;
1409  batch->shared->ntuples += batch->ntuples;
1410  batch->shared->old_ntuples += batch->old_ntuples;
1411  batch->size = 0;
1412  batch->estimated_size = 0;
1413  batch->ntuples = 0;
1414  batch->old_ntuples = 0;
1415  pstate->total_tuples += batch->shared->ntuples;
1416  }
1417  LWLockRelease(&pstate->lock);
1418 }
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1724
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1120
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
int i

◆ ExecParallelHashNextTuple()

static HashJoinTuple ExecParallelHashNextTuple ( HashJoinTable  table,
HashJoinTuple  tuple 
)
inlinestatic

Definition at line 3165 of file nodeHash.c.

References HashJoinTableData::area, Assert, dsa_get_address(), HashJoinTupleData::next, next, HashJoinTableData::parallel_state, and HashJoinTupleData::shared.

Referenced by ExecParallelScanHashBucket().

3166 {
3168 
3169  Assert(hashtable->parallel_state);
3170  next = (HashJoinTuple) dsa_get_address(hashtable->area, tuple->next.shared);
3171 
3172  return next;
3173 }
static int32 next
Definition: blutils.c:210
union HashJoinTupleData::@95 next
dsa_pointer shared
Definition: hashjoin.h:73
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
#define Assert(condition)
Definition: c.h:688

◆ ExecParallelHashPopChunkQueue()

static HashMemoryChunk ExecParallelHashPopChunkQueue ( HashJoinTable  table,
dsa_pointer shared 
)
static

Definition at line 3218 of file nodeHash.c.

References HashJoinTableData::area, ParallelHashJoinState::chunk_work_queue, dsa_get_address(), DsaPointerIsValid, ParallelHashJoinState::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), HashMemoryChunkData::next, HashJoinTableData::parallel_state, and HashMemoryChunkData::shared.

Referenced by ExecParallelHashIncreaseNumBuckets(), and ExecParallelHashRepartitionFirst().

3219 {
3220  ParallelHashJoinState *pstate = hashtable->parallel_state;
3221  HashMemoryChunk chunk;
3222 
3223  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3224  if (DsaPointerIsValid(pstate->chunk_work_queue))
3225  {
3226  *shared = pstate->chunk_work_queue;
3227  chunk = (HashMemoryChunk)
3228  dsa_get_address(hashtable->area, *shared);
3229  pstate->chunk_work_queue = chunk->next.shared;
3230  }
3231  else
3232  chunk = NULL;
3233  LWLockRelease(&pstate->lock);
3234 
3235  return chunk;
3236 }
dsa_pointer chunk_work_queue
Definition: hashjoin.h:242
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1724
dsa_pointer shared
Definition: hashjoin.h:127
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
union HashMemoryChunkData::@96 next
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1120
#define DsaPointerIsValid(x)
Definition: dsa.h:81

◆ ExecParallelHashPushTuple()

static void ExecParallelHashPushTuple ( dsa_pointer_atomic head,
HashJoinTuple  tuple,
dsa_pointer  tuple_shared 
)
inlinestatic

Definition at line 3179 of file nodeHash.c.

References dsa_pointer_atomic_compare_exchange, dsa_pointer_atomic_read, HashJoinTupleData::next, and HashJoinTupleData::shared.

Referenced by ExecParallelHashIncreaseNumBuckets(), ExecParallelHashRepartitionFirst(), ExecParallelHashTableInsert(), and ExecParallelHashTableInsertCurrentBatch().

3182 {
3183  for (;;)
3184  {
3185  tuple->next.shared = dsa_pointer_atomic_read(head);
3187  &tuple->next.shared,
3188  tuple_shared))
3189  break;
3190  }
3191 }
union HashJoinTupleData::@95 next
dsa_pointer shared
Definition: hashjoin.h:73
#define dsa_pointer_atomic_compare_exchange
Definition: dsa.h:68
#define dsa_pointer_atomic_read
Definition: dsa.h:65

◆ ExecParallelHashRepartitionFirst()

static void ExecParallelHashRepartitionFirst ( HashJoinTable  hashtable)
static

Definition at line 1269 of file nodeHash.c.

References HashJoinTableData::area, Assert, HashJoinTableData::batches, HashJoinTableData::buckets, CHECK_FOR_INTERRUPTS, dsa_free(), ParallelHashJoinBatchAccessor::estimated_size, ExecHashGetBucketAndBatch(), ExecParallelHashPopChunkQueue(), ExecParallelHashPushTuple(), ExecParallelHashTupleAlloc(), HASH_CHUNK_DATA, HashJoinTupleData::hashvalue, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, idx(), ParallelHashJoinBatchAccessor::inner_tuples, MAXALIGN, ParallelHashJoinState::nbatch, HashJoinTableData::nbatch, ParallelHashJoinBatchAccessor::ntuples, ParallelHashJoinBatchAccessor::old_ntuples, HashJoinTableData::parallel_state, HashJoinTableData::shared, sts_puttuple(), and MinimalTupleData::t_len.

Referenced by ExecParallelHashIncreaseNumBatches().

1270 {
1271  dsa_pointer chunk_shared;
1272  HashMemoryChunk chunk;
1273 
1274  Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);
1275 
1276  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared)))
1277  {
1278  size_t idx = 0;
1279 
1280  /* Repartition all tuples in this chunk. */
1281  while (idx < chunk->used)
1282  {
1283  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1284  MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
1285  HashJoinTuple copyTuple;
1286  dsa_pointer shared;
1287  int bucketno;
1288  int batchno;
1289 
1290  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1291  &bucketno, &batchno);
1292 
1293  Assert(batchno < hashtable->nbatch);
1294  if (batchno == 0)
1295  {
1296  /* It still belongs in batch 0. Copy to a new chunk. */
1297  copyTuple =
1298  ExecParallelHashTupleAlloc(hashtable,
1299  HJTUPLE_OVERHEAD + tuple->t_len,
1300  &shared);
1301  copyTuple->hashvalue = hashTuple->hashvalue;
1302  memcpy(HJTUPLE_MINTUPLE(copyTuple), tuple, tuple->t_len);
1303  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1304  copyTuple, shared);
1305  }
1306  else
1307  {
1308  size_t tuple_size =
1309  MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1310 
1311  /* It belongs in a later batch. */
1312  hashtable->batches[batchno].estimated_size += tuple_size;
1313  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1314  &hashTuple->hashvalue, tuple);
1315  }
1316 
1317  /* Count this tuple. */
1318  ++hashtable->batches[0].old_ntuples;
1319  ++hashtable->batches[batchno].ntuples;
1320 
1321  idx += MAXALIGN(HJTUPLE_OVERHEAD +
1322  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1323  }
1324 
1325  /* Free this chunk. */
1326  dsa_free(hashtable->area, chunk_shared);
1327 
1329  }
1330 }
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
void sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, MinimalTuple tuple)
Datum idx(PG_FUNCTION_ARGS)
Definition: _int_op.c:264
uint64 dsa_pointer
Definition: dsa.h:62
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_area * area
Definition: hashjoin.h:355
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table, dsa_pointer *shared)
Definition: nodeHash.c:3218
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
#define MAXALIGN(LEN)
Definition: c.h:641
union HashJoinTableData::@97 buckets
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:812
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3179
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2729
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecParallelHashRepartitionRest()

static void ExecParallelHashRepartitionRest ( HashJoinTable  hashtable)
static

Definition at line 1336 of file nodeHash.c.

References HashJoinTableData::area, HashJoinTableData::batches, CHECK_FOR_INTERRUPTS, dsa_get_address(), ParallelHashJoinBatchAccessor::estimated_size, ExecHashGetBucketAndBatch(), ParallelHashJoinState::fileset, HJTUPLE_OVERHEAD, i, ParallelHashJoinBatchAccessor::inner_tuples, MAXALIGN, NthParallelHashJoinBatch, ParallelHashJoinBatchAccessor::ntuples, ParallelHashJoinState::old_batches, ParallelHashJoinState::old_nbatch, ParallelHashJoinBatchAccessor::old_ntuples, palloc0(), HashJoinTableData::parallel_state, ParallelHashJoinBatchInner, ParallelWorkerNumber, pfree(), sts_attach(), sts_begin_parallel_scan(), sts_end_parallel_scan(), sts_parallel_scan_next(), sts_puttuple(), and MinimalTupleData::t_len.

Referenced by ExecParallelHashIncreaseNumBatches().

1337 {
1338  ParallelHashJoinState *pstate = hashtable->parallel_state;
1339  int old_nbatch = pstate->old_nbatch;
1340  SharedTuplestoreAccessor **old_inner_tuples;
1341  ParallelHashJoinBatch *old_batches;
1342  int i;
1343 
1344  /* Get our hands on the previous generation of batches. */
1345  old_batches = (ParallelHashJoinBatch *)
1346  dsa_get_address(hashtable->area, pstate->old_batches);
1347  old_inner_tuples = palloc0(sizeof(SharedTuplestoreAccessor *) * old_nbatch);
1348  for (i = 1; i < old_nbatch; ++i)
1349  {
1350  ParallelHashJoinBatch *shared =
1351  NthParallelHashJoinBatch(old_batches, i);
1352 
1353  old_inner_tuples[i] = sts_attach(ParallelHashJoinBatchInner(shared),
1355  &pstate->fileset);
1356  }
1357 
1358  /* Join in the effort to repartition them. */
1359  for (i = 1; i < old_nbatch; ++i)
1360  {
1361  MinimalTuple tuple;
1362  uint32 hashvalue;
1363 
1364  /* Scan one partition from the previous generation. */
1365  sts_begin_parallel_scan(old_inner_tuples[i]);
1366  while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue)))
1367  {
1368  size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1369  int bucketno;
1370  int batchno;
1371 
1372  /* Decide which partition it goes to in the new generation. */
1373  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
1374  &batchno);
1375 
1376  hashtable->batches[batchno].estimated_size += tuple_size;
1377  ++hashtable->batches[batchno].ntuples;
1378  ++hashtable->batches[i].old_ntuples;
1379 
1380  /* Store the tuple its new batch. */
1381  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1382  &hashvalue, tuple);
1383 
1385  }
1386  sts_end_parallel_scan(old_inner_tuples[i]);
1387  }
1388 
1389  pfree(old_inner_tuples);
1390 }
void sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, MinimalTuple tuple)
SharedFileSet fileset
Definition: hashjoin.h:253
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_area * area
Definition: hashjoin.h:355
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
void pfree(void *pointer)
Definition: mcxt.c:936
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
#define ParallelHashJoinBatchInner(batch)
Definition: hashjoin.h:170
SharedTuplestoreAccessor * sts_attach(SharedTuplestore *sts, int my_participant_number, SharedFileSet *fileset)
int ParallelWorkerNumber
Definition: parallel.c:103
unsigned int uint32
Definition: c.h:314
void sts_begin_parallel_scan(SharedTuplestoreAccessor *accessor)
void * palloc0(Size size)
Definition: mcxt.c:864
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
dsa_pointer old_batches
Definition: hashjoin.h:237
#define NthParallelHashJoinBatch(base, n)
Definition: hashjoin.h:186
#define MAXALIGN(LEN)
Definition: c.h:641
int i
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
MinimalTuple sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data)

◆ ExecParallelHashTableAlloc()

void ExecParallelHashTableAlloc ( HashJoinTable  hashtable,
int  batchno 
)

Definition at line 3035 of file nodeHash.c.

References HashJoinTableData::area, HashJoinTableData::batches, ParallelHashJoinBatch::buckets, dsa_allocate, dsa_get_address(), dsa_pointer_atomic_init, i, InvalidDsaPointer, ParallelHashJoinState::nbuckets, HashJoinTableData::parallel_state, and ParallelHashJoinBatchAccessor::shared.

Referenced by ExecHashTableCreate(), and ExecParallelHashJoinNewBatch().

3036 {
3037  ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
3038  dsa_pointer_atomic *buckets;
3039  int nbuckets = hashtable->parallel_state->nbuckets;
3040  int i;
3041 
3042  batch->buckets =
3043  dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
3044  buckets = (dsa_pointer_atomic *)
3045  dsa_get_address(hashtable->area, batch->buckets);
3046  for (i = 0; i < nbuckets; ++i)
3048 }
#define InvalidDsaPointer
Definition: dsa.h:78
dsa_area * area
Definition: hashjoin.h:355
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
#define dsa_pointer_atomic_init
Definition: dsa.h:64
int i
#define dsa_allocate(area, size)
Definition: dsa.h:84
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecParallelHashTableInsert()

void ExecParallelHashTableInsert ( HashJoinTable  hashtable,
TupleTableSlot slot,
uint32  hashvalue 
)

Definition at line 1673 of file nodeHash.c.

References Assert, BarrierPhase(), HashJoinTableData::batches, HashJoinTableData::buckets, ParallelHashJoinState::build_barrier, ExecFetchSlotMinimalTuple(), ExecHashGetBucketAndBatch(), ExecParallelHashPushTuple(), ExecParallelHashTupleAlloc(), ExecParallelHashTuplePrealloc(), HashJoinTupleData::hashvalue, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, ParallelHashJoinBatchAccessor::inner_tuples, MAXALIGN, ParallelHashJoinBatchAccessor::ntuples, HashJoinTableData::parallel_state, PHJ_BUILD_HASHING_INNER, ParallelHashJoinBatchAccessor::preallocated, HashJoinTableData::shared, sts_puttuple(), and MinimalTupleData::t_len.

Referenced by MultiExecParallelHash().

1676 {
1678  dsa_pointer shared;
1679  int bucketno;
1680  int batchno;
1681 
1682 retry:
1683  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1684 
1685  if (batchno == 0)
1686  {
1687  HashJoinTuple hashTuple;
1688 
1689  /* Try to load it into memory. */
1692  hashTuple = ExecParallelHashTupleAlloc(hashtable,
1693  HJTUPLE_OVERHEAD + tuple->t_len,
1694  &shared);
1695  if (hashTuple == NULL)
1696  goto retry;
1697 
1698  /* Store the hash value in the HashJoinTuple header. */
1699  hashTuple->hashvalue = hashvalue;
1700  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1701 
1702  /* Push it onto the front of the bucket's list */
1703  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1704  hashTuple, shared);
1705  }
1706  else
1707  {
1708  size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1709 
1710  Assert(batchno > 0);
1711 
1712  /* Try to preallocate space in the batch if necessary. */
1713  if (hashtable->batches[batchno].preallocated < tuple_size)
1714  {
1715  if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size))
1716  goto retry;
1717  }
1718 
1719  Assert(hashtable->batches[batchno].preallocated >= tuple_size);
1720  hashtable->batches[batchno].preallocated -= tuple_size;
1721  sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue,
1722  tuple);
1723  }
1724  ++hashtable->batches[batchno].ntuples;
1725 }
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
void sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, MinimalTuple tuple)
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot)
Definition: execTuples.c:688
uint64 dsa_pointer
Definition: dsa.h:62
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size)
Definition: nodeHash.c:3259
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
#define MAXALIGN(LEN)
Definition: c.h:641
union HashJoinTableData::@97 buckets
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3179
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2729
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecParallelHashTableInsertCurrentBatch()

void ExecParallelHashTableInsertCurrentBatch ( HashJoinTable  hashtable,
TupleTableSlot slot,
uint32  hashvalue 
)

Definition at line 1734 of file nodeHash.c.

References Assert, HashJoinTableData::buckets, HashJoinTableData::curbatch, ExecFetchSlotMinimalTuple(), ExecHashGetBucketAndBatch(), ExecParallelHashPushTuple(), ExecParallelHashTupleAlloc(), HashJoinTupleData::hashvalue, HeapTupleHeaderClearMatch, HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, HashJoinTableData::shared, and MinimalTupleData::t_len.

Referenced by ExecParallelHashJoinNewBatch().

1737 {
1739  HashJoinTuple hashTuple;
1740  dsa_pointer shared;
1741  int batchno;
1742  int bucketno;
1743 
1744  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1745  Assert(batchno == hashtable->curbatch);
1746  hashTuple = ExecParallelHashTupleAlloc(hashtable,
1747  HJTUPLE_OVERHEAD + tuple->t_len,
1748  &shared);
1749  hashTuple->hashvalue = hashvalue;
1750  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1752  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1753  hashTuple, shared);
1754 }
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot)
Definition: execTuples.c:688
uint64 dsa_pointer
Definition: dsa.h:62
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1874
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:688
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:532
union HashJoinTableData::@97 buckets
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3179
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2729
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecParallelHashTableSetCurrentBatch()

void ExecParallelHashTableSetCurrentBatch ( HashJoinTable  hashtable,
int  batchno 
)

Definition at line 3197 of file nodeHash.c.

References HashJoinTableData::area, Assert, ParallelHashJoinBatchAccessor::at_least_one_chunk, HashJoinTableData::batches, ParallelHashJoinBatch::buckets, HashJoinTableData::buckets, HashJoinTableData::curbatch, HashJoinTableData::current_chunk, HashJoinTableData::current_chunk_shared, dsa_get_address(), InvalidDsaPointer, HashJoinTableData::log2_nbuckets, my_log2(), ParallelHashJoinState::nbuckets, HashJoinTableData::nbuckets, HashJoinTableData::parallel_state, ParallelHashJoinBatchAccessor::shared, and HashJoinTableData::shared.

Referenced by ExecParallelHashIncreaseNumBatches(), ExecParallelHashIncreaseNumBuckets(), ExecParallelHashJoinNewBatch(), and MultiExecParallelHash().

3198 {
3199  Assert(hashtable->batches[batchno].shared->buckets != InvalidDsaPointer);
3200 
3201  hashtable->curbatch = batchno;
3202  hashtable->buckets.shared = (dsa_pointer_atomic *)
3203  dsa_get_address(hashtable->area,
3204  hashtable->batches[batchno].shared->buckets);
3205  hashtable->nbuckets = hashtable->parallel_state->nbuckets;
3206  hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
3207  hashtable->current_chunk = NULL;
3209  hashtable->batches[batchno].at_least_one_chunk = false;
3210 }
dsa_pointer current_chunk_shared
Definition: hashjoin.h:358
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
#define InvalidDsaPointer
Definition: dsa.h:78
dsa_area * area
Definition: hashjoin.h:355
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
int my_log2(long num)
Definition: dynahash.c:1716
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define Assert(condition)
Definition: c.h:688
union HashJoinTableData::@97 buckets
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
HashMemoryChunk current_chunk
Definition: hashjoin.h:354
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecParallelHashTupleAlloc()

static HashJoinTuple ExecParallelHashTupleAlloc ( HashJoinTable  hashtable,
size_t  size,
dsa_pointer shared 
)
static

Definition at line 2729 of file nodeHash.c.

References HashJoinTableData::area, Assert, ParallelHashJoinBatchAccessor::at_least_one_chunk, BarrierPhase(), HashJoinTableData::batches, ParallelHashJoinState::build_barrier, ParallelHashJoinBatch::chunks, HashJoinTableData::curbatch, HashJoinTableData::current_chunk, HashJoinTableData::current_chunk_shared, dsa_allocate, dsa_get_address(), ExecParallelHashIncreaseNumBatches(), ExecParallelHashIncreaseNumBuckets(), ParallelHashJoinState::growth, HASH_CHUNK_DATA, HASH_CHUNK_HEADER_SIZE, HASH_CHUNK_SIZE, HASH_CHUNK_THRESHOLD, ParallelHashJoinState::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXALIGN, HashMemoryChunkData::maxlen, HashJoinTableData::nbatch, HashJoinTableData::nbuckets, HashMemoryChunkData::next, NTUP_PER_BUCKET, ParallelHashJoinBatch::ntuples, ParallelHashJoinBatchAccessor::ntuples, HashJoinTableData::parallel_state, PHJ_BUILD_HASHING_INNER, PHJ_GROWTH_DISABLED, PHJ_GROWTH_NEED_MORE_BATCHES, PHJ_GROWTH_NEED_MORE_BUCKETS, HashMemoryChunkData::shared, ParallelHashJoinBatchAccessor::shared, ParallelHashJoinBatch::size, ParallelHashJoinState::space_allowed, ParallelHashJoinBatch::space_exhausted, and HashMemoryChunkData::used.

Referenced by ExecParallelHashRepartitionFirst(), ExecParallelHashTableInsert(), and ExecParallelHashTableInsertCurrentBatch().

2731 {
2732  ParallelHashJoinState *pstate = hashtable->parallel_state;
2733  dsa_pointer chunk_shared;
2734  HashMemoryChunk chunk;
2735  Size chunk_size;
2736  HashJoinTuple result;
2737  int curbatch = hashtable->curbatch;
2738 
2739  size = MAXALIGN(size);
2740 
2741  /*
2742  * Fast path: if there is enough space in this backend's current chunk,
2743  * then we can allocate without any locking.
2744  */
2745  chunk = hashtable->current_chunk;
2746  if (chunk != NULL &&
2747  size <= HASH_CHUNK_THRESHOLD &&
2748  chunk->maxlen - chunk->used >= size)
2749  {
2750 
2751  chunk_shared = hashtable->current_chunk_shared;
2752  Assert(chunk == dsa_get_address(hashtable->area, chunk_shared));
2753  *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE + chunk->used;
2754  result = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + chunk->used);
2755  chunk->used += size;
2756 
2757  Assert(chunk->used <= chunk->maxlen);
2758  Assert(result == dsa_get_address(hashtable->area, *shared));
2759 
2760  return result;
2761  }
2762 
2763  /* Slow path: try to allocate a new chunk. */
2764  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
2765 
2766  /*
2767  * Check if we need to help increase the number of buckets or batches.
2768  */
2769  if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
2771  {
2772  ParallelHashGrowth growth = pstate->growth;
2773 
2774  hashtable->current_chunk = NULL;
2775  LWLockRelease(&pstate->lock);
2776 
2777  /* Another participant has commanded us to help grow. */
2778  if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
2780  else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
2782 
2783  /* The caller must retry. */
2784  return NULL;
2785  }
2786 
2787  /* Oversized tuples get their own chunk. */
2788  if (size > HASH_CHUNK_THRESHOLD)
2789  chunk_size = size + HASH_CHUNK_HEADER_SIZE;
2790  else
2791  chunk_size = HASH_CHUNK_SIZE;
2792 
2793  /* Check if it's time to grow batches or buckets. */
2794  if (pstate->growth != PHJ_GROWTH_DISABLED)
2795  {
2796  Assert(curbatch == 0);
2798 
2799  /*
2800  * Check if our space limit would be exceeded. To avoid choking on
2801  * very large tuples or very low work_mem setting, we'll always allow
2802  * each backend to allocate at least one chunk.
2803  */
2804  if (hashtable->batches[0].at_least_one_chunk &&
2805  hashtable->batches[0].shared->size +
2806  chunk_size > pstate->space_allowed)
2807  {
2809  hashtable->batches[0].shared->space_exhausted = true;
2810  LWLockRelease(&pstate->lock);
2811 
2812  return NULL;
2813  }
2814 
2815  /* Check if our load factor limit would be exceeded. */
2816  if (hashtable->nbatch == 1)
2817  {
2818  hashtable->batches[0].shared->ntuples += hashtable->batches[0].ntuples;
2819  hashtable->batches[0].ntuples = 0;
2820  if (hashtable->batches[0].shared->ntuples + 1 >
2821  hashtable->nbuckets * NTUP_PER_BUCKET &&
2822  hashtable->nbuckets < (INT_MAX / 2))
2823  {
2825  LWLockRelease(&pstate->lock);
2826 
2827  return NULL;
2828  }
2829  }
2830  }
2831 
2832  /* We are cleared to allocate a new chunk. */
2833  chunk_shared = dsa_allocate(hashtable->area, chunk_size);
2834  hashtable->batches[curbatch].shared->size += chunk_size;
2835  hashtable->batches[curbatch].at_least_one_chunk = true;
2836 
2837  /* Set up the chunk. */
2838  chunk = (HashMemoryChunk) dsa_get_address(hashtable->area, chunk_shared);
2839  *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE;
2840  chunk->maxlen = chunk_size - HASH_CHUNK_HEADER_SIZE;
2841  chunk->used = size;
2842 
2843  /*
2844  * Push it onto the list of chunks, so that it can be found if we need to
2845  * increase the number of buckets or batches (batch 0 only) and later for
2846  * freeing the memory (all batches).
2847  */
2848  chunk->next.shared = hashtable->batches[curbatch].shared->chunks;
2849  hashtable->batches[curbatch].shared->chunks = chunk_shared;
2850 
2851  if (size <= HASH_CHUNK_THRESHOLD)
2852  {
2853  /*
2854  * Make this the current chunk so that we can use the fast path to
2855  * fill the rest of it up in future calls.
2856  */
2857  hashtable->current_chunk = chunk;
2858  hashtable->current_chunk_shared = chunk_shared;
2859  }
2860  LWLockRelease(&pstate->lock);
2861 
2862  Assert(HASH_CHUNK_DATA(chunk) == dsa_get_address(hashtable->area, *shared));
2863  result = (HashJoinTuple) HASH_CHUNK_DATA(chunk);
2864 
2865  return result;
2866 }
dsa_pointer current_chunk_shared
Definition: hashjoin.h:358
static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:1052
#define HASH_CHUNK_SIZE
Definition: hashjoin.h:139
dsa_pointer chunks
Definition: hashjoin.h:156
uint64 dsa_pointer
Definition: dsa.h:62
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1724
dsa_area * area
Definition: hashjoin.h:355
dsa_pointer shared
Definition: hashjoin.h:127
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:924
union HashMemoryChunkData::@96 next
#define HASH_CHUNK_THRESHOLD
Definition: hashjoin.h:143
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1719
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
#define NTUP_PER_BUCKET
Definition: nodeHash.c:658
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define Assert(condition)
Definition: c.h:688
ParallelHashGrowth growth
Definition: hashjoin.h:241
size_t Size
Definition: c.h:422
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1120
#define MAXALIGN(LEN)
Definition: c.h:641
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
HashMemoryChunk current_chunk
Definition: hashjoin.h:354
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
ParallelHashGrowth
Definition: hashjoin.h:218
static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
Definition: nodeHash.c:1489
#define dsa_allocate(area, size)
Definition: dsa.h:84

◆ ExecParallelHashTuplePrealloc()

static bool ExecParallelHashTuplePrealloc ( HashJoinTable  hashtable,
int  batchno,
size_t  size 
)
static

Definition at line 3259 of file nodeHash.c.

References Assert, ParallelHashJoinBatchAccessor::at_least_one_chunk, HashJoinTableData::batches, ParallelHashJoinBatch::estimated_size, ExecParallelHashIncreaseNumBatches(), ExecParallelHashIncreaseNumBuckets(), ParallelHashJoinState::growth, HASH_CHUNK_HEADER_SIZE, HASH_CHUNK_SIZE, ParallelHashJoinState::lock, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), Max, MAXALIGN, HashJoinTableData::parallel_state, PHJ_GROWTH_DISABLED, PHJ_GROWTH_NEED_MORE_BATCHES, PHJ_GROWTH_NEED_MORE_BUCKETS, ParallelHashJoinBatchAccessor::preallocated, ParallelHashJoinBatchAccessor::shared, ParallelHashJoinState::space_allowed, and ParallelHashJoinBatch::space_exhausted.

Referenced by ExecParallelHashTableInsert().

3260 {
3261  ParallelHashJoinState *pstate = hashtable->parallel_state;
3262  ParallelHashJoinBatchAccessor *batch = &hashtable->batches[batchno];
3263  size_t want = Max(size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);
3264 
3265  Assert(batchno > 0);
3266  Assert(batchno < hashtable->nbatch);
3267  Assert(size == MAXALIGN(size));
3268 
3269  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3270 
3271  /* Has another participant commanded us to help grow? */
3272  if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
3274  {
3275  ParallelHashGrowth growth = pstate->growth;
3276 
3277  LWLockRelease(&pstate->lock);
3278  if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
3280  else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3282 
3283  return false;
3284  }
3285 
3286  if (pstate->growth != PHJ_GROWTH_DISABLED &&
3287  batch->at_least_one_chunk &&
3288  (batch->shared->estimated_size + want + HASH_CHUNK_HEADER_SIZE
3289  > pstate->space_allowed))
3290  {
3291  /*
3292  * We have determined that this batch would exceed the space budget if
3293  * loaded into memory. Command all participants to help repartition.
3294  */
3295  batch->shared->space_exhausted = true;
3297  LWLockRelease(&pstate->lock);
3298 
3299  return false;
3300  }
3301 
3302  batch->at_least_one_chunk = true;
3303  batch->shared->estimated_size += want + HASH_CHUNK_HEADER_SIZE;
3304  batch->preallocated = want;
3305  LWLockRelease(&pstate->lock);
3306 
3307  return true;
3308 }
static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:1052
#define HASH_CHUNK_SIZE
Definition: hashjoin.h:139
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1724
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:356
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:357
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define Max(x, y)
Definition: c.h:840
#define Assert(condition)
Definition: c.h:688
ParallelHashGrowth growth
Definition: hashjoin.h:241
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1120
#define MAXALIGN(LEN)
Definition: c.h:641
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
ParallelHashGrowth
Definition: hashjoin.h:218
static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
Definition: nodeHash.c:1489

◆ ExecParallelScanHashBucket()

bool ExecParallelScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 1967 of file nodeHash.c.

References ExprContext::ecxt_innertuple, ExecParallelHashFirstTuple(), ExecParallelHashNextTuple(), ExecQualAndReset(), ExecStoreMinimalTuple(), HashJoinState::hashclauses, HashJoinTupleData::hashvalue, HashJoinState::hj_CurBucketNo, HashJoinState::hj_CurHashValue, HashJoinState::hj_CurTuple, HashJoinState::hj_HashTable, HashJoinState::hj_HashTupleSlot, and HJTUPLE_MINTUPLE.

Referenced by ExecHashJoinImpl().

1969 {
1970  ExprState *hjclauses = hjstate->hashclauses;
1971  HashJoinTable hashtable = hjstate->hj_HashTable;
1972  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1973  uint32 hashvalue = hjstate->hj_CurHashValue;
1974 
1975  /*
1976  * hj_CurTuple is the address of the tuple last returned from the current
1977  * bucket, or NULL if it's time to start scanning a new bucket.
1978  */
1979  if (hashTuple != NULL)
1980  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
1981  else
1982  hashTuple = ExecParallelHashFirstTuple(hashtable,
1983  hjstate->hj_CurBucketNo);
1984 
1985  while (hashTuple != NULL)
1986  {
1987  if (hashTuple->hashvalue == hashvalue)
1988  {
1989  TupleTableSlot *inntuple;
1990 
1991  /* insert hashtable's tuple into exec slot so ExecQual sees it */
1992  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
1993  hjstate->hj_HashTupleSlot,
1994  false); /* do not pfree */
1995  econtext->ecxt_innertuple = inntuple;
1996 
1997  if (ExecQualAndReset(hjclauses, econtext))
1998  {
1999  hjstate->hj_CurTuple = hashTuple;
2000  return true;
2001  }
2002  }
2003 
2004  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2005  }
2006 
2007  /*
2008  * no match
2009  */
2010  return false;
2011 }
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:420
uint32 hj_CurHashValue
Definition: execnodes.h:1730
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1733
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:212
unsigned int uint32
Definition: c.h:314
int hj_CurBucketNo
Definition: execnodes.h:1731
static bool ExecQualAndReset(ExprState *state, ExprContext *econtext)
Definition: executor.h:389
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
static HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable table, int bucketno)
Definition: nodeHash.c:3149
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:1735
static HashJoinTuple ExecParallelHashNextTuple(HashJoinTable table, HashJoinTuple tuple)
Definition: nodeHash.c:3165
HashJoinTable hj_HashTable
Definition: execnodes.h:1729
uint32 hashvalue
Definition: hashjoin.h:75
ExprState * hashclauses
Definition: execnodes.h:1725

◆ ExecPrepHashTableForUnmatched()

void ExecPrepHashTableForUnmatched ( HashJoinState hjstate)

Definition at line 2018 of file nodeHash.c.

References HashJoinState::hj_CurBucketNo, HashJoinState::hj_CurSkewBucketNo, and HashJoinState::hj_CurTuple.

Referenced by ExecHashJoinImpl().

2019 {
2020  /*----------
2021  * During this scan we use the HashJoinState fields as follows:
2022  *
2023  * hj_CurBucketNo: next regular bucket to scan
2024  * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
2025  * hj_CurTuple: last tuple returned, or NULL to start next bucket
2026  *----------
2027  */
2028  hjstate->hj_CurBucketNo = 0;
2029  hjstate->hj_CurSkewBucketNo = 0;
2030  hjstate->hj_CurTuple = NULL;
2031 }
int hj_CurSkewBucketNo
Definition: execnodes.h:1732
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1733
int hj_CurBucketNo
Definition: execnodes.h:1731

◆ ExecReScanHash()

void ExecReScanHash ( HashState node)

Definition at line 2168 of file nodeHash.c.

References PlanState::chgParam, ExecReScan(), PlanState::lefttree, and HashState::ps.

Referenced by ExecReScan().

2169 {
2170  /*
2171  * if chgParam of subnode is not null then plan will be re-scanned by
2172  * first ExecProcNode.
2173  */
2174  if (node->ps.lefttree->chgParam == NULL)
2175  ExecReScan(node->ps.lefttree);
2176 }
void ExecReScan(PlanState *node)
Definition: execAmi.c:76
struct PlanState * lefttree
Definition: execnodes.h:887
PlanState ps
Definition: execnodes.h:2058
Bitmapset * chgParam
Definition: execnodes.h:896

◆ ExecScanHashBucket()

bool ExecScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 1906 of file nodeHash.c.

References HashJoinTableData::buckets, ExprContext::ecxt_innertuple, ExecQualAndReset(), ExecStoreMinimalTuple(), HashJoinState::hashclauses, HashJoinTupleData::hashvalue, HashJoinState::hj_CurBucketNo, HashJoinState::hj_CurHashValue, HashJoinState::hj_CurSkewBucketNo, HashJoinState::hj_CurTuple, HashJoinState::hj_HashTable, HashJoinState::hj_HashTupleSlot, HJTUPLE_MINTUPLE, INVALID_SKEW_BUCKET_NO, HashJoinTupleData::next, HashJoinTableData::skewBucket, HashSkewBucket::tuples, HashJoinTupleData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashJoinImpl().

1908 {
1909  ExprState *hjclauses = hjstate->hashclauses;
1910  HashJoinTable hashtable = hjstate->hj_HashTable;
1911  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1912  uint32 hashvalue = hjstate->hj_CurHashValue;
1913 
1914  /*
1915  * hj_CurTuple is the address of the tuple last returned from the current
1916  * bucket, or NULL if it's time to start scanning a new bucket.
1917  *
1918  * If the tuple hashed to a skew bucket then scan the skew bucket
1919  * otherwise scan the standard hashtable bucket.
1920  */
1921  if (hashTuple != NULL)
1922  hashTuple = hashTuple->next.unshared;
1923  else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
1924  hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
1925  else
1926  hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
1927 
1928  while (hashTuple != NULL)
1929  {
1930  if (hashTuple->hashvalue == hashvalue)
1931  {
1932  TupleTableSlot *inntuple;
1933 
1934  /* insert hashtable's tuple into exec slot so ExecQual sees it */
1935  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
1936  hjstate->hj_HashTupleSlot,
1937  false); /* do not pfree */
1938  econtext->ecxt_innertuple = inntuple;
1939 
1940  if (ExecQualAndReset(hjclauses, econtext))
1941  {
1942  hjstate->hj_CurTuple = hashTuple;
1943  return true;
1944  }
1945  }
1946 
1947  hashTuple = hashTuple->next.unshared;
1948  }
1949 
1950  /*
1951  * no match
1952  */
1953  return false;
1954 }
#define INVALID_SKEW_BUCKET_NO
Definition: hashjoin.h:109
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:420
union HashJoinTupleData::@95 next
uint32 hj_CurHashValue
Definition: execnodes.h:1730
int hj_CurSkewBucketNo
Definition: execnodes.h:1732
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1733
HashJoinTuple tuples
Definition: hashjoin.h:105
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:212
unsigned int uint32
Definition: c.h:314
int hj_CurBucketNo
Definition: execnodes.h:1731
static bool ExecQualAndReset(ExprState *state, ExprContext *econtext)
Definition: executor.h:389
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
union HashJoinTableData::@97 buckets
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:1735
HashJoinTable hj_HashTable
Definition: execnodes.h:1729
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
uint32 hashvalue
Definition: hashjoin.h:75
ExprState * hashclauses
Definition: execnodes.h:1725

◆ ExecScanHashTableForUnmatched()

bool ExecScanHashTableForUnmatched ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 2042 of file nodeHash.c.

References HashJoinTableData::buckets, CHECK_FOR_INTERRUPTS, ExprContext::ecxt_innertuple, ExecStoreMinimalTuple(), HeapTupleHeaderHasMatch, HashJoinState::hj_CurBucketNo, HashJoinState::hj_CurSkewBucketNo, HashJoinState::hj_CurTuple, HashJoinState::hj_HashTable, HashJoinState::hj_HashTupleSlot, HJTUPLE_MINTUPLE, HashJoinTableData::nbuckets, HashJoinTupleData::next, HashJoinTableData::nSkewBuckets, ResetExprContext, HashJoinTableData::skewBucket, HashJoinTableData::skewBucketNums, HashSkewBucket::tuples, HashJoinTupleData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashJoinImpl().

2043 {
2044  HashJoinTable hashtable = hjstate->hj_HashTable;
2045  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2046 
2047  for (;;)
2048  {
2049  /*
2050  * hj_CurTuple is the address of the tuple last returned from the
2051  * current bucket, or NULL if it's time to start scanning a new
2052  * bucket.
2053  */
2054  if (hashTuple != NULL)
2055  hashTuple = hashTuple->next.unshared;
2056  else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
2057  {
2058  hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
2059  hjstate->hj_CurBucketNo++;
2060  }
2061  else if (hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
2062  {
2063  int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
2064 
2065  hashTuple = hashtable->skewBucket[j]->tuples;
2066  hjstate->hj_CurSkewBucketNo++;
2067  }
2068  else
2069  break; /* finished all buckets */
2070 
2071  while (hashTuple != NULL)
2072  {
2073  if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
2074  {
2075  TupleTableSlot *inntuple;
2076 
2077  /* insert hashtable's tuple into exec slot */
2078  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2079  hjstate->hj_HashTupleSlot,
2080  false); /* do not pfree */
2081  econtext->ecxt_innertuple = inntuple;
2082 
2083  /*
2084  * Reset temp memory each time; although this function doesn't
2085  * do any qual eval, the caller will, so let's keep it
2086  * parallel to ExecScanHashBucket.
2087  */
2088  ResetExprContext(econtext);
2089 
2090  hjstate->hj_CurTuple = hashTuple;
2091  return true;
2092  }
2093 
2094  hashTuple = hashTuple->next.unshared;
2095  }
2096 
2097  /* allow this loop to be cancellable */
2099  }
2100 
2101  /*
2102  * no more unmatched tuples
2103  */
2104  return false;
2105 }
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:420
union HashJoinTupleData::@95 next
int * skewBucketNums
Definition: hashjoin.h:308
int hj_CurSkewBucketNo
Definition: execnodes.h:1732
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1733
HashJoinTuple tuples
Definition: hashjoin.h:105
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:212
int hj_CurBucketNo
Definition: execnodes.h:1731
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define HeapTupleHeaderHasMatch(tup)
Definition: htup_details.h:522
union HashJoinTableData::@97 buckets
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:1735
HashJoinTable hj_HashTable
Definition: execnodes.h:1729
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
#define ResetExprContext(econtext)
Definition: executor.h:484

◆ ExecShutdownHash()

void ExecShutdownHash ( HashState node)

Definition at line 2604 of file nodeHash.c.

References ExecHashGetInstrumentation(), HashState::hashtable, and HashState::hinstrument.

Referenced by ExecShutdownNode().

2605 {
2606  if (node->hinstrument && node->hashtable)
2608 }
void ExecHashGetInstrumentation(HashInstrumentation *instrument, HashJoinTable hashtable)
Definition: nodeHash.c:2635
HashJoinTable hashtable
Definition: execnodes.h:2059
HashInstrumentation * hinstrument
Definition: execnodes.h:2064

◆ MultiExecHash()

Node* MultiExecHash ( HashState node)

Definition at line 105 of file nodeHash.c.

References HashState::hashtable, InstrStartNode(), InstrStopNode(), PlanState::instrument, MultiExecParallelHash(), MultiExecPrivateHash(), HashState::parallel_state, HashJoinTableData::partialTuples, and HashState::ps.

Referenced by MultiExecProcNode().

106 {
107  /* must provide our own instrumentation support */
108  if (node->ps.instrument)
110 
111  if (node->parallel_state != NULL)
112  MultiExecParallelHash(node);
113  else
114  MultiExecPrivateHash(node);
115 
116  /* must provide our own instrumentation support */
117  if (node->ps.instrument)
119 
120  /*
121  * We do not return the hash table directly because it's not a subtype of
122  * Node, and so would violate the MultiExecProcNode API. Instead, our
123  * parent Hashjoin node is expected to know how to fish it out of our node
124  * state. Ugly but not really worth cleaning up, since Hashjoin knows
125  * quite a bit more about Hash besides that.
126  */
127  return NULL;
128 }
struct ParallelHashJoinState * parallel_state
Definition: execnodes.h:2067
void InstrStopNode(Instrumentation *instr, double nTuples)
Definition: instrument.c:80
Instrumentation * instrument
Definition: execnodes.h:878
HashJoinTable hashtable
Definition: execnodes.h:2059
static void MultiExecPrivateHash(HashState *node)
Definition: nodeHash.c:138
double partialTuples
Definition: hashjoin.h:319
void InstrStartNode(Instrumentation *instr)
Definition: instrument.c:63
PlanState ps
Definition: execnodes.h:2058
static void MultiExecParallelHash(HashState *node)
Definition: nodeHash.c:213

◆ MultiExecParallelHash()

static void MultiExecParallelHash ( HashState node)
static

Definition at line 213 of file nodeHash.c.

References Assert, BarrierArriveAndWait(), BarrierAttach(), BarrierDetach(), BarrierPhase(), HashJoinTableData::batches, ParallelHashJoinState::build_barrier, HashJoinTableData::curbatch, ExprContext::ecxt_innertuple, ExecHashGetHashValue(), ExecParallelHashEnsureBatchAccessors(), ExecParallelHashIncreaseNumBatches(), ExecParallelHashIncreaseNumBuckets(), ExecParallelHashMergeCounters(), ExecParallelHashTableInsert(), ExecParallelHashTableSetCurrentBatch(), ExecProcNode(), ParallelHashJoinState::grow_batches_barrier, ParallelHashJoinState::grow_buckets_barrier, ParallelHashJoinState::growth, HashState::hashkeys, HashState::hashtable, i, ParallelHashJoinBatchAccessor::inner_tuples, HashJoinTableData::keepNulls, HashJoinTableData::log2_nbuckets, my_log2(), HashJoinTableData::nbatch, ParallelHashJoinState::nbuckets, HashJoinTableData::nbuckets, outerPlanState, HashJoinTableData::parallel_state, HashJoinTableData::partialTuples, PHJ_BUILD_ALLOCATING, PHJ_BUILD_DONE, PHJ_BUILD_HASHING_INNER, PHJ_BUILD_HASHING_OUTER, PHJ_GROW_BATCHES_ELECTING, PHJ_GROW_BATCHES_PHASE, PHJ_GROW_BUCKETS_ELECTING, PHJ_GROW_BUCKETS_PHASE, PHJ_GROWTH_DISABLED, HashState::ps, PlanState::ps_ExprContext, sts_end_write(), ParallelHashJoinState::total_tuples, HashJoinTableData::totalTuples, TupIsNull, WAIT_EVENT_HASH_BUILD_ALLOCATING, and WAIT_EVENT_HASH_BUILD_HASHING_INNER.

Referenced by MultiExecHash().

214 {
215  ParallelHashJoinState *pstate;
216  PlanState *outerNode;
217  List *hashkeys;
218  HashJoinTable hashtable;
219  TupleTableSlot *slot;
220  ExprContext *econtext;
221  uint32 hashvalue;
222  Barrier *build