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, List *hashCollations, 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 2670 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().

2671 {
2672  HashMemoryChunk newChunk;
2673  char *ptr;
2674 
2675  /* just in case the size is not already aligned properly */
2676  size = MAXALIGN(size);
2677 
2678  /*
2679  * If tuple size is larger than threshold, allocate a separate chunk.
2680  */
2681  if (size > HASH_CHUNK_THRESHOLD)
2682  {
2683  /* allocate new chunk and put it at the beginning of the list */
2684  newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
2685  HASH_CHUNK_HEADER_SIZE + size);
2686  newChunk->maxlen = size;
2687  newChunk->used = size;
2688  newChunk->ntuples = 1;
2689 
2690  /*
2691  * Add this chunk to the list after the first existing chunk, so that
2692  * we don't lose the remaining space in the "current" chunk.
2693  */
2694  if (hashtable->chunks != NULL)
2695  {
2696  newChunk->next = hashtable->chunks->next;
2697  hashtable->chunks->next.unshared = newChunk;
2698  }
2699  else
2700  {
2701  newChunk->next.unshared = hashtable->chunks;
2702  hashtable->chunks = newChunk;
2703  }
2704 
2705  return HASH_CHUNK_DATA(newChunk);
2706  }
2707 
2708  /*
2709  * See if we have enough space for it in the current chunk (if any). If
2710  * not, allocate a fresh chunk.
2711  */
2712  if ((hashtable->chunks == NULL) ||
2713  (hashtable->chunks->maxlen - hashtable->chunks->used) < size)
2714  {
2715  /* allocate new chunk and put it at the beginning of the list */
2716  newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
2718 
2719  newChunk->maxlen = HASH_CHUNK_SIZE;
2720  newChunk->used = size;
2721  newChunk->ntuples = 1;
2722 
2723  newChunk->next.unshared = hashtable->chunks;
2724  hashtable->chunks = newChunk;
2725 
2726  return HASH_CHUNK_DATA(newChunk);
2727  }
2728 
2729  /* There is enough space in the current chunk, let's add the tuple */
2730  ptr = HASH_CHUNK_DATA(hashtable->chunks) + hashtable->chunks->used;
2731  hashtable->chunks->used += size;
2732  hashtable->chunks->ntuples += 1;
2733 
2734  /* return pointer to the start of the tuple memory */
2735  return ptr;
2736 }
#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:349
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define MAXALIGN(LEN)
Definition: c.h:685
HashMemoryChunk chunks
Definition: hashjoin.h:352
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:771
#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 665 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().

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

◆ 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:538
void ExecFreeExprContext(PlanState *planstate)
Definition: execUtils.c:617
#define outerPlanState(node)
Definition: execnodes.h:1039
PlanState ps
Definition: execnodes.h:2222
#define outerPlan(node)
Definition: plannodes.h:170

◆ 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(elevel,...)
Definition: elog.h:226

◆ ExecHashBuildSkewHash()

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

Definition at line 2204 of file nodeHash.c.

References ATTSTATSSLOT_NUMBERS, ATTSTATSSLOT_VALUES, HashJoinTableData::batchCxt, BoolGetDatum, HashJoinTableData::collations, DatumGetUInt32, free_attstatsslot(), FunctionCall1Coll(), 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, STATRELATTINH, HashSkewBucket::tuples, and AttStatsSlot::values.

Referenced by ExecHashTableCreate().

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

◆ ExecHashEstimate()

void ExecHashEstimate ( HashState node,
ParallelContext pcxt 
)

Definition at line 2564 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().

2565 {
2566  size_t size;
2567 
2568  /* don't need this if not instrumenting or no workers */
2569  if (!node->ps.instrument || pcxt->nworkers == 0)
2570  return;
2571 
2572  size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
2573  size = add_size(size, offsetof(SharedHashInfo, hinstrument));
2574  shm_toc_estimate_chunk(&pcxt->estimator, size);
2575  shm_toc_estimate_keys(&pcxt->estimator, 1);
2576 }
Instrumentation * instrument
Definition: execnodes.h:955
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:2222
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:655

◆ ExecHashGetBucketAndBatch()

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

Definition at line 1890 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().

1894 {
1895  uint32 nbuckets = (uint32) hashtable->nbuckets;
1896  uint32 nbatch = (uint32) hashtable->nbatch;
1897 
1898  if (nbatch > 1)
1899  {
1900  /* we can do MOD by masking, DIV by shifting */
1901  *bucketno = hashvalue & (nbuckets - 1);
1902  *batchno = (hashvalue >> hashtable->log2_nbuckets) & (nbatch - 1);
1903  }
1904  else
1905  {
1906  *bucketno = hashvalue & (nbuckets - 1);
1907  *batchno = 0;
1908  }
1909 }
unsigned int uint32
Definition: c.h:358

◆ ExecHashGetHashValue()

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

Definition at line 1786 of file nodeHash.c.

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

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

1792 {
1793  uint32 hashkey = 0;
1794  FmgrInfo *hashfunctions;
1795  ListCell *hk;
1796  int i = 0;
1797  MemoryContext oldContext;
1798 
1799  /*
1800  * We reset the eval context each time to reclaim any memory leaked in the
1801  * hashkey expressions.
1802  */
1803  ResetExprContext(econtext);
1804 
1805  oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
1806 
1807  if (outer_tuple)
1808  hashfunctions = hashtable->outer_hashfunctions;
1809  else
1810  hashfunctions = hashtable->inner_hashfunctions;
1811 
1812  foreach(hk, hashkeys)
1813  {
1814  ExprState *keyexpr = (ExprState *) lfirst(hk);
1815  Datum keyval;
1816  bool isNull;
1817 
1818  /* rotate hashkey left 1 bit at each step */
1819  hashkey = (hashkey << 1) | ((hashkey & 0x80000000) ? 1 : 0);
1820 
1821  /*
1822  * Get the join attribute value of the tuple
1823  */
1824  keyval = ExecEvalExpr(keyexpr, econtext, &isNull);
1825 
1826  /*
1827  * If the attribute is NULL, and the join operator is strict, then
1828  * this tuple cannot pass the join qual so we can reject it
1829  * immediately (unless we're scanning the outside of an outer join, in
1830  * which case we must not reject it). Otherwise we act like the
1831  * hashcode of NULL is zero (this will support operators that act like
1832  * IS NOT DISTINCT, though not any more-random behavior). We treat
1833  * the hash support function as strict even if the operator is not.
1834  *
1835  * Note: currently, all hashjoinable operators must be strict since
1836  * the hash index AM assumes that. However, it takes so little extra
1837  * code here to allow non-strict that we may as well do it.
1838  */
1839  if (isNull)
1840  {
1841  if (hashtable->hashStrict[i] && !keep_nulls)
1842  {
1843  MemoryContextSwitchTo(oldContext);
1844  return false; /* cannot match */
1845  }
1846  /* else, leave hashkey unmodified, equivalent to hashcode 0 */
1847  }
1848  else
1849  {
1850  /* Compute the hash function */
1851  uint32 hkey;
1852 
1853  hkey = DatumGetUInt32(FunctionCall1Coll(&hashfunctions[i], hashtable->collations[i], keyval));
1854  hashkey ^= hkey;
1855  }
1856 
1857  i++;
1858  }
1859 
1860  MemoryContextSwitchTo(oldContext);
1861 
1862  *hashvalue = hashkey;
1863  return true;
1864 }
#define DatumGetUInt32(X)
Definition: postgres.h:486
Definition: fmgr.h:56
MemoryContext ecxt_per_tuple_memory
Definition: execnodes.h:231
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:284
unsigned int uint32
Definition: c.h:358
FmgrInfo * outer_hashfunctions
Definition: hashjoin.h:337
uintptr_t Datum
Definition: postgres.h:367
Datum FunctionCall1Coll(FmgrInfo *flinfo, Oid collation, Datum arg1)
Definition: fmgr.c:1130
#define lfirst(lc)
Definition: pg_list.h:106
int i
bool * hashStrict
Definition: hashjoin.h:339
#define ResetExprContext(econtext)
Definition: executor.h:494

◆ ExecHashGetInstrumentation()

void ExecHashGetInstrumentation ( HashInstrumentation instrument,
HashJoinTable  hashtable 
)

◆ ExecHashGetSkewBucket()

int ExecHashGetSkewBucket ( HashJoinTable  hashtable,
uint32  hashvalue 
)

Definition at line 2359 of file nodeHash.c.

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

Referenced by ExecHashJoinImpl(), and MultiExecPrivateHash().

