PostgreSQL Source Code git master
nodeHash.h File Reference
#include "access/parallel.h"
#include "nodes/execnodes.h"
Include dependency graph for nodeHash.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions

HashStateExecInitHash (Hash *node, EState *estate, int eflags)
 
NodeMultiExecHash (HashState *node)
 
void ExecEndHash (HashState *node)
 
void ExecReScanHash (HashState *node)
 
HashJoinTable ExecHashTableCreate (HashState *state)
 
void ExecParallelHashTableAlloc (HashJoinTable hashtable, int batchno)
 
void ExecHashTableDestroy (HashJoinTable hashtable)
 
void ExecHashTableDetach (HashJoinTable hashtable)
 
void ExecHashTableDetachBatch (HashJoinTable hashtable)
 
void ExecParallelHashTableSetCurrentBatch (HashJoinTable hashtable, int batchno)
 
void ExecHashTableInsert (HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue)
 
void ExecParallelHashTableInsert (HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue)
 
void ExecParallelHashTableInsertCurrentBatch (HashJoinTable hashtable, TupleTableSlot *slot, 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 ExecParallelPrepHashTableForUnmatched (HashJoinState *hjstate)
 
bool ExecScanHashTableForUnmatched (HashJoinState *hjstate, ExprContext *econtext)
 
bool ExecParallelScanHashTableForUnmatched (HashJoinState *hjstate, ExprContext *econtext)
 
void ExecHashTableReset (HashJoinTable hashtable)
 
void ExecHashTableResetMatchFlags (HashJoinTable hashtable)
 
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)
 
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 ExecHashRetrieveInstrumentation (HashState *node)
 
void ExecShutdownHash (HashState *node)
 
void ExecHashAccumInstrumentation (HashInstrumentation *instrument, HashJoinTable hashtable)
 

Function Documentation

◆ 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 658 of file nodeHash.c.

665{
666 int tupsize;
667 double inner_rel_bytes;
668 size_t hash_table_bytes;
669 size_t bucket_bytes;
670 size_t max_pointers;
671 int nbatch = 1;
672 int nbuckets;
673 double dbuckets;
674
675 /* Force a plausible relation size if no info */
676 if (ntuples <= 0.0)
677 ntuples = 1000.0;
678
679 /*
680 * Estimate tupsize based on footprint of tuple in hashtable... note this
681 * does not allow for any palloc overhead. The manipulations of spaceUsed
682 * don't count palloc overhead either.
683 */
684 tupsize = HJTUPLE_OVERHEAD +
686 MAXALIGN(tupwidth);
687 inner_rel_bytes = ntuples * tupsize;
688
689 /*
690 * Compute in-memory hashtable size limit from GUCs.
691 */
692 hash_table_bytes = get_hash_memory_limit();
693
694 /*
695 * Parallel Hash tries to use the combined hash_mem of all workers to
696 * avoid the need to batch. If that won't work, it falls back to hash_mem
697 * per worker and tries to process batches in parallel.
698 */
699 if (try_combined_hash_mem)
700 {
701 /* Careful, this could overflow size_t */
702 double newlimit;
703
704 newlimit = (double) hash_table_bytes * (double) (parallel_workers + 1);
705 newlimit = Min(newlimit, (double) SIZE_MAX);
706 hash_table_bytes = (size_t) newlimit;
707 }
708
709 *space_allowed = hash_table_bytes;
710
711 /*
712 * If skew optimization is possible, estimate the number of skew buckets
713 * that will fit in the memory allowed, and decrement the assumed space
714 * available for the main hash table accordingly.
715 *
716 * We make the optimistic assumption that each skew bucket will contain
717 * one inner-relation tuple. If that turns out to be low, we will recover
718 * at runtime by reducing the number of skew buckets.
719 *
720 * hashtable->skewBucket will have up to 8 times as many HashSkewBucket
721 * pointers as the number of MCVs we allow, since ExecHashBuildSkewHash
722 * will round up to the next power of 2 and then multiply by 4 to reduce
723 * collisions.
724 */
725 if (useskew)
726 {
727 size_t bytes_per_mcv;
728 size_t skew_mcvs;
729
730 /*----------
731 * Compute number of MCVs we could hold in hash_table_bytes
732 *
733 * Divisor is:
734 * size of a hash tuple +
735 * worst-case size of skewBucket[] per MCV +
736 * size of skewBucketNums[] entry +
737 * size of skew bucket struct itself
738 *----------
739 */
740 bytes_per_mcv = tupsize +
741 (8 * sizeof(HashSkewBucket *)) +
742 sizeof(int) +
744 skew_mcvs = hash_table_bytes / bytes_per_mcv;
745
746 /*
747 * Now scale by SKEW_HASH_MEM_PERCENT (we do it in this order so as
748 * not to worry about size_t overflow in the multiplication)
749 */
750 skew_mcvs = (skew_mcvs * SKEW_HASH_MEM_PERCENT) / 100;
751
752 /* Now clamp to integer range */
753 skew_mcvs = Min(skew_mcvs, INT_MAX);
754
755 *num_skew_mcvs = (int) skew_mcvs;
756
757 /* Reduce hash_table_bytes by the amount needed for the skew table */
758 if (skew_mcvs > 0)
759 hash_table_bytes -= skew_mcvs * bytes_per_mcv;
760 }
761 else
762 *num_skew_mcvs = 0;
763
764 /*
765 * Set nbuckets to achieve an average bucket load of NTUP_PER_BUCKET when
766 * memory is filled, assuming a single batch; but limit the value so that
767 * the pointer arrays we'll try to allocate do not exceed hash_table_bytes
768 * nor MaxAllocSize.
769 *
770 * Note that both nbuckets and nbatch must be powers of 2 to make
771 * ExecHashGetBucketAndBatch fast.
772 */
773 max_pointers = hash_table_bytes / sizeof(HashJoinTuple);
774 max_pointers = Min(max_pointers, MaxAllocSize / sizeof(HashJoinTuple));
775 /* If max_pointers isn't a power of 2, must round it down to one */
776 max_pointers = pg_prevpower2_size_t(max_pointers);
777
778 /* Also ensure we avoid integer overflow in nbatch and nbuckets */
779 /* (this step is redundant given the current value of MaxAllocSize) */
780 max_pointers = Min(max_pointers, INT_MAX / 2 + 1);
781
782 dbuckets = ceil(ntuples / NTUP_PER_BUCKET);
783 dbuckets = Min(dbuckets, max_pointers);
784 nbuckets = (int) dbuckets;
785 /* don't let nbuckets be really small, though ... */
786 nbuckets = Max(nbuckets, 1024);
787 /* ... and force it to be a power of 2. */
788 nbuckets = pg_nextpower2_32(nbuckets);
789
790 /*
791 * If there's not enough space to store the projected number of tuples and
792 * the required bucket headers, we will need multiple batches.
793 */
794 bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
795 if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
796 {
797 /* We'll need multiple batches */
798 size_t sbuckets;
799 double dbatch;
800 int minbatch;
801 size_t bucket_size;
802
803 /*
804 * If Parallel Hash with combined hash_mem would still need multiple
805 * batches, we'll have to fall back to regular hash_mem budget.
806 */
807 if (try_combined_hash_mem)
808 {
809 ExecChooseHashTableSize(ntuples, tupwidth, useskew,
810 false, parallel_workers,
811 space_allowed,
812 numbuckets,
813 numbatches,
814 num_skew_mcvs);
815 return;
816 }
817
818 /*
819 * Estimate the number of buckets we'll want to have when hash_mem is
820 * entirely full. Each bucket will contain a bucket pointer plus
821 * NTUP_PER_BUCKET tuples, whose projected size already includes
822 * overhead for the hash code, pointer to the next tuple, etc.
823 */
824 bucket_size = (tupsize * NTUP_PER_BUCKET + sizeof(HashJoinTuple));
825 if (hash_table_bytes <= bucket_size)
826 sbuckets = 1; /* avoid pg_nextpower2_size_t(0) */
827 else
828 sbuckets = pg_nextpower2_size_t(hash_table_bytes / bucket_size);
829 sbuckets = Min(sbuckets, max_pointers);
830 nbuckets = (int) sbuckets;
831 nbuckets = pg_nextpower2_32(nbuckets);
832 bucket_bytes = nbuckets * sizeof(HashJoinTuple);
833
834 /*
835 * Buckets are simple pointers to hashjoin tuples, while tupsize
836 * includes the pointer, hash code, and MinimalTupleData. So buckets
837 * should never really exceed 25% of hash_mem (even for
838 * NTUP_PER_BUCKET=1); except maybe for hash_mem values that are not
839 * 2^N bytes, where we might get more because of doubling. So let's
840 * look for 50% here.
841 */
842 Assert(bucket_bytes <= hash_table_bytes / 2);
843
844 /* Calculate required number of batches. */
845 dbatch = ceil(inner_rel_bytes / (hash_table_bytes - bucket_bytes));
846 dbatch = Min(dbatch, max_pointers);
847 minbatch = (int) dbatch;
848 nbatch = pg_nextpower2_32(Max(2, minbatch));
849 }
850
851 /*
852 * Optimize the total amount of memory consumed by the hash node.
853 *
854 * The nbatch calculation above focuses on the size of the in-memory hash
855 * table, assuming no per-batch overhead. Now adjust the number of batches
856 * and the size of the hash table to minimize total memory consumed by the
857 * hash node.
858 *
859 * Each batch file has a BLCKSZ buffer, and we may need two files per
860 * batch (inner and outer side). So with enough batches this can be
861 * significantly more memory than the hashtable itself.
862 *
863 * The total memory usage may be expressed by this formula:
864 *
865 * (inner_rel_bytes / nbatch) + (2 * nbatch * BLCKSZ) <= hash_table_bytes
866 *
867 * where (inner_rel_bytes / nbatch) is the size of the in-memory hash
868 * table and (2 * nbatch * BLCKSZ) is the amount of memory used by file
869 * buffers. But for sufficiently large values of inner_rel_bytes value
870 * there may not be a nbatch value that would make both parts fit into
871 * hash_table_bytes.
872 *
873 * In this case we can't enforce the memory limit - we're going to exceed
874 * it. We can however minimize the impact and use as little memory as
875 * possible. (We haven't really enforced it before either, as we simply
876 * ignored the batch files.)
877 *
878 * The formula for total memory usage says that given an inner relation of
879 * size inner_rel_bytes, we may divide it into an arbitrary number of
880 * batches. This determines both the size of the in-memory hash table and
881 * the amount of memory needed for batch files. These two terms work in
882 * opposite ways - when one decreases, the other increases.
883 *
884 * For low nbatch values, the hash table takes most of the memory, but at
885 * some point the batch files start to dominate. If you combine these two
886 * terms, the memory consumption (for a fixed size of the inner relation)
887 * has a u-shape, with a minimum at some nbatch value.
888 *
889 * Our goal is to find this nbatch value, minimizing the memory usage. We
890 * calculate the memory usage with half the batches (i.e. nbatch/2), and
891 * if it's lower than the current memory usage we know it's better to use
892 * fewer batches. We repeat this until reducing the number of batches does
893 * not reduce the memory usage - we found the optimum. We know the optimum
894 * exists, thanks to the u-shape.
895 *
896 * We only want to do this when exceeding the memory limit, not every
897 * time. The goal is not to minimize memory usage in every case, but to
898 * minimize the memory usage when we can't stay within the memory limit.
899 *
900 * For this reason we only consider reducing the number of batches. We
901 * could try the opposite direction too, but that would save memory only
902 * when most of the memory is used by the hash table. And the hash table
903 * was used for the initial sizing, so we shouldn't be exceeding the
904 * memory limit too much. We might save memory by using more batches, but
905 * it would result in spilling more batch files, which does not seem like
906 * a great trade off.
907 *
908 * While growing the hashtable, we also adjust the number of buckets, to
909 * not have more than one tuple per bucket (load factor 1). We can only do
910 * this during the initial sizing - once we start building the hash,
911 * nbucket is fixed.
912 */
913 while (nbatch > 0)
914 {
915 /* how much memory are we using with current nbatch value */
916 size_t current_space = hash_table_bytes + (2 * nbatch * BLCKSZ);
917
918 /* how much memory would we use with half the batches */
919 size_t new_space = hash_table_bytes * 2 + (nbatch * BLCKSZ);
920
921 /* If the memory usage would not decrease, we found the optimum. */
922 if (current_space < new_space)
923 break;
924
925 /*
926 * It's better to use half the batches, so do that and adjust the
927 * nbucket in the opposite direction, and double the allowance.
928 */
929 nbatch /= 2;
930 nbuckets *= 2;
931
932 *space_allowed = (*space_allowed) * 2;
933 }
934
935 Assert(nbuckets > 0);
936 Assert(nbatch > 0);
937
938 *numbuckets = nbuckets;
939 *numbatches = nbatch;
940}
#define Min(x, y)
Definition: c.h:975
#define MAXALIGN(LEN)
Definition: c.h:782
#define Max(x, y)
Definition: c.h:969
struct HashJoinTupleData * HashJoinTuple
Definition: execnodes.h:2244
#define MaxAllocSize
Definition: fe_memutils.h:22
Assert(PointerIsAligned(start, uint64))
#define HJTUPLE_OVERHEAD
Definition: hashjoin.h:90
#define SKEW_BUCKET_OVERHEAD
Definition: hashjoin.h:119
#define SKEW_HASH_MEM_PERCENT
Definition: hashjoin.h:121
#define SizeofMinimalTupleHeader
Definition: htup_details.h:699
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:658
#define NTUP_PER_BUCKET
Definition: nodeHash.c:655
size_t get_hash_memory_limit(void)
Definition: nodeHash.c:3616
static uint32 pg_nextpower2_32(uint32 num)
Definition: pg_bitutils.h:189
#define pg_nextpower2_size_t
Definition: pg_bitutils.h:415
#define pg_prevpower2_size_t
Definition: pg_bitutils.h:416

