Hi,
* avoids wasting memory on duplicated hash tables
* avoids wasting disk space on duplicated batch files
* avoids wasting CPU executing duplicate subplans
What's the last one referring to?
+static void
+MultiExecParallelHash(HashState *node)
+{
+ switch (BarrierPhase(build_barrier))
+ {
+ case PHJ_BUILD_ALLOCATING:
+
+ /*
+ * Either I just allocated the initial hash table in
+ * ExecHashTableCreate(), or someone else is doing
that. Either
+ * way, wait for everyone to arrive here so we can
proceed, and
+ * then fall through.
+ */
+ BarrierArriveAndWait(build_barrier,
WAIT_EVENT_HASH_BUILD_ALLOCATING);
Can you add a /* fallthrough */ comment here? Gcc is warning if you
don't. While we currently have lotsa other places not having the
annotation, it seem reasonable to have it in new code.
+ case PHJ_BUILD_HASHING_INNER:
+
+ /*
+ * It's time to begin hashing, or if we just arrived
here then
+ * hashing is already underway, so join in that effort.
While
+ * hashing we have to be prepared to help increase the
number of
+ * batches or buckets at any time, and if we arrived
here when
+ * that was already underway we'll have to help
complete that work
+ * immediately so that it's safe to access batches and
buckets
+ * below.
+ */
+ if
(PHJ_GROW_BATCHES_PHASE(BarrierAttach(&pstate->grow_batches_barrier)) !=
+ PHJ_GROW_BATCHES_ELECTING)
+ ExecParallelHashIncreaseNumBatches(hashtable);
+ if
(PHJ_GROW_BUCKETS_PHASE(BarrierAttach(&pstate->grow_buckets_barrier)) !=
+ PHJ_GROW_BUCKETS_ELECTING)
+ ExecParallelHashIncreaseNumBuckets(hashtable);
+ ExecParallelHashEnsureBatchAccessors(hashtable);
"accessors" sounds a bit weird for a bunch of pointers, but maybe that's
just my ESL senses tingling wrongly.
/* ----------------------------------------------------------------
@@ -240,12 +427,15 @@ ExecEndHash(HashState *node)
* ----------------------------------------------------------------
*/
HashJoinTable
-ExecHashTableCreate(Hash *node, List *hashOperators, bool keepNulls)
+ExecHashTableCreate(HashState *state, List *hashOperators, bool keepNulls)
+ /*
+ * Parallel Hash tries to use the combined work_mem of all workers to
+ * avoid the need to batch. If that won't work, it falls back to
work_mem
+ * per worker and tries to process batches in parallel.
+ */
One day we're going to need a better approach to this. I have no idea
how, but this per-node, and now per_node * max_parallelism, approach has
only implementation simplicity as its benefit.
+static HashJoinTuple
+ExecParallelHashLoadTuple(HashJoinTable hashtable, MinimalTuple tuple,
+ dsa_pointer *shared)
+{
+static void
+ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
+{
+/*
+ * Get the first tuple in a given bucket identified by number.
+ */
+static HashJoinTuple
+ExecHashFirstTupleInBucket(HashJoinTable hashtable, int bucketno)
+{
+ if (hashtable->parallel_state)
+ {
+ dsa_pointer p =
+ dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
Can you make this, and possibly a few other places, more readable by
introducing a temporary variable?
+/*
+ * Insert a tuple at the front of a chain of tuples in DSA memory atomically.
+ */
+static void
+ExecParallelHashPushTuple(dsa_pointer_atomic *head,
+ HashJoinTuple tuple,
+ dsa_pointer tuple_shared)
+{
+ do
+ {
+ tuple->next.shared = dsa_pointer_atomic_read(head);
+ } while (!dsa_pointer_atomic_compare_exchange(head,
+
&tuple->next.shared,
+
tuple_shared));
+}
This is hard to read.
+ * 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
+ * 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 rep
s/rep/repartition/?
#include "access/htup_details.h"
+#include "access/parallel.h"
#include "executor/executor.h"
#include "executor/hashjoin.h"
#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "miscadmin.h"
+#include "pgstat.h"
#include "utils/memutils.h"
-
+#include "utils/sharedtuplestore.h"
deletes a separator newline.
/* ----------------------------------------------------------------
@@ -138,6 +236,18 @@ ExecHashJoin(PlanState *pstate)
/* no chance to not build the hash
table */
node->hj_FirstOuterTupleSlot = NULL;
}
+ else if (hashNode->parallel_state != NULL)
+ {
+ /*
+ * The empty-outer optimization is not
implemented for
+ * shared hash tables, because no one
participant can
+ * determine that there are no outer
tuples, and it's not
+ * yet clear that it's worth the
synchronization overhead
+ * of reaching consensus to figure that
out. So we have
+ * to build the hash table.
+ */
+ node->hj_FirstOuterTupleSlot = NULL;
+ }
Hm. Isn't MultiExecParallelHash already doing so?
- node->hj_JoinState = HJ_NEED_NEW_OUTER;
+ if (hashtable->parallel_state)
+ {
+ Barrier *build_barrier;
+
+ build_barrier =
&hashtable->parallel_state->build_barrier;
+ if (BarrierPhase(build_barrier) ==
PHJ_BUILD_HASHING_OUTER)
+ {
+ /*
+ * If multi-batch, we need to
hash the outer relation
+ * up front.
+ */
+ if (hashtable->nbatch > 1)
+
ExecParallelHashJoinPartitionOuter(node);
+
BarrierArriveAndWait(build_barrier,
+
WAIT_EVENT_HASH_BUILD_HASHING_OUTER);
+ }
+ Assert(BarrierPhase(build_barrier) ==
PHJ_BUILD_DONE);
+
+ /* Each backend should now select a
batch to work on. */
+ hashtable->curbatch = -1;
+ node->hj_JoinState = HJ_NEED_NEW_BATCH;
+
+ continue;
+ }
+ else
+ node->hj_JoinState = HJ_NEED_NEW_OUTER;
You know what I'm going to say about all these branches, and sigh.
If we don't split this into two versions, we at least should store
hashNode->parallel_state in a local var, so the compiler doesn't have to
pull that out of memory after every external function call (of which
there are a lot). In common cases it'll end up in a callee saved
registers, and most of the called functions won't be too register
starved (on x86-64).
+/*
+ * Choose a batch to work on, and attach to it. Returns true if successful,
+ * false if there are no more batches.
+ */
+static bool
+ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
+{
+ /*
+ * This batch is ready to probe.
Return control to
+ * caller. We stay attached to
batch_barrier so that the
+ * hash table stays alive until
everyone's finish probing
*finished?
+ case PHJ_BATCH_DONE:
+
+ /*
+ * Already done. Detach and go around
again (if any
+ * remain).
+ */
+ BarrierDetach(batch_barrier);
+
+ /*
+ * We didn't work on this batch, but we
need to observe
+ * its size for EXPLAIN.
+ */
+
ExecParallelHashUpdateSpacePeak(hashtable, batchno);
+ hashtable->batches[batchno].done = true;
+ hashtable->curbatch = -1;
+ break;
Hm, maybe I'm missing something, but why is it guaranteed that "we
didn't work on this batch"?
+void
+ExecShutdownHashJoin(HashJoinState *node)
+{
+ /*
+ * By the time ExecEndHashJoin runs in a worker, shared memory has been
+ * destroyed. So this is our last chance to do any shared memory
cleanup.
+ */
This comment doesn't really make much sense to me.
+void
+ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *cxt)
+{
could use a header comment.
a) The executor side is starting to look good.
b) This is a lot of code.
c) I'm tired, planner has to wait till tomorrow.
- Andres
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