2360 {
2361  int bucket;
2362 
2363  /*
2364  * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
2365  * particular, this happens after the initial batch is done).
2366  */
2367  if (!hashtable->skewEnabled)
2368  return INVALID_SKEW_BUCKET_NO;
2369 
2370  /*
2371  * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
2372  */
2373  bucket = hashvalue & (hashtable->skewBucketLen - 1);
2374 
2375  /*
2376  * While we have not hit a hole in the hashtable and have not hit the
2377  * desired bucket, we have collided with some other hash value, so try the
2378  * next bucket location.
2379  */
2380  while (hashtable->skewBucket[bucket] != NULL &&
2381  hashtable->skewBucket[bucket]->hashvalue != hashvalue)
2382  bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
2383 
2384  /*
2385  * Found the desired bucket?
2386  */
2387  if (hashtable->skewBucket[bucket] != NULL)
2388  return bucket;
2389 
2390  /*
2391  * There must not be any hashtable entry for this hash value.
2392  */
2393  return INVALID_SKEW_BUCKET_NO;
2394 }
#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 885 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(), printf, repalloc(), HashJoinTableData::spaceUsed, MinimalTupleData::t_len, HashJoinTupleData::unshared, HashMemoryChunkData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashSkewTableInsert(), and ExecHashTableInsert().

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

Referenced by MultiExecPrivateHash().

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

2584 {
2585  size_t size;
2586 
2587  /* don't need this if not instrumenting or no workers */
2588  if (!node->ps.instrument || pcxt->nworkers == 0)
2589  return;
2590 
2591  size = offsetof(SharedHashInfo, hinstrument) +
2592  pcxt->nworkers * sizeof(HashInstrumentation);
2593  node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
2594  memset(node->shared_info, 0, size);
2595  node->shared_info->num_workers = pcxt->nworkers;
2596  shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id,
2597  node->shared_info);
2598 }
Instrumentation * instrument
Definition: execnodes.h:955
struct HashInstrumentation HashInstrumentation
int plan_node_id
Definition: plannodes.h:139
SharedHashInfo * shared_info
Definition: execnodes.h:2227
PlanState ps
Definition: execnodes.h:2222
Plan * plan
Definition: execnodes.h:945
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:655
shm_toc * toc
Definition: parallel.h:44

◆ ExecHashInitializeWorker()

void ExecHashInitializeWorker ( HashState node,
ParallelWorkerContext pwcxt 
)

Definition at line 2605 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().

2606 {
2607  SharedHashInfo *shared_info;
2608 
2609  /* don't need this if not instrumenting */
2610  if (!node->ps.instrument)
2611  return;
2612 
2613  shared_info = (SharedHashInfo *)
2614  shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
2615  node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
2616 }
Instrumentation * instrument
Definition: execnodes.h:955
int plan_node_id
Definition: plannodes.h:139
int ParallelWorkerNumber
Definition: parallel.c:110
PlanState ps
Definition: execnodes.h:2222
HashInstrumentation * hinstrument
Definition: execnodes.h:2228
HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER]
Definition: execnodes.h:2213
Plan * plan
Definition: execnodes.h:945
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 2451 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().

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

◆ ExecHashRetrieveInstrumentation()

void ExecHashRetrieveInstrumentation ( HashState node)

Definition at line 2636 of file nodeHash.c.

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

Referenced by ExecParallelRetrieveInstrumentation().

2637 {
2638  SharedHashInfo *shared_info = node->shared_info;
2639  size_t size;
2640 
2641  if (shared_info == NULL)
2642  return;
2643 
2644  /* Replace node->shared_info with a copy in backend-local memory. */
2645  size = offsetof(SharedHashInfo, hinstrument) +
2646  shared_info->num_workers * sizeof(HashInstrumentation);
2647  node->shared_info = palloc(size);
2648  memcpy(node->shared_info, shared_info, size);
2649 }
struct HashInstrumentation HashInstrumentation
SharedHashInfo * shared_info
Definition: execnodes.h:2227
void * palloc(Size size)
Definition: mcxt.c:924
#define offsetof(type, field)
Definition: c.h:655

◆ ExecHashSkewTableInsert()

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

Definition at line 2405 of file nodeHash.c.

References Assert, HashJoinTableData::batchCxt, ExecFetchSlotMinimalTuple(), ExecHashIncreaseNumBatches(), ExecHashRemoveNextSkewBucket(), HashJoinTupleData::hashvalue, heap_free_minimal_tuple(), 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().

2409 {
2410  bool shouldFree;
2411  MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
2412  HashJoinTuple hashTuple;
2413  int hashTupleSize;
2414 
2415  /* Create the HashJoinTuple */
2416  hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
2417  hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
2418  hashTupleSize);
2419  hashTuple->hashvalue = hashvalue;
2420  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
2422 
2423  /* Push it onto the front of the skew bucket's list */
2424  hashTuple->next.unshared = hashtable->skewBucket[bucketNumber]->tuples;
2425  hashtable->skewBucket[bucketNumber]->tuples = hashTuple;
2426  Assert(hashTuple != hashTuple->next.unshared);
2427 
2428  /* Account for space used, and back off if we've used too much */
2429  hashtable->spaceUsed += hashTupleSize;
2430  hashtable->spaceUsedSkew += hashTupleSize;
2431  if (hashtable->spaceUsed > hashtable->spacePeak)
2432  hashtable->spacePeak = hashtable->spaceUsed;
2433  while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
2434  ExecHashRemoveNextSkewBucket(hashtable);
2435 
2436  /* Check we are not over the total spaceAllowed, either */
2437  if (hashtable->spaceUsed > hashtable->spaceAllowed)
2438  ExecHashIncreaseNumBatches(hashtable);
2439 
2440  if (shouldFree)
2441  heap_free_minimal_tuple(tuple);
2442 }
static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
Definition: nodeHash.c:2451
union HashJoinTupleData::@95 next
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot, bool *shouldFree)
Definition: execTuples.c:1656
static void ExecHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:885
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
void heap_free_minimal_tuple(MinimalTuple mtup)
Definition: heaptuple.c:1427
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1878
Size spaceAllowedSkew
Definition: hashjoin.h:346
HashJoinTuple tuples
Definition: hashjoin.h:105
MemoryContext batchCxt
Definition: hashjoin.h:349
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:732
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:526
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:771
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecHashTableCreate()

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

◆ ExecHashTableDestroy()

void ExecHashTableDestroy ( HashJoinTable  hashtable)

Definition at line 852 of file nodeHash.c.

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

Referenced by ExecEndHashJoin(), and ExecReScanHashJoin().

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

◆ ExecHashTableDetach()

void ExecHashTableDetach ( HashJoinTable  hashtable)

Definition at line 3136 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().

3137 {
3138  if (hashtable->parallel_state)
3139  {
3140  ParallelHashJoinState *pstate = hashtable->parallel_state;
3141  int i;
3142 
3143  /* Make sure any temporary files are closed. */
3144  if (hashtable->batches)
3145  {
3146  for (i = 0; i < hashtable->nbatch; ++i)
3147  {
3148  sts_end_write(hashtable->batches[i].inner_tuples);
3149  sts_end_write(hashtable->batches[i].outer_tuples);
3152  }
3153  }
3154 
3155  /* If we're last to detach, clean up shared memory. */
3156  if (BarrierDetach(&pstate->build_barrier))
3157  {
3158  if (DsaPointerIsValid(pstate->batches))
3159  {
3160  dsa_free(hashtable->area, pstate->batches);
3161  pstate->batches = InvalidDsaPointer;
3162  }
3163  }
3164 
3165  hashtable->parallel_state = NULL;
3166  }
3167 }
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:356
dsa_pointer batches
Definition: hashjoin.h:236
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
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:820
int i
void sts_end_write(SharedTuplestoreAccessor *accessor)

◆ ExecHashTableDetachBatch()

void ExecHashTableDetachBatch ( HashJoinTable  hashtable)

Definition at line 3079 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().