References Assert(), ExecChooseHashTableSize(), 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 ExecChooseHashTableSize(), ExecHashTableCreate(), and initial_cost_hashjoin().

◆ ExecEndHash()

void ExecEndHash ( HashState node)

Definition at line 427 of file nodeHash.c.

428{
430
431 /*
432 * shut down the subplan
433 */
436}
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:562
#define outerPlanState(node)
Definition: execnodes.h:1249
#define outerPlan(node)
Definition: plannodes.h:231

References ExecEndNode(), outerPlan, and outerPlanState.

Referenced by ExecEndNode().

◆ ExecHashAccumInstrumentation()

void ExecHashAccumInstrumentation ( HashInstrumentation instrument,
HashJoinTable  hashtable 
)

Definition at line 2871 of file nodeHash.c.

2873{
2874 instrument->nbuckets = Max(instrument->nbuckets,
2875 hashtable->nbuckets);
2876 instrument->nbuckets_original = Max(instrument->nbuckets_original,
2877 hashtable->nbuckets_original);
2878 instrument->nbatch = Max(instrument->nbatch,
2879 hashtable->nbatch);
2880 instrument->nbatch_original = Max(instrument->nbatch_original,
2881 hashtable->nbatch_original);
2882 instrument->space_peak = Max(instrument->space_peak,
2883 hashtable->spacePeak);
2884}

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

◆ ExecHashEstimate()

void ExecHashEstimate ( HashState node,
ParallelContext pcxt 
)

Definition at line 2755 of file nodeHash.c.

2756{
2757 size_t size;
2758
2759 /* don't need this if not instrumenting or no workers */
2760 if (!node->ps.instrument || pcxt->nworkers == 0)
2761 return;
2762
2763 size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
2764 size = add_size(size, offsetof(SharedHashInfo, hinstrument));
2765 shm_toc_estimate_chunk(&pcxt->estimator, size);
2767}
#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:488
Size mul_size(Size s1, Size s2)
Definition: shmem.c:505
PlanState ps
Definition: execnodes.h:2804
shm_toc_estimator estimator
Definition: parallel.h:41
Instrumentation * instrument
Definition: execnodes.h:1163

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 1954 of file nodeHash.c.

