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 "port/pg_bitutils.h"
#include "utils/dynahash.h"
#include "utils/lsyscache.h"
#include "utils/memutils.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 2680 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().

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

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

◆ ExecEndHash()

void ExecEndHash ( HashState node)

Definition at line 406 of file nodeHash.c.

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

Referenced by ExecEndNode().

407 {
409 
410  /*
411  * free exprcontext
412  */
413  ExecFreeExprContext(&node->ps);
414 
415  /*
416  * shut down the subplan
417  */
418  outerPlan = outerPlanState(node);
419  ExecEndNode(outerPlan);
420 }
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:537
void ExecFreeExprContext(PlanState *planstate)
Definition: execUtils.c:614
#define outerPlanState(node)
Definition: execnodes.h:1037
PlanState ps
Definition: execnodes.h:2278
#define outerPlan(node)
Definition: plannodes.h:172

◆ 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:228

◆ ExecHashBuildSkewHash()

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

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

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

◆ ExecHashEstimate()

void ExecHashEstimate ( HashState node,
ParallelContext pcxt 
)

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

2575 {
2576  size_t size;
2577 
2578  /* don't need this if not instrumenting or no workers */
2579  if (!node->ps.instrument || pcxt->nworkers == 0)
2580  return;
2581 
2582  size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
2583  size = add_size(size, offsetof(SharedHashInfo, hinstrument));
2584  shm_toc_estimate_chunk(&pcxt->estimator, size);
2585  shm_toc_estimate_keys(&pcxt->estimator, 1);
2586 }
Instrumentation * instrument
Definition: execnodes.h:953
shm_toc_estimator estimator
Definition: parallel.h:42
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
PlanState ps
Definition: execnodes.h:2278
Size mul_size(Size s1, Size s2)
Definition: shmem.c:515
Size add_size(Size s1, Size s2)
Definition: shmem.c:498
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
#define offsetof(type, field)
Definition: c.h:661

◆ ExecHashGetBucketAndBatch()

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

Definition at line 1900 of file nodeHash.c.

References HashJoinTableData::log2_nbuckets, HashJoinTableData::nbatch, HashJoinTableData::nbuckets, and pg_rotate_right32().

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

1904 {
1905  uint32 nbuckets = (uint32) hashtable->nbuckets;
1906  uint32 nbatch = (uint32) hashtable->nbatch;
1907 
1908  if (nbatch > 1)
1909  {
1910  *bucketno = hashvalue & (nbuckets - 1);
1911  *batchno = pg_rotate_right32(hashvalue,
1912  hashtable->log2_nbuckets) & (nbatch - 1);
1913  }
1914  else
1915  {
1916  *bucketno = hashvalue & (nbuckets - 1);
1917  *batchno = 0;
1918  }
1919 }
static uint32 pg_rotate_right32(uint32 word, int n)
Definition: pg_bitutils.h:143
unsigned int uint32
Definition: c.h:367

◆ ExecHashGetHashValue()

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

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

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

◆ ExecHashGetInstrumentation()

void ExecHashGetInstrumentation ( HashInstrumentation instrument,
HashJoinTable  hashtable 
)

◆ ExecHashGetSkewBucket()

int ExecHashGetSkewBucket ( HashJoinTable  hashtable,
uint32  hashvalue 
)

Definition at line 2369 of file nodeHash.c.

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

Referenced by ExecHashJoinImpl(), and MultiExecPrivateHash().

2370 {
2371  int bucket;
2372 
2373  /*
2374  * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
2375  * particular, this happens after the initial batch is done).
2376  */
2377  if (!hashtable->skewEnabled)
2378  return INVALID_SKEW_BUCKET_NO;
2379 
2380  /*
2381  * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
2382  */
2383  bucket = hashvalue & (hashtable->skewBucketLen - 1);
2384 
2385  /*
2386  * While we have not hit a hole in the hashtable and have not hit the
2387  * desired bucket, we have collided with some other hash value, so try the
2388  * next bucket location.
2389  */
2390  while (hashtable->skewBucket[bucket] != NULL &&
2391  hashtable->skewBucket[bucket]->hashvalue != hashvalue)
2392  bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
2393 
2394  /*
2395  * Found the desired bucket?
2396  */
2397  if (hashtable->skewBucket[bucket] != NULL)
2398  return bucket;
2399 
2400  /*
2401  * There must not be any hashtable entry for this hash value.
2402  */
2403  return INVALID_SKEW_BUCKET_NO;
2404 }
#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 887 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().

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

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