3080 {
3081  if (hashtable->parallel_state != NULL &&
3082  hashtable->curbatch >= 0)
3083  {
3084  int curbatch = hashtable->curbatch;
3085  ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
3086 
3087  /* Make sure any temporary files are closed. */
3088  sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
3089  sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
3090 
3091  /* Detach from the batch we were last working on. */
3093  {
3094  /*
3095  * Technically we shouldn't access the barrier because we're no
3096  * longer attached, but since there is no way it's moving after
3097  * this point it seems safe to make the following assertion.
3098  */
3100 
3101  /* Free shared chunks and buckets. */
3102  while (DsaPointerIsValid(batch->chunks))
3103  {
3104  HashMemoryChunk chunk =
3105  dsa_get_address(hashtable->area, batch->chunks);
3106  dsa_pointer next = chunk->next.shared;
3107 
3108  dsa_free(hashtable->area, batch->chunks);
3109  batch->chunks = next;
3110  }
3111  if (DsaPointerIsValid(batch->buckets))
3112  {
3113  dsa_free(hashtable->area, batch->buckets);
3114  batch->buckets = InvalidDsaPointer;
3115  }
3116  }
3117 
3118  /*
3119  * Track the largest batch we've been attached to. Though each
3120  * backend might see a different subset of batches, explain.c will
3121  * scan the results from all backends to find the largest value.
3122  */
3123  hashtable->spacePeak =
3124  Max(hashtable->spacePeak,
3125  batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
3126 
3127  /* Remember that we are not attached to a batch. */
3128  hashtable->curbatch = -1;
3129  }
3130 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
#define PHJ_BATCH_DONE
Definition: hashjoin.h:268
static int32 next
Definition: blutils.c:213
#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:356
dsa_pointer shared
Definition: hashjoin.h:127
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
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:357
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define Max(x, y)
Definition: c.h:884
#define Assert(condition)
Definition: c.h:732
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:820
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecHashTableInsert()

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

Definition at line 1592 of file nodeHash.c.

References Assert, HashJoinTableData::buckets, HashJoinTableData::curbatch, dense_alloc(), ExecFetchSlotMinimalTuple(), ExecHashGetBucketAndBatch(), ExecHashIncreaseNumBatches(), ExecHashJoinSaveTuple(), HashJoinTupleData::hashvalue, heap_free_minimal_tuple(), 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().

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

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

Referenced by ExecHashJoinNewBatch().

2130 {
2131  MemoryContext oldcxt;
2132  int nbuckets = hashtable->nbuckets;
2133 
2134  /*
2135  * Release all the hash buckets and tuples acquired in the prior pass, and
2136  * reinitialize the context for a new pass.
2137  */
2138  MemoryContextReset(hashtable->batchCxt);
2139  oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
2140 
2141  /* Reallocate and reinitialize the hash bucket headers. */
2142  hashtable->buckets.unshared = (HashJoinTuple *)
2143  palloc0(nbuckets * sizeof(HashJoinTuple));
2144 
2145  hashtable->spaceUsed = 0;
2146 
2147  MemoryContextSwitchTo(oldcxt);
2148 
2149  /* Forget the chunks (the memory was freed by the context reset above). */
2150  hashtable->chunks = NULL;
2151 }
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:136
MemoryContext batchCxt
Definition: hashjoin.h:349
void * palloc0(Size size)
Definition: mcxt.c:955
union HashJoinTableData::@97 buckets
HashMemoryChunk chunks
Definition: hashjoin.h:352
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297

◆ ExecHashTableResetMatchFlags()

void ExecHashTableResetMatchFlags ( HashJoinTable  hashtable)

Definition at line 2158 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().

2159 {
2160  HashJoinTuple tuple;
2161  int i;
2162 
2163  /* Reset all flags in the main table ... */
2164  for (i = 0; i < hashtable->nbuckets; i++)
2165  {
2166  for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
2167  tuple = tuple->next.unshared)
2169  }
2170 
2171  /* ... and the same for the skew buckets, if any */
2172  for (i = 0; i < hashtable->nSkewBuckets; i++)
2173  {
2174  int j = hashtable->skewBucketNums[i];
2175  HashSkewBucket *skewBucket = hashtable->skewBucket[j];
2176 
2177  for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
2179  }
2180 }
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:526
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, Hash::plan, PlanState::plan, HashState::ps, PlanState::ps_ProjInfo, Plan::qual, PlanState::qual, PlanState::state, and TTSOpsMinimalTuple.

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  */
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:141
ProjectionInfo * ps_ProjInfo
Definition: execnodes.h:985
HashJoinTable hashtable
Definition: execnodes.h:2223
EState * state
Definition: execnodes.h:947
ExprState * ExecInitQual(List *qual, PlanState *parent)
Definition: execExpr.c:206
#define EXEC_FLAG_BACKWARD
Definition: executor.h:57
#define outerPlanState(node)
Definition: execnodes.h:1039
List * hashkeys
Definition: execnodes.h:2224
PlanState ps
Definition: execnodes.h:2222
#define outerPlan(node)
Definition: plannodes.h:170
static TupleTableSlot * ExecHash(PlanState *pstate)
Definition: nodeHash.c:91
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:951
Plan * plan
Definition: execnodes.h:945
#define makeNode(_type_)
Definition: nodes.h:572
#define Assert(condition)
Definition: c.h:732
#define EXEC_FLAG_MARK
Definition: executor.h:58
void ExecAssignExprContext(EState *estate, PlanState *planstate)
Definition: execUtils.c:447
void ExecInitResultTupleSlotTL(PlanState *planstate, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1763
ExprState * qual
Definition: execnodes.h:966
Plan plan
Definition: plannodes.h:901
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:139
const TupleTableSlotOps TTSOpsMinimalTuple
Definition: execTuples.c:85

◆ ExecParallelHashCloseBatchAccessors()

static void ExecParallelHashCloseBatchAccessors ( HashJoinTable  hashtable)
static

Definition at line 2975 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().

2976 {
2977  int i;
2978 
2979  for (i = 0; i < hashtable->nbatch; ++i)
2980  {
2981  /* Make sure no files are left open. */
2982  sts_end_write(hashtable->batches[i].inner_tuples);
2983  sts_end_write(hashtable->batches[i].outer_tuples);
2986  }
2987  pfree(hashtable->batches);
2988  hashtable->batches = NULL;
2989 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
void pfree(void *pointer)
Definition: mcxt.c:1031
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
int i
void sts_end_write(SharedTuplestoreAccessor *accessor)

◆ ExecParallelHashEnsureBatchAccessors()

static void ExecParallelHashEnsureBatchAccessors ( HashJoinTable  hashtable)
static

Definition at line 2996 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().

2997 {
2998  ParallelHashJoinState *pstate = hashtable->parallel_state;
2999  ParallelHashJoinBatch *batches;
3000  MemoryContext oldcxt;
3001  int i;
3002 
3003  if (hashtable->batches != NULL)
3004  {
3005  if (hashtable->nbatch == pstate->nbatch)
3006  return;
3008  }
3009 
3010  /*
3011  * It's possible for a backend to start up very late so that the whole
3012  * join is finished and the shm state for tracking batches has already
3013  * been freed by ExecHashTableDetach(). In that case we'll just leave
3014  * hashtable->batches as NULL so that ExecParallelHashJoinNewBatch() gives
3015  * up early.
3016  */
3017  if (!DsaPointerIsValid(pstate->batches))
3018  return;
3019 
3020  /* Use hash join memory context. */
3021  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
3022 
3023  /* Allocate this backend's accessor array. */
3024  hashtable->nbatch = pstate->nbatch;
3025  hashtable->batches = (ParallelHashJoinBatchAccessor *)
3026  palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
3027 
3028  /* Find the base of the pseudo-array of ParallelHashJoinBatch objects. */
3029  batches = (ParallelHashJoinBatch *)
3030  dsa_get_address(hashtable->area, pstate->batches);
3031 
3032  /* Set up the accessor array and attach to the tuplestores. */
3033  for (i = 0; i < hashtable->nbatch; ++i)
3034  {
3035  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
3036  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
3037 
3038  accessor->shared = shared;
3039  accessor->preallocated = 0;
3040  accessor->done = false;
3041  accessor->inner_tuples =
3044  &pstate->fileset);
3045  accessor->outer_tuples =
3047  pstate->nparticipants),
3049  &pstate->fileset);
3050  }
3051 
3052  MemoryContextSwitchTo(oldcxt);
3053 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2975
#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:356
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
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:110
void * palloc0(Size size)
Definition: mcxt.c:955
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#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:348

◆ ExecParallelHashFirstTuple()

static HashJoinTuple ExecParallelHashFirstTuple ( HashJoinTable  table,
int  bucketno 
)
inlinestatic

Definition at line 3173 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().

3174 {
3175  HashJoinTuple tuple;
3176  dsa_pointer p;
3177 
3178  Assert(hashtable->parallel_state);
3179  p = dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
3180  tuple = (HashJoinTuple) dsa_get_address(hashtable->area, p);
3181 
3182  return tuple;
3183 }
uint64 dsa_pointer
Definition: dsa.h:62
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1878
#define Assert(condition)
Definition: c.h:732
#define dsa_pointer_atomic_read
Definition: dsa.h:65

◆ ExecParallelHashIncreaseNumBatches()

static void ExecParallelHashIncreaseNumBatches ( HashJoinTable  hashtable)
static

Definition at line 1056 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().

1057 {
1058  ParallelHashJoinState *pstate = hashtable->parallel_state;
1059  int i;
1060 
1062 
1063  /*
1064  * It's unlikely, but we need to be prepared for new participants to show
1065  * up while we're in the middle of this operation so we need to switch on
1066  * barrier phase here.
1067  */
1069  {
1071 
1072  /*
1073  * Elect one participant to prepare to grow the number of batches.
1074  * This involves reallocating or resetting the buckets of batch 0
1075  * in preparation for all participants to begin repartitioning the
1076  * tuples.
1077  */
1080  {
1081  dsa_pointer_atomic *buckets;
1082  ParallelHashJoinBatch *old_batch0;
1083  int new_nbatch;
1084  int i;
1085 
1086  /* Move the old batch out of the way. */
1087  old_batch0 = hashtable->batches[0].shared;
1088  pstate->old_batches = pstate->batches;
1089  pstate->old_nbatch = hashtable->nbatch;
1090  pstate->batches = InvalidDsaPointer;
1091 
1092  /* Free this backend's old accessors. */
1094 
1095  /* Figure out how many batches to use. */
1096  if (hashtable->nbatch == 1)
1097  {
1098  /*
1099  * We are going from single-batch to multi-batch. We need
1100  * to switch from one large combined memory budget to the
1101  * regular work_mem budget.
1102  */
1103  pstate->space_allowed = work_mem * 1024L;
1104 
1105  /*
1106  * The combined work_mem of all participants wasn't
1107  * enough. Therefore one batch per participant would be
1108  * approximately equivalent and would probably also be
1109  * insufficient. So try two batches per participant,
1110  * rounded up to a power of two.
1111  */
1112  new_nbatch = 1 << my_log2(pstate->nparticipants * 2);
1113  }
1114  else
1115  {
1116  /*
1117  * We were already multi-batched. Try doubling the number
1118  * of batches.
1119  */
1120  new_nbatch = hashtable->nbatch * 2;
1121  }
1122 
1123  /* Allocate new larger generation of batches. */
1124  Assert(hashtable->nbatch == pstate->nbatch);
1125  ExecParallelHashJoinSetUpBatches(hashtable, new_nbatch);
1126  Assert(hashtable->nbatch == pstate->nbatch);
1127 
1128  /* Replace or recycle batch 0's bucket array. */
1129  if (pstate->old_nbatch == 1)
1130  {
1131  double dtuples;
1132  double dbuckets;
1133  int new_nbuckets;
1134 
1135  /*
1136  * We probably also need a smaller bucket array. How many
1137  * tuples do we expect per batch, assuming we have only
1138  * half of them so far? Normally we don't need to change
1139  * the bucket array's size, because the size of each batch
1140  * stays the same as we add more batches, but in this
1141  * special case we move from a large batch to many smaller
1142  * batches and it would be wasteful to keep the large
1143  * array.
1144  */
1145  dtuples = (old_batch0->ntuples * 2.0) / new_nbatch;
1146  dbuckets = ceil(dtuples / NTUP_PER_BUCKET);
1147  dbuckets = Min(dbuckets,
1148  MaxAllocSize / sizeof(dsa_pointer_atomic));
1149  new_nbuckets = (int) dbuckets;
1150  new_nbuckets = Max(new_nbuckets, 1024);
1151  new_nbuckets = 1 << my_log2(new_nbuckets);
1152  dsa_free(hashtable->area, old_batch0->buckets);
1153  hashtable->batches[0].shared->buckets =
1154  dsa_allocate(hashtable->area,
1155  sizeof(dsa_pointer_atomic) * new_nbuckets);
1156  buckets = (dsa_pointer_atomic *)
1157  dsa_get_address(hashtable->area,
1158  hashtable->batches[0].shared->buckets);
1159  for (i = 0; i < new_nbuckets; ++i)
1161  pstate->nbuckets = new_nbuckets;
1162  }
1163  else
1164  {
1165  /* Recycle the existing bucket array. */
1166  hashtable->batches[0].shared->buckets = old_batch0->buckets;
1167  buckets = (dsa_pointer_atomic *)
1168  dsa_get_address(hashtable->area, old_batch0->buckets);
1169  for (i = 0; i < hashtable->nbuckets; ++i)
1171  }
1172 
1173  /* Move all chunks to the work queue for parallel processing. */
1174  pstate->chunk_work_queue = old_batch0->chunks;
1175 
1176  /* Disable further growth temporarily while we're growing. */
1177  pstate->growth = PHJ_GROWTH_DISABLED;
1178  }
1179  else
1180  {
1181  /* All other participants just flush their tuples to disk. */
1183  }
1184  /* Fall through. */
1185 
1187  /* Wait for the above to be finished. */
1190  /* Fall through. */
1191 
1193  /* Make sure that we have the current dimensions and buckets. */
1196  /* Then partition, flush counters. */
1199  ExecParallelHashMergeCounters(hashtable);
1200  /* Wait for the above to be finished. */
1203  /* Fall through. */
1204 
1206 
1207  /*
1208  * Elect one participant to clean up and decide whether further
1209  * repartitioning is needed, or should be disabled because it's
1210  * not helping.
1211  */
1214  {
1215  bool space_exhausted = false;
1216  bool extreme_skew_detected = false;
1217 
1218  /* Make sure that we have the current dimensions and buckets. */
1221 
1222  /* Are any of the new generation of batches exhausted? */
1223  for (i = 0; i < hashtable->nbatch; ++i)
1224  {
1225  ParallelHashJoinBatch *batch = hashtable->batches[i].shared;
1226 
1227  if (batch->space_exhausted ||
1228  batch->estimated_size > pstate->space_allowed)
1229  {
1230  int parent;
1231 
1232  space_exhausted = true;
1233 
1234  /*
1235  * Did this batch receive ALL of the tuples from its
1236  * parent batch? That would indicate that further
1237  * repartitioning isn't going to help (the hash values
1238  * are probably all the same).
1239  */
1240  parent = i % pstate->old_nbatch;
1241  if (batch->ntuples == hashtable->batches[parent].shared->old_ntuples)
1242  extreme_skew_detected = true;
1243  }
1244  }
1245 
1246  /* Don't keep growing if it's not helping or we'd overflow. */
1247  if (extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2)
1248  pstate->growth = PHJ_GROWTH_DISABLED;
1249  else if (space_exhausted)
1251  else
1252  pstate->growth = PHJ_GROWTH_OK;
1253 
1254  /* Free the old batches in shared memory. */
1255  dsa_free(hashtable->area, pstate->old_batches);
1256  pstate->old_batches = InvalidDsaPointer;
1257  }
1258  /* Fall through. */
1259 
1261  /* Wait for the above to complete. */
1264  }
1265 }
static void ExecParallelHashRepartitionRest(HashJoinTable hashtable)
Definition: nodeHash.c:1340
static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2975
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:3221
dsa_pointer chunks
Definition: hashjoin.h:156
#define Min(x, y)
Definition: c.h:890
static void ExecParallelHashMergeCounters(HashJoinTable hashtable)
Definition: nodeHash.c:1400
static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2996
static void ExecParallelHashRepartitionFirst(HashJoinTable hashtable)
Definition: nodeHash.c:1273
dsa_area * area
Definition: hashjoin.h:356
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
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:1718
#define PHJ_GROW_BATCHES_ALLOCATING
Definition: hashjoin.h:272
#define MaxAllocSize
Definition: memutils.h:40
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
#define NTUP_PER_BUCKET
Definition: nodeHash.c:662
#define PHJ_GROW_BATCHES_REPARTITIONING
Definition: hashjoin.h:273
int work_mem
Definition: globals.c:121
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define Max(x, y)
Definition: c.h:884
#define dsa_pointer_atomic_write
Definition: dsa.h:66
#define Assert(condition)
Definition: c.h:732
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:820
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
static void ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
Definition: nodeHash.c:2898
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 1493 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().

1494 {
1495  ParallelHashJoinState *pstate = hashtable->parallel_state;
1496  int i;
1497  HashMemoryChunk chunk;
1498  dsa_pointer chunk_s;
1499 
1501 
1502  /*
1503  * It's unlikely, but we need to be prepared for new participants to show
1504  * up while we're in the middle of this operation so we need to switch on
1505  * barrier phase here.
1506  */
1508  {
1510  /* Elect one participant to prepare to increase nbuckets. */
1513  {
1514  size_t size;
1515  dsa_pointer_atomic *buckets;
1516 
1517  /* Double the size of the bucket array. */
1518  pstate->nbuckets *= 2;
1519  size = pstate->nbuckets * sizeof(dsa_pointer_atomic);
1520  hashtable->batches[0].shared->size += size / 2;
1521  dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
1522  hashtable->batches[0].shared->buckets =
1523  dsa_allocate(hashtable->area, size);
1524  buckets = (dsa_pointer_atomic *)
1525  dsa_get_address(hashtable->area,
1526  hashtable->batches[0].shared->buckets);
1527  for (i = 0; i < pstate->nbuckets; ++i)
1529 
1530  /* Put the chunk list onto the work queue. */
1531  pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;
1532 
1533  /* Clear the flag. */
1534  pstate->growth = PHJ_GROWTH_OK;
1535  }
1536  /* Fall through. */
1537 
1539  /* Wait for the above to complete. */
1542  /* Fall through. */
1543 
1545  /* Reinsert all tuples into the hash table. */
1548  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_s)))
1549  {
1550  size_t idx = 0;
1551 
1552  while (idx < chunk->used)
1553  {
1554  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1555  dsa_pointer shared = chunk_s + HASH_CHUNK_HEADER_SIZE + idx;
1556  int bucketno;
1557  int batchno;
1558 
1559  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1560  &bucketno, &batchno);
1561  Assert(batchno == 0);
1562 
1563  /* add the tuple to the proper bucket */
1564  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1565  hashTuple, shared);
1566 
1567  /* advance index past the tuple */
1568  idx += MAXALIGN(HJTUPLE_OVERHEAD +
1569  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1570  }
1571 
1572  /* allow this loop to be cancellable */
1574  }
1577  }
1578 }
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:3221
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:2996
uint64 dsa_pointer
Definition: dsa.h:62
#define PHJ_GROW_BUCKETS_REINSERTING
Definition: hashjoin.h:281
dsa_area * area
Definition: hashjoin.h:356
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1890
#define PHJ_GROW_BUCKETS_ALLOCATING
Definition: hashjoin.h:280
Barrier grow_buckets_barrier
Definition: hashjoin.h:250
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1878
static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table, dsa_pointer *shared)
Definition: nodeHash.c:3242
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
#define PHJ_GROW_BUCKETS_ELECTING
Definition: hashjoin.h:279
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define Assert(condition)
Definition: c.h:732
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:685
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:820
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3203
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
int i
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
#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 2898 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().