1958{
1959 uint32 nbuckets = (uint32) hashtable->nbuckets;
1960 uint32 nbatch = (uint32) hashtable->nbatch;
1961
1962 if (nbatch > 1)
1963 {
1964 *bucketno = hashvalue & (nbuckets - 1);
1965 *batchno = pg_rotate_right32(hashvalue,
1966 hashtable->log2_nbuckets) & (nbatch - 1);
1967 }
1968 else
1969 {
1970 *bucketno = hashvalue & (nbuckets - 1);
1971 *batchno = 0;
1972 }
1973}
uint32_t uint32
Definition: c.h:502
static uint32 pg_rotate_right32(uint32 word, int n)
Definition: pg_bitutils.h:396

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

◆ ExecHashGetSkewBucket()

int ExecHashGetSkewBucket ( HashJoinTable  hashtable,
uint32  hashvalue 
)

Definition at line 2549 of file nodeHash.c.

2550{
2551 int bucket;
2552
2553 /*
2554 * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
2555 * particular, this happens after the initial batch is done).
2556 */
2557 if (!hashtable->skewEnabled)
2559
2560 /*
2561 * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
2562 */
2563 bucket = hashvalue & (hashtable->skewBucketLen - 1);
2564
2565 /*
2566 * While we have not hit a hole in the hashtable and have not hit the
2567 * desired bucket, we have collided with some other hash value, so try the
2568 * next bucket location.
2569 */
2570 while (hashtable->skewBucket[bucket] != NULL &&
2571 hashtable->skewBucket[bucket]->hashvalue != hashvalue)
2572 bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
2573
2574 /*
2575 * Found the desired bucket?
2576 */
2577 if (hashtable->skewBucket[bucket] != NULL)
2578 return bucket;
2579
2580 /*
2581 * There must not be any hashtable entry for this hash value.
2582 */
2584}
#define INVALID_SKEW_BUCKET_NO
Definition: hashjoin.h:120
HashSkewBucket ** skewBucket
Definition: hashjoin.h:317
uint32 hashvalue
Definition: hashjoin.h:115

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

Referenced by ExecHashJoinImpl(), and MultiExecPrivateHash().

◆ ExecHashInitializeDSM()

void ExecHashInitializeDSM ( HashState node,
ParallelContext pcxt 
)

Definition at line 2774 of file nodeHash.c.

2775{
2776 size_t size;
2777
2778 /* don't need this if not instrumenting or no workers */
2779 if (!node->ps.instrument || pcxt->nworkers == 0)
2780 return;
2781
2782 size = offsetof(SharedHashInfo, hinstrument) +
2783 pcxt->nworkers * sizeof(HashInstrumentation);
2784 node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
2785
2786 /* Each per-worker area must start out as zeroes. */
2787 memset(node->shared_info, 0, size);
2788
2789 node->shared_info->num_workers = pcxt->nworkers;
2790 shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id,
2791 node->shared_info);
2792}
struct HashInstrumentation HashInstrumentation
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
SharedHashInfo * shared_info
Definition: execnodes.h:2817
shm_toc * toc
Definition: parallel.h:44
Plan * plan
Definition: execnodes.h:1153
int plan_node_id
Definition: plannodes.h:197

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 2799 of file nodeHash.c.

2800{
2801 SharedHashInfo *shared_info;
2802
2803 /* don't need this if not instrumenting */
2804 if (!node->ps.instrument)
2805 return;
2806
2807 /*
2808 * Find our entry in the shared area, and set up a pointer to it so that
2809 * we'll accumulate stats there when shutting down or rebuilding the hash
2810 * table.
2811 */
2812 shared_info = (SharedHashInfo *)
2813 shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
2814 node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
2815}
int ParallelWorkerNumber
Definition: parallel.c:115
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
HashInstrumentation * hinstrument
Definition: execnodes.h:2824
HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER]
Definition: execnodes.h:2795

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

◆ ExecHashRetrieveInstrumentation()

void ExecHashRetrieveInstrumentation ( HashState node)

Definition at line 2840 of file nodeHash.c.

2841{
2842 SharedHashInfo *shared_info = node->shared_info;
2843 size_t size;
2844
2845 if (shared_info == NULL)
2846 return;
2847
2848 /* Replace node->shared_info with a copy in backend-local memory. */
2849 size = offsetof(SharedHashInfo, hinstrument) +
2850 shared_info->num_workers * sizeof(HashInstrumentation);
2851 node->shared_info = palloc(size);
2852 memcpy(node->shared_info, shared_info, size);
2853}
void * palloc(Size size)
Definition: mcxt.c:1317

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

Referenced by ExecParallelRetrieveInstrumentation().

◆ ExecHashTableCreate()

HashJoinTable ExecHashTableCreate ( HashState state)

Definition at line 446 of file nodeHash.c.

