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/guc.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 hashtable, int bucketno)
 
static HashJoinTuple ExecParallelHashNextTuple (HashJoinTable hashtable, 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 hashtable, 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_hash_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 ExecHashAccumInstrumentation (HashInstrumentation *instrument, HashJoinTable hashtable)
 
void ExecParallelHashTableAlloc (HashJoinTable hashtable, int batchno)
 
void ExecHashTableDetachBatch (HashJoinTable hashtable)
 
void ExecHashTableDetach (HashJoinTable hashtable)
 
void ExecParallelHashTableSetCurrentBatch (HashJoinTable hashtable, int batchno)
 
size_t get_hash_memory_limit (void)
 

Macro Definition Documentation

◆ NTUP_PER_BUCKET

#define NTUP_PER_BUCKET   1

Definition at line 660 of file nodeHash.c.

Function Documentation

◆ dense_alloc()

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

Definition at line 2716 of file nodeHash.c.

2717 {
2718  HashMemoryChunk newChunk;
2719  char *ptr;
2720 
2721  /* just in case the size is not already aligned properly */
2722  size = MAXALIGN(size);
2723 
2724  /*
2725  * If tuple size is larger than threshold, allocate a separate chunk.
2726  */
2727  if (size > HASH_CHUNK_THRESHOLD)
2728  {
2729  /* allocate new chunk and put it at the beginning of the list */
2730  newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
2731  HASH_CHUNK_HEADER_SIZE + size);
2732  newChunk->maxlen = size;
2733  newChunk->used = size;
2734  newChunk->ntuples = 1;
2735 
2736  /*
2737  * Add this chunk to the list after the first existing chunk, so that
2738  * we don't lose the remaining space in the "current" chunk.
2739  */
2740  if (hashtable->chunks != NULL)
2741  {
2742  newChunk->next = hashtable->chunks->next;
2743  hashtable->chunks->next.unshared = newChunk;
2744  }
2745  else
2746  {
2747  newChunk->next.unshared = hashtable->chunks;
2748  hashtable->chunks = newChunk;
2749  }
2750 
2751  return HASH_CHUNK_DATA(newChunk);
2752  }
2753 
2754  /*
2755  * See if we have enough space for it in the current chunk (if any). If
2756  * not, allocate a fresh chunk.
2757  */
2758  if ((hashtable->chunks == NULL) ||
2759  (hashtable->chunks->maxlen - hashtable->chunks->used) < size)
2760  {
2761  /* allocate new chunk and put it at the beginning of the list */
2762  newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
2764 
2765  newChunk->maxlen = HASH_CHUNK_SIZE;
2766  newChunk->used = size;
2767  newChunk->ntuples = 1;
2768 
2769  newChunk->next.unshared = hashtable->chunks;
2770  hashtable->chunks = newChunk;
2771 
2772  return HASH_CHUNK_DATA(newChunk);
2773  }
2774 
2775  /* There is enough space in the current chunk, let's add the tuple */
2776  ptr = HASH_CHUNK_DATA(hashtable->chunks) + hashtable->chunks->used;
2777  hashtable->chunks->used += size;
2778  hashtable->chunks->ntuples += 1;
2779 
2780  /* return pointer to the start of the tuple memory */
2781  return ptr;
2782 }
#define MAXALIGN(LEN)
Definition: c.h:747
struct HashMemoryChunkData * HashMemoryChunk
Definition: hashjoin.h:137
#define HASH_CHUNK_DATA(hc)
Definition: hashjoin.h:141
#define HASH_CHUNK_THRESHOLD
Definition: hashjoin.h:143
#define HASH_CHUNK_HEADER_SIZE
Definition: hashjoin.h:140
#define HASH_CHUNK_SIZE
Definition: hashjoin.h:139
void * MemoryContextAlloc(MemoryContext context, Size size)
Definition: mcxt.c:994
HashMemoryChunk chunks
Definition: hashjoin.h:352
MemoryContext batchCxt
Definition: hashjoin.h:349
struct HashMemoryChunkData * unshared
Definition: hashjoin.h:126
union HashMemoryChunkData::@93 next

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().

◆ ExecChooseHashTableSize()

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

Definition at line 663 of file nodeHash.c.

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

References Assert(), get_hash_memory_limit(), HJTUPLE_OVERHEAD, Max, MAXALIGN, MaxAllocSize, Min, NTUP_PER_BUCKET, pg_nextpower2_32(), pg_nextpower2_size_t, pg_prevpower2_size_t, SizeofMinimalTupleHeader, SKEW_BUCKET_OVERHEAD, and SKEW_HASH_MEM_PERCENT.

Referenced by ExecHashTableCreate(), and initial_cost_hashjoin().

◆ ExecEndHash()

void ExecEndHash ( HashState node)

Definition at line 407 of file nodeHash.c.

408 {
410 
411  /*
412  * free exprcontext
413  */
414  ExecFreeExprContext(&node->ps);
415 
416  /*
417  * shut down the subplan
418  */
419  outerPlan = outerPlanState(node);
421 }
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:557
void ExecFreeExprContext(PlanState *planstate)
Definition: execUtils.c:652
#define outerPlanState(node)
Definition: execnodes.h:1125
#define outerPlan(node)
Definition: plannodes.h:183
PlanState ps
Definition: execnodes.h:2640

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

Referenced by ExecEndNode().

◆ ExecHash()

static TupleTableSlot* ExecHash ( PlanState pstate)
static

Definition at line 92 of file nodeHash.c.

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

References elog(), and ERROR.

Referenced by ExecInitHash().

◆ ExecHashAccumInstrumentation()

void ExecHashAccumInstrumentation ( HashInstrumentation instrument,
HashJoinTable  hashtable 
)

Definition at line 2697 of file nodeHash.c.

2699 {
2700  instrument->nbuckets = Max(instrument->nbuckets,
2701  hashtable->nbuckets);
2702  instrument->nbuckets_original = Max(instrument->nbuckets_original,
2703  hashtable->nbuckets_original);
2704  instrument->nbatch = Max(instrument->nbatch,
2705  hashtable->nbatch);
2706  instrument->nbatch_original = Max(instrument->nbatch_original,
2707  hashtable->nbatch_original);
2708  instrument->space_peak = Max(instrument->space_peak,
2709  hashtable->spacePeak);
2710 }

References Max, HashJoinTableData::nbatch, HashInstrumentation::nbatch, HashJoinTableData::nbatch_original, HashInstrumentation::nbatch_original, HashJoinTableData::nbuckets, HashInstrumentation::nbuckets, HashJoinTableData::nbuckets_original, HashInstrumentation::nbuckets_original, HashInstrumentation::space_peak, and HashJoinTableData::spacePeak.

Referenced by ExecReScanHashJoin(), and ExecShutdownHash().

◆ ExecHashBuildSkewHash()

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

Definition at line 2223 of file nodeHash.c.

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

References ATTSTATSSLOT_NUMBERS, ATTSTATSSLOT_VALUES, HashJoinTableData::batchCxt, BoolGetDatum(), HashJoinTableData::collations, DatumGetUInt32(), free_attstatsslot(), FunctionCall1Coll(), get_attstatsslot(), HashSkewBucket::hashvalue, HeapTupleIsValid, i, if(), Int16GetDatum(), InvalidOid, MemoryContextAlloc(), MemoryContextAllocZero(), HashJoinTableData::nSkewBuckets, AttStatsSlot::numbers, AttStatsSlot::nvalues, ObjectIdGetDatum(), OidIsValid, HashJoinTableData::outer_hashfunctions, pg_nextpower2_32(), 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().

◆ ExecHashEstimate()

void ExecHashEstimate ( HashState node,
ParallelContext pcxt 
)

Definition at line 2581 of file nodeHash.c.

2582 {
2583  size_t size;
2584 
2585  /* don't need this if not instrumenting or no workers */
2586  if (!node->ps.instrument || pcxt->nworkers == 0)
2587  return;
2588 
2589  size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
2590  size = add_size(size, offsetof(SharedHashInfo, hinstrument));
2591  shm_toc_estimate_chunk(&pcxt->estimator, size);
2592  shm_toc_estimate_keys(&pcxt->estimator, 1);
2593 }
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
Size add_size(Size s1, Size s2)
Definition: shmem.c:502
Size mul_size(Size s1, Size s2)
Definition: shmem.c:519
shm_toc_estimator estimator
Definition: parallel.h:42
Instrumentation * instrument
Definition: execnodes.h:1039

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

Referenced by ExecParallelEstimate().

◆ ExecHashGetBucketAndBatch()

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

Definition at line 1908 of file nodeHash.c.

1912 {
1913  uint32 nbuckets = (uint32) hashtable->nbuckets;
1914  uint32 nbatch = (uint32) hashtable->nbatch;
1915 
1916  if (nbatch > 1)
1917  {
1918  *bucketno = hashvalue & (nbuckets - 1);
1919  *batchno = pg_rotate_right32(hashvalue,
1920  hashtable->log2_nbuckets) & (nbatch - 1);
1921  }
1922  else
1923  {
1924  *bucketno = hashvalue & (nbuckets - 1);
1925  *batchno = 0;
1926  }
1927 }
static uint32 pg_rotate_right32(uint32 word, int n)
Definition: pg_bitutils.h:271

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().

◆ ExecHashGetHashValue()

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

Definition at line 1800 of file nodeHash.c.

1806 {
1807  uint32 hashkey = 0;
1808  FmgrInfo *hashfunctions;
1809  ListCell *hk;
1810  int i = 0;
1811  MemoryContext oldContext;
1812 
1813  /*
1814  * We reset the eval context each time to reclaim any memory leaked in the
1815  * hashkey expressions.
1816  */
1817  ResetExprContext(econtext);
1818 
1819  oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
1820 
1821  if (outer_tuple)
1822  hashfunctions = hashtable->outer_hashfunctions;
1823  else
1824  hashfunctions = hashtable->inner_hashfunctions;
1825 
1826  foreach(hk, hashkeys)
1827  {
1828  ExprState *keyexpr = (ExprState *) lfirst(hk);
1829  Datum keyval;
1830  bool isNull;
1831 
1832  /* combine successive hashkeys by rotating */
1833  hashkey = pg_rotate_left32(hashkey, 1);
1834 
1835  /*
1836  * Get the join attribute value of the tuple
1837  */
1838  keyval = ExecEvalExpr(keyexpr, econtext, &isNull);
1839 
1840  /*
1841  * If the attribute is NULL, and the join operator is strict, then
1842  * this tuple cannot pass the join qual so we can reject it
1843  * immediately (unless we're scanning the outside of an outer join, in
1844  * which case we must not reject it). Otherwise we act like the
1845  * hashcode of NULL is zero (this will support operators that act like
1846  * IS NOT DISTINCT, though not any more-random behavior). We treat
1847  * the hash support function as strict even if the operator is not.
1848  *
1849  * Note: currently, all hashjoinable operators must be strict since
1850  * the hash index AM assumes that. However, it takes so little extra
1851  * code here to allow non-strict that we may as well do it.
1852  */
1853  if (isNull)
1854  {
1855  if (hashtable->hashStrict[i] && !keep_nulls)
1856  {
1857  MemoryContextSwitchTo(oldContext);
1858  return false; /* cannot match */
1859  }
1860  /* else, leave hashkey unmodified, equivalent to hashcode 0 */
1861  }
1862  else
1863  {
1864  /* Compute the hash function */
1865  uint32 hkey;
1866 
1867  hkey = DatumGetUInt32(FunctionCall1Coll(&hashfunctions[i], hashtable->collations[i], keyval));
1868  hashkey ^= hkey;
1869  }
1870 
1871  i++;
1872  }
1873 
1874  MemoryContextSwitchTo(oldContext);
1875 
1876  *hashvalue = hashkey;
1877  return true;
1878 }
#define ResetExprContext(econtext)
Definition: executor.h:529
static Datum ExecEvalExpr(ExprState *state, ExprContext *econtext, bool *isNull)
Definition: executor.h:318
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:135
static uint32 pg_rotate_left32(uint32 word, int n)
Definition: pg_bitutils.h:277
#define lfirst(lc)
Definition: pg_list.h:170
uintptr_t Datum
Definition: postgres.h:412
MemoryContext ecxt_per_tuple_memory
Definition: execnodes.h:255
bool * hashStrict
Definition: hashjoin.h:339
FmgrInfo * inner_hashfunctions
Definition: hashjoin.h:338

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

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