2594 {
2595  size_t size;
2596 
2597  /* don't need this if not instrumenting or no workers */
2598  if (!node->ps.instrument || pcxt->nworkers == 0)
2599  return;
2600 
2601  size = offsetof(SharedHashInfo, hinstrument) +
2602  pcxt->nworkers * sizeof(HashInstrumentation);
2603  node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
2604  memset(node->shared_info, 0, size);
2605  node->shared_info->num_workers = pcxt->nworkers;
2606  shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id,
2607  node->shared_info);
2608 }
Instrumentation * instrument
Definition: execnodes.h:953
struct HashInstrumentation HashInstrumentation
int plan_node_id
Definition: plannodes.h:141
SharedHashInfo * shared_info
Definition: execnodes.h:2282
PlanState ps
Definition: execnodes.h:2278
Plan * plan
Definition: execnodes.h:943
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:661
shm_toc * toc
Definition: parallel.h:45

◆ ExecHashInitializeWorker()

void ExecHashInitializeWorker ( HashState node,
ParallelWorkerContext pwcxt 
)

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

2616 {
2617  SharedHashInfo *shared_info;
2618 
2619  /* don't need this if not instrumenting */
2620  if (!node->ps.instrument)
2621  return;
2622 
2623  shared_info = (SharedHashInfo *)
2624  shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
2625  node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
2626 }
Instrumentation * instrument
Definition: execnodes.h:953
int plan_node_id
Definition: plannodes.h:141
int ParallelWorkerNumber
Definition: parallel.c:110
PlanState ps
Definition: execnodes.h:2278
HashInstrumentation * hinstrument
Definition: execnodes.h:2283
HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER]
Definition: execnodes.h:2269
Plan * plan
Definition: execnodes.h:943
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 2461 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().

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

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

Referenced by ExecParallelRetrieveInstrumentation().

2647 {
2648  SharedHashInfo *shared_info = node->shared_info;
2649  size_t size;
2650 
2651  if (shared_info == NULL)
2652  return;
2653 
2654  /* Replace node->shared_info with a copy in backend-local memory. */
2655  size = offsetof(SharedHashInfo, hinstrument) +
2656  shared_info->num_workers * sizeof(HashInstrumentation);
2657  node->shared_info = palloc(size);
2658  memcpy(node->shared_info, shared_info, size);
2659 }
struct HashInstrumentation HashInstrumentation
SharedHashInfo * shared_info
Definition: execnodes.h:2282
void * palloc(Size size)
Definition: mcxt.c:949
#define offsetof(type, field)
Definition: c.h:661

◆ ExecHashSkewTableInsert()

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

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

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

◆ ExecHashTableCreate()

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

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

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

◆ ExecHashTableDestroy()

void ExecHashTableDestroy ( HashJoinTable  hashtable)

Definition at line 854 of file nodeHash.c.

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

Referenced by ExecEndHashJoin(), and ExecReScanHashJoin().

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

3147 {
3148  if (hashtable->parallel_state)
3149  {
3150  ParallelHashJoinState *pstate = hashtable->parallel_state;
3151  int i;
3152 
3153  /* Make sure any temporary files are closed. */
3154  if (hashtable->batches)
3155  {
3156  for (i = 0; i < hashtable->nbatch; ++i)
3157  {
3158  sts_end_write(hashtable->batches[i].inner_tuples);
3159  sts_end_write(hashtable->batches[i].outer_tuples);
3162  }
3163  }
3164 
3165  /* If we're last to detach, clean up shared memory. */
3166  if (BarrierDetach(&pstate->build_barrier))
3167  {
3168  if (DsaPointerIsValid(pstate->batches))
3169  {
3170  dsa_free(hashtable->area, pstate->batches);
3171  pstate->batches = InvalidDsaPointer;
3172  }
3173  }
3174 
3175  hashtable->parallel_state = NULL;
3176  }
3177 }
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 3089 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().

