#include "postgres.h"
#include "access/htup_details.h"
#include "access/parallel.h"
#include "executor/executor.h"
#include "executor/hashjoin.h"
#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "miscadmin.h"
#include "utils/lsyscache.h"
#include "utils/sharedtuplestore.h"
#include "utils/wait_event.h"

Include dependency graph for nodeHashjoin.c:

Macros
#define	HJ_BUILD_HASHTABLE 1

#define	HJ_NEED_NEW_OUTER 2

#define	HJ_SCAN_BUCKET 3

#define	HJ_FILL_OUTER_TUPLE 4

#define	HJ_FILL_INNER_TUPLES 5

#define	HJ_NEED_NEW_BATCH 6

#define	HJ_FILL_OUTER(hjstate) ((hjstate)->hj_NullInnerTupleSlot != NULL)

#define	HJ_FILL_INNER(hjstate) ((hjstate)->hj_NullOuterTupleSlot != NULL)

Functions
static TupleTableSlot *	ExecHashJoinOuterGetTuple (PlanState outerNode, HashJoinState hjstate, uint32 *hashvalue)

static TupleTableSlot *	ExecParallelHashJoinOuterGetTuple (PlanState outerNode, HashJoinState hjstate, uint32 *hashvalue)

static TupleTableSlot *	ExecHashJoinGetSavedTuple (HashJoinState hjstate, BufFile file, uint32 hashvalue, TupleTableSlot tupleSlot)

static bool	ExecHashJoinNewBatch (HashJoinState *hjstate)

static bool	ExecParallelHashJoinNewBatch (HashJoinState *hjstate)

static void	ExecParallelHashJoinPartitionOuter (HashJoinState *hjstate)

static pg_attribute_always_inline TupleTableSlot *	ExecHashJoinImpl (PlanState *pstate, bool parallel)

static TupleTableSlot *	ExecHashJoin (PlanState *pstate)

static TupleTableSlot *	ExecParallelHashJoin (PlanState *pstate)

HashJoinState *	ExecInitHashJoin (HashJoin node, EState estate, int eflags)

void	ExecEndHashJoin (HashJoinState *node)

void	ExecHashJoinSaveTuple (MinimalTuple tuple, uint32 hashvalue, BufFile **fileptr, HashJoinTable hashtable)

void	ExecReScanHashJoin (HashJoinState *node)

void	ExecShutdownHashJoin (HashJoinState *node)

void	ExecHashJoinEstimate (HashJoinState state, ParallelContext pcxt)

void	ExecHashJoinInitializeDSM (HashJoinState state, ParallelContext pcxt)

void	ExecHashJoinReInitializeDSM (HashJoinState state, ParallelContext pcxt)

void	ExecHashJoinInitializeWorker (HashJoinState state, ParallelWorkerContext pwcxt)

Macro Definition Documentation

◆ HJ_BUILD_HASHTABLE

#define HJ_BUILD_HASHTABLE 1

Definition at line 180 of file nodeHashjoin.c.

◆ HJ_FILL_INNER

#define HJ_FILL_INNER ( hjstate ) ((hjstate)->hj_NullOuterTupleSlot != NULL)

Definition at line 190 of file nodeHashjoin.c.

◆ HJ_FILL_INNER_TUPLES

#define HJ_FILL_INNER_TUPLES 5

Definition at line 184 of file nodeHashjoin.c.

◆ HJ_FILL_OUTER

#define HJ_FILL_OUTER ( hjstate ) ((hjstate)->hj_NullInnerTupleSlot != NULL)

Definition at line 188 of file nodeHashjoin.c.

◆ HJ_FILL_OUTER_TUPLE

#define HJ_FILL_OUTER_TUPLE 4

Definition at line 183 of file nodeHashjoin.c.

◆ HJ_NEED_NEW_BATCH

#define HJ_NEED_NEW_BATCH 6

Definition at line 185 of file nodeHashjoin.c.

◆ HJ_NEED_NEW_OUTER

#define HJ_NEED_NEW_OUTER 2

Definition at line 181 of file nodeHashjoin.c.

◆ HJ_SCAN_BUCKET

#define HJ_SCAN_BUCKET 3

Definition at line 182 of file nodeHashjoin.c.

Function Documentation

◆ ExecEndHashJoin()

void ExecEndHashJoin ( HashJoinState * node )

Definition at line 948 of file nodeHashjoin.c.

 {
     /*
      * Free hash table
      */
     if (node->hj_HashTable)
     {
         ExecHashTableDestroy(node->hj_HashTable);
         node->hj_HashTable = NULL;
     }
  
     /*
      * clean up subtrees
      */
     ExecEndNode(outerPlanState(node));
     ExecEndNode(innerPlanState(node));
 }

References ExecEndNode(), ExecHashTableDestroy(), HashJoinState::hj_HashTable, innerPlanState, and outerPlanState.

Referenced by ExecEndNode().

◆ ExecHashJoin()

static TupleTableSlot* ExecHashJoin ( PlanState * pstate )

static

Definition at line 684 of file nodeHashjoin.c.

 {
     /*
      * On sufficiently smart compilers this should be inlined with the
      * parallel-aware branches removed.
      */
     return ExecHashJoinImpl(pstate, false);
 }

References ExecHashJoinImpl().

Referenced by ExecInitHashJoin().

◆ ExecHashJoinEstimate()

void ExecHashJoinEstimate	(	HashJoinState *	state,
		ParallelContext *	pcxt
	)

Definition at line 1648 of file nodeHashjoin.c.

 {
     shm_toc_estimate_chunk(&pcxt->estimator, sizeof(ParallelHashJoinState));
     shm_toc_estimate_keys(&pcxt->estimator, 1);
 }

References ParallelContext::estimator, shm_toc_estimate_chunk, and shm_toc_estimate_keys.

Referenced by ExecParallelEstimate().

◆ ExecHashJoinGetSavedTuple()

static TupleTableSlot * ExecHashJoinGetSavedTuple	(	HashJoinState *	hjstate,
		BufFile *	file,
		uint32 *	hashvalue,
		TupleTableSlot *	tupleSlot
	)

static

Definition at line 1455 of file nodeHashjoin.c.

 {
     uint32      header[2];
     size_t      nread;
     MinimalTuple tuple;
  
     /*
      * We check for interrupts here because this is typically taken as an
      * alternative code path to an ExecProcNode() call, which would include
      * such a check.
      */
     CHECK_FOR_INTERRUPTS();
  
     /*
      * Since both the hash value and the MinimalTuple length word are uint32,
      * we can read them both in one BufFileRead() call without any type
      * cheating.
      */
     nread = BufFileReadMaybeEOF(file, header, sizeof(header), true);
     if (nread == 0)             /* end of file */
     {
         ExecClearTuple(tupleSlot);
         return NULL;
     }
     *hashvalue = header[0];
     tuple = (MinimalTuple) palloc(header[1]);
     tuple->t_len = header[1];
     BufFileReadExact(file,
                      (char *) tuple + sizeof(uint32),
                      header[1] - sizeof(uint32));
     ExecForceStoreMinimalTuple(tuple, tupleSlot, true);
     return tupleSlot;
 }

References BufFileReadExact(), BufFileReadMaybeEOF(), CHECK_FOR_INTERRUPTS, ExecClearTuple(), ExecForceStoreMinimalTuple(), palloc(), and MinimalTupleData::t_len.

Referenced by ExecHashJoinNewBatch(), and ExecHashJoinOuterGetTuple().

◆ ExecHashJoinImpl()

static pg_attribute_always_inline TupleTableSlot* ExecHashJoinImpl	(	PlanState *	pstate,
		bool	parallel
	)

static