◆ ExecHashGetSkewBucket()

int ExecHashGetSkewBucket ( HashJoinTable  hashtable,
uint32  hashvalue 
)

Definition at line 2376 of file nodeHash.c.

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

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

Referenced by ExecHashJoinImpl(), and MultiExecPrivateHash().

◆ ExecHashIncreaseNumBatches()

static void ExecHashIncreaseNumBatches ( HashJoinTable  hashtable)
static

Definition at line 904 of file nodeHash.c.

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

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(), Min, HashJoinTableData::nbatch, HashJoinTableData::nbuckets, HashJoinTableData::nbuckets_optimal, HashJoinTupleData::next, HashMemoryChunkData::next, HashJoinTableData::outerBatchFile, palloc0_array, pfree(), PrepareTempTablespaces(), printf, repalloc0_array, repalloc_array, HashJoinTableData::spaceUsed, MinimalTupleData::t_len, HashJoinTupleData::unshared, HashMemoryChunkData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashSkewTableInsert(), and ExecHashTableInsert().

◆ ExecHashIncreaseNumBuckets()

static void ExecHashIncreaseNumBuckets ( HashJoinTable  hashtable)
static

Definition at line 1440 of file nodeHash.c.

1441 {
1442  HashMemoryChunk chunk;
1443 
1444  /* do nothing if not an increase (it's called increase for a reason) */
1445  if (hashtable->nbuckets >= hashtable->nbuckets_optimal)
1446  return;
1447 
1448 #ifdef HJDEBUG
1449  printf("Hashjoin %p: increasing nbuckets %d => %d\n",
1450  hashtable, hashtable->nbuckets, hashtable->nbuckets_optimal);
1451 #endif
1452 
1453  hashtable->nbuckets = hashtable->nbuckets_optimal;
1454  hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
1455 
1456  Assert(hashtable->nbuckets > 1);
1457  Assert(hashtable->nbuckets <= (INT_MAX / 2));
1458  Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets));
1459 
1460  /*
1461  * Just reallocate the proper number of buckets - we don't need to walk
1462  * through them - we can walk the dense-allocated chunks (just like in
1463  * ExecHashIncreaseNumBatches, but without all the copying into new
1464  * chunks)
1465  */
1466  hashtable->buckets.unshared =
1467  repalloc_array(hashtable->buckets.unshared,
1468  HashJoinTuple, hashtable->nbuckets);
1469 
1470  memset(hashtable->buckets.unshared, 0,
1471  hashtable->nbuckets * sizeof(HashJoinTuple));
1472 
1473  /* scan through all tuples in all chunks to rebuild the hash table */
1474  for (chunk = hashtable->chunks; chunk != NULL; chunk = chunk->next.unshared)
1475  {
1476  /* process all tuples stored in this chunk */
1477  size_t idx = 0;
1478 
1479  while (idx < chunk->used)
1480  {
1481  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1482  int bucketno;
1483  int batchno;
1484 
1485  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1486  &bucketno, &batchno);
1487 
1488  /* add the tuple to the proper bucket */
1489  hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1490  hashtable->buckets.unshared[bucketno] = hashTuple;
1491 
1492  /* advance index past the tuple */
1494  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1495  }
1496 
1497  /* allow this loop to be cancellable */
1499  }
1500 }

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_array, HashJoinTupleData::unshared, HashMemoryChunkData::unshared, and HashJoinTableData::unshared.

Referenced by MultiExecPrivateHash().

◆ ExecHashInitializeDSM()

void ExecHashInitializeDSM ( HashState node,
ParallelContext pcxt 
)

Definition at line 2600 of file nodeHash.c.

2601 {
2602  size_t size;
2603 
2604  /* don't need this if not instrumenting or no workers */
2605  if (!node->ps.instrument || pcxt->nworkers == 0)
2606  return;
2607 
2608  size = offsetof(SharedHashInfo, hinstrument) +
2609  pcxt->nworkers * sizeof(HashInstrumentation);
2610  node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
2611 
2612  /* Each per-worker area must start out as zeroes. */
2613  memset(node->shared_info, 0, size);
2614 
2615  node->shared_info->num_workers = pcxt->nworkers;
2616  shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id,
2617  node->shared_info);
2618 }
struct HashInstrumentation HashInstrumentation
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
SharedHashInfo * shared_info
Definition: execnodes.h:2650
shm_toc * toc
Definition: parallel.h:45
Plan * plan
Definition: execnodes.h:1029
int plan_node_id
Definition: plannodes.h:152

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

Referenced by ExecParallelInitializeDSM().

◆ ExecHashInitializeWorker()

void ExecHashInitializeWorker ( HashState node,
ParallelWorkerContext pwcxt 
)

Definition at line 2625 of file nodeHash.c.

2626 {
2627  SharedHashInfo *shared_info;
2628 
2629  /* don't need this if not instrumenting */
2630  if (!node->ps.instrument)
2631  return;
2632 
2633  /*
2634  * Find our entry in the shared area, and set up a pointer to it so that
2635  * we'll accumulate stats there when shutting down or rebuilding the hash
2636  * table.
2637  */
2638  shared_info = (SharedHashInfo *)
2639  shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
2640  node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
2641 }
int ParallelWorkerNumber
Definition: parallel.c:113
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
HashInstrumentation * hinstrument
Definition: execnodes.h:2657
HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER]
Definition: execnodes.h:2631

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().

◆ ExecHashRemoveNextSkewBucket()

static void ExecHashRemoveNextSkewBucket ( HashJoinTable  hashtable)
static

Definition at line 2468 of file nodeHash.c.

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

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().

◆ ExecHashRetrieveInstrumentation()

void ExecHashRetrieveInstrumentation ( HashState node)

Definition at line 2666 of file nodeHash.c.

2667 {
2668  SharedHashInfo *shared_info = node->shared_info;
2669  size_t size;
2670 
2671  if (shared_info == NULL)
2672  return;
2673 
2674  /* Replace node->shared_info with a copy in backend-local memory. */
2675  size = offsetof(SharedHashInfo, hinstrument) +
2676  shared_info->num_workers * sizeof(HashInstrumentation);
2677  node->shared_info = palloc(size);
2678  memcpy(node->shared_info, shared_info, size);
2679 }
void * palloc(Size size)
Definition: mcxt.c:1199

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

Referenced by ExecParallelRetrieveInstrumentation().

◆ ExecHashSkewTableInsert()

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

Definition at line 2422 of file nodeHash.c.

2426 {
2427  bool shouldFree;
2428  MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
2429  HashJoinTuple hashTuple;
2430  int hashTupleSize;
2431 
2432  /* Create the HashJoinTuple */
2433  hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
2434  hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
2435  hashTupleSize);
2436  hashTuple->hashvalue = hashvalue;
2437  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
2439 
2440  /* Push it onto the front of the skew bucket's list */
2441  hashTuple->next.unshared = hashtable->skewBucket[bucketNumber]->tuples;
2442  hashtable->skewBucket[bucketNumber]->tuples = hashTuple;
2443  Assert(hashTuple != hashTuple->next.unshared);
2444 
2445  /* Account for space used, and back off if we've used too much */
2446  hashtable->spaceUsed += hashTupleSize;
2447  hashtable->spaceUsedSkew += hashTupleSize;
2448  if (hashtable->spaceUsed > hashtable->spacePeak)
2449  hashtable->spacePeak = hashtable->spaceUsed;
2450  while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
2451  ExecHashRemoveNextSkewBucket(hashtable);
2452 
2453  /* Check we are not over the total spaceAllowed, either */
2454  if (hashtable->spaceUsed > hashtable->spaceAllowed)
2455  ExecHashIncreaseNumBatches(hashtable);
2456 
2457  if (shouldFree)
2458  heap_free_minimal_tuple(tuple);
2459 }
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot, bool *shouldFree)
Definition: execTuples.c:1692
void heap_free_minimal_tuple(MinimalTuple mtup)
Definition: heaptuple.c:1427
#define HeapTupleHeaderClearMatch(tup)
Definition: htup_details.h:520
static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
Definition: nodeHash.c:2468
static void ExecHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:904
Size spaceAllowedSkew
Definition: hashjoin.h:346

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().

◆ ExecHashTableCreate()

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

Definition at line 431 of file nodeHash.c.

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

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, 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, HashJoinTableData::nSkewBuckets, OidIsValid, op_strict(), HashJoinTableData::outer_hashfunctions, HashJoinTableData::outerBatchFile, outerPlan, palloc0_array, palloc_array, palloc_object, Plan::parallel_aware, HashJoinTableData::parallel_state, HashJoinTableData::partialTuples, PHJ_BUILD_ELECTING, PHJ_GROWTH_OK, Hash::plan, Plan::plan_rows, Plan::plan_width, PrepareTempTablespaces(), printf, Hash::rows_total, SKEW_HASH_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, HashJoinTableData::totalTuples, HashJoinTableData::unshared, and WAIT_EVENT_HASH_BUILD_ELECT.

Referenced by ExecHashJoinImpl().

◆ ExecHashTableDestroy()

void ExecHashTableDestroy ( HashJoinTable  hashtable)

Definition at line 871 of file nodeHash.c.