3090 {
3091  if (hashtable->parallel_state != NULL &&
3092  hashtable->curbatch >= 0)
3093  {
3094  int curbatch = hashtable->curbatch;
3095  ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
3096 
3097  /* Make sure any temporary files are closed. */
3098  sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
3099  sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
3100 
3101  /* Detach from the batch we were last working on. */
3103  {
3104  /*
3105  * Technically we shouldn't access the barrier because we're no
3106  * longer attached, but since there is no way it's moving after
3107  * this point it seems safe to make the following assertion.
3108  */
3110 
3111  /* Free shared chunks and buckets. */
3112  while (DsaPointerIsValid(batch->chunks))
3113  {
3114  HashMemoryChunk chunk =
3115  dsa_get_address(hashtable->area, batch->chunks);
3116  dsa_pointer next = chunk->next.shared;
3117 
3118  dsa_free(hashtable->area, batch->chunks);
3119  batch->chunks = next;
3120  }
3121  if (DsaPointerIsValid(batch->buckets))
3122  {
3123  dsa_free(hashtable->area, batch->buckets);
3124  batch->buckets = InvalidDsaPointer;
3125  }
3126  }
3127 
3128  /*
3129  * Track the largest batch we've been attached to. Though each
3130  * backend might see a different subset of batches, explain.c will
3131  * scan the results from all backends to find the largest value.
3132  */
3133  hashtable->spacePeak =
3134  Max(hashtable->spacePeak,
3135  batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
3136 
3137  /* Remember that we are not attached to a batch. */
3138  hashtable->curbatch = -1;
3139  }
3140 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
#define PHJ_BATCH_DONE
Definition: hashjoin.h:268
static int32 next
Definition: blutils.c:217
#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
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:914
#define Assert(condition)
Definition: c.h:738
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
union HashMemoryChunkData::@94 next
dsa_pointer buckets
Definition: hashjoin.h:153

◆ ExecHashTableInsert()

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

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

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

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

Referenced by ExecHashJoinNewBatch().

2140 {
2141  MemoryContext oldcxt;
2142  int nbuckets = hashtable->nbuckets;
2143 
2144  /*
2145  * Release all the hash buckets and tuples acquired in the prior pass, and
2146  * reinitialize the context for a new pass.
2147  */
2148  MemoryContextReset(hashtable->batchCxt);
2149  oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
2150 
2151  /* Reallocate and reinitialize the hash bucket headers. */
2152  hashtable->buckets.unshared = (HashJoinTuple *)
2153  palloc0(nbuckets * sizeof(HashJoinTuple));
2154 
2155  hashtable->spaceUsed = 0;
2156 
2157  MemoryContextSwitchTo(oldcxt);
2158 
2159  /* Forget the chunks (the memory was freed by the context reset above). */
2160  hashtable->chunks = NULL;
2161 }
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:980
union HashJoinTableData::@95 buckets
HashMemoryChunk chunks
Definition: hashjoin.h:352
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297

◆ ExecHashTableResetMatchFlags()

void ExecHashTableResetMatchFlags ( HashJoinTable  hashtable)

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

2169 {
2170  HashJoinTuple tuple;
2171  int i;
2172 
2173  /* Reset all flags in the main table ... */
2174  for (i = 0; i < hashtable->nbuckets; i++)
2175  {
2176  for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
2177  tuple = tuple->next.unshared)
2179  }
2180 
2181  /* ... and the same for the skew buckets, if any */
2182  for (i = 0; i < hashtable->nSkewBuckets; i++)
2183  {
2184  int j = hashtable->skewBucketNums[i];
2185  HashSkewBucket *skewBucket = hashtable->skewBucket[j];
2186 
2187  for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
2189  }
2190 }
int * skewBucketNums
Definition: hashjoin.h:308
struct HashJoinTupleData * unshared
Definition: hashjoin.h:72
HashJoinTuple tuples
Definition: hashjoin.h:105
union HashJoinTupleData::@93 next
HashSkewBucket ** skewBucket
Definition: hashjoin.h:305
union HashJoinTableData::@95 buckets
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:526
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:297
int i

◆ ExecInitHash()

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

Definition at line 353 of file nodeHash.c.

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

Referenced by ExecInitNode().

354 {
355  HashState *hashstate;
356 
357  /* check for unsupported flags */
358  Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
359 
360  /*
361  * create state structure
362  */
363  hashstate = makeNode(HashState);
364  hashstate->ps.plan = (Plan *) node;
365  hashstate->ps.state = estate;
366  hashstate->ps.ExecProcNode = ExecHash;
367  hashstate->hashtable = NULL;
368  hashstate->hashkeys = NIL; /* will be set by parent HashJoin */
369 
370  /*
371  * Miscellaneous initialization
372  *
373  * create expression context for node
374  */
375  ExecAssignExprContext(estate, &hashstate->ps);
376 
377  /*
378  * initialize child nodes
379  */
380  outerPlanState(hashstate) = ExecInitNode(outerPlan(node), estate, eflags);
381 
382  /*
383  * initialize our result slot and type. No need to build projection
384  * because this node doesn't do projections.
385  */
387  hashstate->ps.ps_ProjInfo = NULL;
388 
389  /*
390  * initialize child expressions
391  */
392  Assert(node->plan.qual == NIL);
393  hashstate->hashkeys =
394  ExecInitExprList(node->hashkeys, (PlanState *) hashstate);
395 
396  return hashstate;
397 }
#define NIL
Definition: pg_list.h:65
List * qual
Definition: plannodes.h:143
ProjectionInfo * ps_ProjInfo
Definition: execnodes.h:983
HashJoinTable hashtable
Definition: execnodes.h:2279
EState * state
Definition: execnodes.h:945
#define EXEC_FLAG_BACKWARD
Definition: executor.h:58
#define outerPlanState(node)
Definition: execnodes.h:1037
List * ExecInitExprList(List *nodes, PlanState *parent)
Definition: execExpr.c:316
List * hashkeys
Definition: execnodes.h:2280
PlanState ps
Definition: execnodes.h:2278
#define outerPlan(node)
Definition: plannodes.h:172
static TupleTableSlot * ExecHash(PlanState *pstate)
Definition: nodeHash.c:91
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:949
List * hashkeys
Definition: plannodes.h:920
Plan * plan
Definition: execnodes.h:943
#define makeNode(_type_)
Definition: nodes.h:573
#define Assert(condition)
Definition: c.h:738
#define EXEC_FLAG_MARK
Definition: executor.h:59
void ExecAssignExprContext(EState *estate, PlanState *planstate)
Definition: execUtils.c:444
void ExecInitResultTupleSlotTL(PlanState *planstate, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1769
Plan plan
Definition: plannodes.h:914
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:138
const TupleTableSlotOps TTSOpsMinimalTuple
Definition: execTuples.c:85

◆ ExecParallelHashCloseBatchAccessors()

static void ExecParallelHashCloseBatchAccessors ( HashJoinTable  hashtable)
static

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

2986 {
2987  int i;
2988 
2989  for (i = 0; i < hashtable->nbatch; ++i)
2990  {
2991  /* Make sure no files are left open. */
2992  sts_end_write(hashtable->batches[i].inner_tuples);
2993  sts_end_write(hashtable->batches[i].outer_tuples);
2996  }
2997  pfree(hashtable->batches);
2998  hashtable->batches = NULL;
2999 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
void pfree(void *pointer)
Definition: mcxt.c:1056
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 3006 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().

3007 {
3008  ParallelHashJoinState *pstate = hashtable->parallel_state;
3009  ParallelHashJoinBatch *batches;
3010  MemoryContext oldcxt;
3011  int i;
3012 
3013  if (hashtable->batches != NULL)
3014  {
3015  if (hashtable->nbatch == pstate->nbatch)
3016  return;
3018  }
3019 
3020  /*
3021  * It's possible for a backend to start up very late so that the whole
3022  * join is finished and the shm state for tracking batches has already
3023  * been freed by ExecHashTableDetach(). In that case we'll just leave
3024  * hashtable->batches as NULL so that ExecParallelHashJoinNewBatch() gives
3025  * up early.
3026  */
3027  if (!DsaPointerIsValid(pstate->batches))
3028  return;
3029 
3030  /* Use hash join memory context. */
3031  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
3032 
3033  /* Allocate this backend's accessor array. */
3034  hashtable->nbatch = pstate->nbatch;
3035  hashtable->batches = (ParallelHashJoinBatchAccessor *)
3036  palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
3037 
3038  /* Find the base of the pseudo-array of ParallelHashJoinBatch objects. */
3039  batches = (ParallelHashJoinBatch *)
3040  dsa_get_address(hashtable->area, pstate->batches);
3041 
3042  /* Set up the accessor array and attach to the tuplestores. */
3043  for (i = 0; i < hashtable->nbatch; ++i)
3044  {
3045  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
3046  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
3047 
3048  accessor->shared = shared;
3049  accessor->preallocated = 0;
3050  accessor->done = false;
3051  accessor->inner_tuples =
3054  &pstate->fileset);
3055  accessor->outer_tuples =
3057  pstate->nparticipants),
3059  &pstate->fileset);
3060  }
3061 
3062  MemoryContextSwitchTo(oldcxt);
3063 }
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:2985
#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:980
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 3183 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().