Definition at line 221 of file nodeHashjoin.c.

 {
     HashJoinState *node = castNode(HashJoinState, pstate);
     PlanState  *outerNode;
     HashState  *hashNode;
     ExprState  *joinqual;
     ExprState  *otherqual;
     ExprContext *econtext;
     HashJoinTable hashtable;
     TupleTableSlot *outerTupleSlot;
     uint32      hashvalue;
     int         batchno;
     ParallelHashJoinState *parallel_state;
  
     /*
      * get information from HashJoin node
      */
     joinqual = node->js.joinqual;
     otherqual = node->js.ps.qual;
     hashNode = (HashState *) innerPlanState(node);
     outerNode = outerPlanState(node);
     hashtable = node->hj_HashTable;
     econtext = node->js.ps.ps_ExprContext;
     parallel_state = hashNode->parallel_state;
  
     /*
      * Reset per-tuple memory context to free any expression evaluation
      * storage allocated in the previous tuple cycle.
      */
     ResetExprContext(econtext);
  
     /*
      * run the hash join state machine
      */
     for (;;)
     {
         /*
          * It's possible to iterate this loop many times before returning a
          * tuple, in some pathological cases such as needing to move much of
          * the current batch to a later batch.  So let's check for interrupts
          * each time through.
          */
         CHECK_FOR_INTERRUPTS();
  
         switch (node->hj_JoinState)
         {
             case HJ_BUILD_HASHTABLE:
  
                 /*
                  * First time through: build hash table for inner relation.
                  */
                 Assert(hashtable == NULL);
  
                 /*
                  * If the outer relation is completely empty, and it's not
                  * right/right-anti/full join, we can quit without building
                  * the hash table.  However, for an inner join it is only a
                  * win to check this when the outer relation's startup cost is
                  * less than the projected cost of building the hash table.
                  * Otherwise it's best to build the hash table first and see
                  * if the inner relation is empty.  (When it's a left join, we
                  * should always make this check, since we aren't going to be
                  * able to skip the join on the strength of an empty inner
                  * relation anyway.)
                  *
                  * If we are rescanning the join, we make use of information
                  * gained on the previous scan: don't bother to try the
                  * prefetch if the previous scan found the outer relation
                  * nonempty. This is not 100% reliable since with new
                  * parameters the outer relation might yield different
                  * results, but it's a good heuristic.
                  *
                  * The only way to make the check is to try to fetch a tuple
                  * from the outer plan node.  If we succeed, we have to stash
                  * it away for later consumption by ExecHashJoinOuterGetTuple.
                  */
                 if (HJ_FILL_INNER(node))
                 {
                     /* no chance to not build the hash table */
                     node->hj_FirstOuterTupleSlot = NULL;
                 }
                 else if (parallel)
                 {
                     /*
                      * The empty-outer optimization is not implemented for
                      * shared hash tables, because no one participant can
                      * determine that there are no outer tuples, and it's not
                      * yet clear that it's worth the synchronization overhead
                      * of reaching consensus to figure that out.  So we have
                      * to build the hash table.
                      */
                     node->hj_FirstOuterTupleSlot = NULL;
                 }
                 else if (HJ_FILL_OUTER(node) ||
                          (outerNode->plan->startup_cost < hashNode->ps.plan->total_cost &&
                           !node->hj_OuterNotEmpty))
                 {
                     node->hj_FirstOuterTupleSlot = ExecProcNode(outerNode);
                     if (TupIsNull(node->hj_FirstOuterTupleSlot))
                     {
                         node->hj_OuterNotEmpty = false;
                         return NULL;
                     }
                     else
                         node->hj_OuterNotEmpty = true;
                 }
                 else
                     node->hj_FirstOuterTupleSlot = NULL;
  
                 /*
                  * Create the hash table.  If using Parallel Hash, then
                  * whoever gets here first will create the hash table and any
                  * later arrivals will merely attach to it.
                  */
                 hashtable = ExecHashTableCreate(hashNode);
                 node->hj_HashTable = hashtable;
  
                 /*
                  * Execute the Hash node, to build the hash table.  If using
                  * Parallel Hash, then we'll try to help hashing unless we
                  * arrived too late.
                  */
                 hashNode->hashtable = hashtable;
                 (void) MultiExecProcNode((PlanState *) hashNode);
  
                 /*
                  * If the inner relation is completely empty, and we're not
                  * doing a left outer join, we can quit without scanning the
                  * outer relation.
                  */
                 if (hashtable->totalTuples == 0 && !HJ_FILL_OUTER(node))
                 {
                     if (parallel)
                     {
                         /*
                          * Advance the build barrier to PHJ_BUILD_RUN before
                          * proceeding so we can negotiate resource cleanup.
                          */
                         Barrier    *build_barrier = &parallel_state->build_barrier;
  
                         while (BarrierPhase(build_barrier) < PHJ_BUILD_RUN)
                             BarrierArriveAndWait(build_barrier, 0);
                     }
                     return NULL;
                 }
  
                 /*
                  * need to remember whether nbatch has increased since we
                  * began scanning the outer relation
                  */
                 hashtable->nbatch_outstart = hashtable->nbatch;
  
                 /*
                  * Reset OuterNotEmpty for scan.  (It's OK if we fetched a
                  * tuple above, because ExecHashJoinOuterGetTuple will
                  * immediately set it again.)
                  */
                 node->hj_OuterNotEmpty = false;
  
                 if (parallel)
                 {
                     Barrier    *build_barrier;
  
                     build_barrier = &parallel_state->build_barrier;
                     Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASH_OUTER ||
                            BarrierPhase(build_barrier) == PHJ_BUILD_RUN ||
                            BarrierPhase(build_barrier) == PHJ_BUILD_FREE);
                     if (BarrierPhase(build_barrier) == PHJ_BUILD_HASH_OUTER)
                     {
                         /*
                          * If multi-batch, we need to hash the outer relation
                          * up front.
                          */
                         if (hashtable->nbatch > 1)
                             ExecParallelHashJoinPartitionOuter(node);
                         BarrierArriveAndWait(build_barrier,
                                              WAIT_EVENT_HASH_BUILD_HASH_OUTER);
                     }
                     else if (BarrierPhase(build_barrier) == PHJ_BUILD_FREE)
                     {
                         /*
                          * If we attached so late that the job is finished and
                          * the batch state has been freed, we can return
                          * immediately.
                          */
                         return NULL;
                     }
  
                     /* Each backend should now select a batch to work on. */
                     Assert(BarrierPhase(build_barrier) == PHJ_BUILD_RUN);
                     hashtable->curbatch = -1;
                     node->hj_JoinState = HJ_NEED_NEW_BATCH;
  
                     continue;
                 }
                 else
                     node->hj_JoinState = HJ_NEED_NEW_OUTER;
  
                 /* FALL THRU */
  
             case HJ_NEED_NEW_OUTER:
  
                 /*
                  * We don't have an outer tuple, try to get the next one
                  */
                 if (parallel)
                     outerTupleSlot =
                         ExecParallelHashJoinOuterGetTuple(outerNode, node,
                                                           &hashvalue);
                 else
                     outerTupleSlot =
                         ExecHashJoinOuterGetTuple(outerNode, node, &hashvalue);
  
                 if (TupIsNull(outerTupleSlot))
                 {
                     /* end of batch, or maybe whole join */
                     if (HJ_FILL_INNER(node))
                     {
                         /* set up to scan for unmatched inner tuples */
                         if (parallel)
                         {
                             /*
                              * Only one process is currently allow to handle
                              * each batch's unmatched tuples, in a parallel
                              * join.
                              */
                             if (ExecParallelPrepHashTableForUnmatched(node))
                                 node->hj_JoinState = HJ_FILL_INNER_TUPLES;
                             else
                                 node->hj_JoinState = HJ_NEED_NEW_BATCH;
                         }
                         else
                         {
                             ExecPrepHashTableForUnmatched(node);
                             node->hj_JoinState = HJ_FILL_INNER_TUPLES;
                         }
                     }
                     else
                         node->hj_JoinState = HJ_NEED_NEW_BATCH;
                     continue;
                 }
  
                 econtext->ecxt_outertuple = outerTupleSlot;
                 node->hj_MatchedOuter = false;
  
                 /*
                  * Find the corresponding bucket for this tuple in the main
                  * hash table or skew hash table.
                  */
                 node->hj_CurHashValue = hashvalue;
                 ExecHashGetBucketAndBatch(hashtable, hashvalue,
                                           &node->hj_CurBucketNo, &batchno);
                 node->hj_CurSkewBucketNo = ExecHashGetSkewBucket(hashtable,
                                                                  hashvalue);
                 node->hj_CurTuple = NULL;
  
                 /*
                  * The tuple might not belong to the current batch (where
                  * "current batch" includes the skew buckets if any).
                  */
                 if (batchno != hashtable->curbatch &&
                     node->hj_CurSkewBucketNo == INVALID_SKEW_BUCKET_NO)
                 {
                     bool        shouldFree;
                     MinimalTuple mintuple = ExecFetchSlotMinimalTuple(outerTupleSlot,
                                                                       &shouldFree);
  
                     /*
                      * Need to postpone this outer tuple to a later batch.
                      * Save it in the corresponding outer-batch file.
                      */
                     Assert(parallel_state == NULL);
                     Assert(batchno > hashtable->curbatch);
                     ExecHashJoinSaveTuple(mintuple, hashvalue,
                                           &hashtable->outerBatchFile[batchno],
                                           hashtable);
  
                     if (shouldFree)
                         heap_free_minimal_tuple(mintuple);
  
                     /* Loop around, staying in HJ_NEED_NEW_OUTER state */
                     continue;
                 }
  
                 /* OK, let's scan the bucket for matches */
                 node->hj_JoinState = HJ_SCAN_BUCKET;
  
                 /* FALL THRU */
  
             case HJ_SCAN_BUCKET:
  
                 /*
                  * Scan the selected hash bucket for matches to current outer
                  */
                 if (parallel)
                 {
                     if (!ExecParallelScanHashBucket(node, econtext))
                     {
                         /* out of matches; check for possible outer-join fill */
                         node->hj_JoinState = HJ_FILL_OUTER_TUPLE;
                         continue;
                     }
                 }
                 else
                 {
                     if (!ExecScanHashBucket(node, econtext))
                     {
                         /* out of matches; check for possible outer-join fill */
                         node->hj_JoinState = HJ_FILL_OUTER_TUPLE;
                         continue;
                     }
                 }
  
                 /*
                  * In a right-semijoin, we only need the first match for each
                  * inner tuple.
                  */
                 if (node->js.jointype == JOIN_RIGHT_SEMI &&
                     HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(node->hj_CurTuple)))
                     continue;
  
                 /*
                  * We've got a match, but still need to test non-hashed quals.
                  * ExecScanHashBucket already set up all the state needed to
                  * call ExecQual.
                  *
                  * If we pass the qual, then save state for next call and have
                  * ExecProject form the projection, store it in the tuple
                  * table, and return the slot.
                  *
                  * Only the joinquals determine tuple match status, but all
                  * quals must pass to actually return the tuple.
                  */
                 if (joinqual == NULL || ExecQual(joinqual, econtext))
                 {
                     node->hj_MatchedOuter = true;
  
                     /*
                      * This is really only needed if HJ_FILL_INNER(node) or if
                      * we are in a right-semijoin, but we'll avoid the branch
                      * and just set it always.
                      */
                     if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(node->hj_CurTuple)))
                         HeapTupleHeaderSetMatch(HJTUPLE_MINTUPLE(node->hj_CurTuple));
  
                     /* In an antijoin, we never return a matched tuple */
                     if (node->js.jointype == JOIN_ANTI)
                     {
                         node->hj_JoinState = HJ_NEED_NEW_OUTER;
                         continue;
                     }
  
                     /*
                      * If we only need to consider the first matching inner
                      * tuple, then advance to next outer tuple after we've
                      * processed this one.
                      */
                     if (node->js.single_match)
                         node->hj_JoinState = HJ_NEED_NEW_OUTER;
  
                     /*
                      * In a right-antijoin, we never return a matched tuple.
                      * If it's not an inner_unique join, we need to stay on
                      * the current outer tuple to continue scanning the inner
                      * side for matches.
                      */
                     if (node->js.jointype == JOIN_RIGHT_ANTI)
                         continue;
  
                     if (otherqual == NULL || ExecQual(otherqual, econtext))
                         return ExecProject(node->js.ps.ps_ProjInfo);
                     else
                         InstrCountFiltered2(node, 1);
                 }
                 else
                     InstrCountFiltered1(node, 1);
                 break;
  
             case HJ_FILL_OUTER_TUPLE:
  
                 /*
                  * The current outer tuple has run out of matches, so check
                  * whether to emit a dummy outer-join tuple.  Whether we emit
                  * one or not, the next state is NEED_NEW_OUTER.
                  */
                 node->hj_JoinState = HJ_NEED_NEW_OUTER;
  
                 if (!node->hj_MatchedOuter &&
                     HJ_FILL_OUTER(node))
                 {
                     /*
                      * Generate a fake join tuple with nulls for the inner
                      * tuple, and return it if it passes the non-join quals.
                      */
                     econtext->ecxt_innertuple = node->hj_NullInnerTupleSlot;
  
                     if (otherqual == NULL || ExecQual(otherqual, econtext))
                         return ExecProject(node->js.ps.ps_ProjInfo);
                     else
                         InstrCountFiltered2(node, 1);
                 }
                 break;
  
             case HJ_FILL_INNER_TUPLES:
  
                 /*
                  * We have finished a batch, but we are doing
                  * right/right-anti/full join, so any unmatched inner tuples
                  * in the hashtable have to be emitted before we continue to
                  * the next batch.
                  */
                 if (!(parallel ? ExecParallelScanHashTableForUnmatched(node, econtext)
                       : ExecScanHashTableForUnmatched(node, econtext)))
                 {
                     /* no more unmatched tuples */
                     node->hj_JoinState = HJ_NEED_NEW_BATCH;
                     continue;
                 }
  
                 /*
                  * Generate a fake join tuple with nulls for the outer tuple,
                  * and return it if it passes the non-join quals.
                  */
                 econtext->ecxt_outertuple = node->hj_NullOuterTupleSlot;
  
                 if (otherqual == NULL || ExecQual(otherqual, econtext))
                     return ExecProject(node->js.ps.ps_ProjInfo);
                 else
                     InstrCountFiltered2(node, 1);
                 break;
  
             case HJ_NEED_NEW_BATCH:
  
                 /*
                  * Try to advance to next batch.  Done if there are no more.
                  */
                 if (parallel)
                 {
                     if (!ExecParallelHashJoinNewBatch(node))
                         return NULL;    /* end of parallel-aware join */
                 }
                 else
                 {
                     if (!ExecHashJoinNewBatch(node))
                         return NULL;    /* end of parallel-oblivious join */
                 }
                 node->hj_JoinState = HJ_NEED_NEW_OUTER;
                 break;
  
             default:
                 elog(ERROR, "unrecognized hashjoin state: %d",
                      (int) node->hj_JoinState);
         }
     }
 }