872 {
873  int i;
874 
875  /*
876  * Make sure all the temp files are closed. We skip batch 0, since it
877  * can't have any temp files (and the arrays might not even exist if
878  * nbatch is only 1). Parallel hash joins don't use these files.
879  */
880  if (hashtable->innerBatchFile != NULL)
881  {
882  for (i = 1; i < hashtable->nbatch; i++)
883  {
884  if (hashtable->innerBatchFile[i])
885  BufFileClose(hashtable->innerBatchFile[i]);
886  if (hashtable->outerBatchFile[i])
887  BufFileClose(hashtable->outerBatchFile[i]);
888  }
889  }
890 
891  /* Release working memory (batchCxt is a child, so it goes away too) */
892  MemoryContextDelete(hashtable->hashCxt);
893 
894  /* And drop the control block */
895  pfree(hashtable);
896 }
void BufFileClose(BufFile *file)
Definition: buffile.c:407
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:376

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

Referenced by ExecEndHashJoin(), and ExecReScanHashJoin().

◆ ExecHashTableDetach()

void ExecHashTableDetach ( HashJoinTable  hashtable)

Definition at line 3182 of file nodeHash.c.

3183 {
3184  if (hashtable->parallel_state)
3185  {
3186  ParallelHashJoinState *pstate = hashtable->parallel_state;
3187  int i;
3188 
3189  /* Make sure any temporary files are closed. */
3190  if (hashtable->batches)
3191  {
3192  for (i = 0; i < hashtable->nbatch; ++i)
3193  {
3194  sts_end_write(hashtable->batches[i].inner_tuples);
3195  sts_end_write(hashtable->batches[i].outer_tuples);
3198  }
3199  }
3200 
3201  /* If we're last to detach, clean up shared memory. */
3202  if (BarrierDetach(&pstate->build_barrier))
3203  {
3204  if (DsaPointerIsValid(pstate->batches))
3205  {
3206  dsa_free(hashtable->area, pstate->batches);
3207  pstate->batches = InvalidDsaPointer;
3208  }
3209  }
3210 
3211  hashtable->parallel_state = NULL;
3212  }
3213 }
bool BarrierDetach(Barrier *barrier)
Definition: barrier.c:256
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:832
#define InvalidDsaPointer
Definition: dsa.h:78
#define DsaPointerIsValid(x)
Definition: dsa.h:81
void sts_end_write(SharedTuplestoreAccessor *accessor)
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:209
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:208
dsa_pointer batches
Definition: hashjoin.h:236

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().

◆ ExecHashTableDetachBatch()

void ExecHashTableDetachBatch ( HashJoinTable  hashtable)

Definition at line 3125 of file nodeHash.c.

3126 {
3127  if (hashtable->parallel_state != NULL &&
3128  hashtable->curbatch >= 0)
3129  {
3130  int curbatch = hashtable->curbatch;
3131  ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
3132 
3133  /* Make sure any temporary files are closed. */
3134  sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
3135  sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
3136 
3137  /* Detach from the batch we were last working on. */
3139  {
3140  /*
3141  * Technically we shouldn't access the barrier because we're no
3142  * longer attached, but since there is no way it's moving after
3143  * this point it seems safe to make the following assertion.
3144  */
3146 
3147  /* Free shared chunks and buckets. */
3148  while (DsaPointerIsValid(batch->chunks))
3149  {
3150  HashMemoryChunk chunk =
3151  dsa_get_address(hashtable->area, batch->chunks);
3152  dsa_pointer next = chunk->next.shared;
3153 
3154  dsa_free(hashtable->area, batch->chunks);
3155  batch->chunks = next;
3156  }
3157  if (DsaPointerIsValid(batch->buckets))
3158  {
3159  dsa_free(hashtable->area, batch->buckets);
3160  batch->buckets = InvalidDsaPointer;
3161  }
3162  }
3163 
3164  /*
3165  * Track the largest batch we've been attached to. Though each
3166  * backend might see a different subset of batches, explain.c will
3167  * scan the results from all backends to find the largest value.
3168  */
3169  hashtable->spacePeak =
3170  Max(hashtable->spacePeak,
3171  batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
3172 
3173  /* Remember that we are not attached to a batch. */
3174  hashtable->curbatch = -1;
3175  }
3176 }
bool BarrierArriveAndDetach(Barrier *barrier)
Definition: barrier.c:203
static int32 next
Definition: blutils.c:219
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:944
uint64 dsa_pointer
Definition: dsa.h:62
#define PHJ_BATCH_DONE
Definition: hashjoin.h:268
dsa_pointer shared
Definition: hashjoin.h:127
ParallelHashJoinBatch * shared
Definition: hashjoin.h:197
dsa_pointer chunks
Definition: hashjoin.h:156
dsa_pointer buckets
Definition: hashjoin.h:153

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, next, HashMemoryChunkData::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().

◆ ExecHashTableInsert()

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

Definition at line 1602 of file nodeHash.c.

1605 {
1606  bool shouldFree;
1607  MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1608  int bucketno;
1609  int batchno;
1610 
1611  ExecHashGetBucketAndBatch(hashtable, hashvalue,
1612  &bucketno, &batchno);
1613 
1614  /*
1615  * decide whether to put the tuple in the hash table or a temp file
1616  */
1617  if (batchno == hashtable->curbatch)
1618  {
1619  /*
1620  * put the tuple in hash table
1621  */
1622  HashJoinTuple hashTuple;
1623  int hashTupleSize;
1624  double ntuples = (hashtable->totalTuples - hashtable->skewTuples);
1625 
1626  /* Create the HashJoinTuple */
1627  hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
1628  hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
1629 
1630  hashTuple->hashvalue = hashvalue;
1631  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1632 
1633  /*
1634  * We always reset the tuple-matched flag on insertion. This is okay
1635  * even when reloading a tuple from a batch file, since the tuple
1636  * could not possibly have been matched to an outer tuple before it
1637  * went into the batch file.
1638  */
1640 
1641  /* Push it onto the front of the bucket's list */
1642  hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1643  hashtable->buckets.unshared[bucketno] = hashTuple;
1644 
1645  /*
1646  * Increase the (optimal) number of buckets if we just exceeded the
1647  * NTUP_PER_BUCKET threshold, but only when there's still a single
1648  * batch.
1649  */
1650  if (hashtable->nbatch == 1 &&
1651  ntuples > (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
1652  {
1653  /* Guard against integer overflow and alloc size overflow */
1654  if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
1655  hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
1656  {
1657  hashtable->nbuckets_optimal *= 2;
1658  hashtable->log2_nbuckets_optimal += 1;
1659  }
1660  }
1661 
1662  /* Account for space used, and back off if we've used too much */
1663  hashtable->spaceUsed += hashTupleSize;
1664  if (hashtable->spaceUsed > hashtable->spacePeak)
1665  hashtable->spacePeak = hashtable->spaceUsed;
1666  if (hashtable->spaceUsed +
1667  hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
1668  > hashtable->spaceAllowed)
1669  ExecHashIncreaseNumBatches(hashtable);
1670  }
1671  else
1672  {
1673  /*
1674  * put the tuple into a temp file for later batches
1675  */
1676  Assert(batchno > hashtable->curbatch);
1677  ExecHashJoinSaveTuple(tuple,
1678  hashvalue,
1679  &hashtable->innerBatchFile[batchno]);
1680  }
1681 
1682  if (shouldFree)
1683  heap_free_minimal_tuple(tuple);
1684 }

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().

◆ ExecHashTableReset()

void ExecHashTableReset ( HashJoinTable  hashtable)

Definition at line 2147 of file nodeHash.c.

2148 {
2149  MemoryContext oldcxt;
2150  int nbuckets = hashtable->nbuckets;
2151 
2152  /*
2153  * Release all the hash buckets and tuples acquired in the prior pass, and
2154  * reinitialize the context for a new pass.
2155  */
2156  MemoryContextReset(hashtable->batchCxt);
2157  oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
2158 
2159  /* Reallocate and reinitialize the hash bucket headers. */
2160  hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets);
2161 
2162  hashtable->spaceUsed = 0;
2163 
2164  MemoryContextSwitchTo(oldcxt);
2165 
2166  /* Forget the chunks (the memory was freed by the context reset above). */
2167  hashtable->chunks = NULL;
2168 }
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:303

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

Referenced by ExecHashJoinNewBatch().

◆ ExecHashTableResetMatchFlags()

void ExecHashTableResetMatchFlags ( HashJoinTable  hashtable)

Definition at line 2175 of file nodeHash.c.

2176 {
2177  HashJoinTuple tuple;
2178  int i;
2179 
2180  /* Reset all flags in the main table ... */
2181  for (i = 0; i < hashtable->nbuckets; i++)
2182  {
2183  for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
2184  tuple = tuple->next.unshared)
2186  }
2187 
2188  /* ... and the same for the skew buckets, if any */
2189  for (i = 0; i < hashtable->nSkewBuckets; i++)
2190  {
2191  int j = hashtable->skewBucketNums[i];
2192  HashSkewBucket *skewBucket = hashtable->skewBucket[j];
2193 
2194  for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
2196  }
2197 }
int j
Definition: isn.c:74

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

Referenced by ExecReScanHashJoin().

◆ ExecInitHash()

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

Definition at line 354 of file nodeHash.c.

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

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

Referenced by ExecInitNode().

◆ ExecParallelHashCloseBatchAccessors()

static void ExecParallelHashCloseBatchAccessors ( HashJoinTable  hashtable)
static

Definition at line 3021 of file nodeHash.c.

3022 {
3023  int i;
3024 
3025  for (i = 0; i < hashtable->nbatch; ++i)
3026  {
3027  /* Make sure no files are left open. */
3028  sts_end_write(hashtable->batches[i].inner_tuples);
3029  sts_end_write(hashtable->batches[i].outer_tuples);
3032  }
3033  pfree(hashtable->batches);
3034  hashtable->batches = NULL;
3035 }

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().

◆ ExecParallelHashEnsureBatchAccessors()

static void ExecParallelHashEnsureBatchAccessors ( HashJoinTable  hashtable)
static

Definition at line 3042 of file nodeHash.c.

3043 {
3044  ParallelHashJoinState *pstate = hashtable->parallel_state;
3045  ParallelHashJoinBatch *batches;
3046  MemoryContext oldcxt;
3047  int i;
3048 
3049  if (hashtable->batches != NULL)
3050  {
3051  if (hashtable->nbatch == pstate->nbatch)
3052  return;
3054  }
3055 
3056  /*
3057  * It's possible for a backend to start up very late so that the whole
3058  * join is finished and the shm state for tracking batches has already
3059  * been freed by ExecHashTableDetach(). In that case we'll just leave
3060  * hashtable->batches as NULL so that ExecParallelHashJoinNewBatch() gives
3061  * up early.
3062  */
3063  if (!DsaPointerIsValid(pstate->batches))
3064  return;
3065 
3066  /* Use hash join memory context. */
3067  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
3068 
3069  /* Allocate this backend's accessor array. */
3070  hashtable->nbatch = pstate->nbatch;
3071  hashtable->batches =
3073 
3074  /* Find the base of the pseudo-array of ParallelHashJoinBatch objects. */
3075  batches = (ParallelHashJoinBatch *)
3076  dsa_get_address(hashtable->area, pstate->batches);
3077 
3078  /* Set up the accessor array and attach to the tuplestores. */
3079  for (i = 0; i < hashtable->nbatch; ++i)
3080  {
3081  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
3082  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
3083 
3084  accessor->shared = shared;
3085  accessor->preallocated = 0;
3086  accessor->done = false;
3087  accessor->inner_tuples =
3090  &pstate->fileset);
3091  accessor->outer_tuples =
3093  pstate->nparticipants),
3095  &pstate->fileset);
3096  }
3097 
3098  MemoryContextSwitchTo(oldcxt);
3099 }
#define ParallelHashJoinBatchInner(batch)
Definition: hashjoin.h:170
#define NthParallelHashJoinBatch(base, n)
Definition: hashjoin.h:186
#define ParallelHashJoinBatchOuter(batch, nparticipants)
Definition: hashjoin.h:175
static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:3021
SharedTuplestoreAccessor * sts_attach(SharedTuplestore *sts, int my_participant_number, SharedFileSet *fileset)
SharedFileSet fileset
Definition: hashjoin.h:253

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_array, HashJoinTableData::parallel_state, ParallelHashJoinBatchInner, ParallelHashJoinBatchOuter, ParallelWorkerNumber, ParallelHashJoinBatchAccessor::preallocated, ParallelHashJoinBatchAccessor::shared, and sts_attach().

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