3184 {
3185  HashJoinTuple tuple;
3186  dsa_pointer p;
3187 
3188  Assert(hashtable->parallel_state);
3189  p = dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
3190  tuple = (HashJoinTuple) dsa_get_address(hashtable->area, p);
3191 
3192  return tuple;
3193 }
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:1935
#define Assert(condition)
Definition: c.h:738
#define dsa_pointer_atomic_read
Definition: dsa.h:65

◆ ExecParallelHashIncreaseNumBatches()

static void ExecParallelHashIncreaseNumBatches ( HashJoinTable  hashtable)
static

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

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

1496 {
1497  ParallelHashJoinState *pstate = hashtable->parallel_state;
1498  int i;
1499  HashMemoryChunk chunk;
1500  dsa_pointer chunk_s;
1501 
1503 
1504  /*
1505  * It's unlikely, but we need to be prepared for new participants to show
1506  * up while we're in the middle of this operation so we need to switch on
1507  * barrier phase here.
1508  */
1510  {
1512  /* Elect one participant to prepare to increase nbuckets. */
1515  {
1516  size_t size;
1517  dsa_pointer_atomic *buckets;
1518 
1519  /* Double the size of the bucket array. */
1520  pstate->nbuckets *= 2;
1521  size = pstate->nbuckets * sizeof(dsa_pointer_atomic);
1522  hashtable->batches[0].shared->size += size / 2;
1523  dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
1524  hashtable->batches[0].shared->buckets =
1525  dsa_allocate(hashtable->area, size);
1526  buckets = (dsa_pointer_atomic *)
1527  dsa_get_address(hashtable->area,
1528  hashtable->batches[0].shared->buckets);
1529  for (i = 0; i < pstate->nbuckets; ++i)
1531 
1532  /* Put the chunk list onto the work queue. */
1533  pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;
1534 
1535  /* Clear the flag. */
1536  pstate->growth = PHJ_GROWTH_OK;
1537  }
1538  /* Fall through. */
1539 
1541  /* Wait for the above to complete. */
1544  /* Fall through. */
1545 
1547  /* Reinsert all tuples into the hash table. */
1550  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_s)))
1551  {
1552  size_t idx = 0;
1553 
1554  while (idx < chunk->used)
1555  {
1556  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1557  dsa_pointer shared = chunk_s + HASH_CHUNK_HEADER_SIZE + idx;
1558  int bucketno;
1559  int batchno;
1560 
1561  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1562  &bucketno, &batchno);
1563  Assert(batchno == 0);
1564 
1565  /* add the tuple to the proper bucket */
1566  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1567  hashTuple, shared);
1568 
1569  /* advance index past the tuple */
1570  idx += MAXALIGN(HJTUPLE_OVERHEAD +
1571  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1572  }
1573 
1574  /* allow this loop to be cancellable */
1576  }
1579  }
1580 }
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:3231
dsa_pointer chunks
Definition: hashjoin.h:156
Datum idx(PG_FUNCTION_ARGS)
Definition: _int_op.c:259
static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:3006
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:1900
#define PHJ_GROW_BUCKETS_ALLOCATING
Definition: hashjoin.h:280
Barrier grow_buckets_barrier
Definition: hashjoin.h:250
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1935
static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table, dsa_pointer *shared)
Definition: nodeHash.c:3252
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
union HashJoinTableData::@95 buckets
#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:738
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:691
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:3213
#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 2908 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().