2899 {
2900  ParallelHashJoinState *pstate = hashtable->parallel_state;
2901  ParallelHashJoinBatch *batches;
2902  MemoryContext oldcxt;
2903  int i;
2904 
2905  Assert(hashtable->batches == NULL);
2906 
2907  /* Allocate space. */
2908  pstate->batches =
2909  dsa_allocate0(hashtable->area,
2910  EstimateParallelHashJoinBatch(hashtable) * nbatch);
2911  pstate->nbatch = nbatch;
2912  batches = dsa_get_address(hashtable->area, pstate->batches);
2913 
2914  /* Use hash join memory context. */
2915  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
2916 
2917  /* Allocate this backend's accessor array. */
2918  hashtable->nbatch = nbatch;
2919  hashtable->batches = (ParallelHashJoinBatchAccessor *)
2920  palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
2921 
2922  /* Set up the shared state, tuplestores and backend-local accessors. */
2923  for (i = 0; i < hashtable->nbatch; ++i)
2924  {
2925  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
2926  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
2927  char name[MAXPGPATH];
2928 
2929  /*
2930  * All members of shared were zero-initialized. We just need to set
2931  * up the Barrier.
2932  */
2933  BarrierInit(&shared->batch_barrier, 0);
2934  if (i == 0)
2935  {
2936  /* Batch 0 doesn't need to be loaded. */
2937  BarrierAttach(&shared->batch_barrier);
2938  while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING)
2939  BarrierArriveAndWait(&shared->batch_barrier, 0);
2940  BarrierDetach(&shared->batch_barrier);
2941  }
2942 
2943  /* Initialize accessor state. All members were zero-initialized. */
2944  accessor->shared = shared;
2945 
2946  /* Initialize the shared tuplestores. */
2947  snprintf(name, sizeof(name), "i%dof%d", i, hashtable->nbatch);
2948  accessor->inner_tuples =
2950  pstate->nparticipants,
2952  sizeof(uint32),
2954  &pstate->fileset,
2955  name);
2956  snprintf(name, sizeof(name), "o%dof%d", i, hashtable->nbatch);
2957  accessor->outer_tuples =
2959  pstate->nparticipants),
2960  pstate->nparticipants,
2962  sizeof(uint32),
2964  &pstate->fileset,
2965  name);
2966  }
2967 
2968  MemoryContextSwitchTo(oldcxt);
2969 }
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
SharedFileSet fileset
Definition: hashjoin.h:253
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_area * area
Definition: hashjoin.h:356
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
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:110
unsigned int uint32
Definition: c.h:358
#define SHARED_TUPLESTORE_SINGLE_PASS
int BarrierAttach(Barrier *barrier)
Definition: barrier.c:214
void * palloc0(Size size)
Definition: mcxt.c:955
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
#define PHJ_BATCH_PROBING
Definition: hashjoin.h:267
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define Assert(condition)
Definition: c.h:732
#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:348
#define snprintf
Definition: port.h:192