◆ ExecParallelHashFirstTuple()

static HashJoinTuple ExecParallelHashFirstTuple ( HashJoinTable  hashtable,
int  bucketno 
)
inlinestatic

Definition at line 3219 of file nodeHash.c.

3220 {
3221  HashJoinTuple tuple;
3222  dsa_pointer p;
3223 
3224  Assert(hashtable->parallel_state);
3225  p = dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
3226  tuple = (HashJoinTuple) dsa_get_address(hashtable->area, p);
3227 
3228  return tuple;
3229 }
#define dsa_pointer_atomic_read
Definition: dsa.h:65
dsa_pointer_atomic * shared
Definition: hashjoin.h:299

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

Referenced by ExecParallelScanHashBucket().

◆ ExecParallelHashIncreaseNumBatches()

static void ExecParallelHashIncreaseNumBatches ( HashJoinTable  hashtable)
static

Definition at line 1067 of file nodeHash.c.

1068 {
1069  ParallelHashJoinState *pstate = hashtable->parallel_state;
1070 
1072 
1073  /*
1074  * It's unlikely, but we need to be prepared for new participants to show
1075  * up while we're in the middle of this operation so we need to switch on
1076  * barrier phase here.
1077  */
1079  {
1081 
1082  /*
1083  * Elect one participant to prepare to grow the number of batches.
1084  * This involves reallocating or resetting the buckets of batch 0
1085  * in preparation for all participants to begin repartitioning the
1086  * tuples.
1087  */
1090  {
1091  dsa_pointer_atomic *buckets;
1092  ParallelHashJoinBatch *old_batch0;
1093  int new_nbatch;
1094  int i;
1095 
1096  /* Move the old batch out of the way. */
1097  old_batch0 = hashtable->batches[0].shared;
1098  pstate->old_batches = pstate->batches;
1099  pstate->old_nbatch = hashtable->nbatch;
1100  pstate->batches = InvalidDsaPointer;
1101 
1102  /* Free this backend's old accessors. */
1104 
1105  /* Figure out how many batches to use. */
1106  if (hashtable->nbatch == 1)
1107  {
1108  /*
1109  * We are going from single-batch to multi-batch. We need
1110  * to switch from one large combined memory budget to the
1111  * regular hash_mem budget.
1112  */
1114 
1115  /*
1116  * The combined hash_mem of all participants wasn't
1117  * enough. Therefore one batch per participant would be
1118  * approximately equivalent and would probably also be
1119  * insufficient. So try two batches per participant,
1120  * rounded up to a power of two.
1121  */
1122  new_nbatch = pg_nextpower2_32(pstate->nparticipants * 2);
1123  }
1124  else
1125  {
1126  /*
1127  * We were already multi-batched. Try doubling the number
1128  * of batches.
1129  */
1130  new_nbatch = hashtable->nbatch * 2;
1131  }
1132 
1133  /* Allocate new larger generation of batches. */
1134  Assert(hashtable->nbatch == pstate->nbatch);
1135  ExecParallelHashJoinSetUpBatches(hashtable, new_nbatch);
1136  Assert(hashtable->nbatch == pstate->nbatch);
1137 
1138  /* Replace or recycle batch 0's bucket array. */
1139  if (pstate->old_nbatch == 1)
1140  {
1141  double dtuples;
1142  double dbuckets;
1143  int new_nbuckets;
1144 
1145  /*
1146  * We probably also need a smaller bucket array. How many
1147  * tuples do we expect per batch, assuming we have only
1148  * half of them so far? Normally we don't need to change
1149  * the bucket array's size, because the size of each batch
1150  * stays the same as we add more batches, but in this
1151  * special case we move from a large batch to many smaller
1152  * batches and it would be wasteful to keep the large
1153  * array.
1154  */
1155  dtuples = (old_batch0->ntuples * 2.0) / new_nbatch;
1156  dbuckets = ceil(dtuples / NTUP_PER_BUCKET);
1157  dbuckets = Min(dbuckets,
1158  MaxAllocSize / sizeof(dsa_pointer_atomic));
1159  new_nbuckets = (int) dbuckets;
1160  new_nbuckets = Max(new_nbuckets, 1024);
1161  new_nbuckets = pg_nextpower2_32(new_nbuckets);
1162  dsa_free(hashtable->area, old_batch0->buckets);
1163  hashtable->batches[0].shared->buckets =
1164  dsa_allocate(hashtable->area,
1165  sizeof(dsa_pointer_atomic) * new_nbuckets);
1166  buckets = (dsa_pointer_atomic *)
1167  dsa_get_address(hashtable->area,
1168  hashtable->batches[0].shared->buckets);
1169  for (i = 0; i < new_nbuckets; ++i)
1171  pstate->nbuckets = new_nbuckets;
1172  }
1173  else
1174  {
1175  /* Recycle the existing bucket array. */
1176  hashtable->batches[0].shared->buckets = old_batch0->buckets;
1177  buckets = (dsa_pointer_atomic *)
1178  dsa_get_address(hashtable->area, old_batch0->buckets);
1179  for (i = 0; i < hashtable->nbuckets; ++i)
1181  }
1182 
1183  /* Move all chunks to the work queue for parallel processing. */
1184  pstate->chunk_work_queue = old_batch0->chunks;
1185 
1186  /* Disable further growth temporarily while we're growing. */
1187  pstate->growth = PHJ_GROWTH_DISABLED;
1188  }
1189  else
1190  {
1191  /* All other participants just flush their tuples to disk. */
1193  }
1194  /* Fall through. */
1195 
1197  /* Wait for the above to be finished. */
1200  /* Fall through. */
1201 
1203  /* Make sure that we have the current dimensions and buckets. */
1206  /* Then partition, flush counters. */
1209  ExecParallelHashMergeCounters(hashtable);
1210  /* Wait for the above to be finished. */
1213  /* Fall through. */
1214 
1216 
1217  /*
1218  * Elect one participant to clean up and decide whether further
1219  * repartitioning is needed, or should be disabled because it's
1220  * not helping.
1221  */
1224  {
1225  bool space_exhausted = false;
1226  bool extreme_skew_detected = false;
1227 
1228  /* Make sure that we have the current dimensions and buckets. */
1231 
1232  /* Are any of the new generation of batches exhausted? */
1233  for (int i = 0; i < hashtable->nbatch; ++i)
1234  {
1235  ParallelHashJoinBatch *batch = hashtable->batches[i].shared;
1236 
1237  if (batch->space_exhausted ||
1238  batch->estimated_size > pstate->space_allowed)
1239  {
1240  int parent;
1241 
1242  space_exhausted = true;
1243 
1244  /*
1245  * Did this batch receive ALL of the tuples from its
1246  * parent batch? That would indicate that further
1247  * repartitioning isn't going to help (the hash values
1248  * are probably all the same).
1249  */
1250  parent = i % pstate->old_nbatch;
1251  if (batch->ntuples == hashtable->batches[parent].shared->old_ntuples)
1252  extreme_skew_detected = true;
1253  }
1254  }
1255 
1256  /* Don't keep growing if it's not helping or we'd overflow. */
1257  if (extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2)
1258  pstate->growth = PHJ_GROWTH_DISABLED;
1259  else if (space_exhausted)
1261  else
1262  pstate->growth = PHJ_GROWTH_OK;
1263 
1264  /* Free the old batches in shared memory. */
1265  dsa_free(hashtable->area, pstate->old_batches);
1266  pstate->old_batches = InvalidDsaPointer;
1267  }
1268  /* Fall through. */
1269 
1271  /* Wait for the above to complete. */
1274  }
1275 }
#define dsa_pointer_atomic_init
Definition: dsa.h:64
#define dsa_allocate(area, size)
Definition: dsa.h:84
#define dsa_pointer_atomic_write
Definition: dsa.h:66
#define PHJ_GROW_BATCHES_FINISHING
Definition: hashjoin.h:275
#define PHJ_BUILD_HASHING_INNER
Definition: hashjoin.h:259
#define PHJ_GROW_BATCHES_ALLOCATING
Definition: hashjoin.h:272
#define PHJ_GROW_BATCHES_DECIDING
Definition: hashjoin.h:274
#define PHJ_GROW_BATCHES_ELECTING
Definition: hashjoin.h:271
#define PHJ_GROW_BATCHES_REPARTITIONING
Definition: hashjoin.h:273
#define PHJ_GROW_BATCHES_PHASE(n)
Definition: hashjoin.h:276
@ PHJ_GROWTH_NEED_MORE_BATCHES
Definition: hashjoin.h:225
@ PHJ_GROWTH_DISABLED
Definition: hashjoin.h:227
void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno)
Definition: nodeHash.c:3267
static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
Definition: nodeHash.c:3042
static void ExecParallelHashRepartitionRest(HashJoinTable hashtable)
Definition: nodeHash.c:1350
static void ExecParallelHashMergeCounters(HashJoinTable hashtable)
Definition: nodeHash.c:1410
static void ExecParallelHashRepartitionFirst(HashJoinTable hashtable)
Definition: nodeHash.c:1283
Barrier grow_batches_barrier
Definition: hashjoin.h:249
dsa_pointer old_batches
Definition: hashjoin.h:237
dsa_pointer chunk_work_queue
Definition: hashjoin.h:242
@ WAIT_EVENT_HASH_GROW_BATCHES_DECIDE
Definition: wait_event.h:101
@ WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATE
Definition: wait_event.h:100
@ WAIT_EVENT_HASH_GROW_BATCHES_FINISH
Definition: wait_event.h:103
@ WAIT_EVENT_HASH_GROW_BATCHES_REPARTITION
Definition: wait_event.h:104
@ WAIT_EVENT_HASH_GROW_BATCHES_ELECT
Definition: wait_event.h:102

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(), get_hash_memory_limit(), ParallelHashJoinState::grow_batches_barrier, ParallelHashJoinState::growth, i, InvalidDsaPointer, Max, MaxAllocSize, Min, 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, pg_nextpower2_32(), 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_ALLOCATE, WAIT_EVENT_HASH_GROW_BATCHES_DECIDE, WAIT_EVENT_HASH_GROW_BATCHES_ELECT, WAIT_EVENT_HASH_GROW_BATCHES_FINISH, and WAIT_EVENT_HASH_GROW_BATCHES_REPARTITION.

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