447{
448 Hash *node;
449 HashJoinTable hashtable;
450 Plan *outerNode;
451 size_t space_allowed;
452 int nbuckets;
453 int nbatch;
454 double rows;
455 int num_skew_mcvs;
456 int log2_nbuckets;
457 MemoryContext oldcxt;
458
459 /*
460 * Get information about the size of the relation to be hashed (it's the
461 * "outer" subtree of this node, but the inner relation of the hashjoin).
462 * Compute the appropriate size of the hash table.
463 */
464 node = (Hash *) state->ps.plan;
465 outerNode = outerPlan(node);
466
467 /*
468 * If this is shared hash table with a partial plan, then we can't use
469 * outerNode->plan_rows to estimate its size. We need an estimate of the
470 * total number of rows across all copies of the partial plan.
471 */
472 rows = node->plan.parallel_aware ? node->rows_total : outerNode->plan_rows;
473
474 ExecChooseHashTableSize(rows, outerNode->plan_width,
475 OidIsValid(node->skewTable),
476 state->parallel_state != NULL,
477 state->parallel_state != NULL ?
478 state->parallel_state->nparticipants - 1 : 0,
479 &space_allowed,
480 &nbuckets, &nbatch, &num_skew_mcvs);
481
482 /* nbuckets must be a power of 2 */
483 log2_nbuckets = my_log2(nbuckets);
484 Assert(nbuckets == (1 << log2_nbuckets));
485
486 /*
487 * Initialize the hash table control block.
488 *
489 * The hashtable control block is just palloc'd from the executor's
490 * per-query memory context. Everything else should be kept inside the
491 * subsidiary hashCxt, batchCxt or spillCxt.
492 */
493 hashtable = palloc_object(HashJoinTableData);
494 hashtable->nbuckets = nbuckets;
495 hashtable->nbuckets_original = nbuckets;
496 hashtable->nbuckets_optimal = nbuckets;
497 hashtable->log2_nbuckets = log2_nbuckets;
498 hashtable->log2_nbuckets_optimal = log2_nbuckets;
499 hashtable->buckets.unshared = NULL;
500 hashtable->skewEnabled = false;
501 hashtable->skewBucket = NULL;
502 hashtable->skewBucketLen = 0;
503 hashtable->nSkewBuckets = 0;
504 hashtable->skewBucketNums = NULL;
505 hashtable->nbatch = nbatch;
506 hashtable->curbatch = 0;
507 hashtable->nbatch_original = nbatch;
508 hashtable->nbatch_outstart = nbatch;
509 hashtable->growEnabled = true;
510 hashtable->totalTuples = 0;
511 hashtable->partialTuples = 0;
512 hashtable->skewTuples = 0;
513 hashtable->innerBatchFile = NULL;
514 hashtable->outerBatchFile = NULL;
515 hashtable->spaceUsed = 0;
516 hashtable->spacePeak = 0;
517 hashtable->spaceAllowed = space_allowed;
518 hashtable->spaceUsedSkew = 0;
519 hashtable->spaceAllowedSkew =
520 hashtable->spaceAllowed * SKEW_HASH_MEM_PERCENT / 100;
521 hashtable->chunks = NULL;
522 hashtable->current_chunk = NULL;
523 hashtable->parallel_state = state->parallel_state;
524 hashtable->area = state->ps.state->es_query_dsa;
525 hashtable->batches = NULL;
526
527#ifdef HJDEBUG
528 printf("Hashjoin %p: initial nbatch = %d, nbuckets = %d\n",
529 hashtable, nbatch, nbuckets);
530#endif
531
532 /*
533 * Create temporary memory contexts in which to keep the hashtable working
534 * storage. See notes in executor/hashjoin.h.
535 */
537 "HashTableContext",
539
540 hashtable->batchCxt = AllocSetContextCreate(hashtable->hashCxt,
541 "HashBatchContext",
543
544 hashtable->spillCxt = AllocSetContextCreate(hashtable->hashCxt,
545 "HashSpillContext",
547
548 /* Allocate data that will live for the life of the hashjoin */
549
550 oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
551
552 if (nbatch > 1 && hashtable->parallel_state == NULL)
553 {
554 MemoryContext oldctx;
555
556 /*
557 * allocate and initialize the file arrays in hashCxt (not needed for
558 * parallel case which uses shared tuplestores instead of raw files)
559 */
560 oldctx = MemoryContextSwitchTo(hashtable->spillCxt);
561
562 hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch);
563 hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch);
564
565 MemoryContextSwitchTo(oldctx);
566
567 /* The files will not be opened until needed... */
568 /* ... but make sure we have temp tablespaces established for them */
570 }
571
572 MemoryContextSwitchTo(oldcxt);
573
574 if (hashtable->parallel_state)
575 {
576 ParallelHashJoinState *pstate = hashtable->parallel_state;
577 Barrier *build_barrier;
578
579 /*
580 * Attach to the build barrier. The corresponding detach operation is
581 * in ExecHashTableDetach. Note that we won't attach to the
582 * batch_barrier for batch 0 yet. We'll attach later and start it out
583 * in PHJ_BATCH_PROBE phase, because batch 0 is allocated up front and
584 * then loaded while hashing (the standard hybrid hash join
585 * algorithm), and we'll coordinate that using build_barrier.
586 */
587 build_barrier = &pstate->build_barrier;
588 BarrierAttach(build_barrier);
589
590 /*
591 * So far we have no idea whether there are any other participants,
592 * and if so, what phase they are working on. The only thing we care
593 * about at this point is whether someone has already created the
594 * SharedHashJoinBatch objects and the hash table for batch 0. One
595 * backend will be elected to do that now if necessary.
596 */
597 if (BarrierPhase(build_barrier) == PHJ_BUILD_ELECT &&
598 BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ELECT))
599 {
600 pstate->nbatch = nbatch;
601 pstate->space_allowed = space_allowed;
602 pstate->growth = PHJ_GROWTH_OK;
603
604 /* Set up the shared state for coordinating batches. */
605 ExecParallelHashJoinSetUpBatches(hashtable, nbatch);
606
607 /*
608 * Allocate batch 0's hash table up front so we can load it
609 * directly while hashing.
610 */
611 pstate->nbuckets = nbuckets;
612 ExecParallelHashTableAlloc(hashtable, 0);
613 }
614
615 /*
616 * The next Parallel Hash synchronization point is in
617 * MultiExecParallelHash(), which will progress it all the way to
618 * PHJ_BUILD_RUN. The caller must not return control from this
619 * executor node between now and then.
620 */
621 }
622 else
623 {
624 /*
625 * Prepare context for the first-scan space allocations; allocate the
626 * hashbucket array therein, and set each bucket "empty".
627 */
629
630 hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets);
631
632 /*
633 * Set up for skew optimization, if possible and there's a need for
634 * more than one batch. (In a one-batch join, there's no point in
635 * it.)
636 */
637 if (nbatch > 1)
638 ExecHashBuildSkewHash(state, hashtable, node, num_skew_mcvs);
639
640 MemoryContextSwitchTo(oldcxt);
641 }
642
643 return hashtable;
644}
void PrepareTempTablespaces(void)
Definition: tablespace.c:1331
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
#define OidIsValid(objectId)
Definition: c.h:746
int my_log2(long num)
Definition: dynahash.c:1794
#define palloc_object(type)
Definition: fe_memutils.h:74
#define palloc0_array(type, count)
Definition: fe_memutils.h:77
@ PHJ_GROWTH_OK
Definition: hashjoin.h:233
#define PHJ_BUILD_ELECT
Definition: hashjoin.h:269
MemoryContext CurrentMemoryContext
Definition: mcxt.c:143
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:160
static void ExecHashBuildSkewHash(HashState *hashstate, HashJoinTable hashtable, Hash *node, int mcvsToUse)
Definition: nodeHash.c:2397
static void ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
Definition: nodeHash.c:3118
void ExecParallelHashTableAlloc(HashJoinTable hashtable, int batchno)
Definition: nodeHash.c:3283
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
#define printf(...)
Definition: port.h:245
struct HashJoinTupleData ** unshared
Definition: hashjoin.h:311
HashMemoryChunk chunks
Definition: hashjoin.h:355
ParallelHashJoinBatchAccessor * batches
Definition: hashjoin.h:361
MemoryContext hashCxt
Definition: hashjoin.h:350
double totalTuples
Definition: hashjoin.h:330
double partialTuples
Definition: hashjoin.h:331
ParallelHashJoinState * parallel_state
Definition: hashjoin.h:360
MemoryContext spillCxt
Definition: hashjoin.h:352
HashMemoryChunk current_chunk
Definition: hashjoin.h:358
Size spaceAllowedSkew
Definition: hashjoin.h:348
int * skewBucketNums
Definition: hashjoin.h:320
BufFile ** innerBatchFile
Definition: hashjoin.h:341
int log2_nbuckets_optimal
Definition: hashjoin.h:305
dsa_area * area
Definition: hashjoin.h:359
BufFile ** outerBatchFile
Definition: hashjoin.h:342
MemoryContext batchCxt
Definition: hashjoin.h:351
union HashJoinTableData::@108 buckets
double skewTuples
Definition: hashjoin.h:332
Oid skewTable
Definition: plannodes.h:1351
Cardinality rows_total
Definition: plannodes.h:1358
Plan plan
Definition: plannodes.h:1342
ParallelHashGrowth growth
Definition: hashjoin.h:253
bool parallel_aware
Definition: plannodes.h:183
int plan_width
Definition: plannodes.h:177
Cardinality plan_rows
Definition: plannodes.h:175
Definition: regguts.h:323

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, HashJoinTableData::area, Assert(), BarrierArriveAndWait(), BarrierAttach(), BarrierPhase(), HashJoinTableData::batchCxt, HashJoinTableData::batches, HashJoinTableData::buckets, ParallelHashJoinState::build_barrier, HashJoinTableData::chunks, HashJoinTableData::curbatch, HashJoinTableData::current_chunk, CurrentMemoryContext, ExecChooseHashTableSize(), ExecHashBuildSkewHash(), ExecParallelHashJoinSetUpBatches(), ExecParallelHashTableAlloc(), HashJoinTableData::growEnabled, ParallelHashJoinState::growth, HashJoinTableData::hashCxt, HashJoinTableData::innerBatchFile, 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, HashJoinTableData::outerBatchFile, outerPlan, palloc0_array, palloc_object, Plan::parallel_aware, HashJoinTableData::parallel_state, HashJoinTableData::partialTuples, PHJ_BUILD_ELECT, 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::spillCxt, HashJoinTableData::totalTuples, and HashJoinTableData::unshared.

Referenced by ExecHashJoinImpl().

◆ ExecHashTableDestroy()

void ExecHashTableDestroy ( HashJoinTable  hashtable)

Definition at line 950 of file nodeHash.c.

951{
952 int i;
953
954 /*
955 * Make sure all the temp files are closed. We skip batch 0, since it
956 * can't have any temp files (and the arrays might not even exist if
957 * nbatch is only 1). Parallel hash joins don't use these files.
958 */
959 if (hashtable->innerBatchFile != NULL)
960 {
961 for (i = 1; i < hashtable->nbatch; i++)
962 {
963 if (hashtable->innerBatchFile[i])
964 BufFileClose(hashtable->innerBatchFile[i]);
965 if (hashtable->outerBatchFile[i])
966 BufFileClose(hashtable->outerBatchFile[i]);
967 }
968 }
969
970 /* Release working memory (batchCxt is a child, so it goes away too) */
971 MemoryContextDelete(hashtable->hashCxt);
972
973 /* And drop the control block */
974 pfree(hashtable);
975}
void BufFileClose(BufFile *file)
Definition: buffile.c:412
int i
Definition: isn.c:74
void pfree(void *pointer)
Definition: mcxt.c:1524
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:454

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 3395 of file nodeHash.c.