◆ ExecParallelHashMergeCounters()

static void ExecParallelHashMergeCounters ( HashJoinTable  hashtable)
static

Definition at line 1400 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().

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

◆ ExecParallelHashNextTuple()

static HashJoinTuple ExecParallelHashNextTuple ( HashJoinTable  table,
HashJoinTuple  tuple 
)
inlinestatic

Definition at line 3189 of file nodeHash.c.

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

Referenced by ExecParallelScanHashBucket().

3190 {
3192 
3193  Assert(hashtable->parallel_state);
3194  next = (HashJoinTuple) dsa_get_address(hashtable->area, tuple->next.shared);
3195 
3196  return next;
3197 }
static int32 next
Definition: blutils.c:213
union HashJoinTupleData::@95 next
dsa_pointer shared
Definition: hashjoin.h:73
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1878
#define Assert(condition)
Definition: c.h:732

◆ ExecParallelHashPopChunkQueue()

static HashMemoryChunk ExecParallelHashPopChunkQueue ( HashJoinTable  table,
dsa_pointer shared 
)
static

Definition at line 3242 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().

3243 {
3244  ParallelHashJoinState *pstate = hashtable->parallel_state;
3245  HashMemoryChunk chunk;
3246 
3247  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3248  if (DsaPointerIsValid(pstate->chunk_work_queue))
3249  {
3250  *shared = pstate->chunk_work_queue;
3251  chunk = (HashMemoryChunk)
3252  dsa_get_address(hashtable->area, *shared);
3253  pstate->chunk_work_queue = chunk->next.shared;
3254  }
3255  else
3256  chunk = NULL;
3257  LWLockRelease(&pstate->lock);
3258 
3259  return chunk;
3260 }
dsa_pointer chunk_work_queue
Definition: hashjoin.h:242
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
dsa_pointer shared
Definition: hashjoin.h:127
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
union HashMemoryChunkData::@96 next
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
#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 3203 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().

3206 {
3207  for (;;)
3208  {
3209  tuple->next.shared = dsa_pointer_atomic_read(head);
3211  &tuple->next.shared,
3212  tuple_shared))
3213  break;
3214  }
3215 }
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 1273 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().

1274 {
1275  dsa_pointer chunk_shared;
1276  HashMemoryChunk chunk;
1277 
1278  Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);
1279 
1280  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared)))
1281  {
1282  size_t idx = 0;
1283 
1284  /* Repartition all tuples in this chunk. */
1285  while (idx < chunk->used)
1286  {
1287  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1288  MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
1289  HashJoinTuple copyTuple;
1290  dsa_pointer shared;
1291  int bucketno;
1292  int batchno;
1293 
1294  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1295  &bucketno, &batchno);
1296 
1297  Assert(batchno < hashtable->nbatch);
1298  if (batchno == 0)
1299  {
1300  /* It still belongs in batch 0. Copy to a new chunk. */
1301  copyTuple =
1302  ExecParallelHashTupleAlloc(hashtable,
1303  HJTUPLE_OVERHEAD + tuple->t_len,
1304  &shared);
1305  copyTuple->hashvalue = hashTuple->hashvalue;
1306  memcpy(HJTUPLE_MINTUPLE(copyTuple), tuple, tuple->t_len);
1307  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1308  copyTuple, shared);
1309  }
1310  else
1311  {
1312  size_t tuple_size =
1313  MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1314 
1315  /* It belongs in a later batch. */
1316  hashtable->batches[batchno].estimated_size += tuple_size;
1317  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1318  &hashTuple->hashvalue, tuple);
1319  }
1320 
1321  /* Count this tuple. */
1322  ++hashtable->batches[0].old_ntuples;
1323  ++hashtable->batches[batchno].ntuples;
1324 
1325  idx += MAXALIGN(HJTUPLE_OVERHEAD +
1326  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1327  }
1328 
1329  /* Free this chunk. */
1330  dsa_free(hashtable->area, chunk_shared);
1331 
1333  }
1334 }
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:356
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1890
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1878
static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table, dsa_pointer *shared)
Definition: nodeHash.c:3242
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:732
#define MAXALIGN(LEN)
Definition: c.h:685
union HashJoinTableData::@97 buckets
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:820
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3203
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2750
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecParallelHashRepartitionRest()

