On Sat, Jul 10, 2021 at 9:13 AM vignesh C <vignes...@gmail.com> wrote:
>
> On Mon, May 31, 2021 at 10:47 AM Greg Nancarrow <gregn4...@gmail.com> wrote:
> >
> > On Sat, Mar 6, 2021 at 12:31 PM Thomas Munro <thomas.mu...@gmail.com> wrote:
> > >
> > > On Tue, Mar 2, 2021 at 11:27 PM Thomas Munro <thomas.mu...@gmail.com>
> > > wrote:
> > > > On Fri, Feb 12, 2021 at 11:02 AM Melanie Plageman
> > > > <melanieplage...@gmail.com> wrote:
> > > > > I just attached the diff.
> > > >
> > > > Squashed into one patch for the cfbot to chew on, with a few minor
> > > > adjustments to a few comments.
> > >
> > > I did some more minor tidying of comments and naming. It's been on my
> > > to-do-list to update some phase names after commit 3048898e, and while
> > > doing that I couldn't resist the opportunity to change DONE to FREE,
> > > which somehow hurts my brain less, and makes much more obvious sense
> > > after the bugfix in CF #3031 that splits DONE into two separate
> > > phases. It also pairs obviously with ALLOCATE. I include a copy of
> > > that bugix here too as 0001, because I'll likely commit that first, so
> > > I rebased the stack of patches that way. 0002 includes the renaming I
> > > propose (master only). Then 0003 is Melanie's patch, using the name
> > > SCAN for the new match bit scan phase. I've attached an updated
> > > version of my "phase diagram" finger painting, to show how it looks
> > > with these three patches. "scan*" is new.
> >
> > Patches 0002, 0003 no longer apply to the master branch, seemingly
> > because of subsequent changes to pgstat, so need rebasing.
>
> I am changing the status to "Waiting on Author" as the patch does not
> apply on Head.
>
> Regards,
> Vignesh
>
>
Rebased patches attached. I will change status back to "Ready for Committer"
From 954bc84ca79e217c216c0ed8160853c49c19b609 Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Wed, 4 Nov 2020 14:25:33 -0800
Subject: [PATCH v7 3/3] Parallel Hash {Full,Right} Outer Join.
Previously, parallel full and right outer joins were not supported due
to a potential deadlock hazard (see discussion).
For now, sidestep the problem by terminating parallelism for the
unmatched inner tuple scan. The last process to arrive at the barrier
prepares for the unmatched inner tuple scan in HJ_NEED_NEW_OUTER and
transitions to HJ_FILL_INNER, scanning the hash table and emitting
unmatched inner tuples. Other processes are free to go and work on
other batches, if there are any.
To make parallel and serial hash join more consistent, change the serial
version to scan match bits in tuple chunk order, instead of doing it in
hash table bucket order.
Author: Melanie Plageman <melanieplage...@gmail.com>
Reviewed-by: Thomas Munro <thomas.mu...@gmail.com>
Discussion: https://postgr.es/m/CA%2BhUKG%2BA6ftXPz4oe92%2Bx8Er%2BxpGZqto70-Q_ERwRaSyA%3DafNg%40mail.gmail.com
---
src/backend/executor/nodeHash.c | 205 ++++++++++++++++++------
src/backend/executor/nodeHashjoin.c | 59 +++----
src/backend/optimizer/path/joinpath.c | 14 +-
src/include/executor/hashjoin.h | 15 +-
src/include/executor/nodeHash.h | 3 +
src/test/regress/expected/join_hash.out | 56 ++++++-
src/test/regress/sql/join_hash.sql | 23 ++-
7 files changed, 283 insertions(+), 92 deletions(-)
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index 3ba688e8e0..e7d420ee12 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -517,6 +517,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
hashtable->spaceAllowed * SKEW_HASH_MEM_PERCENT / 100;
hashtable->chunks = NULL;
hashtable->current_chunk = NULL;
+ hashtable->current_chunk_idx = 0;
hashtable->parallel_state = state->parallel_state;
hashtable->area = state->ps.state->es_query_dsa;
hashtable->batches = NULL;
@@ -2070,16 +2071,72 @@ void
ExecPrepHashTableForUnmatched(HashJoinState *hjstate)
{
/*----------
- * During this scan we use the HashJoinState fields as follows:
+ * During this scan we use the HashJoinTable fields as follows:
*
- * hj_CurBucketNo: next regular bucket to scan
- * hj_CurSkewBucketNo: next skew bucket (an index into skewBucketNums)
- * hj_CurTuple: last tuple returned, or NULL to start next bucket
+ * current_chunk: current HashMemoryChunk to scan
+ * current_chunk_idx: index in current HashMemoryChunk
*----------
*/
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+
hjstate->hj_CurBucketNo = 0;
hjstate->hj_CurSkewBucketNo = 0;
hjstate->hj_CurTuple = NULL;
+ hashtable->current_chunk = hashtable->chunks;
+ hashtable->current_chunk_idx = 0;
+}
+
+/*
+ * ExecParallelPrepHashTableForUnmatched
+ * set up for a series of ExecParallelScanHashTableForUnmatched calls
+ * return true if this worker is elected to do the unmatched inner scan
+ */
+bool
+ExecParallelPrepHashTableForUnmatched(HashJoinState *hjstate)
+{
+ /*----------
+ * During this scan we use the ParallelHashJoinBatchAccessor fields for the
+ * current batch as follows:
+ *
+ * current_chunk: current HashMemoryChunk to scan
+ * current_chunk_idx: index in current HashMemoryChunk
+ *----------
+ */
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+ int curbatch = hashtable->curbatch;
+ ParallelHashJoinBatchAccessor *batch_accessor = &hashtable->batches[curbatch];
+ ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
+ bool last = false;
+
+ hjstate->hj_CurBucketNo = 0;
+ hjstate->hj_CurSkewBucketNo = 0;
+ hjstate->hj_CurTuple = NULL;
+ if (curbatch < 0)
+ return false;
+ last = BarrierArriveAndDetachExceptLast(&batch->batch_barrier);
+ if (!last)
+ {
+ hashtable->batches[hashtable->curbatch].done = true;
+ /* Make sure any temporary files are closed. */
+ sts_end_parallel_scan(hashtable->batches[curbatch].inner_tuples);
+ sts_end_parallel_scan(hashtable->batches[curbatch].outer_tuples);
+
+ /*
+ * Track the largest batch we've been attached to. Though each
+ * backend might see a different subset of batches, explain.c will
+ * scan the results from all backends to find the largest value.
+ */
+ hashtable->spacePeak =
+ Max(hashtable->spacePeak, batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
+ hashtable->curbatch = -1;
+ }
+ else
+ {
+ batch_accessor->shared_chunk = batch->chunks;
+ batch_accessor->current_chunk = dsa_get_address(hashtable->area, batch_accessor->shared_chunk);
+ batch_accessor->current_chunk_idx = 0;
+ }
+ return last;
}
/*
@@ -2093,60 +2150,110 @@ ExecPrepHashTableForUnmatched(HashJoinState *hjstate)
bool
ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
{
+ HashMemoryChunk next;
HashJoinTable hashtable = hjstate->hj_HashTable;
- HashJoinTuple hashTuple = hjstate->hj_CurTuple;
- for (;;)
+ while (hashtable->current_chunk)
{
- /*
- * hj_CurTuple is the address of the tuple last returned from the
- * current bucket, or NULL if it's time to start scanning a new
- * bucket.
- */
- if (hashTuple != NULL)
- hashTuple = hashTuple->next.unshared;
- else if (hjstate->hj_CurBucketNo < hashtable->nbuckets)
- {
- hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
- hjstate->hj_CurBucketNo++;
- }
- else if (hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
+ while (hashtable->current_chunk_idx < hashtable->current_chunk->used)
{
- int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
+ HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(hashtable->current_chunk) +
+ hashtable->current_chunk_idx);
+ MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
+ int hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len);
+
+ /* next tuple in this chunk */
+ hashtable->current_chunk_idx += MAXALIGN(hashTupleSize);
+
+ if (HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
+ continue;
+
+ /* insert hashtable's tuple into exec slot */
+ econtext->ecxt_innertuple =
+ ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ hjstate->hj_HashTupleSlot,
+ false);
+
+ /*
+ * Reset temp memory each time; although this function doesn't do
+ * any qual eval, the caller will, so let's keep it parallel to
+ * ExecScanHashBucket.
+ */
+ ResetExprContext(econtext);
- hashTuple = hashtable->skewBucket[j]->tuples;
- hjstate->hj_CurSkewBucketNo++;
+ hjstate->hj_CurTuple = hashTuple;
+ return true;
}
- else
- break; /* finished all buckets */
- while (hashTuple != NULL)
+ next = hashtable->current_chunk->next.unshared;
+ hashtable->current_chunk = next;
+ hashtable->current_chunk_idx = 0;
+
+ /* allow this loop to be cancellable */
+ CHECK_FOR_INTERRUPTS();
+ }
+
+ /*
+ * no more unmatched tuples
+ */
+ return false;
+}
+
+/*
+ * ExecParallelScanHashTableForUnmatched
+ * scan the hash table for unmatched inner tuples, in parallel
+ *
+ * On success, the inner tuple is stored into hjstate->hj_CurTuple and
+ * econtext->ecxt_innertuple, using hjstate->hj_HashTupleSlot as the slot
+ * for the latter.
+ */
+bool
+ExecParallelScanHashTableForUnmatched(HashJoinState *hjstate,
+ ExprContext *econtext)
+{
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+ int curbatch = hashtable->curbatch;
+ ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[curbatch];
+
+ /*
+ * Only one worker should execute this function. Since tuples have already
+ * been emitted, it is hazardous for workers to wait at the batch_barrier
+ * again. Instead, all workers except the last will detach and the last
+ * will conduct this unmatched inner tuple scan.
+ */
+ Assert(BarrierParticipants(&accessor->shared->batch_barrier) == 1);
+ while (accessor->current_chunk)
+ {
+ while (accessor->current_chunk_idx < accessor->current_chunk->used)
{
- if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
- {
- TupleTableSlot *inntuple;
+ HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(accessor->current_chunk) +
+ accessor->current_chunk_idx);
- /* insert hashtable's tuple into exec slot */
- inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
- hjstate->hj_HashTupleSlot,
- false); /* do not pfree */
- econtext->ecxt_innertuple = inntuple;
+ accessor->current_chunk_idx += MAXALIGN(HJTUPLE_OVERHEAD +
+ HJTUPLE_MINTUPLE(hashTuple)->t_len);
- /*
- * Reset temp memory each time; although this function doesn't
- * do any qual eval, the caller will, so let's keep it
- * parallel to ExecScanHashBucket.
- */
- ResetExprContext(econtext);
+ if (HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
+ continue;
- hjstate->hj_CurTuple = hashTuple;
- return true;
- }
+ /* insert hashtable's tuple into exec slot */
+ econtext->ecxt_innertuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ hjstate->hj_HashTupleSlot, false);
+
+ /*
+ * Reset temp memory each time; although this function doesn't do
+ * any qual eval, the caller will, so let's keep it parallel to
+ * ExecScanHashBucket.
+ */
+ ResetExprContext(econtext);
- hashTuple = hashTuple->next.unshared;
+ hjstate->hj_CurTuple = hashTuple;
+ return true;
}
- /* allow this loop to be cancellable */
+ accessor->shared_chunk = accessor->current_chunk->next.shared;
+ accessor->current_chunk = dsa_get_address(hashtable->area, accessor->shared_chunk);
+ accessor->current_chunk_idx = 0;
+
CHECK_FOR_INTERRUPTS();
}
@@ -3152,13 +3259,6 @@ ExecHashTableDetachBatch(HashJoinTable hashtable)
/* Detach from the batch we were last working on. */
if (BarrierArriveAndDetach(&batch->batch_barrier))
{
- /*
- * Technically we shouldn't access the barrier because we're no
- * longer attached, but since there is no way it's moving after
- * this point it seems safe to make the following assertion.
- */
- Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_FREE);
-
/* Free shared chunks and buckets. */
while (DsaPointerIsValid(batch->chunks))
{
@@ -3305,6 +3405,9 @@ ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno)
hashtable->current_chunk = NULL;
hashtable->current_chunk_shared = InvalidDsaPointer;
hashtable->batches[batchno].at_least_one_chunk = false;
+ hashtable->batches[batchno].shared_chunk = InvalidDsaPointer;
+ hashtable->batches[batchno].current_chunk = NULL;
+ hashtable->batches[batchno].current_chunk_idx = 0;
}
/*
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
index 3b1553fefe..db80bdf80f 100644
--- a/src/backend/executor/nodeHashjoin.c
+++ b/src/backend/executor/nodeHashjoin.c
@@ -86,6 +86,7 @@
* PHJ_BATCH_ALLOCATE* -- one allocates buckets
* PHJ_BATCH_LOAD -- all load the hash table from disk
* PHJ_BATCH_PROBE -- all probe
+ * PHJ_BATCH_SCAN* -- full/right outer scan
* PHJ_BATCH_FREE* -- one frees memory
*
* Batch 0 is a special case, because it starts out in phase
@@ -103,9 +104,10 @@
* to a barrier, unless the barrier has reached a phase that means that no
* process will wait on it again. We emit tuples while attached to the build
* barrier in phase PHJ_BUILD_RUN, and to a per-batch barrier in phase
- * PHJ_BATCH_PROBE. These are advanced to PHJ_BUILD_FREE and PHJ_BATCH_FREE
- * respectively without waiting, using BarrierArriveAndDetach(). The last to
- * detach receives a different return value so that it knows that it's safe to
+ * PHJ_BATCH_PROBE. These are advanced to PHJ_BUILD_FREE and PHJ_BATCH_SCAN
+ * respectively without waiting, using BarrierArriveAndDetach() and
+ * BarrierArriveAndDetachExceptLast() respectively. The last to detach
+ * receives a different return value so that it knows that it's safe to
* clean up. Any straggler process that attaches after that phase is reached
* will see that it's too late to participate or access the relevant shared
* memory objects.
@@ -379,9 +381,19 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
/* end of batch, or maybe whole join */
if (HJ_FILL_INNER(node))
{
- /* set up to scan for unmatched inner tuples */
- ExecPrepHashTableForUnmatched(node);
- node->hj_JoinState = HJ_FILL_INNER_TUPLES;
+ if (parallel)
+ {
+ if (ExecParallelPrepHashTableForUnmatched(node))
+ node->hj_JoinState = HJ_FILL_INNER_TUPLES;
+ else
+ node->hj_JoinState = HJ_NEED_NEW_BATCH;
+ }
+ else
+ {
+ /* set up to scan for unmatched inner tuples */
+ ExecPrepHashTableForUnmatched(node);
+ node->hj_JoinState = HJ_FILL_INNER_TUPLES;
+ }
}
else
node->hj_JoinState = HJ_NEED_NEW_BATCH;
@@ -474,25 +486,13 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
{
node->hj_MatchedOuter = true;
- if (parallel)
- {
- /*
- * Full/right outer joins are currently not supported
- * for parallel joins, so we don't need to set the
- * match bit. Experiments show that it's worth
- * avoiding the shared memory traffic on large
- * systems.
- */
- Assert(!HJ_FILL_INNER(node));
- }
- else
- {
- /*
- * This is really only needed if HJ_FILL_INNER(node),
- * but we'll avoid the branch and just set it always.
- */
+
+ /*
+ * This is really only needed if HJ_FILL_INNER(node), 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)
@@ -550,7 +550,8 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
* so any unmatched inner tuples in the hashtable have to be
* emitted before we continue to the next batch.
*/
- if (!ExecScanHashTableForUnmatched(node, econtext))
+ if (!(parallel ? ExecParallelScanHashTableForUnmatched(node, econtext)
+ : ExecScanHashTableForUnmatched(node, econtext)))
{
/* no more unmatched tuples */
node->hj_JoinState = HJ_NEED_NEW_BATCH;
@@ -1184,13 +1185,15 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
* 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 call
- * BarrierArriveAndDetach() so that the final phase
- * PHJ_BATCH_FREE can be reached.
+ * 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:
+ /* Fall through. */
case PHJ_BATCH_FREE:
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index 6407ede12a..06cee346d1 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -2091,15 +2091,9 @@ hash_inner_and_outer(PlannerInfo *root,
* able to properly guarantee uniqueness. Similarly, we can't handle
* JOIN_FULL and JOIN_RIGHT, because they can produce false null
* extended rows. Also, the resulting path must not be parameterized.
- * We would be able to support JOIN_FULL and JOIN_RIGHT for Parallel
- * Hash, since in that case we're back to a single hash table with a
- * single set of match bits for each batch, but that will require
- * figuring out a deadlock-free way to wait for the probe to finish.
*/
if (joinrel->consider_parallel &&
save_jointype != JOIN_UNIQUE_OUTER &&
- save_jointype != JOIN_FULL &&
- save_jointype != JOIN_RIGHT &&
outerrel->partial_pathlist != NIL &&
bms_is_empty(joinrel->lateral_relids))
{
@@ -2133,9 +2127,13 @@ hash_inner_and_outer(PlannerInfo *root,
* total inner path will also be parallel-safe, but if not, we'll
* have to search for the cheapest safe, unparameterized inner
* path. If doing JOIN_UNIQUE_INNER, we can't use any alternative
- * inner path.
+ * inner path. If full or right join, we can't use parallelism
+ * (building the hash table in each backend) because no one
+ * process has all the match bits.
*/
- if (cheapest_total_inner->parallel_safe)
+ if (save_jointype == JOIN_FULL || save_jointype == JOIN_RIGHT)
+ cheapest_safe_inner = NULL;
+ else if (cheapest_total_inner->parallel_safe)
cheapest_safe_inner = cheapest_total_inner;
else if (save_jointype != JOIN_UNIQUE_INNER)
cheapest_safe_inner =
diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h
index 176fbef149..fb0f1339d3 100644
--- a/src/include/executor/hashjoin.h
+++ b/src/include/executor/hashjoin.h
@@ -205,6 +205,15 @@ typedef struct ParallelHashJoinBatchAccessor
bool at_least_one_chunk; /* has this backend allocated a chunk? */
bool done; /* flag to remember that a batch is done */
+
+ /*
+ * While doing the unmatched inner scan, the assigned worker may emit
+ * tuples. Thus, we must keep track of where it was in the hashtable so it
+ * can return to the correct offset within the correct chunk.
+ */
+ dsa_pointer shared_chunk;
+ HashMemoryChunk current_chunk;
+ size_t current_chunk_idx;
SharedTuplestoreAccessor *inner_tuples;
SharedTuplestoreAccessor *outer_tuples;
} ParallelHashJoinBatchAccessor;
@@ -266,7 +275,8 @@ typedef struct ParallelHashJoinState
#define PHJ_BATCH_ALLOCATE 1
#define PHJ_BATCH_LOAD 2
#define PHJ_BATCH_PROBE 3
-#define PHJ_BATCH_FREE 4
+#define PHJ_BATCH_SCAN 4
+#define PHJ_BATCH_FREE 5
/* The phases of batch growth while hashing, for grow_batches_barrier. */
#define PHJ_GROW_BATCHES_ELECT 0
@@ -352,6 +362,9 @@ typedef struct HashJoinTableData
/* used for dense allocation of tuples (into linked chunks) */
HashMemoryChunk chunks; /* one list for the whole batch */
+ /* index of tuple within current chunk for serial unmatched inner scan */
+ size_t current_chunk_idx;
+
/* Shared and private state for Parallel Hash. */
HashMemoryChunk current_chunk; /* this backend's current chunk */
dsa_area *area; /* DSA area to allocate memory from */
diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h
index 3fbe02e80d..7460dfff64 100644
--- a/src/include/executor/nodeHash.h
+++ b/src/include/executor/nodeHash.h
@@ -56,8 +56,11 @@ extern void ExecHashGetBucketAndBatch(HashJoinTable hashtable,
extern bool ExecScanHashBucket(HashJoinState *hjstate, ExprContext *econtext);
extern bool ExecParallelScanHashBucket(HashJoinState *hjstate, ExprContext *econtext);
extern void ExecPrepHashTableForUnmatched(HashJoinState *hjstate);
+extern bool ExecParallelPrepHashTableForUnmatched(HashJoinState *hjstate);
extern bool ExecScanHashTableForUnmatched(HashJoinState *hjstate,
ExprContext *econtext);
+extern bool ExecParallelScanHashTableForUnmatched(HashJoinState *hjstate,
+ ExprContext *econtext);
extern void ExecHashTableReset(HashJoinTable hashtable);
extern void ExecHashTableResetMatchFlags(HashJoinTable hashtable);
extern void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
diff --git a/src/test/regress/expected/join_hash.out b/src/test/regress/expected/join_hash.out
index 3a91c144a2..4ca0e01756 100644
--- a/src/test/regress/expected/join_hash.out
+++ b/src/test/regress/expected/join_hash.out
@@ -767,8 +767,9 @@ select count(*) from simple r full outer join simple s using (id);
(1 row)
rollback to settings;
--- parallelism not possible with parallel-oblivious outer hash join
+-- parallelism not possible with parallel-oblivious full hash join
savepoint settings;
+set enable_parallel_hash = off;
set local max_parallel_workers_per_gather = 2;
explain (costs off)
select count(*) from simple r full outer join simple s using (id);
@@ -788,6 +789,31 @@ select count(*) from simple r full outer join simple s using (id);
20000
(1 row)
+rollback to settings;
+-- parallelism is possible with parallel-aware full hash join
+savepoint settings;
+set local max_parallel_workers_per_gather = 2;
+explain (costs off)
+ select count(*) from simple r full outer join simple s using (id);
+ QUERY PLAN
+-------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather
+ Workers Planned: 2
+ -> Partial Aggregate
+ -> Parallel Hash Full Join
+ Hash Cond: (r.id = s.id)
+ -> Parallel Seq Scan on simple r
+ -> Parallel Hash
+ -> Parallel Seq Scan on simple s
+(9 rows)
+
+select count(*) from simple r full outer join simple s using (id);
+ count
+-------
+ 20000
+(1 row)
+
rollback to settings;
-- An full outer join where every record is not matched.
-- non-parallel
@@ -812,8 +838,9 @@ select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
(1 row)
rollback to settings;
--- parallelism not possible with parallel-oblivious outer hash join
+-- parallelism not possible with parallel-oblivious full hash join
savepoint settings;
+set enable_parallel_hash = off;
set local max_parallel_workers_per_gather = 2;
explain (costs off)
select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
@@ -833,6 +860,31 @@ select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
40000
(1 row)
+rollback to settings;
+-- parallelism is possible with parallel-aware full hash join
+savepoint settings;
+set local max_parallel_workers_per_gather = 2;
+explain (costs off)
+ select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
+ QUERY PLAN
+-------------------------------------------------------------
+ Finalize Aggregate
+ -> Gather
+ Workers Planned: 2
+ -> Partial Aggregate
+ -> Parallel Hash Full Join
+ Hash Cond: ((0 - s.id) = r.id)
+ -> Parallel Seq Scan on simple s
+ -> Parallel Hash
+ -> Parallel Seq Scan on simple r
+(9 rows)
+
+select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
+ count
+-------
+ 40000
+(1 row)
+
rollback to settings;
-- exercise special code paths for huge tuples (note use of non-strict
-- expression and left join required to get the detoasted tuple into
diff --git a/src/test/regress/sql/join_hash.sql b/src/test/regress/sql/join_hash.sql
index 68c1a8c7b6..504b3611ca 100644
--- a/src/test/regress/sql/join_hash.sql
+++ b/src/test/regress/sql/join_hash.sql
@@ -418,7 +418,16 @@ explain (costs off)
select count(*) from simple r full outer join simple s using (id);
rollback to settings;
--- parallelism not possible with parallel-oblivious outer hash join
+-- parallelism not possible with parallel-oblivious full hash join
+savepoint settings;
+set enable_parallel_hash = off;
+set local max_parallel_workers_per_gather = 2;
+explain (costs off)
+ select count(*) from simple r full outer join simple s using (id);
+select count(*) from simple r full outer join simple s using (id);
+rollback to settings;
+
+-- parallelism is possible with parallel-aware full hash join
savepoint settings;
set local max_parallel_workers_per_gather = 2;
explain (costs off)
@@ -436,14 +445,24 @@ explain (costs off)
select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
rollback to settings;
--- parallelism not possible with parallel-oblivious outer hash join
+-- parallelism not possible with parallel-oblivious full hash join
savepoint settings;
+set enable_parallel_hash = off;
set local max_parallel_workers_per_gather = 2;
explain (costs off)
select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
rollback to settings;
+-- parallelism is possible with parallel-aware full hash join
+savepoint settings;
+set local max_parallel_workers_per_gather = 2;
+explain (costs off)
+ select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
+select count(*) from simple r full outer join simple s on (r.id = 0 - s.id);
+rollback to settings;
+
+
-- exercise special code paths for huge tuples (note use of non-strict
-- expression and left join required to get the detoasted tuple into
-- the hash table)
--
2.27.0
From 783e83d52a9fd80cdd389308201e2b0824efb600 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.mu...@gmail.com>
Date: Fri, 2 Oct 2020 15:53:44 +1300
Subject: [PATCH v7 1/3] Fix race condition in parallel hash join batch
cleanup.
With unlucky timing and parallel_leader_participation off, PHJ could
attempt to access per-batch state just as it was being freed. There was
code intended to prevent that by checking for a cleared pointer, but it
was racy. Fix, by introducing an extra barrier phase. The new phase
PHJ_BUILD_RUNNING means that it's safe to access the per-batch state to
find a batch to help with, and PHJ_BUILD_DONE means that it is too late.
The last to detach will free the array of per-batch state as before, but
now it will also atomically advance the phase at the same time, so that
late attachers can avoid the hazard. This mirrors the way per-batch
hash tables are freed (see phases PHJ_BATCH_PROBING and PHJ_BATCH_DONE).
Revealed by a build farm failure, where BarrierAttach() failed a sanity
check assertion, because the memory had been clobbered by dsa_free().
Back-patch to all supported releases.
Reported-by: Michael Paquier <mich...@paquier.xyz>
Discussion: https://postgr.es/m/20200929061142.GA29096%40paquier.xyz
---
src/backend/executor/nodeHash.c | 47 ++++++++++++++++++++---------
src/backend/executor/nodeHashjoin.c | 40 ++++++++++++++----------
src/include/executor/hashjoin.h | 3 +-
3 files changed, 58 insertions(+), 32 deletions(-)
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index 73eb074cbf..db5d46b23f 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -333,14 +333,21 @@ MultiExecParallelHash(HashState *node)
hashtable->nbuckets = pstate->nbuckets;
hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
hashtable->totalTuples = pstate->total_tuples;
- ExecParallelHashEnsureBatchAccessors(hashtable);
+
+ /*
+ * Unless we're completely done and the batch state has been freed, make
+ * sure we have accessors.
+ */
+ if (BarrierPhase(build_barrier) < PHJ_BUILD_DONE)
+ ExecParallelHashEnsureBatchAccessors(hashtable);
/*
* The next synchronization point is in ExecHashJoin's HJ_BUILD_HASHTABLE
- * case, which will bring the build phase to PHJ_BUILD_DONE (if it isn't
+ * case, which will bring the build phase to PHJ_BUILD_RUNNING (if it isn't
* there already).
*/
Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER ||
+ BarrierPhase(build_barrier) == PHJ_BUILD_RUNNING ||
BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
}
@@ -624,7 +631,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
/*
* The next Parallel Hash synchronization point is in
* MultiExecParallelHash(), which will progress it all the way to
- * PHJ_BUILD_DONE. The caller must not return control from this
+ * PHJ_BUILD_RUNNING. The caller must not return control from this
* executor node between now and then.
*/
}
@@ -3065,14 +3072,11 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
}
/*
- * It's possible for a backend to start up very late so that the whole
- * join is finished and the shm state for tracking batches has already
- * been freed by ExecHashTableDetach(). In that case we'll just leave
- * hashtable->batches as NULL so that ExecParallelHashJoinNewBatch() gives
- * up early.
+ * We should never see a state where the batch-tracking array is freed,
+ * because we should have given up sooner if we join when the build barrier
+ * has reached the PHJ_BUILD_DONE phase.
*/
- if (!DsaPointerIsValid(pstate->batches))
- return;
+ Assert(DsaPointerIsValid(pstate->batches));
/* Use hash join memory context. */
oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
@@ -3192,9 +3196,17 @@ ExecHashTableDetachBatch(HashJoinTable hashtable)
void
ExecHashTableDetach(HashJoinTable hashtable)
{
- if (hashtable->parallel_state)
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+
+ /*
+ * If we're involved in a parallel query, we must either have got all the
+ * way to PHJ_BUILD_RUNNING, or joined too late and be in PHJ_BUILD_DONE.
+ */
+ Assert(!pstate ||
+ BarrierPhase(&pstate->build_barrier) >= PHJ_BUILD_RUNNING);
+
+ if (pstate && BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_RUNNING)
{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
int i;
/* Make sure any temporary files are closed. */
@@ -3210,17 +3222,22 @@ ExecHashTableDetach(HashJoinTable hashtable)
}
/* If we're last to detach, clean up shared memory. */
- if (BarrierDetach(&pstate->build_barrier))
+ if (BarrierArriveAndDetach(&pstate->build_barrier))
{
+ /*
+ * Late joining processes will see this state and give up
+ * immediately.
+ */
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_DONE);
+
if (DsaPointerIsValid(pstate->batches))
{
dsa_free(hashtable->area, pstate->batches);
pstate->batches = InvalidDsaPointer;
}
}
-
- hashtable->parallel_state = NULL;
}
+ hashtable->parallel_state = NULL;
}
/*
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
index 510bdd39ad..a9c263c071 100644
--- a/src/backend/executor/nodeHashjoin.c
+++ b/src/backend/executor/nodeHashjoin.c
@@ -45,7 +45,8 @@
* PHJ_BUILD_ALLOCATING -- one sets up the batches and table 0
* PHJ_BUILD_HASHING_INNER -- all hash the inner rel
* PHJ_BUILD_HASHING_OUTER -- (multi-batch only) all hash the outer
- * PHJ_BUILD_DONE -- building done, probing can begin
+ * PHJ_BUILD_RUNNING -- building done, probing can begin
+ * PHJ_BUILD_DONE -- all work complete, one frees batches
*
* While in the phase PHJ_BUILD_HASHING_INNER a separate pair of barriers may
* be used repeatedly as required to coordinate expansions in the number of
@@ -73,7 +74,7 @@
* batches whenever it encounters them while scanning and probing, which it
* can do because it processes batches in serial order.
*
- * Once PHJ_BUILD_DONE is reached, backends then split up and process
+ * Once PHJ_BUILD_RUNNING is reached, backends then split up and process
* different batches, or gang up and work together on probing batches if there
* aren't enough to go around. For each batch there is a separate barrier
* with the following phases:
@@ -95,11 +96,16 @@
*
* To avoid deadlocks, we never wait for any barrier unless it is known that
* all other backends attached to it are actively executing the node or have
- * already arrived. Practically, that means that we never return a tuple
- * while attached to a barrier, unless the barrier has reached its final
- * state. In the slightly special case of the per-batch barrier, we return
- * tuples while in PHJ_BATCH_PROBING phase, but that's OK because we use
- * BarrierArriveAndDetach() to advance it to PHJ_BATCH_DONE without waiting.
+ * finished. Practically, that means that we never emit a tuple while attached
+ * to a barrier, unless the barrier has reached a phase that means that no
+ * process will wait on it again. We emit tuples while attached to the build
+ * barrier in phase PHJ_BUILD_RUNNING, and to a per-batch barrier in phase
+ * PHJ_BATCH_PROBING. These are advanced to PHJ_BUILD_DONE and PHJ_BATCH_DONE
+ * respectively without waiting, using BarrierArriveAndDetach(). The last to
+ * detach receives a different return value so that it knows that it's safe to
+ * clean up. Any straggler process that attaches after that phase is reached
+ * will see that it's too late to participate or access the relevant shared
+ * memory objects.
*
*-------------------------------------------------------------------------
*/
@@ -317,6 +323,7 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
build_barrier = ¶llel_state->build_barrier;
Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER ||
+ BarrierPhase(build_barrier) == PHJ_BUILD_RUNNING ||
BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
if (BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER)
{
@@ -329,9 +336,18 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
BarrierArriveAndWait(build_barrier,
WAIT_EVENT_HASH_BUILD_HASH_OUTER);
}
- Assert(BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
+ else if (BarrierPhase(build_barrier) == PHJ_BUILD_DONE)
+ {
+ /*
+ * 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_RUNNING);
hashtable->curbatch = -1;
node->hj_JoinState = HJ_NEED_NEW_BATCH;
@@ -1090,14 +1106,6 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
int start_batchno;
int batchno;
- /*
- * If we started up so late that the batch tracking array has been freed
- * already by ExecHashTableDetach(), then we are finished. See also
- * ExecParallelHashEnsureBatchAccessors().
- */
- if (hashtable->batches == NULL)
- return false;
-
/*
* 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
diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h
index d74034f64f..d8edd39923 100644
--- a/src/include/executor/hashjoin.h
+++ b/src/include/executor/hashjoin.h
@@ -258,7 +258,8 @@ typedef struct ParallelHashJoinState
#define PHJ_BUILD_ALLOCATING 1
#define PHJ_BUILD_HASHING_INNER 2
#define PHJ_BUILD_HASHING_OUTER 3
-#define PHJ_BUILD_DONE 4
+#define PHJ_BUILD_RUNNING 4
+#define PHJ_BUILD_DONE 5
/* The phases for probing each batch, used by for batch_barrier. */
#define PHJ_BATCH_ELECTING 0
--
2.27.0
From 8771e5fe531023d86da67b539ff57d4845171cae Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.mu...@gmail.com>
Date: Sat, 6 Mar 2021 12:06:16 +1300
Subject: [PATCH v7 2/3] Improve the naming of Parallel Hash Join phases.
Commit 3048898e dropped -ING from some wait event names. Update the
corresponding barrier phases names to match.
While we're here making cosmetic changes, also rename "DONE" to "FREE".
That pairs better with "ALLOCATE", and describes the activity that
actually happens in that phase (as we do for the other phases) rather
than describing a state. As for the growth barriers, rename their
"ALLOCATE" phase to "REALLOCATE", which is probably a better description
of what happens then.
---
src/backend/executor/nodeHash.c | 68 ++++++++++----------
src/backend/executor/nodeHashjoin.c | 83 +++++++++++++------------
src/backend/utils/activity/wait_event.c | 8 +--
src/include/executor/hashjoin.h | 38 +++++------
src/include/utils/wait_event.h | 4 +-
5 files changed, 102 insertions(+), 99 deletions(-)
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index db5d46b23f..3ba688e8e0 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -246,10 +246,10 @@ MultiExecParallelHash(HashState *node)
*/
pstate = hashtable->parallel_state;
build_barrier = &pstate->build_barrier;
- Assert(BarrierPhase(build_barrier) >= PHJ_BUILD_ALLOCATING);
+ Assert(BarrierPhase(build_barrier) >= PHJ_BUILD_ALLOCATE);
switch (BarrierPhase(build_barrier))
{
- case PHJ_BUILD_ALLOCATING:
+ case PHJ_BUILD_ALLOCATE:
/*
* Either I just allocated the initial hash table in
@@ -259,7 +259,7 @@ MultiExecParallelHash(HashState *node)
BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ALLOCATE);
/* Fall through. */
- case PHJ_BUILD_HASHING_INNER:
+ case PHJ_BUILD_HASH_INNER:
/*
* It's time to begin hashing, or if we just arrived here then
@@ -271,10 +271,10 @@ MultiExecParallelHash(HashState *node)
* below.
*/
if (PHJ_GROW_BATCHES_PHASE(BarrierAttach(&pstate->grow_batches_barrier)) !=
- PHJ_GROW_BATCHES_ELECTING)
+ PHJ_GROW_BATCHES_ELECT)
ExecParallelHashIncreaseNumBatches(hashtable);
if (PHJ_GROW_BUCKETS_PHASE(BarrierAttach(&pstate->grow_buckets_barrier)) !=
- PHJ_GROW_BUCKETS_ELECTING)
+ PHJ_GROW_BUCKETS_ELECT)
ExecParallelHashIncreaseNumBuckets(hashtable);
ExecParallelHashEnsureBatchAccessors(hashtable);
ExecParallelHashTableSetCurrentBatch(hashtable, 0);
@@ -338,17 +338,17 @@ MultiExecParallelHash(HashState *node)
* Unless we're completely done and the batch state has been freed, make
* sure we have accessors.
*/
- if (BarrierPhase(build_barrier) < PHJ_BUILD_DONE)
+ if (BarrierPhase(build_barrier) < PHJ_BUILD_FREE)
ExecParallelHashEnsureBatchAccessors(hashtable);
/*
* The next synchronization point is in ExecHashJoin's HJ_BUILD_HASHTABLE
- * case, which will bring the build phase to PHJ_BUILD_RUNNING (if it isn't
+ * case, which will bring the build phase to PHJ_BUILD_RUN (if it isn't
* there already).
*/
- Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER ||
- BarrierPhase(build_barrier) == PHJ_BUILD_RUNNING ||
- BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
+ Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASH_OUTER ||
+ BarrierPhase(build_barrier) == PHJ_BUILD_RUN ||
+ BarrierPhase(build_barrier) == PHJ_BUILD_FREE);
}
/* ----------------------------------------------------------------
@@ -596,7 +596,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
* Attach to the build barrier. The corresponding detach operation is
* in ExecHashTableDetach. Note that we won't attach to the
* batch_barrier for batch 0 yet. We'll attach later and start it out
- * in PHJ_BATCH_PROBING phase, because batch 0 is allocated up front
+ * in PHJ_BATCH_PROBE phase, because batch 0 is allocated up front
* and then loaded while hashing (the standard hybrid hash join
* algorithm), and we'll coordinate that using build_barrier.
*/
@@ -610,7 +610,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
* SharedHashJoinBatch objects and the hash table for batch 0. One
* backend will be elected to do that now if necessary.
*/
- if (BarrierPhase(build_barrier) == PHJ_BUILD_ELECTING &&
+ if (BarrierPhase(build_barrier) == PHJ_BUILD_ELECT &&
BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ELECT))
{
pstate->nbatch = nbatch;
@@ -631,7 +631,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
/*
* The next Parallel Hash synchronization point is in
* MultiExecParallelHash(), which will progress it all the way to
- * PHJ_BUILD_RUNNING. The caller must not return control from this
+ * PHJ_BUILD_RUN. The caller must not return control from this
* executor node between now and then.
*/
}
@@ -1086,7 +1086,7 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
ParallelHashJoinState *pstate = hashtable->parallel_state;
int i;
- Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASH_INNER);
/*
* It's unlikely, but we need to be prepared for new participants to show
@@ -1095,7 +1095,7 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
*/
switch (PHJ_GROW_BATCHES_PHASE(BarrierPhase(&pstate->grow_batches_barrier)))
{
- case PHJ_GROW_BATCHES_ELECTING:
+ case PHJ_GROW_BATCHES_ELECT:
/*
* Elect one participant to prepare to grow the number of batches.
@@ -1211,13 +1211,13 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
}
/* Fall through. */
- case PHJ_GROW_BATCHES_ALLOCATING:
+ case PHJ_GROW_BATCHES_REALLOCATE:
/* Wait for the above to be finished. */
BarrierArriveAndWait(&pstate->grow_batches_barrier,
- WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATE);
+ WAIT_EVENT_HASH_GROW_BATCHES_REALLOCATE);
/* Fall through. */
- case PHJ_GROW_BATCHES_REPARTITIONING:
+ case PHJ_GROW_BATCHES_REPARTITION:
/* Make sure that we have the current dimensions and buckets. */
ExecParallelHashEnsureBatchAccessors(hashtable);
ExecParallelHashTableSetCurrentBatch(hashtable, 0);
@@ -1230,7 +1230,7 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
WAIT_EVENT_HASH_GROW_BATCHES_REPARTITION);
/* Fall through. */
- case PHJ_GROW_BATCHES_DECIDING:
+ case PHJ_GROW_BATCHES_DECIDE:
/*
* Elect one participant to clean up and decide whether further
@@ -1285,7 +1285,7 @@ ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
}
/* Fall through. */
- case PHJ_GROW_BATCHES_FINISHING:
+ case PHJ_GROW_BATCHES_FINISH:
/* Wait for the above to complete. */
BarrierArriveAndWait(&pstate->grow_batches_barrier,
WAIT_EVENT_HASH_GROW_BATCHES_FINISH);
@@ -1525,7 +1525,7 @@ ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
HashMemoryChunk chunk;
dsa_pointer chunk_s;
- Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASH_INNER);
/*
* It's unlikely, but we need to be prepared for new participants to show
@@ -1534,7 +1534,7 @@ ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
*/
switch (PHJ_GROW_BUCKETS_PHASE(BarrierPhase(&pstate->grow_buckets_barrier)))
{
- case PHJ_GROW_BUCKETS_ELECTING:
+ case PHJ_GROW_BUCKETS_ELECT:
/* Elect one participant to prepare to increase nbuckets. */
if (BarrierArriveAndWait(&pstate->grow_buckets_barrier,
WAIT_EVENT_HASH_GROW_BUCKETS_ELECT))
@@ -1563,13 +1563,13 @@ ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
}
/* Fall through. */
- case PHJ_GROW_BUCKETS_ALLOCATING:
+ case PHJ_GROW_BUCKETS_REALLOCATE:
/* Wait for the above to complete. */
BarrierArriveAndWait(&pstate->grow_buckets_barrier,
- WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE);
+ WAIT_EVENT_HASH_GROW_BUCKETS_REALLOCATE);
/* Fall through. */
- case PHJ_GROW_BUCKETS_REINSERTING:
+ case PHJ_GROW_BUCKETS_REINSERT:
/* Reinsert all tuples into the hash table. */
ExecParallelHashEnsureBatchAccessors(hashtable);
ExecParallelHashTableSetCurrentBatch(hashtable, 0);
@@ -1725,7 +1725,7 @@ retry:
/* Try to load it into memory. */
Assert(BarrierPhase(&hashtable->parallel_state->build_barrier) ==
- PHJ_BUILD_HASHING_INNER);
+ PHJ_BUILD_HASH_INNER);
hashTuple = ExecParallelHashTupleAlloc(hashtable,
HJTUPLE_OVERHEAD + tuple->t_len,
&shared);
@@ -2879,7 +2879,7 @@ ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size,
if (pstate->growth != PHJ_GROWTH_DISABLED)
{
Assert(curbatch == 0);
- Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASH_INNER);
/*
* Check if our space limit would be exceeded. To avoid choking on
@@ -2999,7 +2999,7 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
{
/* Batch 0 doesn't need to be loaded. */
BarrierAttach(&shared->batch_barrier);
- while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING)
+ while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBE)
BarrierArriveAndWait(&shared->batch_barrier, 0);
BarrierDetach(&shared->batch_barrier);
}
@@ -3074,7 +3074,7 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
/*
* We should never see a state where the batch-tracking array is freed,
* because we should have given up sooner if we join when the build barrier
- * has reached the PHJ_BUILD_DONE phase.
+ * has reached the PHJ_BUILD_FREE phase.
*/
Assert(DsaPointerIsValid(pstate->batches));
@@ -3157,7 +3157,7 @@ ExecHashTableDetachBatch(HashJoinTable hashtable)
* longer attached, but since there is no way it's moving after
* this point it seems safe to make the following assertion.
*/
- Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_DONE);
+ Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_FREE);
/* Free shared chunks and buckets. */
while (DsaPointerIsValid(batch->chunks))
@@ -3200,12 +3200,12 @@ ExecHashTableDetach(HashJoinTable hashtable)
/*
* If we're involved in a parallel query, we must either have got all the
- * way to PHJ_BUILD_RUNNING, or joined too late and be in PHJ_BUILD_DONE.
+ * way to PHJ_BUILD_RUN, or joined too late and be in PHJ_BUILD_FREE.
*/
Assert(!pstate ||
- BarrierPhase(&pstate->build_barrier) >= PHJ_BUILD_RUNNING);
+ BarrierPhase(&pstate->build_barrier) >= PHJ_BUILD_RUN);
- if (pstate && BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_RUNNING)
+ if (pstate && BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_RUN)
{
int i;
@@ -3228,7 +3228,7 @@ ExecHashTableDetach(HashJoinTable hashtable)
* Late joining processes will see this state and give up
* immediately.
*/
- Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_DONE);
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_FREE);
if (DsaPointerIsValid(pstate->batches))
{
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
index a9c263c071..3b1553fefe 100644
--- a/src/backend/executor/nodeHashjoin.c
+++ b/src/backend/executor/nodeHashjoin.c
@@ -39,27 +39,30 @@
*
* One barrier called build_barrier is used to coordinate the hashing phases.
* The phase is represented by an integer which begins at zero and increments
- * one by one, but in the code it is referred to by symbolic names as follows:
+ * one by one, but in the code it is referred to by symbolic names as follows.
+ * An asterisk indicates a phase that is performed by a single arbitrarily
+ * chosen process.
*
- * PHJ_BUILD_ELECTING -- initial state
- * PHJ_BUILD_ALLOCATING -- one sets up the batches and table 0
- * PHJ_BUILD_HASHING_INNER -- all hash the inner rel
- * PHJ_BUILD_HASHING_OUTER -- (multi-batch only) all hash the outer
- * PHJ_BUILD_RUNNING -- building done, probing can begin
- * PHJ_BUILD_DONE -- all work complete, one frees batches
+ * PHJ_BUILD_ELECT -- initial state
+ * PHJ_BUILD_ALLOCATE* -- one sets up the batches and table 0
+ * PHJ_BUILD_HASH_INNER -- all hash the inner rel
+ * PHJ_BUILD_HASH_OUTER -- (multi-batch only) all hash the outer
+ * PHJ_BUILD_RUN -- building done, probing can begin
+ * PHJ_BUILD_FREE* -- all work complete, one frees batches
*
- * While in the phase PHJ_BUILD_HASHING_INNER a separate pair of barriers may
+ * While in the phase PHJ_BUILD_HASH_INNER a separate pair of barriers may
* be used repeatedly as required to coordinate expansions in the number of
* batches or buckets. Their phases are as follows:
*
- * PHJ_GROW_BATCHES_ELECTING -- initial state
- * PHJ_GROW_BATCHES_ALLOCATING -- one allocates new batches
- * PHJ_GROW_BATCHES_REPARTITIONING -- all repartition
- * PHJ_GROW_BATCHES_FINISHING -- one cleans up, detects skew
+ * PHJ_GROW_BATCHES_ELECT -- initial state
+ * PHJ_GROW_BATCHES_REALLOCATE* -- one allocates new batches
+ * PHJ_GROW_BATCHES_REPARTITION -- all repartition
+ * PHJ_GROW_BATCHES_DECIDE* -- one detects skew and cleans up
+ * PHJ_GROW_BATCHES_FINISH -- finished one growth cycle
*
- * PHJ_GROW_BUCKETS_ELECTING -- initial state
- * PHJ_GROW_BUCKETS_ALLOCATING -- one allocates new buckets
- * PHJ_GROW_BUCKETS_REINSERTING -- all insert tuples
+ * PHJ_GROW_BUCKETS_ELECT -- initial state
+ * PHJ_GROW_BUCKETS_REALLOCATE* -- one allocates new buckets
+ * PHJ_GROW_BUCKETS_REINSERT -- all insert tuples
*
* If the planner got the number of batches and buckets right, those won't be
* necessary, but on the other hand we might finish up needing to expand the
@@ -67,27 +70,27 @@
* within our memory budget and load factor target. For that reason it's a
* separate pair of barriers using circular phases.
*
- * The PHJ_BUILD_HASHING_OUTER phase is required only for multi-batch joins,
+ * The PHJ_BUILD_HASH_OUTER phase is required only for multi-batch joins,
* because we need to divide the outer relation into batches up front in order
* to be able to process batches entirely independently. In contrast, the
* parallel-oblivious algorithm simply throws tuples 'forward' to 'later'
* batches whenever it encounters them while scanning and probing, which it
* can do because it processes batches in serial order.
*
- * Once PHJ_BUILD_RUNNING is reached, backends then split up and process
+ * Once PHJ_BUILD_RUN is reached, backends then split up and process
* different batches, or gang up and work together on probing batches if there
* aren't enough to go around. For each batch there is a separate barrier
* with the following phases:
*
- * PHJ_BATCH_ELECTING -- initial state
- * PHJ_BATCH_ALLOCATING -- one allocates buckets
- * PHJ_BATCH_LOADING -- all load the hash table from disk
- * PHJ_BATCH_PROBING -- all probe
- * PHJ_BATCH_DONE -- end
+ * PHJ_BATCH_ELECT -- initial state
+ * PHJ_BATCH_ALLOCATE* -- one allocates buckets
+ * PHJ_BATCH_LOAD -- all load the hash table from disk
+ * PHJ_BATCH_PROBE -- all probe
+ * PHJ_BATCH_FREE* -- one frees memory
*
* Batch 0 is a special case, because it starts out in phase
- * PHJ_BATCH_PROBING; populating batch 0's hash table is done during
- * PHJ_BUILD_HASHING_INNER so we can skip loading.
+ * PHJ_BATCH_PROBE; populating batch 0's hash table is done during
+ * PHJ_BUILD_HASH_INNER so we can skip loading.
*
* Initially we try to plan for a single-batch hash join using the combined
* hash_mem of all participants to create a large shared hash table. If that
@@ -99,8 +102,8 @@
* finished. Practically, that means that we never emit a tuple while attached
* to a barrier, unless the barrier has reached a phase that means that no
* process will wait on it again. We emit tuples while attached to the build
- * barrier in phase PHJ_BUILD_RUNNING, and to a per-batch barrier in phase
- * PHJ_BATCH_PROBING. These are advanced to PHJ_BUILD_DONE and PHJ_BATCH_DONE
+ * barrier in phase PHJ_BUILD_RUN, and to a per-batch barrier in phase
+ * PHJ_BATCH_PROBE. These are advanced to PHJ_BUILD_FREE and PHJ_BATCH_FREE
* respectively without waiting, using BarrierArriveAndDetach(). The last to
* detach receives a different return value so that it knows that it's safe to
* clean up. Any straggler process that attaches after that phase is reached
@@ -322,10 +325,10 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
Barrier *build_barrier;
build_barrier = ¶llel_state->build_barrier;
- Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER ||
- BarrierPhase(build_barrier) == PHJ_BUILD_RUNNING ||
- BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
- if (BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER)
+ 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
@@ -336,7 +339,7 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
BarrierArriveAndWait(build_barrier,
WAIT_EVENT_HASH_BUILD_HASH_OUTER);
}
- else if (BarrierPhase(build_barrier) == PHJ_BUILD_DONE)
+ else if (BarrierPhase(build_barrier) == PHJ_BUILD_FREE)
{
/*
* If we attached so late that the job is finished and
@@ -347,7 +350,7 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
}
/* Each backend should now select a batch to work on. */
- Assert(BarrierPhase(build_barrier) == PHJ_BUILD_RUNNING);
+ Assert(BarrierPhase(build_barrier) == PHJ_BUILD_RUN);
hashtable->curbatch = -1;
node->hj_JoinState = HJ_NEED_NEW_BATCH;
@@ -1139,7 +1142,7 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
switch (BarrierAttach(batch_barrier))
{
- case PHJ_BATCH_ELECTING:
+ case PHJ_BATCH_ELECT:
/* One backend allocates the hash table. */
if (BarrierArriveAndWait(batch_barrier,
@@ -1147,13 +1150,13 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
ExecParallelHashTableAlloc(hashtable, batchno);
/* Fall through. */
- case PHJ_BATCH_ALLOCATING:
+ case PHJ_BATCH_ALLOCATE:
/* Wait for allocation to complete. */
BarrierArriveAndWait(batch_barrier,
WAIT_EVENT_HASH_BATCH_ALLOCATE);
/* Fall through. */
- case PHJ_BATCH_LOADING:
+ case PHJ_BATCH_LOAD:
/* Start (or join in) loading tuples. */
ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
inner_tuples = hashtable->batches[batchno].inner_tuples;
@@ -1173,7 +1176,7 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
WAIT_EVENT_HASH_BATCH_LOAD);
/* Fall through. */
- case PHJ_BATCH_PROBING:
+ case PHJ_BATCH_PROBE:
/*
* This batch is ready to probe. Return control to
@@ -1183,13 +1186,13 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
* this barrier again (or else a deadlock could occur).
* All attached participants must eventually call
* BarrierArriveAndDetach() so that the final phase
- * PHJ_BATCH_DONE can be reached.
+ * PHJ_BATCH_FREE can be reached.
*/
ExecParallelHashTableSetCurrentBatch(hashtable, batchno);
sts_begin_parallel_scan(hashtable->batches[batchno].outer_tuples);
return true;
- case PHJ_BATCH_DONE:
+ case PHJ_BATCH_FREE:
/*
* Already done. Detach and go around again (if any
@@ -1516,7 +1519,7 @@ ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *cxt)
/*
* 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_DONE and batch 0's batch_barrier to PHJ_BATCH_PROBING, but
+ * 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
@@ -1535,7 +1538,7 @@ ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *cxt)
/* Clear any shared batch files. */
SharedFileSetDeleteAll(&pstate->fileset);
- /* Reset build_barrier to PHJ_BUILD_ELECTING so we can go around again. */
+ /* Reset build_barrier to PHJ_BUILD_ELECT so we can go around again. */
BarrierInit(&pstate->build_barrier, 0);
}
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index ef7e6bfb77..2f3c1a241b 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -361,8 +361,8 @@ pgstat_get_wait_ipc(WaitEventIPC w)
case WAIT_EVENT_HASH_BUILD_HASH_OUTER:
event_name = "HashBuildHashOuter";
break;
- case WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATE:
- event_name = "HashGrowBatchesAllocate";
+ case WAIT_EVENT_HASH_GROW_BATCHES_REALLOCATE:
+ event_name = "HashGrowBatchesReallocate";
break;
case WAIT_EVENT_HASH_GROW_BATCHES_DECIDE:
event_name = "HashGrowBatchesDecide";
@@ -376,8 +376,8 @@ pgstat_get_wait_ipc(WaitEventIPC w)
case WAIT_EVENT_HASH_GROW_BATCHES_REPARTITION:
event_name = "HashGrowBatchesRepartition";
break;
- case WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE:
- event_name = "HashGrowBucketsAllocate";
+ case WAIT_EVENT_HASH_GROW_BUCKETS_REALLOCATE:
+ event_name = "HashGrowBucketsReallocate";
break;
case WAIT_EVENT_HASH_GROW_BUCKETS_ELECT:
event_name = "HashGrowBucketsElect";
diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h
index d8edd39923..176fbef149 100644
--- a/src/include/executor/hashjoin.h
+++ b/src/include/executor/hashjoin.h
@@ -254,32 +254,32 @@ typedef struct ParallelHashJoinState
} ParallelHashJoinState;
/* The phases for building batches, used by build_barrier. */
-#define PHJ_BUILD_ELECTING 0
-#define PHJ_BUILD_ALLOCATING 1
-#define PHJ_BUILD_HASHING_INNER 2
-#define PHJ_BUILD_HASHING_OUTER 3
-#define PHJ_BUILD_RUNNING 4
-#define PHJ_BUILD_DONE 5
+#define PHJ_BUILD_ELECT 0
+#define PHJ_BUILD_ALLOCATE 1
+#define PHJ_BUILD_HASH_INNER 2
+#define PHJ_BUILD_HASH_OUTER 3
+#define PHJ_BUILD_RUN 4
+#define PHJ_BUILD_FREE 5
/* The phases for probing each batch, used by for batch_barrier. */
-#define PHJ_BATCH_ELECTING 0
-#define PHJ_BATCH_ALLOCATING 1
-#define PHJ_BATCH_LOADING 2
-#define PHJ_BATCH_PROBING 3
-#define PHJ_BATCH_DONE 4
+#define PHJ_BATCH_ELECT 0
+#define PHJ_BATCH_ALLOCATE 1
+#define PHJ_BATCH_LOAD 2
+#define PHJ_BATCH_PROBE 3
+#define PHJ_BATCH_FREE 4
/* The phases of batch growth while hashing, for grow_batches_barrier. */
-#define PHJ_GROW_BATCHES_ELECTING 0
-#define PHJ_GROW_BATCHES_ALLOCATING 1
-#define PHJ_GROW_BATCHES_REPARTITIONING 2
-#define PHJ_GROW_BATCHES_DECIDING 3
-#define PHJ_GROW_BATCHES_FINISHING 4
+#define PHJ_GROW_BATCHES_ELECT 0
+#define PHJ_GROW_BATCHES_REALLOCATE 1
+#define PHJ_GROW_BATCHES_REPARTITION 2
+#define PHJ_GROW_BATCHES_DECIDE 3
+#define PHJ_GROW_BATCHES_FINISH 4
#define PHJ_GROW_BATCHES_PHASE(n) ((n) % 5) /* circular phases */
/* The phases of bucket growth while hashing, for grow_buckets_barrier. */
-#define PHJ_GROW_BUCKETS_ELECTING 0
-#define PHJ_GROW_BUCKETS_ALLOCATING 1
-#define PHJ_GROW_BUCKETS_REINSERTING 2
+#define PHJ_GROW_BUCKETS_ELECT 0
+#define PHJ_GROW_BUCKETS_REALLOCATE 1
+#define PHJ_GROW_BUCKETS_REINSERT 2
#define PHJ_GROW_BUCKETS_PHASE(n) ((n) % 3) /* circular phases */
typedef struct HashJoinTableData
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index 6007827b44..a9c6a9b108 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -96,12 +96,12 @@ typedef enum
WAIT_EVENT_HASH_BUILD_ELECT,
WAIT_EVENT_HASH_BUILD_HASH_INNER,
WAIT_EVENT_HASH_BUILD_HASH_OUTER,
- WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATE,
+ WAIT_EVENT_HASH_GROW_BATCHES_REALLOCATE,
WAIT_EVENT_HASH_GROW_BATCHES_DECIDE,
WAIT_EVENT_HASH_GROW_BATCHES_ELECT,
WAIT_EVENT_HASH_GROW_BATCHES_FINISH,
WAIT_EVENT_HASH_GROW_BATCHES_REPARTITION,
- WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE,
+ WAIT_EVENT_HASH_GROW_BUCKETS_REALLOCATE,
WAIT_EVENT_HASH_GROW_BUCKETS_ELECT,
WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT,
WAIT_EVENT_LOGICAL_SYNC_DATA,
--
2.27.0