2909 {
2910  ParallelHashJoinState *pstate = hashtable->parallel_state;
2911  ParallelHashJoinBatch *batches;
2912  MemoryContext oldcxt;
2913  int i;
2914 
2915  Assert(hashtable->batches == NULL);
2916 
2917  /* Allocate space. */
2918  pstate->batches =
2919  dsa_allocate0(hashtable->area,
2920  EstimateParallelHashJoinBatch(hashtable) * nbatch);
2921  pstate->nbatch = nbatch;
2922  batches = dsa_get_address(hashtable->area, pstate->batches);
2923 
2924  /* Use hash join memory context. */
2925  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
2926 
2927  /* Allocate this backend's accessor array. */
2928  hashtable->nbatch = nbatch;
2929  hashtable->batches = (ParallelHashJoinBatchAccessor *)
2930  palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
2931 
2932  /* Set up the shared state, tuplestores and backend-local accessors. */
2933  for (i = 0; i < hashtable->nbatch; ++i)
2934  {
2935  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
2936  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
2937  char name[MAXPGPATH];
2938 
2939  /*
2940  * All members of shared were zero-initialized. We just need to set
2941  * up the Barrier.
2942  */
2943  BarrierInit(&shared->batch_barrier, 0);
2944  if (i == 0)
2945  {
2946  /* Batch 0 doesn't need to be loaded. */
2947  BarrierAttach(&shared->batch_barrier);
2948  while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING)
2949  BarrierArriveAndWait(&shared->batch_barrier, 0);
2950  BarrierDetach(&shared->batch_barrier);
2951  }
2952 
2953  /* Initialize accessor state. All members were zero-initialized. */
2954  accessor->shared = shared;
2955 
2956  /* Initialize the shared tuplestores. */
2957  snprintf(name, sizeof(name), "i%dof%d", i, hashtable->nbatch);
2958  accessor->inner_tuples =
2960  pstate->nparticipants,
2962  sizeof(uint32),
2964  &pstate->fileset,
2965  name);
2966  snprintf(name, sizeof(name), "o%dof%d", i, hashtable->nbatch);
2967  accessor->outer_tuples =
2969  pstate->nparticipants),
2970  pstate->nparticipants,
2972  sizeof(uint32),
2974  &pstate->fileset,
2975  name);
2976  }
2977 
2978  MemoryContextSwitchTo(oldcxt);
2979 }
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:367
#define SHARED_TUPLESTORE_SINGLE_PASS
int BarrierAttach(Barrier *barrier)
Definition: barrier.c:214
void * palloc0(Size size)
Definition: mcxt.c:980
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:738
#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 1402 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().

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

◆ ExecParallelHashNextTuple()

static HashJoinTuple ExecParallelHashNextTuple ( HashJoinTable  table,
HashJoinTuple  tuple 
)
inlinestatic

Definition at line 3199 of file nodeHash.c.

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

Referenced by ExecParallelScanHashBucket().