static void ExecParallelHashRepartitionRest ( HashJoinTable  hashtable)
static

Definition at line 1340 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().

1341 {
1342  ParallelHashJoinState *pstate = hashtable->parallel_state;
1343  int old_nbatch = pstate->old_nbatch;
1344  SharedTuplestoreAccessor **old_inner_tuples;
1345  ParallelHashJoinBatch *old_batches;
1346  int i;
1347 
1348  /* Get our hands on the previous generation of batches. */
1349  old_batches = (ParallelHashJoinBatch *)
1350  dsa_get_address(hashtable->area, pstate->old_batches);
1351  old_inner_tuples = palloc0(sizeof(SharedTuplestoreAccessor *) * old_nbatch);
1352  for (i = 1; i < old_nbatch; ++i)
1353  {
1354  ParallelHashJoinBatch *shared =
1355  NthParallelHashJoinBatch(old_batches, i);
1356 
1357  old_inner_tuples[i] = sts_attach(ParallelHashJoinBatchInner(shared),
1359  &pstate->fileset);
1360  }
1361 
1362  /* Join in the effort to repartition them. */
1363  for (i = 1; i < old_nbatch; ++i)
1364  {
1365  MinimalTuple tuple;
1366  uint32 hashvalue;
1367 
1368  /* Scan one partition from the previous generation. */
1369  sts_begin_parallel_scan(old_inner_tuples[i]);
1370  while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue)))
1371  {
1372  size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1373  int bucketno;
1374  int batchno;
1375 
1376  /* Decide which partition it goes to in the new generation. */
1377  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
1378  &batchno);
1379 
1380  hashtable->batches[batchno].estimated_size += tuple_size;
1381  ++hashtable->batches[batchno].ntuples;
1382  ++hashtable->batches[i].old_ntuples;
1383 
1384  /* Store the tuple its new batch. */
1385  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1386  &hashvalue, tuple);
1387 
1389  }
1390  sts_end_parallel_scan(old_inner_tuples[i]);
1391  }
1392 
1393  pfree(old_inner_tuples);
1394 }
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:356
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1890
void pfree(void *pointer)
Definition: mcxt.c:1031
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:110
unsigned int uint32
Definition: c.h:358
void sts_begin_parallel_scan(SharedTuplestoreAccessor *accessor)
void * palloc0(Size size)
Definition: mcxt.c:955
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
dsa_pointer old_batches
Definition: hashjoin.h:237
#define NthParallelHashJoinBatch(base, n)
Definition: hashjoin.h:186
#define MAXALIGN(LEN)
Definition: c.h:685
int i
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
MinimalTuple sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data)

◆ ExecParallelHashTableAlloc()

void ExecParallelHashTableAlloc ( HashJoinTable  hashtable,
int  batchno 
)

Definition at line 3059 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().

3060 {
3061  ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
3062  dsa_pointer_atomic *buckets;
3063  int nbuckets = hashtable->parallel_state->nbuckets;
3064  int i;
3065 
3066  batch->buckets =
3067  dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
3068  buckets = (dsa_pointer_atomic *)
3069  dsa_get_address(hashtable->area, batch->buckets);
3070  for (i = 0; i < nbuckets; ++i)
3072 }
#define InvalidDsaPointer
Definition: dsa.h:78
dsa_area * area
Definition: hashjoin.h:356
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
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 1681 of file nodeHash.c.

References Assert, BarrierPhase(), HashJoinTableData::batches, HashJoinTableData::buckets, ParallelHashJoinState::build_barrier, ExecFetchSlotMinimalTuple(), ExecHashGetBucketAndBatch(), ExecParallelHashPushTuple(), ExecParallelHashTupleAlloc(), ExecParallelHashTuplePrealloc(), HashJoinTupleData::hashvalue, heap_free_minimal_tuple(), 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().

1684 {
1685  bool shouldFree;
1686  MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1687  dsa_pointer shared;
1688  int bucketno;
1689  int batchno;
1690 
1691 retry:
1692  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1693 
1694  if (batchno == 0)
1695  {
1696  HashJoinTuple hashTuple;
1697 
1698  /* Try to load it into memory. */
1701  hashTuple = ExecParallelHashTupleAlloc(hashtable,
1702  HJTUPLE_OVERHEAD + tuple->t_len,
1703  &shared);
1704  if (hashTuple == NULL)
1705  goto retry;
1706 
1707  /* Store the hash value in the HashJoinTuple header. */
1708  hashTuple->hashvalue = hashvalue;
1709  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1710 
1711  /* Push it onto the front of the bucket's list */
1712  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1713  hashTuple, shared);
1714  }
1715  else
1716  {
1717  size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1718 
1719  Assert(batchno > 0);
1720 
1721  /* Try to preallocate space in the batch if necessary. */
1722  if (hashtable->batches[batchno].preallocated < tuple_size)
1723  {
1724  if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size))
1725  goto retry;
1726  }
1727 
1728  Assert(hashtable->batches[batchno].preallocated >= tuple_size);
1729  hashtable->batches[batchno].preallocated -= tuple_size;
1730  sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue,
1731  tuple);
1732  }
1733  ++hashtable->batches[batchno].ntuples;
1734 
1735  if (shouldFree)
1736  heap_free_minimal_tuple(tuple);
1737 }
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
void sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, MinimalTuple tuple)
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot, bool *shouldFree)
Definition: execTuples.c:1656
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:1890
void heap_free_minimal_tuple(MinimalTuple mtup)
Definition: heaptuple.c:1427
static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size)
Definition: nodeHash.c:3283
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:732
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
#define MAXALIGN(LEN)
Definition: c.h:685
union HashJoinTableData::@97 buckets
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3203
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2750
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecParallelHashTableInsertCurrentBatch()

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

Definition at line 1746 of file nodeHash.c.

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

Referenced by ExecParallelHashJoinNewBatch().

1749 {
1750  bool shouldFree;
1751  MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1752  HashJoinTuple hashTuple;
1753  dsa_pointer shared;
1754  int batchno;
1755  int bucketno;
1756 
1757  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1758  Assert(batchno == hashtable->curbatch);
1759  hashTuple = ExecParallelHashTupleAlloc(hashtable,
1760  HJTUPLE_OVERHEAD + tuple->t_len,
1761  &shared);
1762  hashTuple->hashvalue = hashvalue;
1763  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1765  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1766  hashTuple, shared);
1767 
1768  if (shouldFree)
1769  heap_free_minimal_tuple(tuple);
1770 }
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot, bool *shouldFree)
Definition: execTuples.c:1656
uint64 dsa_pointer
Definition: dsa.h:62
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1890
void heap_free_minimal_tuple(MinimalTuple mtup)
Definition: heaptuple.c:1427
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:79
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define Assert(condition)
Definition: c.h:732
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:526
union HashJoinTableData::@97 buckets
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3203
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2750
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecParallelHashTableSetCurrentBatch()

void ExecParallelHashTableSetCurrentBatch ( HashJoinTable  hashtable,
int  batchno 
)

Definition at line 3221 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().

3222 {
3223  Assert(hashtable->batches[batchno].shared->buckets != InvalidDsaPointer);
3224 
3225  hashtable->curbatch = batchno;
3226  hashtable->buckets.shared = (dsa_pointer_atomic *)
3227  dsa_get_address(hashtable->area,
3228  hashtable->batches[batchno].shared->buckets);
3229  hashtable->nbuckets = hashtable->parallel_state->nbuckets;
3230  hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
3231  hashtable->current_chunk = NULL;
3233  hashtable->batches[batchno].at_least_one_chunk = false;
3234 }
dsa_pointer current_chunk_shared
Definition: hashjoin.h:359
dsa_pointer_atomic * shared
Definition: hashjoin.h:299
#define InvalidDsaPointer
Definition: dsa.h:78
dsa_area * area
Definition: hashjoin.h:356
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
int my_log2(long num)
Definition: dynahash.c:1718
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define Assert(condition)
Definition: c.h:732
union HashJoinTableData::@97 buckets
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
HashMemoryChunk current_chunk
Definition: hashjoin.h:355
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecParallelHashTupleAlloc()

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