◆ ExecParallelHashIncreaseNumBuckets()

static void ExecParallelHashIncreaseNumBuckets ( HashJoinTable  hashtable)
static

Definition at line 1503 of file nodeHash.c.

1504 {
1505  ParallelHashJoinState *pstate = hashtable->parallel_state;
1506  int i;
1507  HashMemoryChunk chunk;
1508  dsa_pointer chunk_s;
1509 
1511 
1512  /*
1513  * It's unlikely, but we need to be prepared for new participants to show
1514  * up while we're in the middle of this operation so we need to switch on
1515  * barrier phase here.
1516  */
1518  {
1520  /* Elect one participant to prepare to increase nbuckets. */
1523  {
1524  size_t size;
1525  dsa_pointer_atomic *buckets;
1526 
1527  /* Double the size of the bucket array. */
1528  pstate->nbuckets *= 2;
1529  size = pstate->nbuckets * sizeof(dsa_pointer_atomic);
1530  hashtable->batches[0].shared->size += size / 2;
1531  dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
1532  hashtable->batches[0].shared->buckets =
1533  dsa_allocate(hashtable->area, size);
1534  buckets = (dsa_pointer_atomic *)
1535  dsa_get_address(hashtable->area,
1536  hashtable->batches[0].shared->buckets);
1537  for (i = 0; i < pstate->nbuckets; ++i)
1539 
1540  /* Put the chunk list onto the work queue. */
1541  pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;
1542 
1543  /* Clear the flag. */
1544  pstate->growth = PHJ_GROWTH_OK;
1545  }
1546  /* Fall through. */
1547 
1549  /* Wait for the above to complete. */
1552  /* Fall through. */
1553 
1555  /* Reinsert all tuples into the hash table. */
1558  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_s)))
1559  {
1560  size_t idx = 0;
1561 
1562  while (idx < chunk->used)
1563  {
1564  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1565  dsa_pointer shared = chunk_s + HASH_CHUNK_HEADER_SIZE + idx;
1566  int bucketno;
1567  int batchno;
1568 
1569  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1570  &bucketno, &batchno);
1571  Assert(batchno == 0);
1572 
1573  /* add the tuple to the proper bucket */
1574  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1575  hashTuple, shared);
1576 
1577  /* advance index past the tuple */
1579  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1580  }
1581 
1582  /* allow this loop to be cancellable */
1584  }
1587  }
1588 }
pg_atomic_uint64 dsa_pointer_atomic
Definition: dsa.h:63
#define PHJ_GROW_BUCKETS_REINSERTING
Definition: hashjoin.h:281
#define PHJ_GROW_BUCKETS_PHASE(n)
Definition: hashjoin.h:282
#define PHJ_GROW_BUCKETS_ELECTING
Definition: hashjoin.h:279
#define PHJ_GROW_BUCKETS_ALLOCATING
Definition: hashjoin.h:280
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3249
static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable hashtable, dsa_pointer *shared)
Definition: nodeHash.c:3288
Barrier grow_buckets_barrier
Definition: hashjoin.h:250
@ WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE
Definition: wait_event.h:105
@ WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT
Definition: wait_event.h:107
@ WAIT_EVENT_HASH_GROW_BUCKETS_ELECT
Definition: wait_event.h:106

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_ALLOCATE, WAIT_EVENT_HASH_GROW_BUCKETS_ELECT, and WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT.

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

◆ ExecParallelHashJoinSetUpBatches()

static void ExecParallelHashJoinSetUpBatches ( HashJoinTable  hashtable,
int  nbatch 
)
static

Definition at line 2944 of file nodeHash.c.

2945 {
2946  ParallelHashJoinState *pstate = hashtable->parallel_state;
2947  ParallelHashJoinBatch *batches;
2948  MemoryContext oldcxt;
2949  int i;
2950 
2951  Assert(hashtable->batches == NULL);
2952 
2953  /* Allocate space. */
2954  pstate->batches =
2955  dsa_allocate0(hashtable->area,
2956  EstimateParallelHashJoinBatch(hashtable) * nbatch);
2957  pstate->nbatch = nbatch;
2958  batches = dsa_get_address(hashtable->area, pstate->batches);
2959 
2960  /* Use hash join memory context. */
2961  oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
2962 
2963  /* Allocate this backend's accessor array. */
2964  hashtable->nbatch = nbatch;
2965  hashtable->batches =
2967 
2968  /* Set up the shared state, tuplestores and backend-local accessors. */
2969  for (i = 0; i < hashtable->nbatch; ++i)
2970  {
2971  ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
2972  ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
2973  char name[MAXPGPATH];
2974 
2975  /*
2976  * All members of shared were zero-initialized. We just need to set
2977  * up the Barrier.
2978  */
2979  BarrierInit(&shared->batch_barrier, 0);
2980  if (i == 0)
2981  {
2982  /* Batch 0 doesn't need to be loaded. */
2983  BarrierAttach(&shared->batch_barrier);
2984  while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING)
2985  BarrierArriveAndWait(&shared->batch_barrier, 0);
2986  BarrierDetach(&shared->batch_barrier);
2987  }
2988 
2989  /* Initialize accessor state. All members were zero-initialized. */
2990  accessor->shared = shared;
2991 
2992  /* Initialize the shared tuplestores. */
2993  snprintf(name, sizeof(name), "i%dof%d", i, hashtable->nbatch);
2994  accessor->inner_tuples =
2996  pstate->nparticipants,
2998  sizeof(uint32),
3000  &pstate->fileset,
3001  name);
3002  snprintf(name, sizeof(name), "o%dof%d", i, hashtable->nbatch);
3003  accessor->outer_tuples =
3005  pstate->nparticipants),
3006  pstate->nparticipants,
3008  sizeof(uint32),
3010  &pstate->fileset,
3011  name);
3012  }
3013 
3014  MemoryContextSwitchTo(oldcxt);
3015 }
void BarrierInit(Barrier *barrier, int participants)
Definition: barrier.c:100
#define dsa_allocate0(area, size)
Definition: dsa.h:88
const char * name
Definition: encode.c:561
#define PHJ_BATCH_PROBING
Definition: hashjoin.h:267
#define EstimateParallelHashJoinBatch(hashtable)
Definition: hashjoin.h:181
#define MAXPGPATH
#define snprintf
Definition: port.h:238
SharedTuplestoreAccessor * sts_initialize(SharedTuplestore *sts, int participants, int my_participant_number, size_t meta_data_size, int flags, SharedFileSet *fileset, const char *name)
#define SHARED_TUPLESTORE_SINGLE_PASS

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_array, HashJoinTableData::parallel_state, ParallelHashJoinBatchInner, ParallelHashJoinBatchOuter, ParallelWorkerNumber, PHJ_BATCH_PROBING, ParallelHashJoinBatchAccessor::shared, SHARED_TUPLESTORE_SINGLE_PASS, snprintf, and sts_initialize().

Referenced by ExecHashTableCreate(), and ExecParallelHashIncreaseNumBatches().

◆ ExecParallelHashMergeCounters()

static void ExecParallelHashMergeCounters ( HashJoinTable  hashtable)
static

Definition at line 1410 of file nodeHash.c.

1411 {
1412  ParallelHashJoinState *pstate = hashtable->parallel_state;
1413  int i;
1414 
1415  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
1416  pstate->total_tuples = 0;
1417  for (i = 0; i < hashtable->nbatch; ++i)
1418  {
1419  ParallelHashJoinBatchAccessor *batch = &hashtable->batches[i];
1420 
1421  batch->shared->size += batch->size;
1422  batch->shared->estimated_size += batch->estimated_size;
1423  batch->shared->ntuples += batch->ntuples;
1424  batch->shared->old_ntuples += batch->old_ntuples;
1425  batch->size = 0;
1426  batch->estimated_size = 0;
1427  batch->ntuples = 0;
1428  batch->old_ntuples = 0;
1429  pstate->total_tuples += batch->shared->ntuples;
1430  }
1431  LWLockRelease(&pstate->lock);
1432 }
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1194
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1802
@ LW_EXCLUSIVE
Definition: lwlock.h:112

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().

◆ ExecParallelHashNextTuple()

static HashJoinTuple ExecParallelHashNextTuple ( HashJoinTable  hashtable,
HashJoinTuple  tuple 
)
inlinestatic

Definition at line 3235 of file nodeHash.c.

3236 {
3238 
3239  Assert(hashtable->parallel_state);
3240  next = (HashJoinTuple) dsa_get_address(hashtable->area, tuple->next.shared);
3241 
3242  return next;
3243 }
dsa_pointer shared
Definition: hashjoin.h:73

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

Referenced by ExecParallelScanHashBucket().

◆ ExecParallelHashPopChunkQueue()

static HashMemoryChunk ExecParallelHashPopChunkQueue ( HashJoinTable  hashtable,
dsa_pointer shared 
)
static

Definition at line 3288 of file nodeHash.c.

3289 {
3290  ParallelHashJoinState *pstate = hashtable->parallel_state;
3291  HashMemoryChunk chunk;
3292 
3293  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3294  if (DsaPointerIsValid(pstate->chunk_work_queue))
3295  {
3296  *shared = pstate->chunk_work_queue;
3297  chunk = (HashMemoryChunk)
3298  dsa_get_address(hashtable->area, *shared);
3299  pstate->chunk_work_queue = chunk->next.shared;
3300  }
3301  else
3302  chunk = NULL;
3303  LWLockRelease(&pstate->lock);
3304 
3305  return chunk;
3306 }

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().

◆ ExecParallelHashPushTuple()

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

Definition at line 3249 of file nodeHash.c.

3252 {
3253  for (;;)
3254  {
3255  tuple->next.shared = dsa_pointer_atomic_read(head);
3257  &tuple->next.shared,
3258  tuple_shared))
3259  break;
3260  }
3261 }
#define dsa_pointer_atomic_compare_exchange
Definition: dsa.h:68

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

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

◆ ExecParallelHashRepartitionFirst()

static void ExecParallelHashRepartitionFirst ( HashJoinTable  hashtable)
static

Definition at line 1283 of file nodeHash.c.