3200 {
3202 
3203  Assert(hashtable->parallel_state);
3204  next = (HashJoinTuple) dsa_get_address(hashtable->area, tuple->next.shared);
3205 
3206  return next;
3207 }
static int32 next
Definition: blutils.c:217
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:1935
union HashJoinTupleData::@93 next
#define Assert(condition)
Definition: c.h:738

◆ ExecParallelHashPopChunkQueue()

static HashMemoryChunk ExecParallelHashPopChunkQueue ( HashJoinTable  table,
dsa_pointer shared 
)
static

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

3253 {
3254  ParallelHashJoinState *pstate = hashtable->parallel_state;
3255  HashMemoryChunk chunk;
3256 
3257  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3258  if (DsaPointerIsValid(pstate->chunk_work_queue))
3259  {
3260  *shared = pstate->chunk_work_queue;
3261  chunk = (HashMemoryChunk)
3262  dsa_get_address(hashtable->area, *shared);
3263  pstate->chunk_work_queue = chunk->next.shared;
3264  }
3265  else
3266  chunk = NULL;
3267  LWLockRelease(&pstate->lock);
3268 
3269  return chunk;
3270 }
dsa_pointer chunk_work_queue
Definition: hashjoin.h:242
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1727
dsa_pointer shared
Definition: hashjoin.h:127
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:932
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1123
#define DsaPointerIsValid(x)
Definition: dsa.h:81
union HashMemoryChunkData::@94 next

◆ ExecParallelHashPushTuple()

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

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

3216 {
3217  for (;;)
3218  {
3219  tuple->next.shared = dsa_pointer_atomic_read(head);
3221  &tuple->next.shared,
3222  tuple_shared))
3223  break;
3224  }
3225 }
dsa_pointer shared
Definition: hashjoin.h:73
#define dsa_pointer_atomic_compare_exchange
Definition: dsa.h:68
union HashJoinTupleData::@93 next
#define dsa_pointer_atomic_read
Definition: dsa.h:65

◆ ExecParallelHashRepartitionFirst()

static void ExecParallelHashRepartitionFirst ( HashJoinTable  hashtable)
static

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

1276 {
1277  dsa_pointer chunk_shared;
1278  HashMemoryChunk chunk;
1279 
1280  Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);
1281 
1282  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared)))
1283  {
1284  size_t idx = 0;
1285 
1286  /* Repartition all tuples in this chunk. */
1287  while (idx < chunk->used)
1288  {
1289  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1290  MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
1291  HashJoinTuple copyTuple;
1292  dsa_pointer shared;
1293  int bucketno;
1294  int batchno;
1295 
1296  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1297  &bucketno, &batchno);
1298 
1299  Assert(batchno < hashtable->nbatch);
1300  if (batchno == 0)
1301  {
1302  /* It still belongs in batch 0. Copy to a new chunk. */
1303  copyTuple =
1304  ExecParallelHashTupleAlloc(hashtable,
1305  HJTUPLE_OVERHEAD + tuple->t_len,
1306  &shared);
1307  copyTuple->hashvalue = hashTuple->hashvalue;
1308  memcpy(HJTUPLE_MINTUPLE(copyTuple), tuple, tuple->t_len);
1309  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1310  copyTuple, shared);
1311  }
1312  else
1313  {
1314  size_t tuple_size =
1315  MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1316 
1317  /* It belongs in a later batch. */
1318  hashtable->batches[batchno].estimated_size += tuple_size;
1319  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1320  &hashTuple->hashvalue, tuple);
1321  }
1322 
1323  /* Count this tuple. */
1324  ++hashtable->batches[0].old_ntuples;
1325  ++hashtable->batches[batchno].ntuples;
1326 
1327  idx += MAXALIGN(HJTUPLE_OVERHEAD +
1328  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1329  }
1330 
1331  /* Free this chunk. */
1332  dsa_free(hashtable->area, chunk_shared);
1333 
1335  }
1336 }
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:259
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:1900
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:1935
static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table, dsa_pointer *shared)
Definition: nodeHash.c:3252
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
union HashJoinTableData::@95 buckets
#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:738
#define MAXALIGN(LEN)
Definition: c.h:691
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:3213
#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:2760
uint32 hashvalue
Definition: hashjoin.h:75

◆ ExecParallelHashRepartitionRest()

static void ExecParallelHashRepartitionRest ( HashJoinTable  hashtable)
static

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