Definition at line 2750 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, MaxAllocSize, 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().

2752 {
2753  ParallelHashJoinState *pstate = hashtable->parallel_state;
2754  dsa_pointer chunk_shared;
2755  HashMemoryChunk chunk;
2756  Size chunk_size;
2757  HashJoinTuple result;
2758  int curbatch = hashtable->curbatch;
2759 
2760  size = MAXALIGN(size);
2761 
2762  /*
2763  * Fast path: if there is enough space in this backend's current chunk,
2764  * then we can allocate without any locking.
2765  */
2766  chunk = hashtable->current_chunk;
2767  if (chunk != NULL &&
2768  size <= HASH_CHUNK_THRESHOLD &&
2769  chunk->maxlen - chunk->used >= size)
2770  {
2771 
2772  chunk_shared = hashtable->current_chunk_shared;
2773  Assert(chunk == dsa_get_address(hashtable->area, chunk_shared));
2774  *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE + chunk->used;
2775  result = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + chunk->used);
2776  chunk->used += size;
2777 
2778  Assert(chunk->used <= chunk->maxlen);
2779  Assert(result == dsa_get_address(hashtable->area, *shared));
2780 
2781  return result;
2782  }
2783 
2784  /* Slow path: try to allocate a new chunk. */
2785  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
2786 
2787  /*
2788  * Check if we need to help increase the number of buckets or batches.
2789  */
2790  if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
2792  {
2793  ParallelHashGrowth growth = pstate->growth;
2794 
2795  hashtable->current_chunk = NULL;
2796  LWLockRelease(&pstate->lock);
2797 
2798  /* Another participant has commanded us to help grow. */
2799  if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
2801  else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
2803 
2804  /* The caller must retry. */
2805  return NULL;
2806  }
2807 
2808  /* Oversized tuples get their own chunk. */
2809  if (size > HASH_CHUNK_THRESHOLD)
2810  chunk_size = size + HASH_CHUNK_HEADER_SIZE;
2811  else
2812  chunk_size = HASH_CHUNK_SIZE;
2813 
2814  /* Check if it's time to grow batches or buckets. */
2815  if (pstate->growth != PHJ_GROWTH_DISABLED)
2816  {
2817  Assert(curbatch == 0);
2819 
2820  /*
2821  * Check if our space limit would be exceeded. To avoid choking on
2822  * very large tuples or very low work_mem setting, we'll always allow
2823  * each backend to allocate at least one chunk.
2824  */
2825  if (hashtable->batches[0].at_least_one_chunk &&
2826  hashtable->batches[0].shared->size +
2827  chunk_size > pstate->space_allowed)
2828  {
2830  hashtable->batches[0].shared->space_exhausted = true;
2831  LWLockRelease(&pstate->lock);
2832 
2833  return NULL;
2834  }
2835 
2836  /* Check if our load factor limit would be exceeded. */
2837  if (hashtable->nbatch == 1)
2838  {
2839  hashtable->batches[0].shared->ntuples += hashtable->batches[0].ntuples;
2840  hashtable->batches[0].ntuples = 0;
2841  /* Guard against integer overflow and alloc size overflow */
2842  if (hashtable->batches[0].shared->ntuples + 1 >
2843  hashtable->nbuckets * NTUP_PER_BUCKET &&
2844  hashtable->nbuckets < (INT_MAX / 2) &&
2845  hashtable->nbuckets * 2 <=
2846  MaxAllocSize / sizeof(dsa_pointer_atomic))
2847  {
2849  LWLockRelease(&pstate->lock);
2850 
2851  return NULL;
2852  }
2853  }
2854  }
2855 
2856  /* We are cleared to allocate a new chunk. */
2857  chunk_shared = dsa_allocate(hashtable->area, chunk_size);
2858  hashtable->batches[curbatch].shared->size += chunk_size;
2859  hashtable->batches[curbatch].at_least_one_chunk = true;
2860 
2861  /* Set up the chunk. */
2862  chunk = (HashMemoryChunk) dsa_get_address(hashtable->area, chunk_shared);
2863  *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE;
2864  chunk->maxlen = chunk_size - HASH_CHUNK_HEADER_SIZE;
2865  chunk->used = size;
2866 
2867  /*
2868  * Push it onto the list of chunks, so that it can be found if we need to
2869  * increase the number of buckets or batches (batch 0 only) and later for
2870  * freeing the memory (all batches).
2871  */
2872  chunk->next.shared = hashtable->batches[curbatch].shared->chunks;
2873  hashtable->batches[curbatch].shared->chunks = chunk_shared;
2874 
2875  if (size <= HASH_CHUNK_THRESHOLD)
2876  {
2877  /*
2878  * Make this the current chunk so that we can use the fast path to
2879  * fill the rest of it up in future calls.
2880  */
2881  hashtable->current_chunk = chunk;
2882  hashtable->current_chunk_shared = chunk_shared;
2883  }
2884  LWLockRelease(&pstate->lock);
2885 
2886  Assert(HASH_CHUNK_DATA(chunk) == dsa_get_address(hashtable->area, *shared));
2887  result = (HashJoinTuple) HASH_CHUNK_DATA(chunk);
2888 
2889  return result;
2890 }
dsa_pointer current_chunk_shared
Definition: hashjoin.h:359
static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:1056
#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:1726
dsa_area * area
Definition: hashjoin.h:356
dsa_pointer shared
Definition: hashjoin.h:127
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
union HashMemoryChunkData::@96 next
#define HASH_CHUNK_THRESHOLD
Definition: hashjoin.h:143
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1878
#define MaxAllocSize
Definition: memutils.h:40
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
#define NTUP_PER_BUCKET
Definition: nodeHash.c:662
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define Assert(condition)
Definition: c.h:732
ParallelHashGrowth growth
Definition: hashjoin.h:241
size_t Size
Definition: c.h:466
int BarrierPhase(Barrier *barrier)
Definition: barrier.c:243
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
#define MAXALIGN(LEN)
Definition: c.h:685
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
HashMemoryChunk current_chunk
Definition: hashjoin.h:355
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
ParallelHashGrowth
Definition: hashjoin.h:218
static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
Definition: nodeHash.c:1493
#define dsa_allocate(area, size)
Definition: dsa.h:84

◆ ExecParallelHashTuplePrealloc()

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

Definition at line 3283 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().

3284 {
3285  ParallelHashJoinState *pstate = hashtable->parallel_state;
3286  ParallelHashJoinBatchAccessor *batch = &hashtable->batches[batchno];
3287  size_t want = Max(size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);
3288 
3289  Assert(batchno > 0);
3290  Assert(batchno < hashtable->nbatch);
3291  Assert(size == MAXALIGN(size));
3292 
3293  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3294 
3295  /* Has another participant commanded us to help grow? */
3296  if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
3298  {
3299  ParallelHashGrowth growth = pstate->growth;
3300 
3301  LWLockRelease(&pstate->lock);
3302  if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
3304  else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3306 
3307  return false;
3308  }
3309 
3310  if (pstate->growth != PHJ_GROWTH_DISABLED &&
3311  batch->at_least_one_chunk &&
3312  (batch->shared->estimated_size + want + HASH_CHUNK_HEADER_SIZE
3313  > pstate->space_allowed))
3314  {
3315  /*
3316  * We have determined that this batch would exceed the space budget if
3317  * loaded into memory. Command all participants to help repartition.
3318  */
3319  batch->shared->space_exhausted = true;
3321  LWLockRelease(&pstate->lock);
3322 
3323  return false;
3324  }
3325 
3326  batch->at_least_one_chunk = true;
3327  batch->shared->estimated_size += want + HASH_CHUNK_HEADER_SIZE;
3328  batch->preallocated = want;
3329  LWLockRelease(&pstate->lock);
3330 
3331  return true;
3332 }
static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:1056
#define HASH_CHUNK_SIZE
Definition: hashjoin.h:139
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1726
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define Max(x, y)
Definition: c.h:884
#define Assert(condition)
Definition: c.h:732
ParallelHashGrowth growth
Definition: hashjoin.h:241
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1122
#define MAXALIGN(LEN)
Definition: c.h:685
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
ParallelHashGrowth
Definition: hashjoin.h:218
static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
Definition: nodeHash.c:1493

◆ ExecParallelScanHashBucket()

bool ExecParallelScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 1983 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().