Referenced by ExecHashJoin(), and ExecParallelHashJoin().

◆ ExecHashJoinInitializeDSM()

void ExecHashJoinInitializeDSM	(	HashJoinState *	state,
		ParallelContext *	pcxt
	)

Definition at line 1655 of file nodeHashjoin.c.

 {
     int         plan_node_id = state->js.ps.plan->plan_node_id;
     HashState  *hashNode;
     ParallelHashJoinState *pstate;
  
     /*
      * Disable shared hash table mode if we failed to create a real DSM
      * segment, because that means that we don't have a DSA area to work with.
      */
     if (pcxt->seg == NULL)
         return;
  
     ExecSetExecProcNode(&state->js.ps, ExecParallelHashJoin);
  
     /*
      * Set up the state needed to coordinate access to the shared hash
      * table(s), using the plan node ID as the toc key.
      */
     pstate = shm_toc_allocate(pcxt->toc, sizeof(ParallelHashJoinState));
     shm_toc_insert(pcxt->toc, plan_node_id, pstate);
  
     /*
      * Set up the shared hash join state with no batches initially.
      * ExecHashTableCreate() will prepare at least one later and set nbatch
      * and space_allowed.
      */
     pstate->nbatch = 0;
     pstate->space_allowed = 0;
     pstate->batches = InvalidDsaPointer;
     pstate->old_batches = InvalidDsaPointer;
     pstate->nbuckets = 0;
     pstate->growth = PHJ_GROWTH_OK;
     pstate->chunk_work_queue = InvalidDsaPointer;
     pg_atomic_init_u32(&pstate->distributor, 0);
     pstate->nparticipants = pcxt->nworkers + 1;
     pstate->total_tuples = 0;
     LWLockInitialize(&pstate->lock,
                      LWTRANCHE_PARALLEL_HASH_JOIN);
     BarrierInit(&pstate->build_barrier, 0);
     BarrierInit(&pstate->grow_batches_barrier, 0);
     BarrierInit(&pstate->grow_buckets_barrier, 0);
  
     /* Set up the space we'll use for shared temporary files. */
     SharedFileSetInit(&pstate->fileset, pcxt->seg);
  
     /* Initialize the shared state in the hash node. */
     hashNode = (HashState *) innerPlanState(state);
     hashNode->parallel_state = pstate;
 }

