From da98217b98a04b3f84512a6bf267b3cd107e39d7 Mon Sep 17 00:00:00 2001 From: Thomas Munro Date: Wed, 29 Mar 2023 17:20:13 +1300 Subject: [PATCH] Make multi-batch Parallel Hash Full Join fairer. The initial commit of Parallel Hash Full Join never allowed more than one process to scan a hashtable for unmatched inner tuples. Relax that for multi-batch joins, so that work is more evenly distributed. An atomic variable hands out ranges of buckets to scan within each process. We currently don't see a reasonable deadlock-free way to allow that for single-batch joins (it seems to require major executor changes). For multi-batch joinsing ones, we still have to drop down to one process at the end of the probe of each batch, but processes can safely join any existing unmatched scan that has already started in another batch, if it can find one. Suggested-by: Melanie Plageman Discussion: https://postgr.es/m/CA%2BhUKG%2BA6ftXPz4oe92%2Bx8Er%2BxpGZqto70-Q_ERwRaSyA%3DafNg%40mail.gmail.com --- src/backend/executor/nodeHash.c | 36 +++++++++++++++++++++-------- src/backend/executor/nodeHashjoin.c | 35 +++++++++++++++++++--------- src/include/executor/hashjoin.h | 1 + src/include/nodes/execnodes.h | 2 ++ 4 files changed, 53 insertions(+), 21 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index a45bd3a315..83d33fd618 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -2067,14 +2067,13 @@ ExecPrepHashTableForUnmatched(HashJoinState *hjstate) *---------- */ hjstate->hj_CurBucketNo = 0; + hjstate->hj_MaxBucketNo = 0; /* for parallel scan */ hjstate->hj_CurSkewBucketNo = 0; hjstate->hj_CurTuple = NULL; } /* - * Decide if this process is allowed to run the unmatched scan. If so, the - * batch barrier is advanced to PHJ_BATCH_SCAN and true is returned. - * Otherwise the batch is detached and false is returned. + * Decide if this process is allowed to run the unmatched scan. */ bool ExecParallelPrepHashTableForUnmatched(HashJoinState *hjstate) @@ -2083,17 +2082,20 @@ ExecParallelPrepHashTableForUnmatched(HashJoinState *hjstate) int curbatch = hashtable->curbatch; ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared; - Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE); + Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE || + BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_SCAN); /* - * It would not be deadlock-free to wait on the batch barrier, because it + * It would not be deadlock-free to wait on the batch barrier when it * is in PHJ_BATCH_PROBE phase, and thus processes attached to it have * already emitted tuples. Therefore, we'll hold a wait-free election: * only one process can continue to the next phase, and all others detach * from this batch. They can still go any work on other batches, if there - * are any. + * are any. If we got here when it's already in PHJ_BATCH_SCAN phase, + * we can proceed without further ado. */ - if (!BarrierArriveAndDetachExceptLast(&batch->batch_barrier)) + if (BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE && + !BarrierArriveAndDetachExceptLast(&batch->batch_barrier)) { /* This process considers the batch to be done. */ hashtable->batches[hashtable->curbatch].done = true; @@ -2113,9 +2115,7 @@ ExecParallelPrepHashTableForUnmatched(HashJoinState *hjstate) return false; } - /* Now we are alone with this batch. */ Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_SCAN); - Assert(BarrierParticipants(&batch->batch_barrier) == 1); /* * Has another process decided to give up early and command all processes @@ -2232,9 +2232,24 @@ ExecParallelScanHashTableForUnmatched(HashJoinState *hjstate, */ if (hashTuple != NULL) hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple); - else if (hjstate->hj_CurBucketNo < hashtable->nbuckets) + else if (hjstate->hj_CurBucketNo < hjstate->hj_MaxBucketNo) hashTuple = ExecParallelHashFirstTuple(hashtable, hjstate->hj_CurBucketNo++); + else if (hjstate->hj_CurBucketNo < hashtable->nbuckets) + { + /* + * Allocate a few cachelines' worth of buckets, and loop around. + * Testing shows that 8 is a good multiplier. + */ + size_t step = (PG_CACHE_LINE_SIZE * 8) / sizeof(dsa_pointer_atomic); + ParallelHashJoinBatch *batch; + + batch = hashtable->batches[hashtable->curbatch].shared; + hjstate->hj_CurBucketNo = + pg_atomic_fetch_add_u32(&batch->bucket, step); + hjstate->hj_MaxBucketNo = + Min(hjstate->hj_CurBucketNo + step, hashtable->nbuckets); + } else break; /* finished all buckets */ @@ -3113,6 +3128,7 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch) * up the Barrier. */ BarrierInit(&shared->batch_barrier, 0); + pg_atomic_init_u32(&shared->bucket, 0); if (i == 0) { /* Batch 0 doesn't need to be loaded. */ diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 52ed05c6f5..1f5a10bfbc 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -597,6 +597,20 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) { if (!ExecParallelHashJoinNewBatch(node)) return NULL; /* end of parallel-aware join */ + + /* + * If we've attached to a batch that is already in the + * inner scan phase, we'll help with that. + */ + if (BarrierPhase(&hashtable->batches[hashtable->curbatch].shared->batch_barrier) == PHJ_BATCH_SCAN) + { + if (HJ_FILL_INNER(node)) + node->hj_JoinState = HJ_FILL_INNER_TUPLES; + else + node->hj_JoinState = HJ_NEED_NEW_BATCH; + break; + } + Assert(BarrierPhase(&hashtable->batches[hashtable->curbatch].shared->batch_barrier) == PHJ_BATCH_PROBE); } else { @@ -1216,20 +1230,19 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate) 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. + * Join in with inner scan, if we have not been asked to + * skip it by another process. */ ExecParallelHashTableSetCurrentBatch(hashtable, batchno); + if (ExecParallelPrepHashTableForUnmatched(hjstate)) + return true; + + /* + * Otherwise, we've been detached and we need to go around + * again. + */ hashtable->batches[batchno].done = true; - ExecHashTableDetachBatch(hashtable); + hashtable->curbatch = -1; break; case PHJ_BATCH_FREE: diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 8ee59d2c71..44cb85a89a 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -160,6 +160,7 @@ typedef struct ParallelHashJoinBatch size_t old_ntuples; /* number of tuples before repartitioning */ bool space_exhausted; bool skip_unmatched; /* whether to abandon unmatched scan */ + pg_atomic_uint32 bucket; /* cursor for unmatched inner scan */ /* * Variable-sized SharedTuplestore objects follow this struct in memory. diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index d97f5a8e7d..2529d7a4a6 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -2066,6 +2066,7 @@ typedef struct MergeJoinState * (NULL if table not built yet) * hj_CurHashValue hash value for current outer tuple * hj_CurBucketNo regular bucket# for current outer tuple + * hj_MaxBucketNo bucket range allocated to parallel process * hj_CurSkewBucketNo skew bucket# for current outer tuple * hj_CurTuple last inner tuple matched to current outer * tuple, or NULL if starting search @@ -2096,6 +2097,7 @@ typedef struct HashJoinState HashJoinTable hj_HashTable; uint32 hj_CurHashValue; int hj_CurBucketNo; + int hj_MaxBucketNo; int hj_CurSkewBucketNo; HashJoinTuple hj_CurTuple; TupleTableSlot *hj_OuterTupleSlot; -- 2.40.0