0001-Make-multi-batch-Parallel-Hash-Full-Join-fairer.patch

From da98217b98a04b3f84512a6bf267b3cd107e39d7 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.munro@gmail.com>
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 <melanieplageman@gmail.com>
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