1343 {
1344  ParallelHashJoinState *pstate = hashtable->parallel_state;
1345  int old_nbatch = pstate->old_nbatch;
1346  SharedTuplestoreAccessor **old_inner_tuples;
1347  ParallelHashJoinBatch *old_batches;
1348  int i;
1349 
1350  /* Get our hands on the previous generation of batches. */
1351  old_batches = (ParallelHashJoinBatch *)
1352  dsa_get_address(hashtable->area, pstate->old_batches);
1353  old_inner_tuples = palloc0(sizeof(SharedTuplestoreAccessor *) * old_nbatch);
1354  for (i = 1; i < old_nbatch; ++i)
1355  {
1356  ParallelHashJoinBatch *shared =
1357  NthParallelHashJoinBatch(old_batches, i);
1358 
1359  old_inner_tuples[i] = sts_attach(ParallelHashJoinBatchInner(shared),
1361  &pstate->fileset);
1362  }
1363 
1364  /* Join in the effort to repartition them. */
1365  for (i = 1; i < old_nbatch; ++i)
1366  {
1367  MinimalTuple tuple;
1368  uint32 hashvalue;
1369 
1370  /* Scan one partition from the previous generation. */
1371  sts_begin_parallel_scan(old_inner_tuples[i]);
1372  while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue)))
1373  {
1374  size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1375  int bucketno;
1376  int batchno;
1377 
1378  /* Decide which partition it goes to in the new generation. */
1379  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
1380  &batchno);
1381 
1382  hashtable->batches[batchno].estimated_size += tuple_size;
1383  ++hashtable->batches[batchno].ntuples;
1384  ++hashtable->batches[i].old_ntuples;
1385 
1386  /* Store the tuple its new batch. */
1387  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1388  &hashvalue, tuple);
1389 
1391  }
1392  sts_end_parallel_scan(old_inner_tuples[i]);
1393  }
1394 
1395  pfree(old_inner_tuples);
1396 }
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:1900
void pfree(void *pointer)
Definition: mcxt.c:1056
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:367
void sts_begin_parallel_scan(SharedTuplestoreAccessor *accessor)
void * palloc0(Size size)
Definition: mcxt.c:980
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:691
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 3069 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().

3070 {
3071  ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
3072  dsa_pointer_atomic *buckets;
3073  int nbuckets = hashtable->parallel_state->nbuckets;
3074  int i;
3075 
3076  batch->buckets =
3077  dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
3078  buckets = (dsa_pointer_atomic *)
3079  dsa_get_address(hashtable->area, batch->buckets);
3080  for (i = 0; i < nbuckets; ++i)
3082 }
#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 1683 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().

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

◆ ExecParallelHashTableInsertCurrentBatch()

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

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

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

◆ ExecParallelHashTableSetCurrentBatch()

void ExecParallelHashTableSetCurrentBatch ( HashJoinTable  hashtable,
int  batchno 
)

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

3232 {
3233  Assert(hashtable->batches[batchno].shared->buckets != InvalidDsaPointer);
3234 
3235  hashtable->curbatch = batchno;
3236  hashtable->buckets.shared = (dsa_pointer_atomic *)
3237  dsa_get_address(hashtable->area,
3238  hashtable->batches[batchno].shared->buckets);
3239  hashtable->nbuckets = hashtable->parallel_state->nbuckets;
3240  hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
3241  hashtable->current_chunk = NULL;
3243  hashtable->batches[batchno].at_least_one_chunk = false;
3244 }
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:1719
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:357
union HashJoinTableData::@95 buckets
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:358
#define Assert(condition)
Definition: c.h:738
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 2760 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().

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

◆ ExecParallelHashTuplePrealloc()

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

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

3294 {
3295  ParallelHashJoinState *pstate = hashtable->parallel_state;
3296  ParallelHashJoinBatchAccessor *batch = &hashtable->batches[batchno];
3297  size_t want = Max(size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);
3298 
3299  Assert(batchno > 0);
3300  Assert(batchno < hashtable->nbatch);
3301  Assert(size == MAXALIGN(size));
3302 
3303  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3304 
3305  /* Has another participant commanded us to help grow? */
3306  if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
3308  {
3309  ParallelHashGrowth growth = pstate->growth;
3310 
3311  LWLockRelease(&pstate->lock);
3312  if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
3314  else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3316 
3317  return false;
3318  }
3319 
3320  if (pstate->growth != PHJ_GROWTH_DISABLED &&
3321  batch->at_least_one_chunk &&
3322  (batch->shared->estimated_size + want + HASH_CHUNK_HEADER_SIZE
3323  > pstate->space_allowed))
3324  {
3325  /*
3326  * We have determined that this batch would exceed the space budget if
3327  * loaded into memory. Command all participants to help repartition.
3328  */
3329  batch->shared->space_exhausted = true;
3331  LWLockRelease(&pstate->lock);
3332 
3333  return false;
3334  }
3335 
3336  batch->at_least_one_chunk = true;
3337  batch->shared->estimated_size += want + HASH_CHUNK_HEADER_SIZE;
3338  batch->preallocated = want;
3339  LWLockRelease(&pstate->lock);
3340 
3341  return true;
3342 }
static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:1058
#define HASH_CHUNK_SIZE
Definition: hashjoin.h:139
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1727
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:914
#define Assert(condition)
Definition: c.h:738
ParallelHashGrowth growth
Definition: hashjoin.h:241
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1123
#define MAXALIGN(LEN)
Definition: c.h:691
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
ParallelHashGrowth
Definition: hashjoin.h:218
static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
Definition: nodeHash.c:1495