1284 {
1285  dsa_pointer chunk_shared;
1286  HashMemoryChunk chunk;
1287 
1288  Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);
1289 
1290  while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared)))
1291  {
1292  size_t idx = 0;
1293 
1294  /* Repartition all tuples in this chunk. */
1295  while (idx < chunk->used)
1296  {
1297  HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
1298  MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
1299  HashJoinTuple copyTuple;
1300  dsa_pointer shared;
1301  int bucketno;
1302  int batchno;
1303 
1304  ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
1305  &bucketno, &batchno);
1306 
1307  Assert(batchno < hashtable->nbatch);
1308  if (batchno == 0)
1309  {
1310  /* It still belongs in batch 0. Copy to a new chunk. */
1311  copyTuple =
1312  ExecParallelHashTupleAlloc(hashtable,
1313  HJTUPLE_OVERHEAD + tuple->t_len,
1314  &shared);
1315  copyTuple->hashvalue = hashTuple->hashvalue;
1316  memcpy(HJTUPLE_MINTUPLE(copyTuple), tuple, tuple->t_len);
1317  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1318  copyTuple, shared);
1319  }
1320  else
1321  {
1322  size_t tuple_size =
1323  MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1324 
1325  /* It belongs in a later batch. */
1326  hashtable->batches[batchno].estimated_size += tuple_size;
1327  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1328  &hashTuple->hashvalue, tuple);
1329  }
1330 
1331  /* Count this tuple. */
1332  ++hashtable->batches[0].old_ntuples;
1333  ++hashtable->batches[batchno].ntuples;
1334 
1336  HJTUPLE_MINTUPLE(hashTuple)->t_len);
1337  }
1338 
1339  /* Free this chunk. */
1340  dsa_free(hashtable->area, chunk_shared);
1341 
1343  }
1344 }
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2796
void sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, MinimalTuple tuple)

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().

◆ ExecParallelHashRepartitionRest()

static void ExecParallelHashRepartitionRest ( HashJoinTable  hashtable)
static

Definition at line 1350 of file nodeHash.c.

1351 {
1352  ParallelHashJoinState *pstate = hashtable->parallel_state;
1353  int old_nbatch = pstate->old_nbatch;
1354  SharedTuplestoreAccessor **old_inner_tuples;
1355  ParallelHashJoinBatch *old_batches;
1356  int i;
1357 
1358  /* Get our hands on the previous generation of batches. */
1359  old_batches = (ParallelHashJoinBatch *)
1360  dsa_get_address(hashtable->area, pstate->old_batches);
1361  old_inner_tuples = palloc0_array(SharedTuplestoreAccessor *, old_nbatch);
1362  for (i = 1; i < old_nbatch; ++i)
1363  {
1364  ParallelHashJoinBatch *shared =
1365  NthParallelHashJoinBatch(old_batches, i);
1366 
1367  old_inner_tuples[i] = sts_attach(ParallelHashJoinBatchInner(shared),
1369  &pstate->fileset);
1370  }
1371 
1372  /* Join in the effort to repartition them. */
1373  for (i = 1; i < old_nbatch; ++i)
1374  {
1375  MinimalTuple tuple;
1376  uint32 hashvalue;
1377 
1378  /* Scan one partition from the previous generation. */
1379  sts_begin_parallel_scan(old_inner_tuples[i]);
1380  while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue)))
1381  {
1382  size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1383  int bucketno;
1384  int batchno;
1385 
1386  /* Decide which partition it goes to in the new generation. */
1387  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
1388  &batchno);
1389 
1390  hashtable->batches[batchno].estimated_size += tuple_size;
1391  ++hashtable->batches[batchno].ntuples;
1392  ++hashtable->batches[i].old_ntuples;
1393 
1394  /* Store the tuple its new batch. */
1395  sts_puttuple(hashtable->batches[batchno].inner_tuples,
1396  &hashvalue, tuple);
1397 
1399  }
1400  sts_end_parallel_scan(old_inner_tuples[i]);
1401  }
1402 
1403  pfree(old_inner_tuples);
1404 }
MinimalTuple sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data)
void sts_begin_parallel_scan(SharedTuplestoreAccessor *accessor)

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_array, 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().

◆ ExecParallelHashTableAlloc()

void ExecParallelHashTableAlloc ( HashJoinTable  hashtable,
int  batchno 
)

Definition at line 3105 of file nodeHash.c.

3106 {
3107  ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
3108  dsa_pointer_atomic *buckets;
3109  int nbuckets = hashtable->parallel_state->nbuckets;
3110  int i;
3111 
3112  batch->buckets =
3113  dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
3114  buckets = (dsa_pointer_atomic *)
3115  dsa_get_address(hashtable->area, batch->buckets);
3116  for (i = 0; i < nbuckets; ++i)
3118 }

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().

◆ ExecParallelHashTableInsert()

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

Definition at line 1691 of file nodeHash.c.

1694 {
1695  bool shouldFree;
1696  MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1697  dsa_pointer shared;
1698  int bucketno;
1699  int batchno;
1700 
1701 retry:
1702  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1703 
1704  if (batchno == 0)
1705  {
1706  HashJoinTuple hashTuple;
1707 
1708  /* Try to load it into memory. */
1711  hashTuple = ExecParallelHashTupleAlloc(hashtable,
1712  HJTUPLE_OVERHEAD + tuple->t_len,
1713  &shared);
1714  if (hashTuple == NULL)
1715  goto retry;
1716 
1717  /* Store the hash value in the HashJoinTuple header. */
1718  hashTuple->hashvalue = hashvalue;
1719  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1720 
1721  /* Push it onto the front of the bucket's list */
1722  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1723  hashTuple, shared);
1724  }
1725  else
1726  {
1727  size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1728 
1729  Assert(batchno > 0);
1730 
1731  /* Try to preallocate space in the batch if necessary. */
1732  if (hashtable->batches[batchno].preallocated < tuple_size)
1733  {
1734  if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size))
1735  goto retry;
1736  }
1737 
1738  Assert(hashtable->batches[batchno].preallocated >= tuple_size);
1739  hashtable->batches[batchno].preallocated -= tuple_size;
1740  sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue,
1741  tuple);
1742  }
1743  ++hashtable->batches[batchno].ntuples;
1744 
1745  if (shouldFree)
1746  heap_free_minimal_tuple(tuple);
1747 }
static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size)
Definition: nodeHash.c:3329

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().

◆ ExecParallelHashTableInsertCurrentBatch()

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

Definition at line 1756 of file nodeHash.c.

1759 {
1760  bool shouldFree;
1761  MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1762  HashJoinTuple hashTuple;
1763  dsa_pointer shared;
1764  int batchno;
1765  int bucketno;
1766 
1767  ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1768  Assert(batchno == hashtable->curbatch);
1769  hashTuple = ExecParallelHashTupleAlloc(hashtable,
1770  HJTUPLE_OVERHEAD + tuple->t_len,
1771  &shared);
1772  hashTuple->hashvalue = hashvalue;
1773  memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1775  ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1776  hashTuple, shared);
1777 
1778  if (shouldFree)
1779  heap_free_minimal_tuple(tuple);
1780 }

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().

◆ ExecParallelHashTableSetCurrentBatch()

void ExecParallelHashTableSetCurrentBatch ( HashJoinTable  hashtable,
int  batchno 
)

Definition at line 3267 of file nodeHash.c.

3268 {
3269  Assert(hashtable->batches[batchno].shared->buckets != InvalidDsaPointer);
3270 
3271  hashtable->curbatch = batchno;
3272  hashtable->buckets.shared = (dsa_pointer_atomic *)
3273  dsa_get_address(hashtable->area,
3274  hashtable->batches[batchno].shared->buckets);
3275  hashtable->nbuckets = hashtable->parallel_state->nbuckets;
3276  hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
3277  hashtable->current_chunk = NULL;
3279  hashtable->batches[batchno].at_least_one_chunk = false;
3280 }
dsa_pointer current_chunk_shared
Definition: hashjoin.h:359

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().

◆ ExecParallelHashTupleAlloc()

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

Definition at line 2796 of file nodeHash.c.

2798 {
2799  ParallelHashJoinState *pstate = hashtable->parallel_state;
2800  dsa_pointer chunk_shared;
2801  HashMemoryChunk chunk;
2802  Size chunk_size;
2803  HashJoinTuple result;
2804  int curbatch = hashtable->curbatch;
2805 
2806  size = MAXALIGN(size);
2807 
2808  /*
2809  * Fast path: if there is enough space in this backend's current chunk,
2810  * then we can allocate without any locking.
2811  */
2812  chunk = hashtable->current_chunk;
2813  if (chunk != NULL &&
2814  size <= HASH_CHUNK_THRESHOLD &&
2815  chunk->maxlen - chunk->used >= size)
2816  {
2817 
2818  chunk_shared = hashtable->current_chunk_shared;
2819  Assert(chunk == dsa_get_address(hashtable->area, chunk_shared));
2820  *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE + chunk->used;
2821  result = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + chunk->used);
2822  chunk->used += size;
2823 
2824  Assert(chunk->used <= chunk->maxlen);
2825  Assert(result == dsa_get_address(hashtable->area, *shared));
2826 
2827  return result;
2828  }
2829 
2830  /* Slow path: try to allocate a new chunk. */
2831  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
2832 
2833  /*
2834  * Check if we need to help increase the number of buckets or batches.
2835  */
2836  if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
2838  {
2839  ParallelHashGrowth growth = pstate->growth;
2840 
2841  hashtable->current_chunk = NULL;
2842  LWLockRelease(&pstate->lock);
2843 
2844  /* Another participant has commanded us to help grow. */
2845  if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
2847  else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
2849 
2850  /* The caller must retry. */
2851  return NULL;
2852  }
2853 
2854  /* Oversized tuples get their own chunk. */
2855  if (size > HASH_CHUNK_THRESHOLD)
2856  chunk_size = size + HASH_CHUNK_HEADER_SIZE;
2857  else
2858  chunk_size = HASH_CHUNK_SIZE;
2859 
2860  /* Check if it's time to grow batches or buckets. */
2861  if (pstate->growth != PHJ_GROWTH_DISABLED)
2862  {
2863  Assert(curbatch == 0);
2865 
2866  /*
2867  * Check if our space limit would be exceeded. To avoid choking on
2868  * very large tuples or very low hash_mem setting, we'll always allow
2869  * each backend to allocate at least one chunk.
2870  */
2871  if (hashtable->batches[0].at_least_one_chunk &&
2872  hashtable->batches[0].shared->size +
2873  chunk_size > pstate->space_allowed)
2874  {
2876  hashtable->batches[0].shared->space_exhausted = true;
2877  LWLockRelease(&pstate->lock);
2878 
2879  return NULL;
2880  }
2881 
2882  /* Check if our load factor limit would be exceeded. */
2883  if (hashtable->nbatch == 1)
2884  {
2885  hashtable->batches[0].shared->ntuples += hashtable->batches[0].ntuples;
2886  hashtable->batches[0].ntuples = 0;
2887  /* Guard against integer overflow and alloc size overflow */
2888  if (hashtable->batches[0].shared->ntuples + 1 >
2889  hashtable->nbuckets * NTUP_PER_BUCKET &&
2890  hashtable->nbuckets < (INT_MAX / 2) &&
2891  hashtable->nbuckets * 2 <=
2892  MaxAllocSize / sizeof(dsa_pointer_atomic))
2893  {
2895  LWLockRelease(&pstate->lock);
2896 
2897  return NULL;
2898  }
2899  }
2900  }
2901 
2902  /* We are cleared to allocate a new chunk. */
2903  chunk_shared = dsa_allocate(hashtable->area, chunk_size);
2904  hashtable->batches[curbatch].shared->size += chunk_size;
2905  hashtable->batches[curbatch].at_least_one_chunk = true;
2906 
2907  /* Set up the chunk. */
2908  chunk = (HashMemoryChunk) dsa_get_address(hashtable->area, chunk_shared);
2909  *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE;
2910  chunk->maxlen = chunk_size - HASH_CHUNK_HEADER_SIZE;
2911  chunk->used = size;
2912 
2913  /*
2914  * Push it onto the list of chunks, so that it can be found if we need to
2915  * increase the number of buckets or batches (batch 0 only) and later for
2916  * freeing the memory (all batches).
2917  */
2918  chunk->next.shared = hashtable->batches[curbatch].shared->chunks;
2919  hashtable->batches[curbatch].shared->chunks = chunk_shared;
2920 
2921  if (size <= HASH_CHUNK_THRESHOLD)
2922  {
2923  /*
2924  * Make this the current chunk so that we can use the fast path to
2925  * fill the rest of it up in future calls.
2926  */
2927  hashtable->current_chunk = chunk;
2928  hashtable->current_chunk_shared = chunk_shared;
2929  }
2930  LWLockRelease(&pstate->lock);
2931 
2932  Assert(HASH_CHUNK_DATA(chunk) == dsa_get_address(hashtable->area, *shared));
2933  result = (HashJoinTuple) HASH_CHUNK_DATA(chunk);
2934 
2935  return result;
2936 }
ParallelHashGrowth
Definition: hashjoin.h:219
@ PHJ_GROWTH_NEED_MORE_BUCKETS
Definition: hashjoin.h:223
static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
Definition: nodeHash.c:1503
static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:1067

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().