Referenced by ExecParallelInitializeDSM().

◆ ExecHashJoinInitializeWorker()

void ExecHashJoinInitializeWorker	(	HashJoinState *	state,
		ParallelWorkerContext *	pwcxt
	)

Definition at line 1746 of file nodeHashjoin.c.

 {
     HashState  *hashNode;
     int         plan_node_id = state->js.ps.plan->plan_node_id;
     ParallelHashJoinState *pstate =
         shm_toc_lookup(pwcxt->toc, plan_node_id, false);
  
     /* Attach to the space for shared temporary files. */
     SharedFileSetAttach(&pstate->fileset, pwcxt->seg);
  
     /* Attach to the shared state in the hash node. */
     hashNode = (HashState *) innerPlanState(state);
     hashNode->parallel_state = pstate;
  
     ExecSetExecProcNode(&state->js.ps, ExecParallelHashJoin);
 }

References ExecParallelHashJoin(), ExecSetExecProcNode(), ParallelHashJoinState::fileset, innerPlanState, HashState::parallel_state, ParallelWorkerContext::seg, SharedFileSetAttach(), shm_toc_lookup(), and ParallelWorkerContext::toc.

Referenced by ExecParallelInitializeWorker().

◆ ExecHashJoinNewBatch()

static bool ExecHashJoinNewBatch ( HashJoinState * hjstate )

static

Definition at line 1130 of file nodeHashjoin.c.

 {
     HashJoinTable hashtable = hjstate->hj_HashTable;
     int         nbatch;
     int         curbatch;
     BufFile    *innerFile;
     TupleTableSlot *slot;
     uint32      hashvalue;
  
     nbatch = hashtable->nbatch;
     curbatch = hashtable->curbatch;
  
     if (curbatch > 0)
     {
         /*
          * We no longer need the previous outer batch file; close it right
          * away to free disk space.
          */
         if (hashtable->outerBatchFile[curbatch])
             BufFileClose(hashtable->outerBatchFile[curbatch]);
         hashtable->outerBatchFile[curbatch] = NULL;
     }
     else                        /* we just finished the first batch */
     {
         /*
          * Reset some of the skew optimization state variables, since we no
          * longer need to consider skew tuples after the first batch. The
          * memory context reset we are about to do will release the skew
          * hashtable itself.
          */
         hashtable->skewEnabled = false;
         hashtable->skewBucket = NULL;
         hashtable->skewBucketNums = NULL;
         hashtable->nSkewBuckets = 0;
         hashtable->spaceUsedSkew = 0;
     }
  
     /*
      * We can always skip over any batches that are completely empty on both
      * sides.  We can sometimes skip over batches that are empty on only one
      * side, but there are exceptions:
      *
      * 1. In a left/full outer join, we have to process outer batches even if
      * the inner batch is empty.  Similarly, in a right/right-anti/full outer
      * join, we have to process inner batches even if the outer batch is
      * empty.
      *
      * 2. If we have increased nbatch since the initial estimate, we have to
      * scan inner batches since they might contain tuples that need to be
      * reassigned to later inner batches.
      *
      * 3. Similarly, if we have increased nbatch since starting the outer
      * scan, we have to rescan outer batches in case they contain tuples that
      * need to be reassigned.
      */
     curbatch++;
     while (curbatch < nbatch &&
            (hashtable->outerBatchFile[curbatch] == NULL ||
             hashtable->innerBatchFile[curbatch] == NULL))
     {
         if (hashtable->outerBatchFile[curbatch] &&
             HJ_FILL_OUTER(hjstate))
             break;              /* must process due to rule 1 */
         if (hashtable->innerBatchFile[curbatch] &&
             HJ_FILL_INNER(hjstate))
             break;              /* must process due to rule 1 */
         if (hashtable->innerBatchFile[curbatch] &&
             nbatch != hashtable->nbatch_original)
             break;              /* must process due to rule 2 */
         if (hashtable->outerBatchFile[curbatch] &&
             nbatch != hashtable->nbatch_outstart)
             break;              /* must process due to rule 3 */
         /* We can ignore this batch. */
         /* Release associated temp files right away. */
         if (hashtable->innerBatchFile[curbatch])
             BufFileClose(hashtable->innerBatchFile[curbatch]);
         hashtable->innerBatchFile[curbatch] = NULL;
         if (hashtable->outerBatchFile[curbatch])
             BufFileClose(hashtable->outerBatchFile[curbatch]);
         hashtable->outerBatchFile[curbatch] = NULL;
         curbatch++;
     }
  
     if (curbatch >= nbatch)
         return false;           /* no more batches */
  
     hashtable->curbatch = curbatch;
  
     /*
      * Reload the hash table with the new inner batch (which could be empty)
      */
     ExecHashTableReset(hashtable);
  
     innerFile = hashtable->innerBatchFile[curbatch];
  
     if (innerFile != NULL)
     {
         if (BufFileSeek(innerFile, 0, 0, SEEK_SET))
             ereport(ERROR,
                     (errcode_for_file_access(),
                      errmsg("could not rewind hash-join temporary file")));
  
         while ((slot = ExecHashJoinGetSavedTuple(hjstate,
                                                  innerFile,
                                                  &hashvalue,
                                                  hjstate->hj_HashTupleSlot)))
         {
             /*
              * NOTE: some tuples may be sent to future batches.  Also, it is
              * possible for hashtable->nbatch to be increased here!
              */
             ExecHashTableInsert(hashtable, slot, hashvalue);
         }
  
         /*
          * after we build the hash table, the inner batch file is no longer
          * needed
          */
         BufFileClose(innerFile);
         hashtable->innerBatchFile[curbatch] = NULL;
     }
  
     /*
      * Rewind outer batch file (if present), so that we can start reading it.
      */
     if (hashtable->outerBatchFile[curbatch] != NULL)
     {
         if (BufFileSeek(hashtable->outerBatchFile[curbatch], 0, 0, SEEK_SET))
             ereport(ERROR,
                     (errcode_for_file_access(),
                      errmsg("could not rewind hash-join temporary file")));
     }
  
     return true;
 }

Referenced by ExecHashJoinImpl().

◆ ExecHashJoinOuterGetTuple()

static TupleTableSlot * ExecHashJoinOuterGetTuple	(	PlanState *	outerNode,
		HashJoinState *	hjstate,
		uint32 *	hashvalue
	)

static

Definition at line 979 of file nodeHashjoin.c.

 {
     HashJoinTable hashtable = hjstate->hj_HashTable;
     int         curbatch = hashtable->curbatch;
     TupleTableSlot *slot;
  
     if (curbatch == 0)          /* if it is the first pass */
     {
         /*
          * Check to see if first outer tuple was already fetched by
          * ExecHashJoin() and not used yet.
          */
         slot = hjstate->hj_FirstOuterTupleSlot;
         if (!TupIsNull(slot))
             hjstate->hj_FirstOuterTupleSlot = NULL;
         else
             slot = ExecProcNode(outerNode);
  
         while (!TupIsNull(slot))
         {
             bool        isnull;
  
             /*
              * We have to compute the tuple's hash value.
              */
             ExprContext *econtext = hjstate->js.ps.ps_ExprContext;
  
             econtext->ecxt_outertuple = slot;
  
             ResetExprContext(econtext);
  
             *hashvalue = DatumGetUInt32(ExecEvalExprSwitchContext(hjstate->hj_OuterHash,
                                                                   econtext,
                                                                   &isnull));
  
             if (!isnull)
             {
                 /* remember outer relation is not empty for possible rescan */
                 hjstate->hj_OuterNotEmpty = true;
  
                 return slot;
             }
  
             /*
              * That tuple couldn't match because of a NULL, so discard it and
              * continue with the next one.
              */
             slot = ExecProcNode(outerNode);
         }
     }
     else if (curbatch < hashtable->nbatch)
     {
         BufFile    *file = hashtable->outerBatchFile[curbatch];
  
         /*
          * In outer-join cases, we could get here even though the batch file
          * is empty.
          */
         if (file == NULL)
             return NULL;
  
         slot = ExecHashJoinGetSavedTuple(hjstate,
                                          file,
                                          hashvalue,
                                          hjstate->hj_OuterTupleSlot);
         if (!TupIsNull(slot))
             return slot;
     }
  
     /* End of this batch */
     return NULL;
 }