◆ ExecParallelScanHashBucket()

bool ExecParallelScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

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

1995 {
1996  ExprState *hjclauses = hjstate->hashclauses;
1997  HashJoinTable hashtable = hjstate->hj_HashTable;
1998  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1999  uint32 hashvalue = hjstate->hj_CurHashValue;
2000 
2001  /*
2002  * hj_CurTuple is the address of the tuple last returned from the current
2003  * bucket, or NULL if it's time to start scanning a new bucket.
2004  */
2005  if (hashTuple != NULL)
2006  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2007  else
2008  hashTuple = ExecParallelHashFirstTuple(hashtable,
2009  hjstate->hj_CurBucketNo);
2010 
2011  while (hashTuple != NULL)
2012  {
2013  if (hashTuple->hashvalue == hashvalue)
2014  {
2015  TupleTableSlot *inntuple;
2016 
2017  /* insert hashtable's tuple into exec slot so ExecQual sees it */
2018  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2019  hjstate->hj_HashTupleSlot,
2020  false); /* do not pfree */
2021  econtext->ecxt_innertuple = inntuple;
2022 
2023  if (ExecQualAndReset(hjclauses, econtext))
2024  {
2025  hjstate->hj_CurTuple = hashTuple;
2026  return true;
2027  }
2028  }
2029 
2030  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2031  }
2032 
2033  /*
2034  * no match
2035  */
2036  return false;
2037 }
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1416
uint32 hj_CurHashValue
Definition: execnodes.h:1946
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1949
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:226
unsigned int uint32
Definition: c.h:367
int hj_CurBucketNo
Definition: execnodes.h:1947
static bool ExecQualAndReset(ExprState *state, ExprContext *econtext)
Definition: executor.h:397
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:80
static HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable table, int bucketno)
Definition: nodeHash.c:3183
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:1951
static HashJoinTuple ExecParallelHashNextTuple(HashJoinTable table, HashJoinTuple tuple)
Definition: nodeHash.c:3199
HashJoinTable hj_HashTable
Definition: execnodes.h:1945
uint32 hashvalue
Definition: hashjoin.h:75
ExprState * hashclauses
Definition: execnodes.h:1941

◆ ExecPrepHashTableForUnmatched()

void ExecPrepHashTableForUnmatched ( HashJoinState hjstate)

Definition at line 2044 of file nodeHash.c.

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

Referenced by ExecHashJoinImpl().

2045 {
2046  /*----------
2047  * During this scan we use the HashJoinState fields as follows:
2048  *
2049  * hj_CurBucketNo: next regular bucket to scan
2050  * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
2051  * hj_CurTuple: last tuple returned, or NULL to start next bucket
2052  *----------
2053  */
2054  hjstate->hj_CurBucketNo = 0;
2055  hjstate->hj_CurSkewBucketNo = 0;
2056  hjstate->hj_CurTuple = NULL;
2057 }
int hj_CurSkewBucketNo
Definition: execnodes.h:1948
HashJoinTuple hj_CurTuple
Definition: execnodes.h:1949
int hj_CurBucketNo
Definition: execnodes.h:1947

◆ ExecReScanHash()

void ExecReScanHash ( HashState node)

Definition at line 2194 of file nodeHash.c.

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

Referenced by ExecReScan().

2195 {
2196  /*
2197  * if chgParam of subnode is not null then plan will be re-scanned by
2198  * first ExecProcNode.
2199  */
2200  if (node->ps.lefttree->chgParam == NULL)
2201  ExecReScan(node->ps.lefttree);
2202 }
void ExecReScan(PlanState *node)
Definition: execAmi.c:75
struct PlanState * lefttree
Definition: execnodes.h:965
PlanState ps
Definition: execnodes.h:2278
Bitmapset * chgParam
Definition: execnodes.h:975

◆ ExecScanHashBucket()

bool ExecScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

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

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

◆ ExecScanHashTableForUnmatched()

bool ExecScanHashTableForUnmatched ( HashJoinState hjstate,
ExprContext econtext 
)

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

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

◆ ExecShutdownHash()

void ExecShutdownHash ( HashState node)

Definition at line 2635 of file nodeHash.c.

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

Referenced by ExecShutdownNode().

2636 {
2637  if (node->hinstrument && node->hashtable)
2639 }
void ExecHashGetInstrumentation(HashInstrumentation *instrument, HashJoinTable hashtable)
Definition: nodeHash.c:2666
HashJoinTable hashtable
Definition: execnodes.h:2279
HashInstrumentation * hinstrument
Definition: execnodes.h:2283

◆ 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)
118