◆ ExecParallelHashTuplePrealloc()

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

Definition at line 3329 of file nodeHash.c.

3330 {
3331  ParallelHashJoinState *pstate = hashtable->parallel_state;
3332  ParallelHashJoinBatchAccessor *batch = &hashtable->batches[batchno];
3333  size_t want = Max(size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);
3334 
3335  Assert(batchno > 0);
3336  Assert(batchno < hashtable->nbatch);
3337  Assert(size == MAXALIGN(size));
3338 
3339  LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
3340 
3341  /* Has another participant commanded us to help grow? */
3342  if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
3344  {
3345  ParallelHashGrowth growth = pstate->growth;
3346 
3347  LWLockRelease(&pstate->lock);
3348  if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
3350  else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
3352 
3353  return false;
3354  }
3355 
3356  if (pstate->growth != PHJ_GROWTH_DISABLED &&
3357  batch->at_least_one_chunk &&
3358  (batch->shared->estimated_size + want + HASH_CHUNK_HEADER_SIZE
3359  > pstate->space_allowed))
3360  {
3361  /*
3362  * We have determined that this batch would exceed the space budget if
3363  * loaded into memory. Command all participants to help repartition.
3364  */
3365  batch->shared->space_exhausted = true;
3367  LWLockRelease(&pstate->lock);
3368 
3369  return false;
3370  }
3371 
3372  batch->at_least_one_chunk = true;
3373  batch->shared->estimated_size += want + HASH_CHUNK_HEADER_SIZE;
3374  batch->preallocated = want;
3375  LWLockRelease(&pstate->lock);
3376 
3377  return true;
3378 }

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().

◆ ExecParallelScanHashBucket()

bool ExecParallelScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 2001 of file nodeHash.c.

2003 {
2004  ExprState *hjclauses = hjstate->hashclauses;
2005  HashJoinTable hashtable = hjstate->hj_HashTable;
2006  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2007  uint32 hashvalue = hjstate->hj_CurHashValue;
2008 
2009  /*
2010  * hj_CurTuple is the address of the tuple last returned from the current
2011  * bucket, or NULL if it's time to start scanning a new bucket.
2012  */
2013  if (hashTuple != NULL)
2014  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2015  else
2016  hashTuple = ExecParallelHashFirstTuple(hashtable,
2017  hjstate->hj_CurBucketNo);
2018 
2019  while (hashTuple != NULL)
2020  {
2021  if (hashTuple->hashvalue == hashvalue)
2022  {
2023  TupleTableSlot *inntuple;
2024 
2025  /* insert hashtable's tuple into exec slot so ExecQual sees it */
2026  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2027  hjstate->hj_HashTupleSlot,
2028  false); /* do not pfree */
2029  econtext->ecxt_innertuple = inntuple;
2030 
2031  if (ExecQualAndReset(hjclauses, econtext))
2032  {
2033  hjstate->hj_CurTuple = hashTuple;
2034  return true;
2035  }
2036  }
2037 
2038  hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2039  }
2040 
2041  /*
2042  * no match
2043  */
2044  return false;
2045 }
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1446
static bool ExecQualAndReset(ExprState *state, ExprContext *econtext)
Definition: executor.h:425
static HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable hashtable, int bucketno)
Definition: nodeHash.c:3219
static HashJoinTuple ExecParallelHashNextTuple(HashJoinTable hashtable, HashJoinTuple tuple)
Definition: nodeHash.c:3235
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:249
HashJoinTuple hj_CurTuple
Definition: execnodes.h:2093
ExprState * hashclauses
Definition: execnodes.h:2085
uint32 hj_CurHashValue
Definition: execnodes.h:2090
int hj_CurBucketNo
Definition: execnodes.h:2091
HashJoinTable hj_HashTable
Definition: execnodes.h:2089
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:2095

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().

◆ ExecPrepHashTableForUnmatched()

void ExecPrepHashTableForUnmatched ( HashJoinState hjstate)

Definition at line 2052 of file nodeHash.c.

2053 {
2054  /*----------
2055  * During this scan we use the HashJoinState fields as follows:
2056  *
2057  * hj_CurBucketNo: next regular bucket to scan
2058  * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
2059  * hj_CurTuple: last tuple returned, or NULL to start next bucket
2060  *----------
2061  */
2062  hjstate->hj_CurBucketNo = 0;
2063  hjstate->hj_CurSkewBucketNo = 0;
2064  hjstate->hj_CurTuple = NULL;
2065 }
int hj_CurSkewBucketNo
Definition: execnodes.h:2092

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

Referenced by ExecHashJoinImpl().

◆ ExecReScanHash()

void ExecReScanHash ( HashState node)

Definition at line 2201 of file nodeHash.c.

2202 {
2204 
2205  /*
2206  * if chgParam of subnode is not null then plan will be re-scanned by
2207  * first ExecProcNode.
2208  */
2209  if (outerPlan->chgParam == NULL)
2211 }
void ExecReScan(PlanState *node)
Definition: execAmi.c:78

References ExecReScan(), outerPlan, and outerPlanState.

Referenced by ExecReScan().

◆ ExecScanHashBucket()

bool ExecScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 1940 of file nodeHash.c.

1942 {
1943  ExprState *hjclauses = hjstate->hashclauses;
1944  HashJoinTable hashtable = hjstate->hj_HashTable;
1945  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1946  uint32 hashvalue = hjstate->hj_CurHashValue;
1947 
1948  /*
1949  * hj_CurTuple is the address of the tuple last returned from the current
1950  * bucket, or NULL if it's time to start scanning a new bucket.
1951  *
1952  * If the tuple hashed to a skew bucket then scan the skew bucket
1953  * otherwise scan the standard hashtable bucket.
1954  */
1955  if (hashTuple != NULL)
1956  hashTuple = hashTuple->next.unshared;
1957  else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
1958  hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
1959  else
1960  hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
1961 
1962  while (hashTuple != NULL)
1963  {
1964  if (hashTuple->hashvalue == hashvalue)
1965  {
1966  TupleTableSlot *inntuple;
1967 
1968  /* insert hashtable's tuple into exec slot so ExecQual sees it */
1969  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
1970  hjstate->hj_HashTupleSlot,
1971  false); /* do not pfree */
1972  econtext->ecxt_innertuple = inntuple;
1973 
1974  if (ExecQualAndReset(hjclauses, econtext))
1975  {
1976  hjstate->hj_CurTuple = hashTuple;
1977  return true;
1978  }
1979  }
1980 
1981  hashTuple = hashTuple->next.unshared;
1982  }
1983 
1984  /*
1985  * no match
1986  */
1987  return false;
1988 }

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().

◆ ExecScanHashTableForUnmatched()

bool ExecScanHashTableForUnmatched ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 2076 of file nodeHash.c.

2077 {
2078  HashJoinTable hashtable = hjstate->hj_HashTable;
2079  HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2080 
2081  for (;;)
2082  {
2083  /*
2084  * hj_CurTuple is the address of the tuple last returned from the
2085  * current bucket, or NULL if it's time to start scanning a new
2086  * bucket.
2087  */
2088  if (hashTuple != NULL)
2089  hashTuple = hashTuple->next.unshared;
2090  else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
2091  {
2092  hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
2093  hjstate->hj_CurBucketNo++;
2094  }
2095  else if (hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
2096  {
2097  int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
2098 
2099  hashTuple = hashtable->skewBucket[j]->tuples;
2100  hjstate->hj_CurSkewBucketNo++;
2101  }
2102  else
2103  break; /* finished all buckets */
2104 
2105  while (hashTuple != NULL)
2106  {
2107  if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
2108  {
2109  TupleTableSlot *inntuple;
2110 
2111  /* insert hashtable's tuple into exec slot */
2112  inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2113  hjstate->hj_HashTupleSlot,
2114  false); /* do not pfree */
2115  econtext->ecxt_innertuple = inntuple;
2116 
2117  /*
2118  * Reset temp memory each time; although this function doesn't
2119  * do any qual eval, the caller will, so let's keep it
2120  * parallel to ExecScanHashBucket.
2121  */
2122  ResetExprContext(econtext);
2123 
2124  hjstate->hj_CurTuple = hashTuple;
2125  return true;
2126  }
2127 
2128  hashTuple = hashTuple->next.unshared;
2129  }
2130 
2131  /* allow this loop to be cancellable */
2133  }
2134 
2135  /*
2136  * no more unmatched tuples
2137  */
2138  return false;
2139 }
#define HeapTupleHeaderHasMatch(tup)
Definition: htup_details.h:510

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, j, HashJoinTableData::nbuckets, HashJoinTupleData::next, HashJoinTableData::nSkewBuckets, ResetExprContext, HashJoinTableData::skewBucket, HashJoinTableData::skewBucketNums, HashSkewBucket::tuples, HashJoinTupleData::unshared, and HashJoinTableData::unshared.