References HashJoinTableData::curbatch, DatumGetUInt32(), ExprContext::ecxt_outertuple, ExecEvalExprSwitchContext(), ExecHashJoinGetSavedTuple(), ExecProcNode(), HashJoinState::hj_FirstOuterTupleSlot, HashJoinState::hj_HashTable, HashJoinState::hj_OuterHash, HashJoinState::hj_OuterNotEmpty, HashJoinState::hj_OuterTupleSlot, HashJoinState::js, HashJoinTableData::outerBatchFile, JoinState::ps, PlanState::ps_ExprContext, ResetExprContext, and TupIsNull.

Referenced by ExecHashJoinImpl().

◆ ExecHashJoinReInitializeDSM()

void ExecHashJoinReInitializeDSM	(	HashJoinState *	state,
		ParallelContext *	pcxt
	)

Definition at line 1713 of file nodeHashjoin.c.

 {
     int         plan_node_id = state->js.ps.plan->plan_node_id;
     ParallelHashJoinState *pstate =
         shm_toc_lookup(pcxt->toc, plan_node_id, false);
  
     /*
      * It would be possible to reuse the shared hash table in single-batch
      * cases by resetting and then fast-forwarding build_barrier to
      * PHJ_BUILD_FREE and batch 0's batch_barrier to PHJ_BATCH_PROBE, but
      * currently shared hash tables are already freed by now (by the last
      * participant to detach from the batch).  We could consider keeping it
      * around for single-batch joins.  We'd also need to adjust
      * finalize_plan() so that it doesn't record a dummy dependency for
      * Parallel Hash nodes, preventing the rescan optimization.  For now we
      * don't try.
      */
  
     /* Detach, freeing any remaining shared memory. */
     if (state->hj_HashTable != NULL)
     {
         ExecHashTableDetachBatch(state->hj_HashTable);
         ExecHashTableDetach(state->hj_HashTable);
     }
  
     /* Clear any shared batch files. */
     SharedFileSetDeleteAll(&pstate->fileset);
  
     /* Reset build_barrier to PHJ_BUILD_ELECT so we can go around again. */
     BarrierInit(&pstate->build_barrier, 0);
 }

References BarrierInit(), ParallelHashJoinState::build_barrier, ExecHashTableDetach(), ExecHashTableDetachBatch(), ParallelHashJoinState::fileset, SharedFileSetDeleteAll(), shm_toc_lookup(), and ParallelContext::toc.

Referenced by ExecParallelReInitializeDSM().

◆ ExecHashJoinSaveTuple()

void ExecHashJoinSaveTuple	(	MinimalTuple	tuple,
		uint32	hashvalue,
		BufFile **	fileptr,
		HashJoinTable	hashtable
	)

Definition at line 1414 of file nodeHashjoin.c.

 {
     BufFile    *file = *fileptr;
  
     /*
      * The batch file is lazily created. If this is the first tuple written to
      * this batch, the batch file is created and its buffer is allocated in
      * the spillCxt context, NOT in the batchCxt.
      *
      * During the build phase, buffered files are created for inner batches.
      * Each batch's buffered file is closed (and its buffer freed) after the
      * batch is loaded into memory during the outer side scan. Therefore, it
      * is necessary to allocate the batch file buffer in a memory context
      * which outlives the batch itself.
      *
      * Also, we use spillCxt instead of hashCxt for a better accounting of the
      * spilling memory consumption.
      */
     if (file == NULL)
     {
         MemoryContext oldctx = MemoryContextSwitchTo(hashtable->spillCxt);
  
         file = BufFileCreateTemp(false);
         *fileptr = file;
  
         MemoryContextSwitchTo(oldctx);
     }
  
     BufFileWrite(file, &hashvalue, sizeof(uint32));
     BufFileWrite(file, tuple, tuple->t_len);
 }

References BufFileCreateTemp(), BufFileWrite(), MemoryContextSwitchTo(), HashJoinTableData::spillCxt, and MinimalTupleData::t_len.

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

◆ ExecInitHashJoin()

HashJoinState* ExecInitHashJoin	(	HashJoin *	node,
		EState *	estate,
		int	eflags
	)