3396{
3397 ParallelHashJoinState *pstate = hashtable->parallel_state;
3398
3399 /*
3400 * If we're involved in a parallel query, we must either have gotten all
3401 * the way to PHJ_BUILD_RUN, or joined too late and be in PHJ_BUILD_FREE.
3402 */
3403 Assert(!pstate ||
3405
3406 if (pstate && BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_RUN)
3407 {
3408 int i;
3409
3410 /* Make sure any temporary files are closed. */
3411 if (hashtable->batches)
3412 {
3413 for (i = 0; i < hashtable->nbatch; ++i)
3414 {
3415 sts_end_write(hashtable->batches[i].inner_tuples);
3416 sts_end_write(hashtable->batches[i].outer_tuples);
3419 }
3420 }
3421
3422 /* If we're last to detach, clean up shared memory. */
3424 {
3425 /*
3426 * Late joining processes will see this state and give up
3427 * immediately.
3428 */
3430
3431 if (DsaPointerIsValid(pstate->batches))
3432 {
3433 dsa_free(hashtable->area, pstate->batches);
3434 pstate->batches = InvalidDsaPointer;
3435 }
3436 }
3437 }
3438 hashtable->parallel_state = NULL;
3439}
bool BarrierArriveAndDetach(Barrier *barrier)
Definition: barrier.c:203
void dsa_free(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:826
#define InvalidDsaPointer
Definition: dsa.h:78
#define DsaPointerIsValid(x)
Definition: dsa.h:106
#define PHJ_BUILD_FREE
Definition: hashjoin.h:274
#define PHJ_BUILD_RUN
Definition: hashjoin.h:273
void sts_end_write(SharedTuplestoreAccessor *accessor)
void sts_end_parallel_scan(SharedTuplestoreAccessor *accessor)
SharedTuplestoreAccessor * outer_tuples
Definition: hashjoin.h:221
SharedTuplestoreAccessor * inner_tuples
Definition: hashjoin.h:220
dsa_pointer batches
Definition: hashjoin.h:248

References HashJoinTableData::area, Assert(), BarrierArriveAndDetach(), BarrierPhase(), ParallelHashJoinState::batches, HashJoinTableData::batches, ParallelHashJoinState::build_barrier, dsa_free(), DsaPointerIsValid, i, ParallelHashJoinBatchAccessor::inner_tuples, InvalidDsaPointer, HashJoinTableData::nbatch, ParallelHashJoinBatchAccessor::outer_tuples, HashJoinTableData::parallel_state, PHJ_BUILD_FREE, PHJ_BUILD_RUN, sts_end_parallel_scan(), and sts_end_write().

Referenced by ExecHashJoinReInitializeDSM(), and ExecShutdownHashJoin().

◆ ExecHashTableDetachBatch()

void ExecHashTableDetachBatch ( HashJoinTable  hashtable)

Definition at line 3303 of file nodeHash.c.

3304{
3305 if (hashtable->parallel_state != NULL &&
3306 hashtable->curbatch >= 0)
3307 {
3308 int curbatch = hashtable->curbatch;
3309 ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
3310 bool attached = true;
3311
3312 /* Make sure any temporary files are closed. */
3313 sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
3314 sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
3315
3316 /* After attaching we always get at least to PHJ_BATCH_PROBE. */
3319
3320 /*
3321 * If we're abandoning the PHJ_BATCH_PROBE phase early without having
3322 * reached the end of it, it means the plan doesn't want any more
3323 * tuples, and it is happy to abandon any tuples buffered in this
3324 * process's subplans. For correctness, we can't allow any process to
3325 * execute the PHJ_BATCH_SCAN phase, because we will never have the
3326 * complete set of match bits. Therefore we skip emitting unmatched
3327 * tuples in all backends (if this is a full/right join), as if those
3328 * tuples were all due to be emitted by this process and it has
3329 * abandoned them too.
3330 */
3331 if (BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE &&
3332 !hashtable->batches[curbatch].outer_eof)
3333 {
3334 /*
3335 * This flag may be written to by multiple backends during
3336 * PHJ_BATCH_PROBE phase, but will only be read in PHJ_BATCH_SCAN
3337 * phase so requires no extra locking.
3338 */
3339 batch->skip_unmatched = true;
3340 }
3341
3342 /*
3343 * Even if we aren't doing a full/right outer join, we'll step through
3344 * the PHJ_BATCH_SCAN phase just to maintain the invariant that
3345 * freeing happens in PHJ_BATCH_FREE, but that'll be wait-free.
3346 */
3349 if (attached && BarrierArriveAndDetach(&batch->batch_barrier))
3350 {
3351 /*
3352 * We are not longer attached to the batch barrier, but we're the
3353 * process that was chosen to free resources and it's safe to
3354 * assert the current phase. The ParallelHashJoinBatch can't go
3355 * away underneath us while we are attached to the build barrier,
3356 * making this access safe.
3357 */
3359
3360 /* Free shared chunks and buckets. */
3361 while (DsaPointerIsValid(batch->chunks))
3362 {
3363 HashMemoryChunk chunk =
3364 dsa_get_address(hashtable->area, batch->chunks);
3365 dsa_pointer next = chunk->next.shared;
3366
3367 dsa_free(hashtable->area, batch->chunks);
3368 batch->chunks = next;
3369 }
3370 if (DsaPointerIsValid(batch->buckets))
3371 {
3372 dsa_free(hashtable->area, batch->buckets);
3373 batch->buckets = InvalidDsaPointer;
3374 }
3375 }
3376
3377 /*
3378 * Track the largest batch we've been attached to. Though each
3379 * backend might see a different subset of batches, explain.c will
3380 * scan the results from all backends to find the largest value.
3381 */
3382 hashtable->spacePeak =
3383 Max(hashtable->spacePeak,
3384 batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
3385
3386 /* Remember that we are not attached to a batch. */
3387 hashtable->curbatch = -1;
3388 }
3389}
bool BarrierArriveAndDetachExceptLast(Barrier *barrier)
Definition: barrier.c:213
static int32 next
Definition: blutils.c:224
void * dsa_get_address(dsa_area *area, dsa_pointer dp)
Definition: dsa.c:942
uint64 dsa_pointer
Definition: dsa.h:62
#define PHJ_BATCH_SCAN
Definition: hashjoin.h:281
#define PHJ_BATCH_PROBE
Definition: hashjoin.h:280
#define PHJ_BATCH_FREE
Definition: hashjoin.h:282
union HashMemoryChunkData::@107 next
dsa_pointer shared
Definition: hashjoin.h:138
ParallelHashJoinBatch * shared
Definition: hashjoin.h:209
dsa_pointer chunks
Definition: hashjoin.h:167
dsa_pointer buckets
Definition: hashjoin.h:164

References HashJoinTableData::area, Assert(), BarrierArriveAndDetach(), BarrierArriveAndDetachExceptLast(), 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_eof, ParallelHashJoinBatchAccessor::outer_tuples, HashJoinTableData::parallel_state, PHJ_BATCH_FREE, PHJ_BATCH_PROBE, PHJ_BATCH_SCAN, HashMemoryChunkData::shared, ParallelHashJoinBatchAccessor::shared, ParallelHashJoinBatch::size, ParallelHashJoinBatch::skip_unmatched, HashJoinTableData::spacePeak, and sts_end_parallel_scan().

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

◆ ExecHashTableInsert()

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

Definition at line 1743 of file nodeHash.c.

1746{
1747 bool shouldFree;
1748 MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1749 int bucketno;
1750 int batchno;
1751
1752 ExecHashGetBucketAndBatch(hashtable, hashvalue,
1753 &bucketno, &batchno);
1754
1755 /*
1756 * decide whether to put the tuple in the hash table or a temp file
1757 */
1758 if (batchno == hashtable->curbatch)
1759 {
1760 /*
1761 * put the tuple in hash table
1762 */
1763 HashJoinTuple hashTuple;
1764 int hashTupleSize;
1765 double ntuples = (hashtable->totalTuples - hashtable->skewTuples);
1766
1767 /* Create the HashJoinTuple */
1768 hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
1769 hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
1770
1771 hashTuple->hashvalue = hashvalue;
1772 memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1773
1774 /*
1775 * We always reset the tuple-matched flag on insertion. This is okay
1776 * even when reloading a tuple from a batch file, since the tuple
1777 * could not possibly have been matched to an outer tuple before it
1778 * went into the batch file.
1779 */
1781
1782 /* Push it onto the front of the bucket's list */
1783 hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
1784 hashtable->buckets.unshared[bucketno] = hashTuple;
1785
1786 /*
1787 * Increase the (optimal) number of buckets if we just exceeded the
1788 * NTUP_PER_BUCKET threshold, but only when there's still a single
1789 * batch.
1790 */
1791 if (hashtable->nbatch == 1 &&
1792 ntuples > (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
1793 {
1794 /* Guard against integer overflow and alloc size overflow */
1795 if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
1796 hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
1797 {
1798 hashtable->nbuckets_optimal *= 2;
1799 hashtable->log2_nbuckets_optimal += 1;
1800 }
1801 }
1802
1803 /* Account for space used, and back off if we've used too much */
1804 hashtable->spaceUsed += hashTupleSize;
1805 if (hashtable->spaceUsed > hashtable->spacePeak)
1806 hashtable->spacePeak = hashtable->spaceUsed;
1807 if (hashtable->spaceUsed +
1808 hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
1809 > hashtable->spaceAllowed)
1810 ExecHashIncreaseNumBatches(hashtable);
1811 }
1812 else
1813 {
1814 /*
1815 * put the tuple into a temp file for later batches
1816 */
1817 Assert(batchno > hashtable->curbatch);
1819 hashvalue,
1820 &hashtable->innerBatchFile[batchno],
1821 hashtable);
1822 }
1823
1824 if (shouldFree)
1826}
MinimalTuple ExecFetchSlotMinimalTuple(TupleTableSlot *slot, bool *shouldFree)
Definition: execTuples.c:1879
#define HJTUPLE_MINTUPLE(hjtup)
Definition: hashjoin.h:91
void heap_free_minimal_tuple(MinimalTuple mtup)
Definition: heaptuple.c:1524
static void HeapTupleHeaderClearMatch(MinimalTupleData *tup)
Definition: htup_details.h:718
static void * dense_alloc(HashJoinTable hashtable, Size size)
Definition: nodeHash.c:2890
static void ExecHashIncreaseNumBatches(HashJoinTable hashtable)
Definition: nodeHash.c:1024
void ExecHashGetBucketAndBatch(HashJoinTable hashtable, uint32 hashvalue, int *bucketno, int *batchno)
Definition: nodeHash.c:1954
void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, BufFile **fileptr, HashJoinTable hashtable)
union HashJoinTupleData::@106 next
uint32 hashvalue
Definition: hashjoin.h:86
struct HashJoinTupleData * unshared
Definition: hashjoin.h:83

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 2321 of file nodeHash.c.

2322{
2323 MemoryContext oldcxt;
2324 int nbuckets = hashtable->nbuckets;
2325
2326 /*
2327 * Release all the hash buckets and tuples acquired in the prior pass, and
2328 * reinitialize the context for a new pass.
2329 */
2330 MemoryContextReset(hashtable->batchCxt);
2331 oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
2332
2333 /* Reallocate and reinitialize the hash bucket headers. */
2334 hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets);
2335
2336 hashtable->spaceUsed = 0;
2337
2338 MemoryContextSwitchTo(oldcxt);
2339
2340 /* Forget the chunks (the memory was freed by the context reset above). */
2341 hashtable->chunks = NULL;
2342}
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:383

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 2349 of file nodeHash.c.

2350{
2351 HashJoinTuple tuple;
2352 int i;
2353
2354 /* Reset all flags in the main table ... */
2355 for (i = 0; i < hashtable->nbuckets; i++)
2356 {
2357 for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
2358 tuple = tuple->next.unshared)
2360 }
2361
2362 /* ... and the same for the skew buckets, if any */
2363 for (i = 0; i < hashtable->nSkewBuckets; i++)
2364 {
2365 int j = hashtable->skewBucketNums[i];
2366 HashSkewBucket *skewBucket = hashtable->skewBucket[j];
2367
2368 for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
2370 }
2371}
int j
Definition: isn.c:75
HashJoinTuple tuples
Definition: hashjoin.h:116

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 370 of file nodeHash.c.

371{
372 HashState *hashstate;
373
374 /* check for unsupported flags */
376
377 /*
378 * create state structure
379 */
380 hashstate = makeNode(HashState);
381 hashstate->ps.plan = (Plan *) node;
382 hashstate->ps.state = estate;
383 hashstate->ps.ExecProcNode = ExecHash;
384 /* delay building hashtable until ExecHashTableCreate() in executor run */
385 hashstate->hashtable = NULL;
386
387 /*
388 * Miscellaneous initialization
389 *
390 * create expression context for node
391 */
392 ExecAssignExprContext(estate, &hashstate->ps);
393
394 /*
395 * initialize child nodes
396 */
397 outerPlanState(hashstate) = ExecInitNode(outerPlan(node), estate, eflags);
398
399 /*
400 * initialize our result slot and type. No need to build projection
401 * because this node doesn't do projections.
402 */
404 hashstate->ps.ps_ProjInfo = NULL;
405
406 Assert(node->plan.qual == NIL);
407
408 /*
409 * Delay initialization of hash_expr until ExecInitHashJoin(). We cannot
410 * build the ExprState here as we don't yet know the join type we're going
411 * to be hashing values for and we need to know that before calling
412 * ExecBuildHash32Expr as the keep_nulls parameter depends on the join
413 * type.
414 */
415 hashstate->hash_expr = NULL;
416
417 return hashstate;
418}
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:142
void ExecInitResultTupleSlotTL(PlanState *planstate, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1986
const TupleTableSlotOps TTSOpsMinimalTuple
Definition: execTuples.c:86
void ExecAssignExprContext(EState *estate, PlanState *planstate)
Definition: execUtils.c:487
#define EXEC_FLAG_BACKWARD
Definition: executor.h:69
#define EXEC_FLAG_MARK
Definition: executor.h:70
static TupleTableSlot * ExecHash(PlanState *pstate)
Definition: nodeHash.c:91
#define makeNode(_type_)
Definition: nodes.h:157
#define NIL
Definition: pg_list.h:68
HashJoinTable hashtable
Definition: execnodes.h:2805
ExprState * hash_expr
Definition: execnodes.h:2806
EState * state
Definition: execnodes.h:1155
ProjectionInfo * ps_ProjInfo
Definition: execnodes.h:1193
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:1159
List * qual
Definition: plannodes.h:201

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

Referenced by ExecInitNode().

◆ ExecParallelHashTableAlloc()

void ExecParallelHashTableAlloc ( HashJoinTable  hashtable,
int  batchno 
)

Definition at line 3283 of file nodeHash.c.

3284{
3285 ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
3286 dsa_pointer_atomic *buckets;
3287 int nbuckets = hashtable->parallel_state->nbuckets;
3288 int i;
3289
3290 batch->buckets =
3291 dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
3292 buckets = (dsa_pointer_atomic *)
3293 dsa_get_address(hashtable->area, batch->buckets);
3294 for (i = 0; i < nbuckets; ++i)
3296}
#define dsa_pointer_atomic_init
Definition: dsa.h:64
#define dsa_allocate(area, size)
Definition: dsa.h:109

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 1833 of file nodeHash.c.

1836{
1837 bool shouldFree;
1838 MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1839 dsa_pointer shared;
1840 int bucketno;
1841 int batchno;
1842
1843retry:
1844 ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1845
1846 if (batchno == 0)
1847 {
1848 HashJoinTuple hashTuple;
1849
1850 /* Try to load it into memory. */
1853 hashTuple = ExecParallelHashTupleAlloc(hashtable,
1854 HJTUPLE_OVERHEAD + tuple->t_len,
1855 &shared);
1856 if (hashTuple == NULL)
1857 goto retry;
1858
1859 /* Store the hash value in the HashJoinTuple header. */
1860 hashTuple->hashvalue = hashvalue;
1861 memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1863
1864 /* Push it onto the front of the bucket's list */
1865 ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1866 hashTuple, shared);
1867 }
1868 else
1869 {
1870 size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
1871
1872 Assert(batchno > 0);
1873
1874 /* Try to preallocate space in the batch if necessary. */
1875 if (hashtable->batches[batchno].preallocated < tuple_size)
1876 {
1877 if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size))
1878 goto retry;
1879 }
1880
1881 Assert(hashtable->batches[batchno].preallocated >= tuple_size);
1882 hashtable->batches[batchno].preallocated -= tuple_size;
1883 sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue,
1884 tuple);
1885 }
1886 ++hashtable->batches[batchno].ntuples;
1887
1888 if (shouldFree)
1890}
#define PHJ_BUILD_HASH_INNER
Definition: hashjoin.h:271
static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size)
Definition: nodeHash.c:3555
static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size, dsa_pointer *shared)
Definition: nodeHash.c:2970
static void ExecParallelHashPushTuple(dsa_pointer_atomic *head, HashJoinTuple tuple, dsa_pointer tuple_shared)
Definition: nodeHash.c:3475
void sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data, MinimalTuple tuple)
dsa_pointer_atomic * shared
Definition: hashjoin.h:313