Referenced by ExecHashJoinImpl().

◆ ExecShutdownHash()

void ExecShutdownHash ( HashState node)

Definition at line 2651 of file nodeHash.c.

2652 {
2653  /* Allocate save space if EXPLAIN'ing and we didn't do so already */
2654  if (node->ps.instrument && !node->hinstrument)
2656  /* Now accumulate data for the current (final) hash table */
2657  if (node->hinstrument && node->hashtable)
2659 }
#define palloc0_object(type)
Definition: fe_memutils.h:63
void ExecHashAccumInstrumentation(HashInstrumentation *instrument, HashJoinTable hashtable)
Definition: nodeHash.c:2697

References ExecHashAccumInstrumentation(), HashState::hashtable, HashState::hinstrument, PlanState::instrument, palloc0_object, and HashState::ps.

Referenced by ExecShutdownNode_walker().

◆ get_hash_memory_limit()

size_t get_hash_memory_limit ( void  )

Definition at line 3390 of file nodeHash.c.

3391 {
3392  double mem_limit;
3393 
3394  /* Do initial calculation in double arithmetic */
3395  mem_limit = (double) work_mem * hash_mem_multiplier * 1024.0;
3396 
3397  /* Clamp in case it doesn't fit in size_t */
3398  mem_limit = Min(mem_limit, (double) SIZE_MAX);
3399 
3400  return (size_t) mem_limit;
3401 }
double hash_mem_multiplier
Definition: globals.c:126
int work_mem
Definition: globals.c:125

References hash_mem_multiplier, Min, and work_mem.

Referenced by BuildTupleHashTableExt(), choose_hashed_setop(), consider_groupingsets_paths(), cost_memoize_rescan(), create_unique_path(), ExecChooseHashTableSize(), ExecInitMemoize(), ExecParallelHashIncreaseNumBatches(), final_cost_hashjoin(), hash_agg_set_limits(), hash_choose_num_partitions(), subpath_is_hashable(), and subplan_is_hashable().

◆ MultiExecHash()

Node* MultiExecHash ( HashState node)

Definition at line 106 of file nodeHash.c.

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

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

Referenced by MultiExecProcNode().

◆ MultiExecParallelHash()

static void MultiExecParallelHash ( HashState node)
static

Definition at line 215 of file nodeHash.c.

216 {
217  ParallelHashJoinState *pstate;
218  PlanState *outerNode;
219  List *hashkeys;
220  HashJoinTable hashtable;
221  TupleTableSlot *slot;
222  ExprContext *econtext;
223  uint32 hashvalue;
224  Barrier *build_barrier;
225  int i;
226 
227  /*
228  * get state info from node
229  */
230  outerNode = outerPlanState(node);
231  hashtable = node->hashtable;
232 
233  /*
234  * set expression context
235  */
236  hashkeys = node->hashkeys;
237  econtext = node->ps.ps_ExprContext;
238 
239  /*
240  * Synchronize the parallel hash table build. At this stage we know that
241  * the shared hash table has been or is being set up by
242  * ExecHashTableCreate(), but we don't know if our peers have returned
243  * from there or are here in MultiExecParallelHash(), and if so how far
244  * through they are. To find out, we check the build_barrier phase then
245  * and jump to the right step in the build algorithm.
246  */
247  pstate = hashtable->parallel_state;
248  build_barrier = &pstate->build_barrier;
249  Assert(BarrierPhase(build_barrier) >= PHJ_BUILD_ALLOCATING);
250  switch (BarrierPhase(build_barrier))
251  {
253 
254  /*
255  * Either I just allocated the initial hash table in
256  * ExecHashTableCreate(), or someone else is doing that. Either
257  * way, wait for everyone to arrive here so we can proceed.
258  */
260  /* Fall through. */
261 
263 
264  /*
265  * It's time to begin hashing, or if we just arrived here then
266  * hashing is already underway, so join in that effort. While
267  * hashing we have to be prepared to help increase the number of
268  * batches or buckets at any time, and if we arrived here when
269  * that was already underway we'll have to help complete that work
270  * immediately so that it's safe to access batches and buckets
271  * below.
272  */
281  for (;;)
282  {
283  slot = ExecProcNode(outerNode);
284  if (TupIsNull(slot))
285  break;
286  econtext->ecxt_outertuple = slot;
287  if (ExecHashGetHashValue(hashtable, econtext, hashkeys,
288  false, hashtable->keepNulls,
289  &hashvalue))
290  ExecParallelHashTableInsert(hashtable, slot, hashvalue);
291  hashtable->partialTuples++;
292  }
293 
294  /*
295  * Make sure that any tuples we wrote to disk are visible to
296  * others before anyone tries to load them.
297  */
298  for (i = 0; i < hashtable->nbatch; ++i)
299  sts_end_write(hashtable->batches[i].inner_tuples);
300 
301  /*
302  * Update shared counters. We need an accurate total tuple count
303  * to control the empty table optimization.
304  */
306 
309 
310  /*
311  * Wait for everyone to finish building and flushing files and
312  * counters.
313  */
314  if (BarrierArriveAndWait(build_barrier,
316  {
317  /*
318  * Elect one backend to disable any further growth. Batches
319  * are now fixed. While building them we made sure they'd fit
320  * in our memory budget when we load them back in later (or we
321  * tried to do that and gave up because we detected extreme
322  * skew).
323  */
324  pstate->growth = PHJ_GROWTH_DISABLED;
325  }
326  }
327 
328  /*
329  * We're not yet attached to a batch. We all agree on the dimensions and
330  * number of inner tuples (for the empty table optimization).
331  */
332  hashtable->curbatch = -1;
333  hashtable->nbuckets = pstate->nbuckets;
334  hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
335  hashtable->totalTuples = pstate->total_tuples;
337 
338  /*
339  * The next synchronization point is in ExecHashJoin's HJ_BUILD_HASHTABLE
340  * case, which will bring the build phase to PHJ_BUILD_DONE (if it isn't
341  * there already).
342  */
343  Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER ||
344  BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
345 }
static TupleTableSlot * ExecProcNode(PlanState *node)
Definition: executor.h:254
#define PHJ_BUILD_DONE
Definition: hashjoin.h:261
#define PHJ_BUILD_HASHING_OUTER
Definition: hashjoin.h:260
#define PHJ_BUILD_ALLOCATING
Definition: hashjoin.h:258
void ExecParallelHashTableInsert(HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue)
Definition: nodeHash.c:1691
bool ExecHashGetHashValue(HashJoinTable hashtable, ExprContext *econtext, List *hashkeys, bool outer_tuple, bool keep_nulls, uint32 *hashvalue)
Definition: nodeHash.c:1800
TupleTableSlot * ecxt_outertuple
Definition: execnodes.h:251
Definition: pg_list.h:52
ExprContext * ps_ExprContext
Definition: execnodes.h:1068
#define TupIsNull(slot)
Definition: tuptable.h:300
@ WAIT_EVENT_HASH_BUILD_HASH_INNER
Definition: wait_event.h:98
@ WAIT_EVENT_HASH_BUILD_ALLOCATE
Definition: wait_event.h:96

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

Referenced by MultiExecHash().

◆ MultiExecPrivateHash()

static void MultiExecPrivateHash ( HashState node)
static

Definition at line 139 of file nodeHash.c.

140 {
141  PlanState *outerNode;
142  List *hashkeys;
143  HashJoinTable hashtable;
144  TupleTableSlot *slot;
145  ExprContext *econtext;
146  uint32 hashvalue;
147 
148  /*
149  * get state info from node
150  */
151  outerNode = outerPlanState(node);
152  hashtable = node->hashtable;
153 
154  /*
155  * set expression context
156  */
157  hashkeys = node->hashkeys;
158  econtext = node->ps.ps_ExprContext;
159 
160  /*
161  * Get all tuples from the node below the Hash node and insert into the
162  * hash table (or temp files).
163  */
164  for (;;)
165  {
166  slot = ExecProcNode(outerNode);
167  if (TupIsNull(slot))
168  break;
169  /* We have to compute the hash value */
170  econtext->ecxt_outertuple = slot;
171  if (ExecHashGetHashValue(hashtable, econtext, hashkeys,
172  false, hashtable->keepNulls,
173  &hashvalue))
174  {
175  int bucketNumber;
176 
177  bucketNumber = ExecHashGetSkewBucket(hashtable, hashvalue);
178  if (bucketNumber != INVALID_SKEW_BUCKET_NO)
179  {
180  /* It's a skew tuple, so put it into that hash table */
181  ExecHashSkewTableInsert(hashtable, slot, hashvalue,
182  bucketNumber);
183  hashtable->skewTuples += 1;
184  }
185  else
186  {
187  /* Not subject to skew optimization, so insert normally */
188  ExecHashTableInsert(hashtable, slot, hashvalue);
189  }
190  hashtable->totalTuples += 1;
191  }
192  }
193 
194  /* resize the hash table if needed (NTUP_PER_BUCKET exceeded) */
195  if (hashtable->nbuckets != hashtable->nbuckets_optimal)
196  ExecHashIncreaseNumBuckets(hashtable);
197 
198  /* Account for the buckets in spaceUsed (reported in EXPLAIN ANALYZE) */
199  hashtable->spaceUsed += hashtable->nbuckets * sizeof(HashJoinTuple);
200  if (hashtable->spaceUsed > hashtable->spacePeak)
201  hashtable->spacePeak = hashtable->spaceUsed;
202 
203  hashtable->partialTuples = hashtable->totalTuples;
204 }
static void ExecHashIncreaseNumBuckets(HashJoinTable hashtable)
Definition: nodeHash.c:1440
int ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue)
Definition: nodeHash.c:2376
static void ExecHashSkewTableInsert(HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue, int bucketNumber)
Definition: nodeHash.c:2422
void ExecHashTableInsert(HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue)
Definition: nodeHash.c:1602

References ExprContext::ecxt_outertuple, ExecHashGetHashValue(), ExecHashGetSkewBucket(), ExecHashIncreaseNumBuckets(), ExecHashSkewTableInsert(), ExecHashTableInsert(), ExecProcNode(), HashState::hashkeys, HashState::hashtable, INVALID_SKEW_BUCKET_NO, HashJoinTableData::keepNulls, HashJoinTableData::nbuckets, HashJoinTableData::nbuckets_optimal, outerPlanState, HashJoinTableData::partialTuples, HashState::ps, PlanState::ps_ExprContext, HashJoinTableData::skewTuples, HashJoinTableData::spacePeak, HashJoinTableData::spaceUsed, HashJoinTableData::totalTuples, and TupIsNull.

Referenced by MultiExecHash().