Definition at line 716 of file nodeHashjoin.c.

 {
     HashJoinState *hjstate;
     Plan       *outerNode;
     Hash       *hashNode;
     TupleDesc   outerDesc,
                 innerDesc;
     const TupleTableSlotOps *ops;
  
     /* check for unsupported flags */
     Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
  
     /*
      * create state structure
      */
     hjstate = makeNode(HashJoinState);
     hjstate->js.ps.plan = (Plan *) node;
     hjstate->js.ps.state = estate;
  
     /*
      * See ExecHashJoinInitializeDSM() and ExecHashJoinInitializeWorker()
      * where this function may be replaced with a parallel version, if we
      * managed to launch a parallel query.
      */
     hjstate->js.ps.ExecProcNode = ExecHashJoin;
     hjstate->js.jointype = node->join.jointype;
  
     /*
      * Miscellaneous initialization
      *
      * create expression context for node
      */
     ExecAssignExprContext(estate, &hjstate->js.ps);
  
     /*
      * initialize child nodes
      *
      * Note: we could suppress the REWIND flag for the inner input, which
      * would amount to betting that the hash will be a single batch.  Not
      * clear if this would be a win or not.
      */
     outerNode = outerPlan(node);
     hashNode = (Hash *) innerPlan(node);
  
     outerPlanState(hjstate) = ExecInitNode(outerNode, estate, eflags);
     outerDesc = ExecGetResultType(outerPlanState(hjstate));
     innerPlanState(hjstate) = ExecInitNode((Plan *) hashNode, estate, eflags);
     innerDesc = ExecGetResultType(innerPlanState(hjstate));
  
     /*
      * Initialize result slot, type and projection.
      */
     ExecInitResultTupleSlotTL(&hjstate->js.ps, &TTSOpsVirtual);
     ExecAssignProjectionInfo(&hjstate->js.ps, NULL);
  
     /*
      * tuple table initialization
      */
     ops = ExecGetResultSlotOps(outerPlanState(hjstate), NULL);
     hjstate->hj_OuterTupleSlot = ExecInitExtraTupleSlot(estate, outerDesc,
                                                         ops);
  
     /*
      * detect whether we need only consider the first matching inner tuple
      */
     hjstate->js.single_match = (node->join.inner_unique ||
                                 node->join.jointype == JOIN_SEMI);
  
     /* set up null tuples for outer joins, if needed */
     switch (node->join.jointype)
     {
         case JOIN_INNER:
         case JOIN_SEMI:
         case JOIN_RIGHT_SEMI:
             break;
         case JOIN_LEFT:
         case JOIN_ANTI:
             hjstate->hj_NullInnerTupleSlot =
                 ExecInitNullTupleSlot(estate, innerDesc, &TTSOpsVirtual);
             break;
         case JOIN_RIGHT:
         case JOIN_RIGHT_ANTI:
             hjstate->hj_NullOuterTupleSlot =
                 ExecInitNullTupleSlot(estate, outerDesc, &TTSOpsVirtual);
             break;
         case JOIN_FULL:
             hjstate->hj_NullOuterTupleSlot =
                 ExecInitNullTupleSlot(estate, outerDesc, &TTSOpsVirtual);
             hjstate->hj_NullInnerTupleSlot =
                 ExecInitNullTupleSlot(estate, innerDesc, &TTSOpsVirtual);
             break;
         default:
             elog(ERROR, "unrecognized join type: %d",
                  (int) node->join.jointype);
     }
  
     /*
      * now for some voodoo.  our temporary tuple slot is actually the result
      * tuple slot of the Hash node (which is our inner plan).  we can do this
      * because Hash nodes don't return tuples via ExecProcNode() -- instead
      * the hash join node uses ExecScanHashBucket() to get at the contents of
      * the hash table.  -cim 6/9/91
      */
     {
         HashState  *hashstate = (HashState *) innerPlanState(hjstate);
         Hash       *hash = (Hash *) hashstate->ps.plan;
         TupleTableSlot *slot = hashstate->ps.ps_ResultTupleSlot;
         Oid        *outer_hashfuncid;
         Oid        *inner_hashfuncid;
         bool       *hash_strict;
         ListCell   *lc;
         int         nkeys;
  
  
         hjstate->hj_HashTupleSlot = slot;
  
         /*
          * Build ExprStates to obtain hash values for either side of the join.
          * This must be done here as ExecBuildHash32Expr needs to know how to
          * handle NULL inputs and the required handling of that depends on the
          * jointype.  We don't know the join type in ExecInitHash() and we
          * must build the ExprStates before ExecHashTableCreate() so we
          * properly attribute any SubPlans that exist in the hash expressions
          * to the correct PlanState.
          */
         nkeys = list_length(node->hashoperators);
  
         outer_hashfuncid = palloc_array(Oid, nkeys);
         inner_hashfuncid = palloc_array(Oid, nkeys);
         hash_strict = palloc_array(bool, nkeys);
  
         /*
          * Determine the hash function for each side of the join for the given
          * hash operator.
          */
         foreach(lc, node->hashoperators)
         {
             Oid         hashop = lfirst_oid(lc);
             int         i = foreach_current_index(lc);
  
             if (!get_op_hash_functions(hashop,
                                        &outer_hashfuncid[i],
                                        &inner_hashfuncid[i]))
                 elog(ERROR,
                      "could not find hash function for hash operator %u",
                      hashop);
             hash_strict[i] = op_strict(hashop);
         }
  
         /*
          * Build an ExprState to generate the hash value for the expressions
          * on the outer of the join.  This ExprState must finish generating
          * the hash value when HJ_FILL_OUTER() is true.  Otherwise,
          * ExecBuildHash32Expr will set up the ExprState to abort early if it
          * finds a NULL.  In these cases, we don't need to store these tuples
          * in the hash table as the jointype does not require it.
          */
         hjstate->hj_OuterHash =
             ExecBuildHash32Expr(hjstate->js.ps.ps_ResultTupleDesc,
                                 hjstate->js.ps.resultops,
                                 outer_hashfuncid,
                                 node->hashcollations,
                                 node->hashkeys,
                                 hash_strict,
                                 &hjstate->js.ps,
                                 0,
                                 HJ_FILL_OUTER(hjstate));
  
         /* As above, but for the inner side of the join */
         hashstate->hash_expr =
             ExecBuildHash32Expr(hashstate->ps.ps_ResultTupleDesc,
                                 hashstate->ps.resultops,
                                 inner_hashfuncid,
                                 node->hashcollations,
                                 hash->hashkeys,
                                 hash_strict,
                                 &hashstate->ps,
                                 0,
                                 HJ_FILL_INNER(hjstate));
  
         /*
          * Set up the skew table hash function while we have a record of the
          * first key's hash function Oid.
          */
         if (OidIsValid(hash->skewTable))
         {
             hashstate->skew_hashfunction = palloc0(sizeof(FmgrInfo));
             hashstate->skew_collation = linitial_oid(node->hashcollations);
             fmgr_info(outer_hashfuncid[0], hashstate->skew_hashfunction);
         }
  
         /* no need to keep these */
         pfree(outer_hashfuncid);
         pfree(inner_hashfuncid);
         pfree(hash_strict);
     }
  
     /*
      * initialize child expressions
      */
     hjstate->js.ps.qual =
         ExecInitQual(node->join.plan.qual, (PlanState *) hjstate);
     hjstate->js.joinqual =
         ExecInitQual(node->join.joinqual, (PlanState *) hjstate);
     hjstate->hashclauses =
         ExecInitQual(node->hashclauses, (PlanState *) hjstate);
  
     /*
      * initialize hash-specific info
      */
     hjstate->hj_HashTable = NULL;
     hjstate->hj_FirstOuterTupleSlot = NULL;
  
     hjstate->hj_CurHashValue = 0;
     hjstate->hj_CurBucketNo = 0;
     hjstate->hj_CurSkewBucketNo = INVALID_SKEW_BUCKET_NO;
     hjstate->hj_CurTuple = NULL;
  
     hjstate->hj_JoinState = HJ_BUILD_HASHTABLE;
     hjstate->hj_MatchedOuter = false;
     hjstate->hj_OuterNotEmpty = false;
  
     return hjstate;
 }

References Assert, elog, ERROR, EXEC_FLAG_BACKWARD, EXEC_FLAG_MARK, ExecAssignExprContext(), ExecAssignProjectionInfo(), ExecBuildHash32Expr(), ExecGetResultSlotOps(), ExecGetResultType(), ExecHashJoin(), ExecInitExtraTupleSlot(), ExecInitNode(), ExecInitNullTupleSlot(), ExecInitQual(), ExecInitResultTupleSlotTL(), PlanState::ExecProcNode, fmgr_info(), foreach_current_index, get_op_hash_functions(), hash(), HashState::hash_expr, HashJoinState::hashclauses, HashJoin::hashclauses, HashJoin::hashcollations, HashJoin::hashkeys, HashJoin::hashoperators, HJ_BUILD_HASHTABLE, HashJoinState::hj_CurBucketNo, HashJoinState::hj_CurHashValue, HashJoinState::hj_CurSkewBucketNo, HashJoinState::hj_CurTuple, HJ_FILL_INNER, HJ_FILL_OUTER, HashJoinState::hj_FirstOuterTupleSlot, HashJoinState::hj_HashTable, HashJoinState::hj_HashTupleSlot, HashJoinState::hj_JoinState, HashJoinState::hj_MatchedOuter, HashJoinState::hj_NullInnerTupleSlot, HashJoinState::hj_NullOuterTupleSlot, HashJoinState::hj_OuterHash, HashJoinState::hj_OuterNotEmpty, HashJoinState::hj_OuterTupleSlot, i, Join::inner_unique, innerPlan, innerPlanState, INVALID_SKEW_BUCKET_NO, HashJoin::join, JOIN_ANTI, JOIN_FULL, JOIN_INNER, JOIN_LEFT, JOIN_RIGHT, JOIN_RIGHT_ANTI, JOIN_RIGHT_SEMI, JOIN_SEMI, JoinState::joinqual, Join::joinqual, JoinState::jointype, Join::jointype, HashJoinState::js, lfirst_oid, linitial_oid, list_length(), makeNode, OidIsValid, op_strict(), outerPlan, outerPlanState, palloc0(), palloc_array, pfree(), PlanState::plan, JoinState::ps, HashState::ps, PlanState::ps_ResultTupleDesc, PlanState::ps_ResultTupleSlot, PlanState::qual, PlanState::resultops, JoinState::single_match, HashState::skew_collation, HashState::skew_hashfunction, PlanState::state, and TTSOpsVirtual.

Referenced by ExecInitNode().

◆ ExecParallelHashJoin()

static TupleTableSlot* ExecParallelHashJoin ( PlanState * pstate )

static

Definition at line 700 of file nodeHashjoin.c.

 {
     /*
      * On sufficiently smart compilers this should be inlined with the
      * parallel-oblivious branches removed.
      */
     return ExecHashJoinImpl(pstate, true);
 }

References ExecHashJoinImpl().

Referenced by ExecHashJoinInitializeDSM(), and ExecHashJoinInitializeWorker().

◆ ExecParallelHashJoinNewBatch()

static bool ExecParallelHashJoinNewBatch ( HashJoinState * hjstate )

static