References Assert(), BarrierPhase(), HashJoinTableData::batches, HashJoinTableData::buckets, ParallelHashJoinState::build_barrier, ExecFetchSlotMinimalTuple(), ExecHashGetBucketAndBatch(), ExecParallelHashPushTuple(), ExecParallelHashTupleAlloc(), ExecParallelHashTuplePrealloc(), HashJoinTupleData::hashvalue, heap_free_minimal_tuple(), HeapTupleHeaderClearMatch(), HJTUPLE_MINTUPLE, HJTUPLE_OVERHEAD, ParallelHashJoinBatchAccessor::inner_tuples, MAXALIGN, ParallelHashJoinBatchAccessor::ntuples, HashJoinTableData::parallel_state, PHJ_BUILD_HASH_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 1899 of file nodeHash.c.

1902{
1903 bool shouldFree;
1904 MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
1905 HashJoinTuple hashTuple;
1906 dsa_pointer shared;
1907 int batchno;
1908 int bucketno;
1909
1910 ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
1911 Assert(batchno == hashtable->curbatch);
1912 hashTuple = ExecParallelHashTupleAlloc(hashtable,
1913 HJTUPLE_OVERHEAD + tuple->t_len,
1914 &shared);
1915 hashTuple->hashvalue = hashvalue;
1916 memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
1918 ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
1919 hashTuple, shared);
1920
1921 if (shouldFree)
1923}

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 
)