1985 {
1986  ExprState *hjclauses = hjstate->hashclauses;
1987  HashJoinTable hashtable = hjstate->hj_HashTable;
1988  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1989  uint32 hashvalue = hjstate->hj_CurHashValue;
1990 
1991  /*
1992  * hj_CurTuple is the address of the tuple last returned from the current
1993  * bucket, or NULL if it's time to start scanning a new bucket.
1994  */
1995  if (hashTuple != NULL)
1996  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
1997  else
1998  hashTuple = ExecParallelHashFirstTuple(hashtable,
1999  hjstate->hj_CurBucketNo);
2000 
2001  while (hashTuple != NULL)
2002  {
2003  if (hashTuple->hashvalue == hashvalue)
2004  {
2005  TupleTableSlot *inntuple;
2006 
2007  /* insert hashtable's tuple into exec slot so ExecQual sees it */
2008  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2009  hjstate->hj_HashTupleSlot,
2010  false); /* do not pfree */
2011  econtext->ecxt_innertuple = inntuple;
2012 
2013  if (ExecQualAndReset(hjclauses, econtext))
2014  {
2015  hjstate->hj_CurTuple = hashTuple;
2016  return true;
2017  }
2018  }
2019 
2020  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2021  }
2022 
2023  /*
2024  * no match
2025  */
2026  return false;
2027 }
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1410
uint32 hj_CurHashValue
Definition: execnodes.h:1890
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1893
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:225
unsigned int uint32
Definition: c.h:358
int hj_CurBucketNo
Definition: execnodes.h:1891
static bool ExecQualAndReset(ExprState *state, ExprContext *econtext)
Definition: executor.h:391
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
static HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable table, int bucketno)
Definition: nodeHash.c:3173
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:1895
static HashJoinTuple ExecParallelHashNextTuple(HashJoinTable table, HashJoinTuple tuple)
Definition: nodeHash.c:3189
HashJoinTable hj_HashTable
Definition: execnodes.h:1889
uint32 hashvalue
Definition: hashjoin.h:75
ExprState * hashclauses
Definition: execnodes.h:1884

◆ ExecPrepHashTableForUnmatched()

void ExecPrepHashTableForUnmatched ( HashJoinState hjstate)

Definition at line 2034 of file nodeHash.c.

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

Referenced by ExecHashJoinImpl().

2035 {
2036  /*----------
2037  * During this scan we use the HashJoinState fields as follows:
2038  *
2039  * hj_CurBucketNo: next regular bucket to scan
2040  * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
2041  * hj_CurTuple: last tuple returned, or NULL to start next bucket
2042  *----------
2043  */
2044  hjstate->hj_CurBucketNo = 0;
2045  hjstate->hj_CurSkewBucketNo = 0;
2046  hjstate->hj_CurTuple = NULL;
2047 }
int hj_CurSkewBucketNo
Definition: execnodes.h:1892
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1893
int hj_CurBucketNo
Definition: execnodes.h:1891

◆ ExecReScanHash()

void ExecReScanHash ( HashState node)

Definition at line 2184 of file nodeHash.c.

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

Referenced by ExecReScan().

2185 {
2186  /*
2187  * if chgParam of subnode is not null then plan will be re-scanned by
2188  * first ExecProcNode.
2189  */
2190  if (node->ps.lefttree->chgParam == NULL)
2191  ExecReScan(node->ps.lefttree);
2192 }
void ExecReScan(PlanState *node)
Definition: execAmi.c:77
struct PlanState * lefttree
Definition: execnodes.h:967
PlanState ps
Definition: execnodes.h:2222
Bitmapset * chgParam
Definition: execnodes.h:977

◆ ExecScanHashBucket()

bool ExecScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 1922 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().

1924 {
1925  ExprState *hjclauses = hjstate->hashclauses;
1926  HashJoinTable hashtable = hjstate->hj_HashTable;
1927  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1928  uint32 hashvalue = hjstate->hj_CurHashValue;
1929 
1930  /*
1931  * hj_CurTuple is the address of the tuple last returned from the current
1932  * bucket, or NULL if it's time to start scanning a new bucket.
1933  *
1934  * If the tuple hashed to a skew bucket then scan the skew bucket
1935  * otherwise scan the standard hashtable bucket.
1936  */
1937  if (hashTuple != NULL)
1938  hashTuple = hashTuple->next.unshared;
1939  else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
1940  hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
1941  else
1942  hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
1943 
1944  while (hashTuple != NULL)
1945  {
1946  if (hashTuple->hashvalue == hashvalue)
1947  {
1948  TupleTableSlot *inntuple;
1949 
1950  /* insert hashtable's tuple into exec slot so ExecQual sees it */
1951  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
1952  hjstate->hj_HashTupleSlot,
1953  false); /* do not pfree */
1954  econtext->ecxt_innertuple = inntuple;
1955 
1956  if (ExecQualAndReset(hjclauses, econtext))
1957  {
1958  hjstate->hj_CurTuple = hashTuple;
1959  return true;
1960  }
1961  }
1962 
1963  hashTuple = hashTuple->next.unshared;
1964  }
1965 
1966  /*
1967  * no match
1968  */
1969  return false;
1970 }
#define INVALID_SKEW_BUCKET_NO
Definition: hashjoin.h:109
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1410
union HashJoinTupleData::@95 next
uint32 hj_CurHashValue
Definition: execnodes.h:1890
int hj_CurSkewBucketNo
Definition: execnodes.h:1892
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1893
HashJoinTuple tuples
Definition: hashjoin.h:105
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:225
unsigned int uint32
Definition: c.h:358
int hj_CurBucketNo
Definition: execnodes.h:1891
static bool ExecQualAndReset(ExprState *state, ExprContext *econtext)
Definition: executor.h:391
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
union HashJoinTableData::@97 buckets
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:1895
HashJoinTable hj_HashTable
Definition: execnodes.h:1889
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
uint32 hashvalue
Definition: hashjoin.h:75
ExprState * hashclauses
Definition: execnodes.h:1884

◆ ExecScanHashTableForUnmatched()

bool ExecScanHashTableForUnmatched ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 2058 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().

2059 {
2060  HashJoinTable hashtable = hjstate->hj_HashTable;
2061  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2062 
2063  for (;;)
2064  {
2065  /*
2066  * hj_CurTuple is the address of the tuple last returned from the
2067  * current bucket, or NULL if it's time to start scanning a new
2068  * bucket.
2069  */
2070  if (hashTuple != NULL)
2071  hashTuple = hashTuple->next.unshared;
2072  else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
2073  {
2074  hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
2075  hjstate->hj_CurBucketNo++;
2076  }
2077  else if (hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
2078  {
2079  int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
2080 
2081  hashTuple = hashtable->skewBucket[j]->tuples;
2082  hjstate->hj_CurSkewBucketNo++;
2083  }
2084  else
2085  break; /* finished all buckets */
2086 
2087  while (hashTuple != NULL)
2088  {
2089  if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
2090  {
2091  TupleTableSlot *inntuple;
2092 
2093  /* insert hashtable's tuple into exec slot */
2094  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2095  hjstate->hj_HashTupleSlot,
2096  false); /* do not pfree */
2097  econtext->ecxt_innertuple = inntuple;
2098 
2099  /*
2100  * Reset temp memory each time; although this function doesn't
2101  * do any qual eval, the caller will, so let's keep it
2102  * parallel to ExecScanHashBucket.
2103  */
2104  ResetExprContext(econtext);
2105 
2106  hjstate->hj_CurTuple = hashTuple;
2107  return true;
2108  }
2109 
2110  hashTuple = hashTuple->next.unshared;
2111  }
2112 
2113  /* allow this loop to be cancellable */
2115  }
2116 
2117  /*
2118  * no more unmatched tuples
2119  */
2120  return false;
2121 }
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1410
union HashJoinTupleData::@95 next
int * skewBucketNums
Definition: hashjoin.h:308
int hj_CurSkewBucketNo
Definition: execnodes.h:1892
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1893
HashJoinTuple tuples
Definition: hashjoin.h:105
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:225
int hj_CurBucketNo
Definition: execnodes.h:1891
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define HeapTupleHeaderHasMatch(tup)
Definition: htup_details.h:516
union HashJoinTableData::@97 buckets
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:1895
HashJoinTable hj_HashTable
Definition: execnodes.h:1889
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
#define ResetExprContext(econtext)
Definition: executor.h:494

◆ ExecShutdownHash()

void ExecShutdownHash ( HashState node)

Definition at line 2625 of file nodeHash.c.

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

Referenced by ExecShutdownNode().

2626 {
2627  if (node->hinstrument && node->hashtable)
2629 }
void ExecHashGetInstrumentation(HashInstrumentation *instrument, HashJoinTable hashtable)
Definition: nodeHash.c:2656
HashJoinTable hashtable
Definition: execnodes.h:2223
HashInstrumentation * hinstrument
Definition: execnodes.h:2228

◆ 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