Definition at line 1271 of file nodeHashjoin.c.

 {
     HashJoinTable hashtable = hjstate->hj_HashTable;
     int         start_batchno;
     int         batchno;
  
     /*
      * If we were already attached to a batch, remember not to bother checking
      * it again, and detach from it (possibly freeing the hash table if we are
      * last to detach).
      */
     if (hashtable->curbatch >= 0)
     {
         hashtable->batches[hashtable->curbatch].done = true;
         ExecHashTableDetachBatch(hashtable);
     }
  
     /*
      * Search for a batch that isn't done.  We use an atomic counter to start
      * our search at a different batch in every participant when there are
      * more batches than participants.
      */
     batchno = start_batchno =
         pg_atomic_fetch_add_u32(&hashtable->parallel_state->distributor, 1) %
         hashtable->nbatch;
     do
     {
         uint32      hashvalue;
         MinimalTuple tuple;
         TupleTableSlot *slot;
  
         if (!hashtable->batches[batchno].done)
         {
             SharedTuplestoreAccessor *inner_tuples;
             Barrier    *batch_barrier =
                 &hashtable->batches[batchno].shared->batch_barrier;
  
             switch (BarrierAttach(batch_barrier))
             {
                 case PHJ_BATCH_ELECT:
  
                     /* One backend allocates the hash table. */
                     if (BarrierArriveAndWait(batch_barrier,
                                              WAIT_EVENT_HASH_BATCH_ELECT))
                         ExecParallelHashTableAlloc(hashtable, batchno);
                     /* Fall through. */
  
                 case PHJ_BATCH_ALLOCATE:
                     /* Wait for allocation to complete. */
                     BarrierArriveAndWait(batch_barrier,
                                          WAIT_EVENT_HASH_BATCH_ALLOCATE);
                     /* Fall through. */
  
                 case PHJ_BATCH_LOAD:
                     /* Start (or join in) loading tuples. */
                     ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
                     inner_tuples = hashtable->batches[batchno].inner_tuples;
                     sts_begin_parallel_scan(inner_tuples);
                     while ((tuple = sts_parallel_scan_next(inner_tuples,
                                                            &hashvalue)))
                     {
                         ExecForceStoreMinimalTuple(tuple,
                                                    hjstate->hj_HashTupleSlot,
                                                    false);
                         slot = hjstate->hj_HashTupleSlot;
                         ExecParallelHashTableInsertCurrentBatch(hashtable, slot,
                                                                 hashvalue);
                     }
                     sts_end_parallel_scan(inner_tuples);
                     BarrierArriveAndWait(batch_barrier,
                                          WAIT_EVENT_HASH_BATCH_LOAD);
                     /* Fall through. */
  
                 case PHJ_BATCH_PROBE:
  
                     /*
                      * This batch is ready to probe.  Return control to
                      * caller. We stay attached to batch_barrier so that the
                      * hash table stays alive until everyone's finished
                      * probing it, but no participant is allowed to wait at
                      * this barrier again (or else a deadlock could occur).
                      * All attached participants must eventually detach from
                      * the barrier and one worker must advance the phase so
                      * that the final phase is reached.
                      */
                     ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
                     sts_begin_parallel_scan(hashtable->batches[batchno].outer_tuples);
  
                     return true;
                 case PHJ_BATCH_SCAN:
  
                     /*
                      * In principle, we could help scan for unmatched tuples,
                      * since that phase is already underway (the thing we
                      * can't do under current deadlock-avoidance rules is wait
                      * for others to arrive at PHJ_BATCH_SCAN, because
                      * PHJ_BATCH_PROBE emits tuples, but in this case we just
                      * got here without waiting).  That is not yet done.  For
                      * now, we just detach and go around again.  We have to
                      * use ExecHashTableDetachBatch() because there's a small
                      * chance we'll be the last to detach, and then we're
                      * responsible for freeing memory.
                      */
                     ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
                     hashtable->batches[batchno].done = true;
                     ExecHashTableDetachBatch(hashtable);
                     break;
  
                 case PHJ_BATCH_FREE:
  
                     /*
                      * Already done.  Detach and go around again (if any
                      * remain).
                      */
                     BarrierDetach(batch_barrier);
                     hashtable->batches[batchno].done = true;
                     hashtable->curbatch = -1;
                     break;
  
                 default:
                     elog(ERROR, "unexpected batch phase %d",
                          BarrierPhase(batch_barrier));
             }
         }
         batchno = (batchno + 1) % hashtable->nbatch;
     } while (batchno != start_batchno);
  
     return false;
 }

References BarrierArriveAndWait(), BarrierAttach(), BarrierDetach(), BarrierPhase(), ParallelHashJoinBatch::batch_barrier, HashJoinTableData::batches, HashJoinTableData::curbatch, ParallelHashJoinState::distributor, ParallelHashJoinBatchAccessor::done, elog, ERROR, ExecForceStoreMinimalTuple(), ExecHashTableDetachBatch(), ExecParallelHashTableAlloc(), ExecParallelHashTableInsertCurrentBatch(), ExecParallelHashTableSetCurrentBatch(), HashJoinState::hj_HashTable, HashJoinState::hj_HashTupleSlot, ParallelHashJoinBatchAccessor::inner_tuples, HashJoinTableData::nbatch, ParallelHashJoinBatchAccessor::outer_tuples, HashJoinTableData::parallel_state, pg_atomic_fetch_add_u32(), PHJ_BATCH_ALLOCATE, PHJ_BATCH_ELECT, PHJ_BATCH_FREE, PHJ_BATCH_LOAD, PHJ_BATCH_PROBE, PHJ_BATCH_SCAN, ParallelHashJoinBatchAccessor::shared, sts_begin_parallel_scan(), sts_end_parallel_scan(), and sts_parallel_scan_next().

Referenced by ExecHashJoinImpl().

◆ ExecParallelHashJoinOuterGetTuple()

static TupleTableSlot * ExecParallelHashJoinOuterGetTuple	(	PlanState *	outerNode,
		HashJoinState *	hjstate,
		uint32 *	hashvalue
	)

static

Definition at line 1058 of file nodeHashjoin.c.

 {
     HashJoinTable hashtable = hjstate->hj_HashTable;
     int         curbatch = hashtable->curbatch;
     TupleTableSlot *slot;
  
     /*
      * In the Parallel Hash case we only run the outer plan directly for
      * single-batch hash joins.  Otherwise we have to go to batch files, even
      * for batch 0.
      */
     if (curbatch == 0 && hashtable->nbatch == 1)
     {
         slot = ExecProcNode(outerNode);
  
         while (!TupIsNull(slot))
         {
             bool        isnull;
  
             ExprContext *econtext = hjstate->js.ps.ps_ExprContext;
  
             econtext->ecxt_outertuple = slot;
  
             ResetExprContext(econtext);
  
             *hashvalue = DatumGetUInt32(ExecEvalExprSwitchContext(hjstate->hj_OuterHash,
                                                                   econtext,
                                                                   &isnull));
  
             if (!isnull)
                 return slot;
  
             /*
              * That tuple couldn't match because of a NULL, so discard it and
              * continue with the next one.
              */
             slot = ExecProcNode(outerNode);
         }
     }
     else if (curbatch < hashtable->nbatch)
     {
         MinimalTuple tuple;
  
         tuple = sts_parallel_scan_next(hashtable->batches[curbatch].outer_tuples,
                                        hashvalue);
         if (tuple != NULL)
         {
             ExecForceStoreMinimalTuple(tuple,
                                        hjstate->hj_OuterTupleSlot,
                                        false);
             slot = hjstate->hj_OuterTupleSlot;
             return slot;
         }
         else
             ExecClearTuple(hjstate->hj_OuterTupleSlot);
     }
  
     /* End of this batch */
     hashtable->batches[curbatch].outer_eof = true;
  
     return NULL;
 }

References HashJoinTableData::batches, HashJoinTableData::curbatch, DatumGetUInt32(), ExprContext::ecxt_outertuple, ExecClearTuple(), ExecEvalExprSwitchContext(), ExecForceStoreMinimalTuple(), ExecProcNode(), HashJoinState::hj_HashTable, HashJoinState::hj_OuterHash, HashJoinState::hj_OuterTupleSlot, HashJoinState::js, HashJoinTableData::nbatch, ParallelHashJoinBatchAccessor::outer_eof, ParallelHashJoinBatchAccessor::outer_tuples, JoinState::ps, PlanState::ps_ExprContext, ResetExprContext, sts_parallel_scan_next(), and TupIsNull.