◆ ExecParallelPrepHashTableForUnmatched()

bool ExecParallelPrepHashTableForUnmatched ( HashJoinState hjstate)

Definition at line 2119 of file nodeHash.c.

2120{
2121 HashJoinTable hashtable = hjstate->hj_HashTable;
2122 int curbatch = hashtable->curbatch;
2123 ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
2124
2126
2127 /*
2128 * It would not be deadlock-free to wait on the batch barrier, because it
2129 * is in PHJ_BATCH_PROBE phase, and thus processes attached to it have
2130 * already emitted tuples. Therefore, we'll hold a wait-free election:
2131 * only one process can continue to the next phase, and all others detach
2132 * from this batch. They can still go any work on other batches, if there
2133 * are any.
2134 */
2136 {
2137 /* This process considers the batch to be done. */
2138 hashtable->batches[hashtable->curbatch].done = true;
2139
2140 /* Make sure any temporary files are closed. */
2141 sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
2142 sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
2143
2144 /*
2145 * Track largest batch we've seen, which would normally happen in
2146 * ExecHashTableDetachBatch().
2147 */
2148 hashtable->spacePeak =
2149 Max(hashtable->spacePeak,
2150 batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
2151 hashtable->curbatch = -1;
2152 return false;
2153 }
2154
2155 /* Now we are alone with this batch. */
2157
2158 /*
2159 * Has another process decided to give up early and command all processes
2160 * to skip the unmatched scan?
2161 */
2162 if (batch->skip_unmatched)
2163 {
2164 hashtable->batches[hashtable->curbatch].done = true;
2165 ExecHashTableDetachBatch(hashtable);
2166 return false;
2167 }
2168
2169 /* Now prepare the process local state, just as for non-parallel join. */
2171
2172 return true;
2173}
void ExecHashTableDetachBatch(HashJoinTable hashtable)
Definition: nodeHash.c:3303
void ExecPrepHashTableForUnmatched(HashJoinState *hjstate)
Definition: nodeHash.c:2098
HashJoinTable hj_HashTable
Definition: execnodes.h:2252

References Assert(), BarrierArriveAndDetachExceptLast(), BarrierPhase(), ParallelHashJoinBatch::batch_barrier, HashJoinTableData::batches, HashJoinTableData::curbatch, ParallelHashJoinBatchAccessor::done, ExecHashTableDetachBatch(), ExecPrepHashTableForUnmatched(), HashJoinState::hj_HashTable, ParallelHashJoinBatchAccessor::inner_tuples, Max, HashJoinTableData::nbuckets, ParallelHashJoinBatchAccessor::outer_tuples, PHJ_BATCH_PROBE, PHJ_BATCH_SCAN, ParallelHashJoinBatchAccessor::shared, ParallelHashJoinBatch::size, ParallelHashJoinBatch::skip_unmatched, HashJoinTableData::spacePeak, and sts_end_parallel_scan().

Referenced by ExecHashJoinImpl().

◆ ExecParallelScanHashBucket()

bool ExecParallelScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 2047 of file nodeHash.c.

2049{
2050 ExprState *hjclauses = hjstate->hashclauses;
2051 HashJoinTable hashtable = hjstate->hj_HashTable;
2052 HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2053 uint32 hashvalue = hjstate->hj_CurHashValue;
2054
2055 /*
2056 * hj_CurTuple is the address of the tuple last returned from the current
2057 * bucket, or NULL if it's time to start scanning a new bucket.
2058 */
2059 if (hashTuple != NULL)
2060 hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2061 else
2062 hashTuple = ExecParallelHashFirstTuple(hashtable,
2063 hjstate->hj_CurBucketNo);
2064
2065 while (hashTuple != NULL)
2066 {
2067 if (hashTuple->hashvalue == hashvalue)
2068 {
2069 TupleTableSlot *inntuple;
2070
2071 /* insert hashtable's tuple into exec slot so ExecQual sees it */
2072 inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2073 hjstate->hj_HashTupleSlot,
2074 false); /* do not pfree */
2075 econtext->ecxt_innertuple = inntuple;
2076
2077 if (ExecQualAndReset(hjclauses, econtext))
2078 {
2079 hjstate->hj_CurTuple = hashTuple;
2080 return true;
2081 }
2082 }
2083
2084 hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2085 }
2086
2087 /*
2088 * no match
2089 */
2090 return false;
2091}
TupleTableSlot * ExecStoreMinimalTuple(MinimalTuple mtup, TupleTableSlot *slot, bool shouldFree)
Definition: execTuples.c:1633
static bool ExecQualAndReset(ExprState *state, ExprContext *econtext)
Definition: executor.h:527
static HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable hashtable, int bucketno)
Definition: nodeHash.c:3445
static HashJoinTuple ExecParallelHashNextTuple(HashJoinTable hashtable, HashJoinTuple tuple)
Definition: nodeHash.c:3461
TupleTableSlot * ecxt_innertuple
Definition: execnodes.h:270
HashJoinTuple hj_CurTuple
Definition: execnodes.h:2256
ExprState * hashclauses
Definition: execnodes.h:2250
uint32 hj_CurHashValue
Definition: execnodes.h:2253
int hj_CurBucketNo
Definition: execnodes.h:2254
TupleTableSlot * hj_HashTupleSlot
Definition: execnodes.h:2258

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

◆ ExecParallelScanHashTableForUnmatched()

bool ExecParallelScanHashTableForUnmatched ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 2258 of file nodeHash.c.

