Here is a rebased and lightly hacked-upon version that I'm testing.
0001-Scan-for-unmatched-hash-join-tuples-in-memory-order.patch
* this change can stand on its own, separately from any PHJ changes
* renamed hashtable->current_chunk[_idx] to unmatched_scan_{chunk,idx}
* introduced a local variable to avoid some x->y->z stuff
* removed some references to no-longer-relevant hj_XXX variables in
the Prep function
I haven't attempted to prove anything about the performance of this
one yet, but it seems fairly obvious that it can't be worse than what
we're doing today. I have suppressed the urge to look into improving
locality and software prefetching.
0002-Parallel-Hash-Full-Join.patch
* reuse the same umatched_scan_{chunk,idx} variables as above
* rename the list of chunks to scan to work_queue
* fix race/memory leak if we see PHJ_BATCH_SCAN when we attach (it
wasn't OK to just fall through)
That "work queue" name/concept already exists in other places that
need to process every chunk, namely rebucketing and repartitioning.
In later work, I'd like to harmonise these work queues, but I'm not
trying to increase the size of this patch set at this time, I just
want to use consistent naming.
I don't love the way that both ExecHashTableDetachBatch() and
ExecParallelPrepHashTableForUnmatched() duplicate logic relating to
the _SCAN/_FREE protocol, but I'm struggling to find a better idea.
Perhaps I just need more coffee.
I think your idea of opportunistically joining the scan if it's
already running makes sense to explore for a later step, ie to make
multi-batch PHFJ fully fair, and I think that should be a fairly easy
code change, and I put in some comments where changes would be needed.
Continuing to test, more soon.
From 8b526377eb4a4685628624e75743aedf37dd5bfe Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.mu...@gmail.com>
Date: Fri, 24 Mar 2023 14:19:07 +1300
Subject: [PATCH v12 1/2] Scan for unmatched hash join tuples in memory order.
In a full/right outer join, we need to scan every tuple in the hash
table to find the ones that were not matched while probing, so that we
can emit null-extended inner tuples. Previously we did that in hash
table bucket order, which means that we dereferenced pointers to tuples
that were randomly scattered in memory (ie in an order derived from the
hash of the join key).
Change to a memory-order scan, using the linked list of memory chunks
that hold the tuples. This isn't really being done for performance
reasons (a subject for later work), but it certainly can't be worse than
the previous random order. The goal here is to harmonize the scan logic
with later work that will parallelize full joins, and works in chunks.
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 | 85 +++++++++++++--------------------
src/include/executor/hashjoin.h | 4 ++
2 files changed, 38 insertions(+), 51 deletions(-)
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index 748c9b0024..91fd806c97 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -517,6 +517,8 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
hashtable->spaceAllowed * SKEW_HASH_MEM_PERCENT / 100;
hashtable->chunks = NULL;
hashtable->current_chunk = NULL;
+ hashtable->unmatched_scan_chunk = NULL;
+ hashtable->unmatched_scan_idx = 0;
hashtable->parallel_state = state->parallel_state;
hashtable->area = state->ps.state->es_query_dsa;
hashtable->batches = NULL;
@@ -2058,16 +2060,10 @@ ExecParallelScanHashBucket(HashJoinState *hjstate,
void
ExecPrepHashTableForUnmatched(HashJoinState *hjstate)
{
- /*----------
- * During this scan we use the HashJoinState 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
- *----------
- */
- hjstate->hj_CurBucketNo = 0;
- hjstate->hj_CurSkewBucketNo = 0;
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+
+ hashtable->unmatched_scan_chunk = hashtable->chunks;
+ hashtable->unmatched_scan_idx = 0;
hjstate->hj_CurTuple = NULL;
}
@@ -2083,58 +2079,45 @@ bool
ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
{
HashJoinTable hashtable = hjstate->hj_HashTable;
- HashJoinTuple hashTuple = hjstate->hj_CurTuple;
+ HashMemoryChunk chunk;
- for (;;)
+ while ((chunk = hashtable->unmatched_scan_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->unmatched_scan_idx < chunk->used)
{
- int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
+ HashJoinTuple hashTuple = (HashJoinTuple)
+ (HASH_CHUNK_DATA(hashtable->unmatched_scan_chunk) +
+ hashtable->unmatched_scan_idx);
- hashTuple = hashtable->skewBucket[j]->tuples;
- hjstate->hj_CurSkewBucketNo++;
- }
- else
- break; /* finished all buckets */
+ MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
+ int hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len);
- while (hashTuple != NULL)
- {
- if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
- {
- TupleTableSlot *inntuple;
+ /* next tuple in this chunk */
+ hashtable->unmatched_scan_idx += MAXALIGN(hashTupleSize);
- /* insert hashtable's tuple into exec slot */
- inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
- hjstate->hj_HashTupleSlot,
- false); /* do not pfree */
- econtext->ecxt_innertuple = inntuple;
+ if (HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
+ continue;
- /*
- * 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);
+ /* insert hashtable's tuple into exec slot */
+ econtext->ecxt_innertuple =
+ ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ hjstate->hj_HashTupleSlot,
+ false);
- hjstate->hj_CurTuple = hashTuple;
- return true;
- }
+ /*
+ * 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;
}
+ hashtable->unmatched_scan_chunk = chunk->next.unshared;
+ hashtable->unmatched_scan_idx = 0;
+
/* allow this loop to be cancellable */
CHECK_FOR_INTERRUPTS();
}
diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h
index acb7592ca0..0abd888d1e 100644
--- a/src/include/executor/hashjoin.h
+++ b/src/include/executor/hashjoin.h
@@ -352,6 +352,10 @@ typedef struct HashJoinTableData
/* used for dense allocation of tuples (into linked chunks) */
HashMemoryChunk chunks; /* one list for the whole batch */
+ /* current position in unmatched scan (full/right join only) */
+ HashMemoryChunk unmatched_scan_chunk;
+ size_t unmatched_scan_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 */
--
2.39.2
From 6f4e82f0569e5b388440ca0ef268dd307388e8f8 Mon Sep 17 00:00:00 2001
From: Thomas Munro <thomas.mu...@gmail.com>
Date: Fri, 24 Mar 2023 15:23:14 +1300
Subject: [PATCH v12 2/2] Parallel Hash Full Join.
Full and right outer joins were not supported in the initial
implementation of Parallel Hash Join, because of deadlock hazards (see
discussion). Therefore FULL JOIN inhibited page-based parallelism,
as the other join strategies can't do it either.
Add a new PHJ phase PHJ_BATCH_SCAN that scans for unmatched tuples on
the inner side of one batch's hash table. For now, sidestep the
deadlock problem by terminating parallelism there. The last process to
arrive at that phase emits the unmatched tuples, while others detach and
are free to go and work on other batches, if there are any, but
otherwise they finish the join early.
That unfairness is considered acceptable for now, because it's better
than no parallelism at all. The build and probe phases are run in
parallel, and the new scan-for-unmatched phase, while serial, is usually
applied to the smaller of the two relations and is either limited by
some multiple of work_mem, or it's too big and is partitioned into
batches and then the situation is improved by batch-level parallelism.
In future work on deadlock avoidance strategies, we may find a way to
parallelize the new phase safely.
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 | 163 +++++++++++++++++++++++-
src/backend/executor/nodeHashjoin.c | 91 +++++++++----
src/backend/optimizer/path/joinpath.c | 14 +-
src/include/executor/hashjoin.h | 7 +-
src/include/executor/nodeHash.h | 3 +
src/test/regress/expected/join_hash.out | 58 ++++++++-
src/test/regress/sql/join_hash.sql | 25 +++-
7 files changed, 314 insertions(+), 47 deletions(-)
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index 91fd806c97..58789be71a 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -2067,6 +2067,66 @@ ExecPrepHashTableForUnmatched(HashJoinState *hjstate)
hjstate->hj_CurTuple = NULL;
}
+/*
+ * 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)
+{
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+ int curbatch = hashtable->curbatch;
+ ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
+
+ hjstate->hj_CurTuple = NULL;
+ if (curbatch < 0)
+ return false;
+
+ /*
+ * It would not be deadlock-free to wait on the batch barrier, because 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.
+ */
+ Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE);
+ if (!BarrierArriveAndDetachExceptLast(&batch->batch_barrier))
+ {
+ /* This process considers the batch to be done. */
+ 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 largest batch we've seen, which would normally happen in
+ * ExecHashTableDetachBatch().
+ */
+ hashtable->spacePeak =
+ Max(hashtable->spacePeak,
+ batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);
+ hashtable->curbatch = -1;
+ return false;
+ }
+
+ /* Now we are alone with this batch. */
+ Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_SCAN);
+ Assert(BarrierParticipants(&batch->batch_barrier) == 1);
+
+ /*
+ * See also ExecParallelHashJoinNewBatch()'s assertion that
+ * batch->work_queue == batch->chunks. That is, we are now ready to start
+ * processing all chunks by consuming from work_queue.
+ */
+ hashtable->unmatched_scan_chunk = NULL;
+ hashtable->unmatched_scan_idx = 0;
+
+ return true;
+}
+
/*
* ExecScanHashTableForUnmatched
* scan the hash table for unmatched inner tuples
@@ -2128,6 +2188,80 @@ ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
return false;
}
+/*
+ * ExecParallelScanHashTableForUnmatched
+ * scan the hash table for unmatched inner tuples, in parallel join
+ *
+ * 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];
+ HashMemoryChunk chunk;
+
+ for (;;)
+ {
+ if (!(chunk = hashtable->unmatched_scan_chunk))
+ {
+
+ /*
+ * Since only one process can run this currently, we don't need to
+ * bother with interlocking, when popping chunks off the work
+ * queue.
+ */
+ chunk = (HashMemoryChunk)
+ dsa_get_address(hashtable->area, accessor->shared->work_queue);
+ if (!chunk)
+ break;
+ accessor->shared->work_queue = chunk->next.shared;
+ hashtable->unmatched_scan_chunk = chunk;
+ hashtable->unmatched_scan_idx = 0;
+ }
+
+ while (hashtable->unmatched_scan_idx < chunk->used)
+ {
+ HashJoinTuple hashTuple = (HashJoinTuple)
+ (HASH_CHUNK_DATA(chunk) + hashtable->unmatched_scan_idx);
+
+ hashtable->unmatched_scan_idx += MAXALIGN(HJTUPLE_OVERHEAD +
+ HJTUPLE_MINTUPLE(hashTuple)->t_len);
+
+ 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);
+
+ hjstate->hj_CurTuple = hashTuple;
+ return true;
+ }
+
+ hashtable->unmatched_scan_chunk = NULL;
+
+ CHECK_FOR_INTERRUPTS();
+ }
+
+ /*
+ * no more unmatched tuples
+ */
+ return false;
+}
+
/*
* ExecHashTableReset
*
@@ -2908,6 +3042,12 @@ ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size,
chunk->next.shared = hashtable->batches[curbatch].shared->chunks;
hashtable->batches[curbatch].shared->chunks = chunk_shared;
+ /*
+ * Also make this the head of the work_queue list. This is used as a
+ * cursor for scanning all chunks in the batch.
+ */
+ hashtable->batches[curbatch].shared->work_queue = chunk_shared;
+
if (size <= HASH_CHUNK_THRESHOLD)
{
/*
@@ -3116,18 +3256,31 @@ ExecHashTableDetachBatch(HashJoinTable hashtable)
{
int curbatch = hashtable->curbatch;
ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
+ bool attached = 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);
- /* Detach from the batch we were last working on. */
- if (BarrierArriveAndDetach(&batch->batch_barrier))
+ /* After attaching we always get at least to PHJ_BATCH_PROBE. */
+ Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE ||
+ BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_SCAN);
+
+ /*
+ * Even if we aren't doing a full/right outer join, we'll step through
+ * the PHJ_BATCH_SCAN phase just to maintain the invariant that freeing
+ * happens in PHJ_BATCH_FREE, but that'll be wait-free.
+ */
+ if (BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_PROBE)
+ attached = BarrierArriveAndDetachExceptLast(&batch->batch_barrier);
+ if (attached && 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.
+ * We are not longer attached to the batch barrier, but we're the
+ * process that was chosen to free resources and it's safe to
+ * assert the current phase. The ParallelHashJoinBatch can't go
+ * away underneath us while we are attached to the build barrier,
+ * making this access safe.
*/
Assert(BarrierPhase(&batch->batch_barrier) == PHJ_BATCH_FREE);
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
index f189fb4d28..93bf0ad6e9 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* -- one does full/right unmatched 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.
@@ -393,8 +395,23 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
if (HJ_FILL_INNER(node))
{
/* set up to scan for unmatched inner tuples */
- ExecPrepHashTableForUnmatched(node);
- node->hj_JoinState = HJ_FILL_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;
@@ -487,25 +504,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)
@@ -563,7 +568,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;
@@ -1197,13 +1203,44 @@ 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);
+
+ /*
+ * This is a good place to assert that the batch's
+ * work_queue has been set to point to the head of the
+ * chunk list. No new chunks are created once this phase
+ * is reached, and we don't want to have to do any extra
+ * IPC just to set up the work_queue that the later
+ * PHJ_BATCH_SCAN phase wants, so instead require that it
+ * has been maintained correctly by the earlier phases.
+ */
+ Assert(hashtable->batches[batchno].shared->work_queue ==
+ hashtable->batches[batchno].shared->chunks);
+
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:
diff --git a/src/backend/optimizer/path/joinpath.c b/src/backend/optimizer/path/joinpath.c
index e6ef0deb23..bd51e4f972 100644
--- a/src/backend/optimizer/path/joinpath.c
+++ b/src/backend/optimizer/path/joinpath.c
@@ -2193,15 +2193,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))
{
@@ -2235,9 +2229,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 0abd888d1e..7615025d73 100644
--- a/src/include/executor/hashjoin.h
+++ b/src/include/executor/hashjoin.h
@@ -154,6 +154,7 @@ typedef struct ParallelHashJoinBatch
Barrier batch_barrier; /* synchronization for joining this batch */
dsa_pointer chunks; /* chunks of tuples loaded */
+ dsa_pointer work_queue; /* cursor for processing all chunks */
size_t size; /* size of buckets + chunks in memory */
size_t estimated_size; /* size of buckets + chunks while writing */
size_t ntuples; /* number of tuples loaded */
@@ -266,7 +267,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
@@ -356,6 +358,9 @@ typedef struct HashJoinTableData
HashMemoryChunk unmatched_scan_chunk;
size_t unmatched_scan_idx;
+ /* 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 d7634af05c..56d5350c61 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 3ec07bc1af..027f3888b0 100644
--- a/src/test/regress/expected/join_hash.out
+++ b/src/test/regress/expected/join_hash.out
@@ -784,8 +784,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);
@@ -806,7 +807,32 @@ select count(*) from simple r full outer join simple s using (id);
(1 row)
rollback to settings;
--- An full outer join where every record is not matched.
+-- 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;
+-- A full outer join where every record is not matched.
-- non-parallel
savepoint settings;
set local max_parallel_workers_per_gather = 0;
@@ -829,8 +855,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);
@@ -850,6 +877,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 77dbc182d5..ba1b3e6e1b 100644
--- a/src/test/regress/sql/join_hash.sql
+++ b/src/test/regress/sql/join_hash.sql
@@ -435,15 +435,24 @@ 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;
--- An full outer join where every record is not matched.
+-- 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);
+select count(*) from simple r full outer join simple s using (id);
+rollback to settings;
+
+-- A full outer join where every record is not matched.
-- non-parallel
savepoint settings;
@@ -453,14 +462,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.39.2