Referenced by ExecHashJoinImpl().

◆ ExecParallelHashJoinPartitionOuter()

static void ExecParallelHashJoinPartitionOuter ( HashJoinState * hjstate )

static

Definition at line 1597 of file nodeHashjoin.c.

 {
     PlanState  *outerState = outerPlanState(hjstate);
     ExprContext *econtext = hjstate->js.ps.ps_ExprContext;
     HashJoinTable hashtable = hjstate->hj_HashTable;
     TupleTableSlot *slot;
     uint32      hashvalue;
     int         i;
  
     Assert(hjstate->hj_FirstOuterTupleSlot == NULL);
  
     /* Execute outer plan, writing all tuples to shared tuplestores. */
     for (;;)
     {
         bool        isnull;
  
         slot = ExecProcNode(outerState);
         if (TupIsNull(slot))
             break;
         econtext->ecxt_outertuple = slot;
  
         ResetExprContext(econtext);
  
         hashvalue = DatumGetUInt32(ExecEvalExprSwitchContext(hjstate->hj_OuterHash,
                                                              econtext,
                                                              &isnull));
  
         if (!isnull)
         {
             int         batchno;
             int         bucketno;
             bool        shouldFree;
             MinimalTuple mintup = ExecFetchSlotMinimalTuple(slot, &shouldFree);
  
             ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
                                       &batchno);
             sts_puttuple(hashtable->batches[batchno].outer_tuples,
                          &hashvalue, mintup);
  
             if (shouldFree)
                 heap_free_minimal_tuple(mintup);
         }
         CHECK_FOR_INTERRUPTS();
     }
  
     /* Make sure all outer partitions are readable by any backend. */
     for (i = 0; i < hashtable->nbatch; ++i)
         sts_end_write(hashtable->batches[i].outer_tuples);
 }

References Assert, HashJoinTableData::batches, CHECK_FOR_INTERRUPTS, DatumGetUInt32(), ExprContext::ecxt_outertuple, ExecEvalExprSwitchContext(), ExecFetchSlotMinimalTuple(), ExecHashGetBucketAndBatch(), ExecProcNode(), heap_free_minimal_tuple(), HashJoinState::hj_FirstOuterTupleSlot, HashJoinState::hj_HashTable, HashJoinState::hj_OuterHash, i, HashJoinState::js, HashJoinTableData::nbatch, ParallelHashJoinBatchAccessor::outer_tuples, outerPlanState, JoinState::ps, PlanState::ps_ExprContext, ResetExprContext, sts_end_write(), sts_puttuple(), and TupIsNull.

Referenced by ExecHashJoinImpl().

◆ ExecReScanHashJoin()

void ExecReScanHashJoin ( HashJoinState * node )

Definition at line 1494 of file nodeHashjoin.c.

 {
     PlanState  *outerPlan = outerPlanState(node);
     PlanState  *innerPlan = innerPlanState(node);
  
     /*
      * In a multi-batch join, we currently have to do rescans the hard way,
      * primarily because batch temp files may have already been released. But
      * if it's a single-batch join, and there is no parameter change for the
      * inner subnode, then we can just re-use the existing hash table without
      * rebuilding it.
      */
     if (node->hj_HashTable != NULL)
     {
         if (node->hj_HashTable->nbatch == 1 &&
             innerPlan->chgParam == NULL)
         {
             /*
              * Okay to reuse the hash table; needn't rescan inner, either.
              *
              * However, if it's a right/right-anti/full join, we'd better
              * reset the inner-tuple match flags contained in the table.
              */
             if (HJ_FILL_INNER(node))
                 ExecHashTableResetMatchFlags(node->hj_HashTable);
  
             /*
              * Also, we need to reset our state about the emptiness of the
              * outer relation, so that the new scan of the outer will update
              * it correctly if it turns out to be empty this time. (There's no
              * harm in clearing it now because ExecHashJoin won't need the
              * info.  In the other cases, where the hash table doesn't exist
              * or we are destroying it, we leave this state alone because
              * ExecHashJoin will need it the first time through.)
              */
             node->hj_OuterNotEmpty = false;
  
             /* ExecHashJoin can skip the BUILD_HASHTABLE step */
             node->hj_JoinState = HJ_NEED_NEW_OUTER;
         }
         else
         {
             /* must destroy and rebuild hash table */
             HashState  *hashNode = castNode(HashState, innerPlan);
  
             Assert(hashNode->hashtable == node->hj_HashTable);
             /* accumulate stats from old hash table, if wanted */
             /* (this should match ExecShutdownHash) */
             if (hashNode->ps.instrument && !hashNode->hinstrument)
                 hashNode->hinstrument = (HashInstrumentation *)
                     palloc0(sizeof(HashInstrumentation));
             if (hashNode->hinstrument)
                 ExecHashAccumInstrumentation(hashNode->hinstrument,
                                              hashNode->hashtable);
             /* for safety, be sure to clear child plan node's pointer too */
             hashNode->hashtable = NULL;
  
             ExecHashTableDestroy(node->hj_HashTable);
             node->hj_HashTable = NULL;
             node->hj_JoinState = HJ_BUILD_HASHTABLE;
  
             /*
              * if chgParam of subnode is not null then plan will be re-scanned
              * by first ExecProcNode.
              */
             if (innerPlan->chgParam == NULL)
                 ExecReScan(innerPlan);
         }
     }
  
     /* Always reset intra-tuple state */
     node->hj_CurHashValue = 0;
     node->hj_CurBucketNo = 0;
     node->hj_CurSkewBucketNo = INVALID_SKEW_BUCKET_NO;
     node->hj_CurTuple = NULL;
  
     node->hj_MatchedOuter = false;
     node->hj_FirstOuterTupleSlot = NULL;
  
     /*
      * if chgParam of subnode is not null then plan will be re-scanned by
      * first ExecProcNode.
      */
     if (outerPlan->chgParam == NULL)
         ExecReScan(outerPlan);
 }

Referenced by ExecReScan().

◆ ExecShutdownHashJoin()

void ExecShutdownHashJoin ( HashJoinState * node )

Definition at line 1582 of file nodeHashjoin.c.

 {
     if (node->hj_HashTable)
     {
         /*
          * Detach from shared state before DSM memory goes away.  This makes
          * sure that we don't have any pointers into DSM memory by the time
          * ExecEndHashJoin runs.
          */
         ExecHashTableDetachBatch(node->hj_HashTable);
         ExecHashTableDetach(node->hj_HashTable);
     }
 }

References ExecHashTableDetach(), ExecHashTableDetachBatch(), and HashJoinState::hj_HashTable.

Referenced by ExecShutdownNode_walker().

Macros

Functions

Macro Definition Documentation

◆ HJ_BUILD_HASHTABLE

◆ HJ_FILL_INNER

◆ HJ_FILL_INNER_TUPLES

◆ HJ_FILL_OUTER

◆ HJ_FILL_OUTER_TUPLE

◆ HJ_NEED_NEW_BATCH

◆ HJ_NEED_NEW_OUTER

◆ HJ_SCAN_BUCKET

Function Documentation

◆ ExecEndHashJoin()

◆ ExecHashJoin()

◆ ExecHashJoinEstimate()

◆ ExecHashJoinGetSavedTuple()

◆ ExecHashJoinImpl()

◆ ExecHashJoinInitializeDSM()

◆ ExecHashJoinInitializeWorker()

◆ ExecHashJoinNewBatch()

◆ ExecHashJoinOuterGetTuple()

◆ ExecHashJoinReInitializeDSM()

◆ ExecHashJoinSaveTuple()

◆ ExecInitHashJoin()

◆ ExecParallelHashJoin()

◆ ExecParallelHashJoinNewBatch()

◆ ExecParallelHashJoinOuterGetTuple()

◆ ExecParallelHashJoinPartitionOuter()

◆ ExecReScanHashJoin()

◆ ExecShutdownHashJoin()