2260{
2261 HashJoinTable hashtable = hjstate->hj_HashTable;
2262 HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2263
2264 for (;;)
2265 {
2266 /*
2267 * hj_CurTuple is the address of the tuple last returned from the
2268 * current bucket, or NULL if it's time to start scanning a new
2269 * bucket.
2270 */
2271 if (hashTuple != NULL)
2272 hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2273 else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
2274 hashTuple = ExecParallelHashFirstTuple(hashtable,
2275 hjstate->hj_CurBucketNo++);
2276 else
2277 break; /* finished all buckets */
2278
2279 while (hashTuple != NULL)
2280 {
2282 {
2283 TupleTableSlot *inntuple;
2284
2285 /* insert hashtable's tuple into exec slot */
2286 inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2287 hjstate->hj_HashTupleSlot,
2288 false); /* do not pfree */
2289 econtext->ecxt_innertuple = inntuple;
2290
2291 /*
2292 * Reset temp memory each time; although this function doesn't
2293 * do any qual eval, the caller will, so let's keep it
2294 * parallel to ExecScanHashBucket.
2295 */
2296 ResetExprContext(econtext);
2297
2298 hjstate->hj_CurTuple = hashTuple;
2299 return true;
2300 }
2301
2302 hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
2303 }
2304
2305 /* allow this loop to be cancellable */
2307 }
2308
2309 /*
2310 * no more unmatched tuples
2311 */
2312 return false;
2313}
#define ResetExprContext(econtext)
Definition: executor.h:631
static bool HeapTupleHeaderHasMatch(const MinimalTupleData *tup)
Definition: htup_details.h:706
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122

References CHECK_FOR_INTERRUPTS, ExprContext::ecxt_innertuple, ExecParallelHashFirstTuple(), ExecParallelHashNextTuple(), ExecStoreMinimalTuple(), HeapTupleHeaderHasMatch(), HashJoinState::hj_CurBucketNo, HashJoinState::hj_CurTuple, HashJoinState::hj_HashTable, HashJoinState::hj_HashTupleSlot, HJTUPLE_MINTUPLE, HashJoinTableData::nbuckets, and ResetExprContext.

Referenced by ExecHashJoinImpl().

◆ ExecPrepHashTableForUnmatched()

void ExecPrepHashTableForUnmatched ( HashJoinState hjstate)

Definition at line 2098 of file nodeHash.c.

2099{
2100 /*----------
2101 * During this scan we use the HashJoinState fields as follows:
2102 *
2103 * hj_CurBucketNo: next regular bucket to scan
2104 * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
2105 * hj_CurTuple: last tuple returned, or NULL to start next bucket
2106 *----------
2107 */
2108 hjstate->hj_CurBucketNo = 0;
2109 hjstate->hj_CurSkewBucketNo = 0;
2110 hjstate->hj_CurTuple = NULL;
2111}
int hj_CurSkewBucketNo
Definition: execnodes.h:2255

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

Referenced by ExecHashJoinImpl(), and ExecParallelPrepHashTableForUnmatched().

◆ ExecReScanHash()

void ExecReScanHash ( HashState node)

Definition at line 2375 of file nodeHash.c.

2376{
2378
2379 /*
2380 * if chgParam of subnode is not null then plan will be re-scanned by
2381 * first ExecProcNode.
2382 */
2383 if (outerPlan->chgParam == NULL)
2385}
void ExecReScan(PlanState *node)
Definition: execAmi.c:77

References ExecReScan(), outerPlan, and outerPlanState.

Referenced by ExecReScan().

◆ ExecScanHashBucket()

bool ExecScanHashBucket ( HashJoinState hjstate,
ExprContext econtext 
)

Definition at line 1986 of file nodeHash.c.

1988{
1989 ExprState *hjclauses = hjstate->hashclauses;
1990 HashJoinTable hashtable = hjstate->hj_HashTable;
1991 HashJoinTuple hashTuple = hjstate->hj_CurTuple;
1992 uint32 hashvalue = hjstate->hj_CurHashValue;
1993
1994 /*
1995 * hj_CurTuple is the address of the tuple last returned from the current
1996 * bucket, or NULL if it's time to start scanning a new bucket.
1997 *
1998 * If the tuple hashed to a skew bucket then scan the skew bucket
1999 * otherwise scan the standard hashtable bucket.
2000 */
2001 if (hashTuple != NULL)
2002 hashTuple = hashTuple->next.unshared;
2003 else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
2004 hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
2005 else
2006 hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
2007
2008 while (hashTuple != NULL)
2009 {
2010 if (hashTuple->hashvalue == hashvalue)
2011 {
2012 TupleTableSlot *inntuple;
2013
2014 /* insert hashtable's tuple into exec slot so ExecQual sees it */
2015 inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2016 hjstate->hj_HashTupleSlot,
2017 false); /* do not pfree */
2018 econtext->ecxt_innertuple = inntuple;
2019
2020 if (ExecQualAndReset(hjclauses, econtext))
2021 {
2022 hjstate->hj_CurTuple = hashTuple;
2023 return true;
2024 }
2025 }
2026
2027 hashTuple = hashTuple->next.unshared;
2028 }
2029
2030 /*
2031 * no match
2032 */
2033 return false;
2034}

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 2184 of file nodeHash.c.

2185{
2186 HashJoinTable hashtable = hjstate->hj_HashTable;
2187 HashJoinTuple hashTuple = hjstate->hj_CurTuple;
2188
2189 for (;;)
2190 {
2191 /*
2192 * hj_CurTuple is the address of the tuple last returned from the
2193 * current bucket, or NULL if it's time to start scanning a new
2194 * bucket.
2195 */
2196 if (hashTuple != NULL)
2197 hashTuple = hashTuple->next.unshared;
2198 else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
2199 {
2200 hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
2201 hjstate->hj_CurBucketNo++;
2202 }
2203 else if (hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
2204 {
2205 int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
2206
2207 hashTuple = hashtable->skewBucket[j]->tuples;
2208 hjstate->hj_CurSkewBucketNo++;
2209 }
2210 else
2211 break; /* finished all buckets */
2212
2213 while (hashTuple != NULL)
2214 {
2216 {
2217 TupleTableSlot *inntuple;
2218
2219 /* insert hashtable's tuple into exec slot */
2220 inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
2221 hjstate->hj_HashTupleSlot,
2222 false); /* do not pfree */
2223 econtext->ecxt_innertuple = inntuple;
2224
2225 /*
2226 * Reset temp memory each time; although this function doesn't
2227 * do any qual eval, the caller will, so let's keep it
2228 * parallel to ExecScanHashBucket.
2229 */
2230 ResetExprContext(econtext);
2231
2232 hjstate->hj_CurTuple = hashTuple;
2233 return true;
2234 }
2235
2236 hashTuple = hashTuple->next.unshared;
2237 }
2238
2239 /* allow this loop to be cancellable */
2241 }
2242
2243 /*
2244 * no more unmatched tuples
2245 */
2246 return false;
2247}

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 2825 of file nodeHash.c.

2826{
2827 /* Allocate save space if EXPLAIN'ing and we didn't do so already */
2828 if (node->ps.instrument && !node->hinstrument)
2830 /* Now accumulate data for the current (final) hash table */
2831 if (node->hinstrument && node->hashtable)
2833}
#define palloc0_object(type)
Definition: fe_memutils.h:75
void ExecHashAccumInstrumentation(HashInstrumentation *instrument, HashJoinTable hashtable)
Definition: nodeHash.c:2871

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

Referenced by ExecShutdownNode_walker().

◆ MultiExecHash()

Node * MultiExecHash ( HashState node)

Definition at line 105 of file nodeHash.c.

106{
107 /* must provide our own instrumentation support */
108 if (node->ps.instrument)
110
111 if (node->parallel_state != NULL)
113 else
115
116 /* must provide our own instrumentation support */
117 if (node->ps.instrument)
119
120 /*
121 * We do not return the hash table directly because it's not a subtype of
122 * Node, and so would violate the MultiExecProcNode API. Instead, our
123 * parent Hashjoin node is expected to know how to fish it out of our node
124 * state. Ugly but not really worth cleaning up, since Hashjoin knows
125 * quite a bit more about Hash besides that.
126 */
127 return NULL;
128}
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:219
static void MultiExecPrivateHash(HashState *node)
Definition: nodeHash.c:138
struct ParallelHashJoinState * parallel_state
Definition: execnodes.h:2827

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

Referenced by MultiExecProcNode().