agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Andres Freund <[email protected]>
Subject: [PATCH] Merge Hash nodes into their associated HashJoin node.
Date: Mon, 28 Oct 2019 16:12:12 -0700
---
src/backend/commands/explain.c | 73 +-
src/backend/executor/Makefile | 2 +-
src/backend/executor/execAmi.c | 5 -
src/backend/executor/execParallel.c | 30 +-
src/backend/executor/execProcnode.c | 17 -
src/backend/executor/nodeHash.c | 3338 -----------------------
src/backend/executor/nodeHashjoin.c | 3303 +++++++++++++++++++++-
src/backend/nodes/copyfuncs.c | 38 +-
src/backend/nodes/outfuncs.c | 27 +-
src/backend/nodes/readfuncs.c | 30 +-
src/backend/optimizer/path/costsize.c | 2 +-
src/backend/optimizer/plan/createplan.c | 91 +-
src/backend/optimizer/plan/setrefs.c | 62 +-
src/backend/optimizer/plan/subselect.c | 1 -
src/include/executor/nodeHash.h | 79 -
src/include/executor/nodeHashjoin.h | 13 +-
src/include/nodes/execnodes.h | 73 +-
src/include/nodes/nodes.h | 2 -
src/include/nodes/plannodes.h | 46 +-
src/test/regress/expected/join_hash.out | 28 +-
src/test/regress/sql/join_hash.sql | 8 +-
21 files changed, 3488 insertions(+), 3780 deletions(-)
delete mode 100644 src/backend/executor/nodeHash.c
delete mode 100644 src/include/executor/nodeHash.h
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 62fb3434a32..b9130ce351b 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -18,7 +18,7 @@
#include "commands/createas.h"
#include "commands/defrem.h"
#include "commands/prepare.h"
-#include "executor/nodeHash.h"
+#include "executor/nodeHashjoin.h"
#include "foreign/fdwapi.h"
#include "jit/jit.h"
#include "nodes/extensible.h"
@@ -101,7 +101,7 @@ static void show_sortorder_options(StringInfo buf, Node *sortexpr,
static void show_tablesample(TableSampleClause *tsc, PlanState *planstate,
List *ancestors, ExplainState *es);
static void show_sort_info(SortState *sortstate, ExplainState *es);
-static void show_hash_info(HashState *hashstate, ExplainState *es);
+static void show_hash_info(HashJoinState *hashjoinstate, ExplainState *es);
static void show_tidbitmap_info(BitmapHeapScanState *planstate,
ExplainState *es);
static void show_instrumentation_count(const char *qlabel, int which,
@@ -1291,9 +1291,6 @@ ExplainNode(PlanState *planstate, List *ancestors,
case T_Limit:
pname = sname = "Limit";
break;
- case T_Hash:
- pname = sname = "Hash";
- break;
default:
pname = sname = "???";
break;
@@ -1822,6 +1819,10 @@ ExplainNode(PlanState *planstate, List *ancestors,
if (plan->qual)
show_instrumentation_count("Rows Removed by Filter", 2,
planstate, es);
+
+ show_hash_info(castNode(HashJoinState, planstate), es);
+ break;
+
break;
case T_Agg:
show_agg_keys(castNode(AggState, planstate), ancestors, es);
@@ -1857,9 +1858,6 @@ ExplainNode(PlanState *planstate, List *ancestors,
show_modifytable_info(castNode(ModifyTableState, planstate), ancestors,
es);
break;
- case T_Hash:
- show_hash_info(castNode(HashState, planstate), es);
- break;
default:
break;
}
@@ -1966,11 +1964,66 @@ ExplainNode(PlanState *planstate, List *ancestors,
ExplainNode(outerPlanState(planstate), ancestors,
"Outer", NULL, es);
+ /* XXX: Temporary hack to hide explain changes */
+ if (IsA(plan, HashJoin))
+ {
+ if (es->format == EXPLAIN_FORMAT_TEXT)
+ {
+ if (es->indent)
+ {
+ appendStringInfoSpaces(es->str, es->indent * 2);
+ appendStringInfoString(es->str, "-> ");
+ es->indent += 2;
+ }
+ if (plan->parallel_aware)
+ appendStringInfoString(es->str, "Parallel ");
+ appendStringInfoString(es->str, "Hash\n");
+ es->indent++;
+ }
+
+ /* fake up Hash targetlist */
+ if (es->verbose)
+ {
+ List *context;
+ List *result = NIL;
+ bool useprefix;
+ ListCell *lc;
+
+ /* Set up deparsing context */
+ context = set_deparse_context_planstate(es->deparse_cxt,
+ (Node *) planstate,
+ ancestors);
+ useprefix = list_length(es->rtable) > 1;
+
+ /* Deparse each result column (we now include resjunk ones) */
+ foreach(lc, ((HashJoin *) plan)->inner_tlist)
+ {
+ TargetEntry *tle = (TargetEntry *) lfirst(lc);
+
+ result = lappend(result,
+ deparse_expression((Node *) tle->expr, context,
+ useprefix, false));
+ }
+
+ /* Print results */
+ ExplainPropertyList("Output", result, es);
+ }
+ }
+
/* righttree */
if (innerPlanState(planstate))
ExplainNode(innerPlanState(planstate), ancestors,
"Inner", NULL, es);
+ /* XXX: Temporary hack to hide explain changes */
+ if (IsA(plan, HashJoin))
+ {
+ if (es->format == EXPLAIN_FORMAT_TEXT)
+ {
+ es->indent -= 3;
+ }
+ }
+
/* special child plans */
switch (nodeTag(plan))
{
@@ -2616,7 +2669,7 @@ show_sort_info(SortState *sortstate, ExplainState *es)
* Show information on hash buckets/batches.
*/
static void
-show_hash_info(HashState *hashstate, ExplainState *es)
+show_hash_info(HashJoinState *hashstate, ExplainState *es)
{
HashInstrumentation hinstrument = {0};
@@ -2628,7 +2681,7 @@ show_hash_info(HashState *hashstate, ExplainState *es)
* prepared to get instrumentation data from all participants.
*/
if (hashstate->hashtable)
- ExecHashGetInstrumentation(&hinstrument, hashstate->hashtable);
+ ExecHashJoinGetInstrumentation(&hinstrument, hashstate->hashtable);
/*
* Merge results from workers. In the parallel-oblivious case, the
diff --git a/src/backend/executor/Makefile b/src/backend/executor/Makefile
index cc09895fa5c..057a8b72f72 100644
--- a/src/backend/executor/Makefile
+++ b/src/backend/executor/Makefile
@@ -20,7 +20,7 @@ OBJS = execAmi.o execCurrent.o execExpr.o execExprInterp.o \
nodeBitmapAnd.o nodeBitmapOr.o \
nodeBitmapHeapscan.o nodeBitmapIndexscan.o \
nodeCustom.o nodeFunctionscan.o nodeGather.o \
- nodeHash.o nodeHashjoin.o nodeIndexscan.o nodeIndexonlyscan.o \
+ nodeHashjoin.o nodeIndexscan.o nodeIndexonlyscan.o \
nodeLimit.o nodeLockRows.o nodeGatherMerge.o \
nodeMaterial.o nodeMergeAppend.o nodeMergejoin.o nodeModifyTable.o \
nodeNestloop.o nodeProjectSet.o nodeRecursiveunion.o nodeResult.o \
diff --git a/src/backend/executor/execAmi.c b/src/backend/executor/execAmi.c
index 1f18e5d3a2f..d98a274346b 100644
--- a/src/backend/executor/execAmi.c
+++ b/src/backend/executor/execAmi.c
@@ -29,7 +29,6 @@
#include "executor/nodeGatherMerge.h"
#include "executor/nodeGroup.h"
#include "executor/nodeGroup.h"
-#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "executor/nodeIndexonlyscan.h"
#include "executor/nodeIndexscan.h"
@@ -270,10 +269,6 @@ ExecReScan(PlanState *node)
ExecReScanUnique((UniqueState *) node);
break;
- case T_HashState:
- ExecReScanHash((HashState *) node);
- break;
-
case T_SetOpState:
ExecReScanSetOp((SetOpState *) node);
break;
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index 53cd2fc666b..af13356525d 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -29,7 +29,6 @@
#include "executor/nodeBitmapHeapscan.h"
#include "executor/nodeCustom.h"
#include "executor/nodeForeignscan.h"
-#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "executor/nodeIndexscan.h"
#include "executor/nodeIndexonlyscan.h"
@@ -268,13 +267,9 @@ ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e)
e->pcxt);
break;
case T_HashJoinState:
- if (planstate->plan->parallel_aware)
- ExecHashJoinEstimate((HashJoinState *) planstate,
- e->pcxt);
- break;
- case T_HashState:
/* even when not parallel-aware, for EXPLAIN ANALYZE */
- ExecHashEstimate((HashState *) planstate, e->pcxt);
+ ExecHashJoinEstimate((HashJoinState *) planstate,
+ e->pcxt);
break;
case T_SortState:
/* even when not parallel-aware, for EXPLAIN ANALYZE */
@@ -481,13 +476,9 @@ ExecParallelInitializeDSM(PlanState *planstate,
d->pcxt);
break;
case T_HashJoinState:
- if (planstate->plan->parallel_aware)
- ExecHashJoinInitializeDSM((HashJoinState *) planstate,
- d->pcxt);
- break;
- case T_HashState:
/* even when not parallel-aware, for EXPLAIN ANALYZE */
- ExecHashInitializeDSM((HashState *) planstate, d->pcxt);
+ ExecHashJoinInitializeDSM((HashJoinState *) planstate,
+ d->pcxt);
break;
case T_SortState:
/* even when not parallel-aware, for EXPLAIN ANALYZE */
@@ -953,7 +944,6 @@ ExecParallelReInitializeDSM(PlanState *planstate,
ExecHashJoinReInitializeDSM((HashJoinState *) planstate,
pcxt);
break;
- case T_HashState:
case T_SortState:
/* these nodes have DSM state, but no reinitialization is required */
break;
@@ -1015,8 +1005,8 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate,
case T_SortState:
ExecSortRetrieveInstrumentation((SortState *) planstate);
break;
- case T_HashState:
- ExecHashRetrieveInstrumentation((HashState *) planstate);
+ case T_HashJoinState:
+ ExecHashJoinRetrieveInstrumentation((HashJoinState *) planstate);
break;
default:
break;
@@ -1289,13 +1279,9 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt)
pwcxt);
break;
case T_HashJoinState:
- if (planstate->plan->parallel_aware)
- ExecHashJoinInitializeWorker((HashJoinState *) planstate,
- pwcxt);
- break;
- case T_HashState:
/* even when not parallel-aware, for EXPLAIN ANALYZE */
- ExecHashInitializeWorker((HashState *) planstate, pwcxt);
+ ExecHashJoinInitializeWorker((HashJoinState *) planstate,
+ pwcxt);
break;
case T_SortState:
/* even when not parallel-aware, for EXPLAIN ANALYZE */
diff --git a/src/backend/executor/execProcnode.c b/src/backend/executor/execProcnode.c
index c227282975a..0979715ba05 100644
--- a/src/backend/executor/execProcnode.c
+++ b/src/backend/executor/execProcnode.c
@@ -86,7 +86,6 @@
#include "executor/nodeGather.h"
#include "executor/nodeGatherMerge.h"
#include "executor/nodeGroup.h"
-#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "executor/nodeIndexonlyscan.h"
#include "executor/nodeIndexscan.h"
@@ -344,11 +343,6 @@ ExecInitNode(Plan *node, EState *estate, int eflags)
estate, eflags);
break;
- case T_Hash:
- result = (PlanState *) ExecInitHash((Hash *) node,
- estate, eflags);
- break;
-
case T_SetOp:
result = (PlanState *) ExecInitSetOp((SetOp *) node,
estate, eflags);
@@ -497,10 +491,6 @@ MultiExecProcNode(PlanState *node)
* Only node types that actually support multiexec will be listed
*/
- case T_HashState:
- result = MultiExecHash((HashState *) node);
- break;
-
case T_BitmapIndexScanState:
result = MultiExecBitmapIndexScan((BitmapIndexScanState *) node);
break;
@@ -710,10 +700,6 @@ ExecEndNode(PlanState *node)
ExecEndUnique((UniqueState *) node);
break;
- case T_HashState:
- ExecEndHash((HashState *) node);
- break;
-
case T_SetOpState:
ExecEndSetOp((SetOpState *) node);
break;
@@ -775,9 +761,6 @@ ExecShutdownNode(PlanState *node)
case T_GatherMergeState:
ExecShutdownGatherMerge((GatherMergeState *) node);
break;
- case T_HashState:
- ExecShutdownHash((HashState *) node);
- break;
case T_HashJoinState:
ExecShutdownHashJoin((HashJoinState *) node);
break;
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
deleted file mode 100644
index 224cbb32bad..00000000000
--- a/src/backend/executor/nodeHash.c
+++ /dev/null
@@ -1,3338 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * nodeHash.c
- * Routines to hash relations for hashjoin
- *
- * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- *
- * IDENTIFICATION
- * src/backend/executor/nodeHash.c
- *
- * See note on parallelism in nodeHashjoin.c.
- *
- *-------------------------------------------------------------------------
- */
-/*
- * INTERFACE ROUTINES
- * MultiExecHash - generate an in-memory hash table of the relation
- * ExecInitHash - initialize node and subnodes
- * ExecEndHash - shutdown node and subnodes
- */
-
-#include "postgres.h"
-
-#include <math.h>
-#include <limits.h>
-
-#include "access/htup_details.h"
-#include "access/parallel.h"
-#include "catalog/pg_statistic.h"
-#include "commands/tablespace.h"
-#include "executor/execdebug.h"
-#include "executor/hashjoin.h"
-#include "executor/nodeHash.h"
-#include "executor/nodeHashjoin.h"
-#include "miscadmin.h"
-#include "pgstat.h"
-#include "port/atomics.h"
-#include "utils/dynahash.h"
-#include "utils/memutils.h"
-#include "utils/lsyscache.h"
-#include "utils/syscache.h"
-
-
-static void ExecHashIncreaseNumBatches(HashJoinTable hashtable);
-static void ExecHashIncreaseNumBuckets(HashJoinTable hashtable);
-static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable);
-static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable);
-static void ExecHashBuildSkewHash(HashJoinTable hashtable, Hash *node,
- int mcvsToUse);
-static void ExecHashSkewTableInsert(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue,
- int bucketNumber);
-static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable);
-
-static void *dense_alloc(HashJoinTable hashtable, Size size);
-static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable,
- size_t size,
- dsa_pointer *shared);
-static void MultiExecPrivateHash(HashState *node);
-static void MultiExecParallelHash(HashState *node);
-static inline HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable table,
- int bucketno);
-static inline HashJoinTuple ExecParallelHashNextTuple(HashJoinTable table,
- HashJoinTuple tuple);
-static inline void ExecParallelHashPushTuple(dsa_pointer_atomic *head,
- HashJoinTuple tuple,
- dsa_pointer tuple_shared);
-static void ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch);
-static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable);
-static void ExecParallelHashRepartitionFirst(HashJoinTable hashtable);
-static void ExecParallelHashRepartitionRest(HashJoinTable hashtable);
-static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table,
- dsa_pointer *shared);
-static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable,
- int batchno,
- size_t size);
-static void ExecParallelHashMergeCounters(HashJoinTable hashtable);
-static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable);
-
-
-/* ----------------------------------------------------------------
- * ExecHash
- *
- * stub for pro forma compliance
- * ----------------------------------------------------------------
- */
-static TupleTableSlot *
-ExecHash(PlanState *pstate)
-{
- elog(ERROR, "Hash node does not support ExecProcNode call convention");
- return NULL;
-}
-
-/* ----------------------------------------------------------------
- * MultiExecHash
- *
- * build hash table for hashjoin, doing partitioning if more
- * than one batch is required.
- * ----------------------------------------------------------------
- */
-Node *
-MultiExecHash(HashState *node)
-{
- /* must provide our own instrumentation support */
- if (node->ps.instrument)
- InstrStartNode(node->ps.instrument);
-
- if (node->parallel_state != NULL)
- MultiExecParallelHash(node);
- else
- MultiExecPrivateHash(node);
-
- /* must provide our own instrumentation support */
- if (node->ps.instrument)
- InstrStopNode(node->ps.instrument, node->hashtable->partialTuples);
-
- /*
- * We do not return the hash table directly because it's not a subtype of
- * Node, and so would violate the MultiExecProcNode API. Instead, our
- * parent Hashjoin node is expected to know how to fish it out of our node
- * state. Ugly but not really worth cleaning up, since Hashjoin knows
- * quite a bit more about Hash besides that.
- */
- return NULL;
-}
-
-/* ----------------------------------------------------------------
- * MultiExecPrivateHash
- *
- * parallel-oblivious version, building a backend-private
- * hash table and (if necessary) batch files.
- * ----------------------------------------------------------------
- */
-static void
-MultiExecPrivateHash(HashState *node)
-{
- PlanState *outerNode;
- List *hashkeys;
- HashJoinTable hashtable;
- TupleTableSlot *slot;
- ExprContext *econtext;
- uint32 hashvalue;
-
- /*
- * get state info from node
- */
- outerNode = outerPlanState(node);
- hashtable = node->hashtable;
-
- /*
- * set expression context
- */
- hashkeys = node->hashkeys;
- econtext = node->ps.ps_ExprContext;
-
- /*
- * Get all tuples from the node below the Hash node and insert into the
- * hash table (or temp files).
- */
- for (;;)
- {
- slot = ExecProcNode(outerNode);
- if (TupIsNull(slot))
- break;
- /* We have to compute the hash value */
- econtext->ecxt_outertuple = slot;
- if (ExecHashGetHashValue(hashtable, econtext, hashkeys,
- false, hashtable->keepNulls,
- &hashvalue))
- {
- int bucketNumber;
-
- bucketNumber = ExecHashGetSkewBucket(hashtable, hashvalue);
- if (bucketNumber != INVALID_SKEW_BUCKET_NO)
- {
- /* It's a skew tuple, so put it into that hash table */
- ExecHashSkewTableInsert(hashtable, slot, hashvalue,
- bucketNumber);
- hashtable->skewTuples += 1;
- }
- else
- {
- /* Not subject to skew optimization, so insert normally */
- ExecHashTableInsert(hashtable, slot, hashvalue);
- }
- hashtable->totalTuples += 1;
- }
- }
-
- /* resize the hash table if needed (NTUP_PER_BUCKET exceeded) */
- if (hashtable->nbuckets != hashtable->nbuckets_optimal)
- ExecHashIncreaseNumBuckets(hashtable);
-
- /* Account for the buckets in spaceUsed (reported in EXPLAIN ANALYZE) */
- hashtable->spaceUsed += hashtable->nbuckets * sizeof(HashJoinTuple);
- if (hashtable->spaceUsed > hashtable->spacePeak)
- hashtable->spacePeak = hashtable->spaceUsed;
-
- hashtable->partialTuples = hashtable->totalTuples;
-}
-
-/* ----------------------------------------------------------------
- * MultiExecParallelHash
- *
- * parallel-aware version, building a shared hash table and
- * (if necessary) batch files using the combined effort of
- * a set of co-operating backends.
- * ----------------------------------------------------------------
- */
-static void
-MultiExecParallelHash(HashState *node)
-{
- ParallelHashJoinState *pstate;
- PlanState *outerNode;
- List *hashkeys;
- HashJoinTable hashtable;
- TupleTableSlot *slot;
- ExprContext *econtext;
- uint32 hashvalue;
- Barrier *build_barrier;
- int i;
-
- /*
- * get state info from node
- */
- outerNode = outerPlanState(node);
- hashtable = node->hashtable;
-
- /*
- * set expression context
- */
- hashkeys = node->hashkeys;
- econtext = node->ps.ps_ExprContext;
-
- /*
- * Synchronize the parallel hash table build. At this stage we know that
- * the shared hash table has been or is being set up by
- * ExecHashTableCreate(), but we don't know if our peers have returned
- * from there or are here in MultiExecParallelHash(), and if so how far
- * through they are. To find out, we check the build_barrier phase then
- * and jump to the right step in the build algorithm.
- */
- pstate = hashtable->parallel_state;
- build_barrier = &pstate->build_barrier;
- Assert(BarrierPhase(build_barrier) >= PHJ_BUILD_ALLOCATING);
- 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.
- */
- BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ALLOCATING);
- /* Fall through. */
-
- 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);
- ExecParallelHashTableSetCurrentBatch(hashtable, 0);
- for (;;)
- {
- slot = ExecProcNode(outerNode);
- if (TupIsNull(slot))
- break;
- econtext->ecxt_outertuple = slot;
- if (ExecHashGetHashValue(hashtable, econtext, hashkeys,
- false, hashtable->keepNulls,
- &hashvalue))
- ExecParallelHashTableInsert(hashtable, slot, hashvalue);
- hashtable->partialTuples++;
- }
-
- /*
- * Make sure that any tuples we wrote to disk are visible to
- * others before anyone tries to load them.
- */
- for (i = 0; i < hashtable->nbatch; ++i)
- sts_end_write(hashtable->batches[i].inner_tuples);
-
- /*
- * Update shared counters. We need an accurate total tuple count
- * to control the empty table optimization.
- */
- ExecParallelHashMergeCounters(hashtable);
-
- BarrierDetach(&pstate->grow_buckets_barrier);
- BarrierDetach(&pstate->grow_batches_barrier);
-
- /*
- * Wait for everyone to finish building and flushing files and
- * counters.
- */
- if (BarrierArriveAndWait(build_barrier,
- WAIT_EVENT_HASH_BUILD_HASHING_INNER))
- {
- /*
- * Elect one backend to disable any further growth. Batches
- * are now fixed. While building them we made sure they'd fit
- * in our memory budget when we load them back in later (or we
- * tried to do that and gave up because we detected extreme
- * skew).
- */
- pstate->growth = PHJ_GROWTH_DISABLED;
- }
- }
-
- /*
- * We're not yet attached to a batch. We all agree on the dimensions and
- * number of inner tuples (for the empty table optimization).
- */
- hashtable->curbatch = -1;
- hashtable->nbuckets = pstate->nbuckets;
- hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
- hashtable->totalTuples = pstate->total_tuples;
- 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
- * there already).
- */
- Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER ||
- BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
-}
-
-/* ----------------------------------------------------------------
- * ExecInitHash
- *
- * Init routine for Hash node
- * ----------------------------------------------------------------
- */
-HashState *
-ExecInitHash(Hash *node, EState *estate, int eflags)
-{
- HashState *hashstate;
-
- /* check for unsupported flags */
- Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
-
- /*
- * create state structure
- */
- hashstate = makeNode(HashState);
- hashstate->ps.plan = (Plan *) node;
- hashstate->ps.state = estate;
- hashstate->ps.ExecProcNode = ExecHash;
- hashstate->hashtable = NULL;
- hashstate->hashkeys = NIL; /* will be set by parent HashJoin */
-
- /*
- * Miscellaneous initialization
- *
- * create expression context for node
- */
- ExecAssignExprContext(estate, &hashstate->ps);
-
- /*
- * initialize child nodes
- */
- outerPlanState(hashstate) = ExecInitNode(outerPlan(node), estate, eflags);
-
- /*
- * initialize our result slot and type. No need to build projection
- * because this node doesn't do projections.
- */
- ExecInitResultTupleSlotTL(&hashstate->ps, &TTSOpsMinimalTuple);
- hashstate->ps.ps_ProjInfo = NULL;
-
- /*
- * initialize child expressions
- */
- Assert(node->plan.qual == NIL);
- hashstate->hashkeys =
- ExecInitExprList(node->hashkeys, (PlanState *) hashstate);
-
- return hashstate;
-}
-
-/* ---------------------------------------------------------------
- * ExecEndHash
- *
- * clean up routine for Hash node
- * ----------------------------------------------------------------
- */
-void
-ExecEndHash(HashState *node)
-{
- PlanState *outerPlan;
-
- /*
- * free exprcontext
- */
- ExecFreeExprContext(&node->ps);
-
- /*
- * shut down the subplan
- */
- outerPlan = outerPlanState(node);
- ExecEndNode(outerPlan);
-}
-
-
-/* ----------------------------------------------------------------
- * ExecHashTableCreate
- *
- * create an empty hashtable data structure for hashjoin.
- * ----------------------------------------------------------------
- */
-HashJoinTable
-ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations, bool keepNulls)
-{
- Hash *node;
- HashJoinTable hashtable;
- Plan *outerNode;
- size_t space_allowed;
- int nbuckets;
- int nbatch;
- double rows;
- int num_skew_mcvs;
- int log2_nbuckets;
- int nkeys;
- int i;
- ListCell *ho;
- ListCell *hc;
- MemoryContext oldcxt;
-
- /*
- * Get information about the size of the relation to be hashed (it's the
- * "outer" subtree of this node, but the inner relation of the hashjoin).
- * Compute the appropriate size of the hash table.
- */
- node = (Hash *) state->ps.plan;
- outerNode = outerPlan(node);
-
- /*
- * If this is shared hash table with a partial plan, then we can't use
- * outerNode->plan_rows to estimate its size. We need an estimate of the
- * total number of rows across all copies of the partial plan.
- */
- rows = node->plan.parallel_aware ? node->rows_total : outerNode->plan_rows;
-
- ExecChooseHashTableSize(rows, outerNode->plan_width,
- OidIsValid(node->skewTable),
- state->parallel_state != NULL,
- state->parallel_state != NULL ?
- state->parallel_state->nparticipants - 1 : 0,
- &space_allowed,
- &nbuckets, &nbatch, &num_skew_mcvs);
-
- /* nbuckets must be a power of 2 */
- log2_nbuckets = my_log2(nbuckets);
- Assert(nbuckets == (1 << log2_nbuckets));
-
- /*
- * Initialize the hash table control block.
- *
- * The hashtable control block is just palloc'd from the executor's
- * per-query memory context. Everything else should be kept inside the
- * subsidiary hashCxt or batchCxt.
- */
- hashtable = (HashJoinTable) palloc(sizeof(HashJoinTableData));
- hashtable->nbuckets = nbuckets;
- hashtable->nbuckets_original = nbuckets;
- hashtable->nbuckets_optimal = nbuckets;
- hashtable->log2_nbuckets = log2_nbuckets;
- hashtable->log2_nbuckets_optimal = log2_nbuckets;
- hashtable->buckets.unshared = NULL;
- hashtable->keepNulls = keepNulls;
- hashtable->skewEnabled = false;
- hashtable->skewBucket = NULL;
- hashtable->skewBucketLen = 0;
- hashtable->nSkewBuckets = 0;
- hashtable->skewBucketNums = NULL;
- hashtable->nbatch = nbatch;
- hashtable->curbatch = 0;
- hashtable->nbatch_original = nbatch;
- hashtable->nbatch_outstart = nbatch;
- hashtable->growEnabled = true;
- hashtable->totalTuples = 0;
- hashtable->partialTuples = 0;
- hashtable->skewTuples = 0;
- hashtable->innerBatchFile = NULL;
- hashtable->outerBatchFile = NULL;
- hashtable->spaceUsed = 0;
- hashtable->spacePeak = 0;
- hashtable->spaceAllowed = space_allowed;
- hashtable->spaceUsedSkew = 0;
- hashtable->spaceAllowedSkew =
- hashtable->spaceAllowed * SKEW_WORK_MEM_PERCENT / 100;
- hashtable->chunks = NULL;
- hashtable->current_chunk = NULL;
- hashtable->parallel_state = state->parallel_state;
- hashtable->area = state->ps.state->es_query_dsa;
- hashtable->batches = NULL;
-
-#ifdef HJDEBUG
- printf("Hashjoin %p: initial nbatch = %d, nbuckets = %d\n",
- hashtable, nbatch, nbuckets);
-#endif
-
- /*
- * Create temporary memory contexts in which to keep the hashtable working
- * storage. See notes in executor/hashjoin.h.
- */
- hashtable->hashCxt = AllocSetContextCreate(CurrentMemoryContext,
- "HashTableContext",
- ALLOCSET_DEFAULT_SIZES);
-
- hashtable->batchCxt = AllocSetContextCreate(hashtable->hashCxt,
- "HashBatchContext",
- ALLOCSET_DEFAULT_SIZES);
-
- /* Allocate data that will live for the life of the hashjoin */
-
- oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
-
- /*
- * Get info about the hash functions to be used for each hash key. Also
- * remember whether the join operators are strict.
- */
- nkeys = list_length(hashOperators);
- hashtable->outer_hashfunctions =
- (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
- hashtable->inner_hashfunctions =
- (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
- hashtable->hashStrict = (bool *) palloc(nkeys * sizeof(bool));
- hashtable->collations = (Oid *) palloc(nkeys * sizeof(Oid));
- i = 0;
- forboth(ho, hashOperators, hc, hashCollations)
- {
- Oid hashop = lfirst_oid(ho);
- Oid left_hashfn;
- Oid right_hashfn;
-
- if (!get_op_hash_functions(hashop, &left_hashfn, &right_hashfn))
- elog(ERROR, "could not find hash function for hash operator %u",
- hashop);
- fmgr_info(left_hashfn, &hashtable->outer_hashfunctions[i]);
- fmgr_info(right_hashfn, &hashtable->inner_hashfunctions[i]);
- hashtable->hashStrict[i] = op_strict(hashop);
- hashtable->collations[i] = lfirst_oid(hc);
- i++;
- }
-
- if (nbatch > 1 && hashtable->parallel_state == NULL)
- {
- /*
- * allocate and initialize the file arrays in hashCxt (not needed for
- * parallel case which uses shared tuplestores instead of raw files)
- */
- hashtable->innerBatchFile = (BufFile **)
- palloc0(nbatch * sizeof(BufFile *));
- hashtable->outerBatchFile = (BufFile **)
- palloc0(nbatch * sizeof(BufFile *));
- /* The files will not be opened until needed... */
- /* ... but make sure we have temp tablespaces established for them */
- PrepareTempTablespaces();
- }
-
- MemoryContextSwitchTo(oldcxt);
-
- if (hashtable->parallel_state)
- {
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- Barrier *build_barrier;
-
- /*
- * 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
- * and then loaded while hashing (the standard hybrid hash join
- * algorithm), and we'll coordinate that using build_barrier.
- */
- build_barrier = &pstate->build_barrier;
- BarrierAttach(build_barrier);
-
- /*
- * So far we have no idea whether there are any other participants,
- * and if so, what phase they are working on. The only thing we care
- * about at this point is whether someone has already created the
- * 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 &&
- BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ELECTING))
- {
- pstate->nbatch = nbatch;
- pstate->space_allowed = space_allowed;
- pstate->growth = PHJ_GROWTH_OK;
-
- /* Set up the shared state for coordinating batches. */
- ExecParallelHashJoinSetUpBatches(hashtable, nbatch);
-
- /*
- * Allocate batch 0's hash table up front so we can load it
- * directly while hashing.
- */
- pstate->nbuckets = nbuckets;
- ExecParallelHashTableAlloc(hashtable, 0);
- }
-
- /*
- * 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
- * executor node between now and then.
- */
- }
- else
- {
- /*
- * Prepare context for the first-scan space allocations; allocate the
- * hashbucket array therein, and set each bucket "empty".
- */
- MemoryContextSwitchTo(hashtable->batchCxt);
-
- hashtable->buckets.unshared = (HashJoinTuple *)
- palloc0(nbuckets * sizeof(HashJoinTuple));
-
- /*
- * Set up for skew optimization, if possible and there's a need for
- * more than one batch. (In a one-batch join, there's no point in
- * it.)
- */
- if (nbatch > 1)
- ExecHashBuildSkewHash(hashtable, node, num_skew_mcvs);
-
- MemoryContextSwitchTo(oldcxt);
- }
-
- return hashtable;
-}
-
-
-/*
- * Compute appropriate size for hashtable given the estimated size of the
- * relation to be hashed (number of rows and average row width).
- *
- * This is exported so that the planner's costsize.c can use it.
- */
-
-/* Target bucket loading (tuples per bucket) */
-#define NTUP_PER_BUCKET 1
-
-void
-ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
- bool try_combined_work_mem,
- int parallel_workers,
- size_t *space_allowed,
- int *numbuckets,
- int *numbatches,
- int *num_skew_mcvs)
-{
- int tupsize;
- double inner_rel_bytes;
- long bucket_bytes;
- long hash_table_bytes;
- long skew_table_bytes;
- long max_pointers;
- long mppow2;
- int nbatch = 1;
- int nbuckets;
- double dbuckets;
-
- /* Force a plausible relation size if no info */
- if (ntuples <= 0.0)
- ntuples = 1000.0;
-
- /*
- * Estimate tupsize based on footprint of tuple in hashtable... note this
- * does not allow for any palloc overhead. The manipulations of spaceUsed
- * don't count palloc overhead either.
- */
- tupsize = HJTUPLE_OVERHEAD +
- MAXALIGN(SizeofMinimalTupleHeader) +
- MAXALIGN(tupwidth);
- inner_rel_bytes = ntuples * tupsize;
-
- /*
- * Target in-memory hashtable size is work_mem kilobytes.
- */
- hash_table_bytes = work_mem * 1024L;
-
- /*
- * 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.
- */
- if (try_combined_work_mem)
- hash_table_bytes += hash_table_bytes * parallel_workers;
-
- *space_allowed = hash_table_bytes;
-
- /*
- * If skew optimization is possible, estimate the number of skew buckets
- * that will fit in the memory allowed, and decrement the assumed space
- * available for the main hash table accordingly.
- *
- * We make the optimistic assumption that each skew bucket will contain
- * one inner-relation tuple. If that turns out to be low, we will recover
- * at runtime by reducing the number of skew buckets.
- *
- * hashtable->skewBucket will have up to 8 times as many HashSkewBucket
- * pointers as the number of MCVs we allow, since ExecHashBuildSkewHash
- * will round up to the next power of 2 and then multiply by 4 to reduce
- * collisions.
- */
- if (useskew)
- {
- skew_table_bytes = hash_table_bytes * SKEW_WORK_MEM_PERCENT / 100;
-
- /*----------
- * Divisor is:
- * size of a hash tuple +
- * worst-case size of skewBucket[] per MCV +
- * size of skewBucketNums[] entry +
- * size of skew bucket struct itself
- *----------
- */
- *num_skew_mcvs = skew_table_bytes / (tupsize +
- (8 * sizeof(HashSkewBucket *)) +
- sizeof(int) +
- SKEW_BUCKET_OVERHEAD);
- if (*num_skew_mcvs > 0)
- hash_table_bytes -= skew_table_bytes;
- }
- else
- *num_skew_mcvs = 0;
-
- /*
- * Set nbuckets to achieve an average bucket load of NTUP_PER_BUCKET when
- * memory is filled, assuming a single batch; but limit the value so that
- * the pointer arrays we'll try to allocate do not exceed work_mem nor
- * MaxAllocSize.
- *
- * Note that both nbuckets and nbatch must be powers of 2 to make
- * ExecHashGetBucketAndBatch fast.
- */
- max_pointers = *space_allowed / sizeof(HashJoinTuple);
- max_pointers = Min(max_pointers, MaxAllocSize / sizeof(HashJoinTuple));
- /* If max_pointers isn't a power of 2, must round it down to one */
- mppow2 = 1L << my_log2(max_pointers);
- if (max_pointers != mppow2)
- max_pointers = mppow2 / 2;
-
- /* Also ensure we avoid integer overflow in nbatch and nbuckets */
- /* (this step is redundant given the current value of MaxAllocSize) */
- max_pointers = Min(max_pointers, INT_MAX / 2);
-
- dbuckets = ceil(ntuples / NTUP_PER_BUCKET);
- dbuckets = Min(dbuckets, max_pointers);
- nbuckets = (int) dbuckets;
- /* don't let nbuckets be really small, though ... */
- nbuckets = Max(nbuckets, 1024);
- /* ... and force it to be a power of 2. */
- nbuckets = 1 << my_log2(nbuckets);
-
- /*
- * If there's not enough space to store the projected number of tuples and
- * the required bucket headers, we will need multiple batches.
- */
- bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
- if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
- {
- /* We'll need multiple batches */
- long lbuckets;
- double dbatch;
- int minbatch;
- long bucket_size;
-
- /*
- * If Parallel Hash with combined work_mem would still need multiple
- * batches, we'll have to fall back to regular work_mem budget.
- */
- if (try_combined_work_mem)
- {
- ExecChooseHashTableSize(ntuples, tupwidth, useskew,
- false, parallel_workers,
- space_allowed,
- numbuckets,
- numbatches,
- num_skew_mcvs);
- return;
- }
-
- /*
- * Estimate the number of buckets we'll want to have when work_mem is
- * entirely full. Each bucket will contain a bucket pointer plus
- * NTUP_PER_BUCKET tuples, whose projected size already includes
- * overhead for the hash code, pointer to the next tuple, etc.
- */
- bucket_size = (tupsize * NTUP_PER_BUCKET + sizeof(HashJoinTuple));
- lbuckets = 1L << my_log2(hash_table_bytes / bucket_size);
- lbuckets = Min(lbuckets, max_pointers);
- nbuckets = (int) lbuckets;
- nbuckets = 1 << my_log2(nbuckets);
- bucket_bytes = nbuckets * sizeof(HashJoinTuple);
-
- /*
- * Buckets are simple pointers to hashjoin tuples, while tupsize
- * includes the pointer, hash code, and MinimalTupleData. So buckets
- * should never really exceed 25% of work_mem (even for
- * NTUP_PER_BUCKET=1); except maybe for work_mem values that are not
- * 2^N bytes, where we might get more because of doubling. So let's
- * look for 50% here.
- */
- Assert(bucket_bytes <= hash_table_bytes / 2);
-
- /* Calculate required number of batches. */
- dbatch = ceil(inner_rel_bytes / (hash_table_bytes - bucket_bytes));
- dbatch = Min(dbatch, max_pointers);
- minbatch = (int) dbatch;
- nbatch = 2;
- while (nbatch < minbatch)
- nbatch <<= 1;
- }
-
- Assert(nbuckets > 0);
- Assert(nbatch > 0);
-
- *numbuckets = nbuckets;
- *numbatches = nbatch;
-}
-
-
-/* ----------------------------------------------------------------
- * ExecHashTableDestroy
- *
- * destroy a hash table
- * ----------------------------------------------------------------
- */
-void
-ExecHashTableDestroy(HashJoinTable hashtable)
-{
- int i;
-
- /*
- * Make sure all the temp files are closed. We skip batch 0, since it
- * can't have any temp files (and the arrays might not even exist if
- * nbatch is only 1). Parallel hash joins don't use these files.
- */
- if (hashtable->innerBatchFile != NULL)
- {
- for (i = 1; i < hashtable->nbatch; i++)
- {
- if (hashtable->innerBatchFile[i])
- BufFileClose(hashtable->innerBatchFile[i]);
- if (hashtable->outerBatchFile[i])
- BufFileClose(hashtable->outerBatchFile[i]);
- }
- }
-
- /* Release working memory (batchCxt is a child, so it goes away too) */
- MemoryContextDelete(hashtable->hashCxt);
-
- /* And drop the control block */
- pfree(hashtable);
-}
-
-/*
- * ExecHashIncreaseNumBatches
- * increase the original number of batches in order to reduce
- * current memory consumption
- */
-static void
-ExecHashIncreaseNumBatches(HashJoinTable hashtable)
-{
- int oldnbatch = hashtable->nbatch;
- int curbatch = hashtable->curbatch;
- int nbatch;
- MemoryContext oldcxt;
- long ninmemory;
- long nfreed;
- HashMemoryChunk oldchunks;
-
- /* do nothing if we've decided to shut off growth */
- if (!hashtable->growEnabled)
- return;
-
- /* safety check to avoid overflow */
- if (oldnbatch > Min(INT_MAX / 2, MaxAllocSize / (sizeof(void *) * 2)))
- return;
-
- nbatch = oldnbatch * 2;
- Assert(nbatch > 1);
-
-#ifdef HJDEBUG
- printf("Hashjoin %p: increasing nbatch to %d because space = %zu\n",
- hashtable, nbatch, hashtable->spaceUsed);
-#endif
-
- oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
-
- if (hashtable->innerBatchFile == NULL)
- {
- /* we had no file arrays before */
- hashtable->innerBatchFile = (BufFile **)
- palloc0(nbatch * sizeof(BufFile *));
- hashtable->outerBatchFile = (BufFile **)
- palloc0(nbatch * sizeof(BufFile *));
- /* time to establish the temp tablespaces, too */
- PrepareTempTablespaces();
- }
- else
- {
- /* enlarge arrays and zero out added entries */
- hashtable->innerBatchFile = (BufFile **)
- repalloc(hashtable->innerBatchFile, nbatch * sizeof(BufFile *));
- hashtable->outerBatchFile = (BufFile **)
- repalloc(hashtable->outerBatchFile, nbatch * sizeof(BufFile *));
- MemSet(hashtable->innerBatchFile + oldnbatch, 0,
- (nbatch - oldnbatch) * sizeof(BufFile *));
- MemSet(hashtable->outerBatchFile + oldnbatch, 0,
- (nbatch - oldnbatch) * sizeof(BufFile *));
- }
-
- MemoryContextSwitchTo(oldcxt);
-
- hashtable->nbatch = nbatch;
-
- /*
- * Scan through the existing hash table entries and dump out any that are
- * no longer of the current batch.
- */
- ninmemory = nfreed = 0;
-
- /* If know we need to resize nbuckets, we can do it while rebatching. */
- if (hashtable->nbuckets_optimal != hashtable->nbuckets)
- {
- /* we never decrease the number of buckets */
- Assert(hashtable->nbuckets_optimal > hashtable->nbuckets);
-
- hashtable->nbuckets = hashtable->nbuckets_optimal;
- hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
-
- hashtable->buckets.unshared =
- repalloc(hashtable->buckets.unshared,
- sizeof(HashJoinTuple) * hashtable->nbuckets);
- }
-
- /*
- * We will scan through the chunks directly, so that we can reset the
- * buckets now and not have to keep track which tuples in the buckets have
- * already been processed. We will free the old chunks as we go.
- */
- memset(hashtable->buckets.unshared, 0,
- sizeof(HashJoinTuple) * hashtable->nbuckets);
- oldchunks = hashtable->chunks;
- hashtable->chunks = NULL;
-
- /* so, let's scan through the old chunks, and all tuples in each chunk */
- while (oldchunks != NULL)
- {
- HashMemoryChunk nextchunk = oldchunks->next.unshared;
-
- /* position within the buffer (up to oldchunks->used) */
- size_t idx = 0;
-
- /* process all tuples stored in this chunk (and then free it) */
- while (idx < oldchunks->used)
- {
- HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(oldchunks) + idx);
- MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
- int hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len);
- int bucketno;
- int batchno;
-
- ninmemory++;
- ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
- &bucketno, &batchno);
-
- if (batchno == curbatch)
- {
- /* keep tuple in memory - copy it into the new chunk */
- HashJoinTuple copyTuple;
-
- copyTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
- memcpy(copyTuple, hashTuple, hashTupleSize);
-
- /* and add it back to the appropriate bucket */
- copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
- hashtable->buckets.unshared[bucketno] = copyTuple;
- }
- else
- {
- /* dump it out */
- Assert(batchno > curbatch);
- ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple),
- hashTuple->hashvalue,
- &hashtable->innerBatchFile[batchno]);
-
- hashtable->spaceUsed -= hashTupleSize;
- nfreed++;
- }
-
- /* next tuple in this chunk */
- idx += MAXALIGN(hashTupleSize);
-
- /* allow this loop to be cancellable */
- CHECK_FOR_INTERRUPTS();
- }
-
- /* we're done with this chunk - free it and proceed to the next one */
- pfree(oldchunks);
- oldchunks = nextchunk;
- }
-
-#ifdef HJDEBUG
- printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n",
- hashtable, nfreed, ninmemory, hashtable->spaceUsed);
-#endif
-
- /*
- * If we dumped out either all or none of the tuples in the table, disable
- * further expansion of nbatch. This situation implies that we have
- * enough tuples of identical hashvalues to overflow spaceAllowed.
- * Increasing nbatch will not fix it since there's no way to subdivide the
- * group any more finely. We have to just gut it out and hope the server
- * has enough RAM.
- */
- if (nfreed == 0 || nfreed == ninmemory)
- {
- hashtable->growEnabled = false;
-#ifdef HJDEBUG
- printf("Hashjoin %p: disabling further increase of nbatch\n",
- hashtable);
-#endif
- }
-}
-
-/*
- * ExecParallelHashIncreaseNumBatches
- * Every participant attached to grow_batches_barrier must run this
- * function when it observes growth == PHJ_GROWTH_NEED_MORE_BATCHES.
- */
-static void
-ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- int i;
-
- Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
-
- /*
- * It's unlikely, but we need to be prepared for new participants to show
- * up while we're in the middle of this operation so we need to switch on
- * barrier phase here.
- */
- switch (PHJ_GROW_BATCHES_PHASE(BarrierPhase(&pstate->grow_batches_barrier)))
- {
- case PHJ_GROW_BATCHES_ELECTING:
-
- /*
- * Elect one participant to prepare to grow the number of batches.
- * This involves reallocating or resetting the buckets of batch 0
- * in preparation for all participants to begin repartitioning the
- * tuples.
- */
- if (BarrierArriveAndWait(&pstate->grow_batches_barrier,
- WAIT_EVENT_HASH_GROW_BATCHES_ELECTING))
- {
- dsa_pointer_atomic *buckets;
- ParallelHashJoinBatch *old_batch0;
- int new_nbatch;
- int i;
-
- /* Move the old batch out of the way. */
- old_batch0 = hashtable->batches[0].shared;
- pstate->old_batches = pstate->batches;
- pstate->old_nbatch = hashtable->nbatch;
- pstate->batches = InvalidDsaPointer;
-
- /* Free this backend's old accessors. */
- ExecParallelHashCloseBatchAccessors(hashtable);
-
- /* Figure out how many batches to use. */
- if (hashtable->nbatch == 1)
- {
- /*
- * We are going from single-batch to multi-batch. We need
- * to switch from one large combined memory budget to the
- * regular work_mem budget.
- */
- pstate->space_allowed = work_mem * 1024L;
-
- /*
- * The combined work_mem of all participants wasn't
- * enough. Therefore one batch per participant would be
- * approximately equivalent and would probably also be
- * insufficient. So try two batches per participant,
- * rounded up to a power of two.
- */
- new_nbatch = 1 << my_log2(pstate->nparticipants * 2);
- }
- else
- {
- /*
- * We were already multi-batched. Try doubling the number
- * of batches.
- */
- new_nbatch = hashtable->nbatch * 2;
- }
-
- /* Allocate new larger generation of batches. */
- Assert(hashtable->nbatch == pstate->nbatch);
- ExecParallelHashJoinSetUpBatches(hashtable, new_nbatch);
- Assert(hashtable->nbatch == pstate->nbatch);
-
- /* Replace or recycle batch 0's bucket array. */
- if (pstate->old_nbatch == 1)
- {
- double dtuples;
- double dbuckets;
- int new_nbuckets;
-
- /*
- * We probably also need a smaller bucket array. How many
- * tuples do we expect per batch, assuming we have only
- * half of them so far? Normally we don't need to change
- * the bucket array's size, because the size of each batch
- * stays the same as we add more batches, but in this
- * special case we move from a large batch to many smaller
- * batches and it would be wasteful to keep the large
- * array.
- */
- dtuples = (old_batch0->ntuples * 2.0) / new_nbatch;
- dbuckets = ceil(dtuples / NTUP_PER_BUCKET);
- dbuckets = Min(dbuckets,
- MaxAllocSize / sizeof(dsa_pointer_atomic));
- new_nbuckets = (int) dbuckets;
- new_nbuckets = Max(new_nbuckets, 1024);
- new_nbuckets = 1 << my_log2(new_nbuckets);
- dsa_free(hashtable->area, old_batch0->buckets);
- hashtable->batches[0].shared->buckets =
- dsa_allocate(hashtable->area,
- sizeof(dsa_pointer_atomic) * new_nbuckets);
- buckets = (dsa_pointer_atomic *)
- dsa_get_address(hashtable->area,
- hashtable->batches[0].shared->buckets);
- for (i = 0; i < new_nbuckets; ++i)
- dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
- pstate->nbuckets = new_nbuckets;
- }
- else
- {
- /* Recycle the existing bucket array. */
- hashtable->batches[0].shared->buckets = old_batch0->buckets;
- buckets = (dsa_pointer_atomic *)
- dsa_get_address(hashtable->area, old_batch0->buckets);
- for (i = 0; i < hashtable->nbuckets; ++i)
- dsa_pointer_atomic_write(&buckets[i], InvalidDsaPointer);
- }
-
- /* Move all chunks to the work queue for parallel processing. */
- pstate->chunk_work_queue = old_batch0->chunks;
-
- /* Disable further growth temporarily while we're growing. */
- pstate->growth = PHJ_GROWTH_DISABLED;
- }
- else
- {
- /* All other participants just flush their tuples to disk. */
- ExecParallelHashCloseBatchAccessors(hashtable);
- }
- /* Fall through. */
-
- case PHJ_GROW_BATCHES_ALLOCATING:
- /* Wait for the above to be finished. */
- BarrierArriveAndWait(&pstate->grow_batches_barrier,
- WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATING);
- /* Fall through. */
-
- case PHJ_GROW_BATCHES_REPARTITIONING:
- /* Make sure that we have the current dimensions and buckets. */
- ExecParallelHashEnsureBatchAccessors(hashtable);
- ExecParallelHashTableSetCurrentBatch(hashtable, 0);
- /* Then partition, flush counters. */
- ExecParallelHashRepartitionFirst(hashtable);
- ExecParallelHashRepartitionRest(hashtable);
- ExecParallelHashMergeCounters(hashtable);
- /* Wait for the above to be finished. */
- BarrierArriveAndWait(&pstate->grow_batches_barrier,
- WAIT_EVENT_HASH_GROW_BATCHES_REPARTITIONING);
- /* Fall through. */
-
- case PHJ_GROW_BATCHES_DECIDING:
-
- /*
- * Elect one participant to clean up and decide whether further
- * repartitioning is needed, or should be disabled because it's
- * not helping.
- */
- if (BarrierArriveAndWait(&pstate->grow_batches_barrier,
- WAIT_EVENT_HASH_GROW_BATCHES_DECIDING))
- {
- bool space_exhausted = false;
- bool extreme_skew_detected = false;
-
- /* Make sure that we have the current dimensions and buckets. */
- ExecParallelHashEnsureBatchAccessors(hashtable);
- ExecParallelHashTableSetCurrentBatch(hashtable, 0);
-
- /* Are any of the new generation of batches exhausted? */
- for (i = 0; i < hashtable->nbatch; ++i)
- {
- ParallelHashJoinBatch *batch = hashtable->batches[i].shared;
-
- if (batch->space_exhausted ||
- batch->estimated_size > pstate->space_allowed)
- {
- int parent;
-
- space_exhausted = true;
-
- /*
- * Did this batch receive ALL of the tuples from its
- * parent batch? That would indicate that further
- * repartitioning isn't going to help (the hash values
- * are probably all the same).
- */
- parent = i % pstate->old_nbatch;
- if (batch->ntuples == hashtable->batches[parent].shared->old_ntuples)
- extreme_skew_detected = true;
- }
- }
-
- /* Don't keep growing if it's not helping or we'd overflow. */
- if (extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2)
- pstate->growth = PHJ_GROWTH_DISABLED;
- else if (space_exhausted)
- pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
- else
- pstate->growth = PHJ_GROWTH_OK;
-
- /* Free the old batches in shared memory. */
- dsa_free(hashtable->area, pstate->old_batches);
- pstate->old_batches = InvalidDsaPointer;
- }
- /* Fall through. */
-
- case PHJ_GROW_BATCHES_FINISHING:
- /* Wait for the above to complete. */
- BarrierArriveAndWait(&pstate->grow_batches_barrier,
- WAIT_EVENT_HASH_GROW_BATCHES_FINISHING);
- }
-}
-
-/*
- * Repartition the tuples currently loaded into memory for inner batch 0
- * because the number of batches has been increased. Some tuples are retained
- * in memory and some are written out to a later batch.
- */
-static void
-ExecParallelHashRepartitionFirst(HashJoinTable hashtable)
-{
- dsa_pointer chunk_shared;
- HashMemoryChunk chunk;
-
- Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);
-
- while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared)))
- {
- size_t idx = 0;
-
- /* Repartition all tuples in this chunk. */
- while (idx < chunk->used)
- {
- HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
- MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
- HashJoinTuple copyTuple;
- dsa_pointer shared;
- int bucketno;
- int batchno;
-
- ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
- &bucketno, &batchno);
-
- Assert(batchno < hashtable->nbatch);
- if (batchno == 0)
- {
- /* It still belongs in batch 0. Copy to a new chunk. */
- copyTuple =
- ExecParallelHashTupleAlloc(hashtable,
- HJTUPLE_OVERHEAD + tuple->t_len,
- &shared);
- copyTuple->hashvalue = hashTuple->hashvalue;
- memcpy(HJTUPLE_MINTUPLE(copyTuple), tuple, tuple->t_len);
- ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
- copyTuple, shared);
- }
- else
- {
- size_t tuple_size =
- MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
-
- /* It belongs in a later batch. */
- hashtable->batches[batchno].estimated_size += tuple_size;
- sts_puttuple(hashtable->batches[batchno].inner_tuples,
- &hashTuple->hashvalue, tuple);
- }
-
- /* Count this tuple. */
- ++hashtable->batches[0].old_ntuples;
- ++hashtable->batches[batchno].ntuples;
-
- idx += MAXALIGN(HJTUPLE_OVERHEAD +
- HJTUPLE_MINTUPLE(hashTuple)->t_len);
- }
-
- /* Free this chunk. */
- dsa_free(hashtable->area, chunk_shared);
-
- CHECK_FOR_INTERRUPTS();
- }
-}
-
-/*
- * Help repartition inner batches 1..n.
- */
-static void
-ExecParallelHashRepartitionRest(HashJoinTable hashtable)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- int old_nbatch = pstate->old_nbatch;
- SharedTuplestoreAccessor **old_inner_tuples;
- ParallelHashJoinBatch *old_batches;
- int i;
-
- /* Get our hands on the previous generation of batches. */
- old_batches = (ParallelHashJoinBatch *)
- dsa_get_address(hashtable->area, pstate->old_batches);
- old_inner_tuples = palloc0(sizeof(SharedTuplestoreAccessor *) * old_nbatch);
- for (i = 1; i < old_nbatch; ++i)
- {
- ParallelHashJoinBatch *shared =
- NthParallelHashJoinBatch(old_batches, i);
-
- old_inner_tuples[i] = sts_attach(ParallelHashJoinBatchInner(shared),
- ParallelWorkerNumber + 1,
- &pstate->fileset);
- }
-
- /* Join in the effort to repartition them. */
- for (i = 1; i < old_nbatch; ++i)
- {
- MinimalTuple tuple;
- uint32 hashvalue;
-
- /* Scan one partition from the previous generation. */
- sts_begin_parallel_scan(old_inner_tuples[i]);
- while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue)))
- {
- size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
- int bucketno;
- int batchno;
-
- /* Decide which partition it goes to in the new generation. */
- ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
- &batchno);
-
- hashtable->batches[batchno].estimated_size += tuple_size;
- ++hashtable->batches[batchno].ntuples;
- ++hashtable->batches[i].old_ntuples;
-
- /* Store the tuple its new batch. */
- sts_puttuple(hashtable->batches[batchno].inner_tuples,
- &hashvalue, tuple);
-
- CHECK_FOR_INTERRUPTS();
- }
- sts_end_parallel_scan(old_inner_tuples[i]);
- }
-
- pfree(old_inner_tuples);
-}
-
-/*
- * Transfer the backend-local per-batch counters to the shared totals.
- */
-static void
-ExecParallelHashMergeCounters(HashJoinTable hashtable)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- int i;
-
- LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
- pstate->total_tuples = 0;
- for (i = 0; i < hashtable->nbatch; ++i)
- {
- ParallelHashJoinBatchAccessor *batch = &hashtable->batches[i];
-
- batch->shared->size += batch->size;
- batch->shared->estimated_size += batch->estimated_size;
- batch->shared->ntuples += batch->ntuples;
- batch->shared->old_ntuples += batch->old_ntuples;
- batch->size = 0;
- batch->estimated_size = 0;
- batch->ntuples = 0;
- batch->old_ntuples = 0;
- pstate->total_tuples += batch->shared->ntuples;
- }
- LWLockRelease(&pstate->lock);
-}
-
-/*
- * ExecHashIncreaseNumBuckets
- * increase the original number of buckets in order to reduce
- * number of tuples per bucket
- */
-static void
-ExecHashIncreaseNumBuckets(HashJoinTable hashtable)
-{
- HashMemoryChunk chunk;
-
- /* do nothing if not an increase (it's called increase for a reason) */
- if (hashtable->nbuckets >= hashtable->nbuckets_optimal)
- return;
-
-#ifdef HJDEBUG
- printf("Hashjoin %p: increasing nbuckets %d => %d\n",
- hashtable, hashtable->nbuckets, hashtable->nbuckets_optimal);
-#endif
-
- hashtable->nbuckets = hashtable->nbuckets_optimal;
- hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
-
- Assert(hashtable->nbuckets > 1);
- Assert(hashtable->nbuckets <= (INT_MAX / 2));
- Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets));
-
- /*
- * Just reallocate the proper number of buckets - we don't need to walk
- * through them - we can walk the dense-allocated chunks (just like in
- * ExecHashIncreaseNumBatches, but without all the copying into new
- * chunks)
- */
- hashtable->buckets.unshared =
- (HashJoinTuple *) repalloc(hashtable->buckets.unshared,
- hashtable->nbuckets * sizeof(HashJoinTuple));
-
- memset(hashtable->buckets.unshared, 0,
- hashtable->nbuckets * sizeof(HashJoinTuple));
-
- /* scan through all tuples in all chunks to rebuild the hash table */
- for (chunk = hashtable->chunks; chunk != NULL; chunk = chunk->next.unshared)
- {
- /* process all tuples stored in this chunk */
- size_t idx = 0;
-
- while (idx < chunk->used)
- {
- HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
- int bucketno;
- int batchno;
-
- ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
- &bucketno, &batchno);
-
- /* add the tuple to the proper bucket */
- hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
- hashtable->buckets.unshared[bucketno] = hashTuple;
-
- /* advance index past the tuple */
- idx += MAXALIGN(HJTUPLE_OVERHEAD +
- HJTUPLE_MINTUPLE(hashTuple)->t_len);
- }
-
- /* allow this loop to be cancellable */
- CHECK_FOR_INTERRUPTS();
- }
-}
-
-static void
-ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- int i;
- HashMemoryChunk chunk;
- dsa_pointer chunk_s;
-
- Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
-
- /*
- * It's unlikely, but we need to be prepared for new participants to show
- * up while we're in the middle of this operation so we need to switch on
- * barrier phase here.
- */
- switch (PHJ_GROW_BUCKETS_PHASE(BarrierPhase(&pstate->grow_buckets_barrier)))
- {
- case PHJ_GROW_BUCKETS_ELECTING:
- /* Elect one participant to prepare to increase nbuckets. */
- if (BarrierArriveAndWait(&pstate->grow_buckets_barrier,
- WAIT_EVENT_HASH_GROW_BUCKETS_ELECTING))
- {
- size_t size;
- dsa_pointer_atomic *buckets;
-
- /* Double the size of the bucket array. */
- pstate->nbuckets *= 2;
- size = pstate->nbuckets * sizeof(dsa_pointer_atomic);
- hashtable->batches[0].shared->size += size / 2;
- dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
- hashtable->batches[0].shared->buckets =
- dsa_allocate(hashtable->area, size);
- buckets = (dsa_pointer_atomic *)
- dsa_get_address(hashtable->area,
- hashtable->batches[0].shared->buckets);
- for (i = 0; i < pstate->nbuckets; ++i)
- dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
-
- /* Put the chunk list onto the work queue. */
- pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;
-
- /* Clear the flag. */
- pstate->growth = PHJ_GROWTH_OK;
- }
- /* Fall through. */
-
- case PHJ_GROW_BUCKETS_ALLOCATING:
- /* Wait for the above to complete. */
- BarrierArriveAndWait(&pstate->grow_buckets_barrier,
- WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATING);
- /* Fall through. */
-
- case PHJ_GROW_BUCKETS_REINSERTING:
- /* Reinsert all tuples into the hash table. */
- ExecParallelHashEnsureBatchAccessors(hashtable);
- ExecParallelHashTableSetCurrentBatch(hashtable, 0);
- while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_s)))
- {
- size_t idx = 0;
-
- while (idx < chunk->used)
- {
- HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
- dsa_pointer shared = chunk_s + HASH_CHUNK_HEADER_SIZE + idx;
- int bucketno;
- int batchno;
-
- ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
- &bucketno, &batchno);
- Assert(batchno == 0);
-
- /* add the tuple to the proper bucket */
- ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
- hashTuple, shared);
-
- /* advance index past the tuple */
- idx += MAXALIGN(HJTUPLE_OVERHEAD +
- HJTUPLE_MINTUPLE(hashTuple)->t_len);
- }
-
- /* allow this loop to be cancellable */
- CHECK_FOR_INTERRUPTS();
- }
- BarrierArriveAndWait(&pstate->grow_buckets_barrier,
- WAIT_EVENT_HASH_GROW_BUCKETS_REINSERTING);
- }
-}
-
-/*
- * ExecHashTableInsert
- * insert a tuple into the hash table depending on the hash value
- * it may just go to a temp file for later batches
- *
- * Note: the passed TupleTableSlot may contain a regular, minimal, or virtual
- * tuple; the minimal case in particular is certain to happen while reloading
- * tuples from batch files. We could save some cycles in the regular-tuple
- * case by not forcing the slot contents into minimal form; not clear if it's
- * worth the messiness required.
- */
-void
-ExecHashTableInsert(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue)
-{
- bool shouldFree;
- MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
- int bucketno;
- int batchno;
-
- ExecHashGetBucketAndBatch(hashtable, hashvalue,
- &bucketno, &batchno);
-
- /*
- * decide whether to put the tuple in the hash table or a temp file
- */
- if (batchno == hashtable->curbatch)
- {
- /*
- * put the tuple in hash table
- */
- HashJoinTuple hashTuple;
- int hashTupleSize;
- double ntuples = (hashtable->totalTuples - hashtable->skewTuples);
-
- /* Create the HashJoinTuple */
- hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
- hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
-
- hashTuple->hashvalue = hashvalue;
- memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
-
- /*
- * We always reset the tuple-matched flag on insertion. This is okay
- * even when reloading a tuple from a batch file, since the tuple
- * could not possibly have been matched to an outer tuple before it
- * went into the batch file.
- */
- HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
-
- /* Push it onto the front of the bucket's list */
- hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
- hashtable->buckets.unshared[bucketno] = hashTuple;
-
- /*
- * Increase the (optimal) number of buckets if we just exceeded the
- * NTUP_PER_BUCKET threshold, but only when there's still a single
- * batch.
- */
- if (hashtable->nbatch == 1 &&
- ntuples > (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
- {
- /* Guard against integer overflow and alloc size overflow */
- if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
- hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
- {
- hashtable->nbuckets_optimal *= 2;
- hashtable->log2_nbuckets_optimal += 1;
- }
- }
-
- /* Account for space used, and back off if we've used too much */
- hashtable->spaceUsed += hashTupleSize;
- if (hashtable->spaceUsed > hashtable->spacePeak)
- hashtable->spacePeak = hashtable->spaceUsed;
- if (hashtable->spaceUsed +
- hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
- > hashtable->spaceAllowed)
- ExecHashIncreaseNumBatches(hashtable);
- }
- else
- {
- /*
- * put the tuple into a temp file for later batches
- */
- Assert(batchno > hashtable->curbatch);
- ExecHashJoinSaveTuple(tuple,
- hashvalue,
- &hashtable->innerBatchFile[batchno]);
- }
-
- if (shouldFree)
- heap_free_minimal_tuple(tuple);
-}
-
-/*
- * ExecParallelHashTableInsert
- * insert a tuple into a shared hash table or shared batch tuplestore
- */
-void
-ExecParallelHashTableInsert(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue)
-{
- bool shouldFree;
- MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
- dsa_pointer shared;
- int bucketno;
- int batchno;
-
-retry:
- ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
-
- if (batchno == 0)
- {
- HashJoinTuple hashTuple;
-
- /* Try to load it into memory. */
- Assert(BarrierPhase(&hashtable->parallel_state->build_barrier) ==
- PHJ_BUILD_HASHING_INNER);
- hashTuple = ExecParallelHashTupleAlloc(hashtable,
- HJTUPLE_OVERHEAD + tuple->t_len,
- &shared);
- if (hashTuple == NULL)
- goto retry;
-
- /* Store the hash value in the HashJoinTuple header. */
- hashTuple->hashvalue = hashvalue;
- memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
-
- /* Push it onto the front of the bucket's list */
- ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
- hashTuple, shared);
- }
- else
- {
- size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
-
- Assert(batchno > 0);
-
- /* Try to preallocate space in the batch if necessary. */
- if (hashtable->batches[batchno].preallocated < tuple_size)
- {
- if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size))
- goto retry;
- }
-
- Assert(hashtable->batches[batchno].preallocated >= tuple_size);
- hashtable->batches[batchno].preallocated -= tuple_size;
- sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue,
- tuple);
- }
- ++hashtable->batches[batchno].ntuples;
-
- if (shouldFree)
- heap_free_minimal_tuple(tuple);
-}
-
-/*
- * Insert a tuple into the current hash table. Unlike
- * ExecParallelHashTableInsert, this version is not prepared to send the tuple
- * to other batches or to run out of memory, and should only be called with
- * tuples that belong in the current batch once growth has been disabled.
- */
-void
-ExecParallelHashTableInsertCurrentBatch(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue)
-{
- bool shouldFree;
- MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
- HashJoinTuple hashTuple;
- dsa_pointer shared;
- int batchno;
- int bucketno;
-
- ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
- Assert(batchno == hashtable->curbatch);
- hashTuple = ExecParallelHashTupleAlloc(hashtable,
- HJTUPLE_OVERHEAD + tuple->t_len,
- &shared);
- hashTuple->hashvalue = hashvalue;
- memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
- HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
- ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
- hashTuple, shared);
-
- if (shouldFree)
- heap_free_minimal_tuple(tuple);
-}
-
-/*
- * ExecHashGetHashValue
- * Compute the hash value for a tuple
- *
- * The tuple to be tested must be in econtext->ecxt_outertuple (thus Vars in
- * the hashkeys expressions need to have OUTER_VAR as varno). If outer_tuple
- * is false (meaning it's the HashJoin's inner node, Hash), econtext,
- * hashkeys, and slot need to be from Hash, with hashkeys/slot referencing and
- * being suitable for tuples from the node below the Hash. Conversely, if
- * outer_tuple is true, econtext is from HashJoin, and hashkeys/slot need to
- * be appropriate for tuples from HashJoin's outer node.
- *
- * A true result means the tuple's hash value has been successfully computed
- * and stored at *hashvalue. A false result means the tuple cannot match
- * because it contains a null attribute, and hence it should be discarded
- * immediately. (If keep_nulls is true then false is never returned.)
- */
-bool
-ExecHashGetHashValue(HashJoinTable hashtable,
- ExprContext *econtext,
- List *hashkeys,
- bool outer_tuple,
- bool keep_nulls,
- uint32 *hashvalue)
-{
- uint32 hashkey = 0;
- FmgrInfo *hashfunctions;
- ListCell *hk;
- int i = 0;
- MemoryContext oldContext;
-
- /*
- * We reset the eval context each time to reclaim any memory leaked in the
- * hashkey expressions.
- */
- ResetExprContext(econtext);
-
- oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
-
- if (outer_tuple)
- hashfunctions = hashtable->outer_hashfunctions;
- else
- hashfunctions = hashtable->inner_hashfunctions;
-
- foreach(hk, hashkeys)
- {
- ExprState *keyexpr = (ExprState *) lfirst(hk);
- Datum keyval;
- bool isNull;
-
- /* rotate hashkey left 1 bit at each step */
- hashkey = (hashkey << 1) | ((hashkey & 0x80000000) ? 1 : 0);
-
- /*
- * Get the join attribute value of the tuple
- */
- keyval = ExecEvalExpr(keyexpr, econtext, &isNull);
-
- /*
- * If the attribute is NULL, and the join operator is strict, then
- * this tuple cannot pass the join qual so we can reject it
- * immediately (unless we're scanning the outside of an outer join, in
- * which case we must not reject it). Otherwise we act like the
- * hashcode of NULL is zero (this will support operators that act like
- * IS NOT DISTINCT, though not any more-random behavior). We treat
- * the hash support function as strict even if the operator is not.
- *
- * Note: currently, all hashjoinable operators must be strict since
- * the hash index AM assumes that. However, it takes so little extra
- * code here to allow non-strict that we may as well do it.
- */
- if (isNull)
- {
- if (hashtable->hashStrict[i] && !keep_nulls)
- {
- MemoryContextSwitchTo(oldContext);
- return false; /* cannot match */
- }
- /* else, leave hashkey unmodified, equivalent to hashcode 0 */
- }
- else
- {
- /* Compute the hash function */
- uint32 hkey;
-
- hkey = DatumGetUInt32(FunctionCall1Coll(&hashfunctions[i], hashtable->collations[i], keyval));
- hashkey ^= hkey;
- }
-
- i++;
- }
-
- MemoryContextSwitchTo(oldContext);
-
- *hashvalue = hashkey;
- return true;
-}
-
-/*
- * ExecHashGetBucketAndBatch
- * Determine the bucket number and batch number for a hash value
- *
- * Note: on-the-fly increases of nbatch must not change the bucket number
- * for a given hash code (since we don't move tuples to different hash
- * chains), and must only cause the batch number to remain the same or
- * increase. Our algorithm is
- * bucketno = hashvalue MOD nbuckets
- * batchno = (hashvalue DIV nbuckets) MOD nbatch
- * where nbuckets and nbatch are both expected to be powers of 2, so we can
- * do the computations by shifting and masking. (This assumes that all hash
- * functions are good about randomizing all their output bits, else we are
- * likely to have very skewed bucket or batch occupancy.)
- *
- * nbuckets and log2_nbuckets may change while nbatch == 1 because of dynamic
- * bucket count growth. Once we start batching, the value is fixed and does
- * not change over the course of the join (making it possible to compute batch
- * number the way we do here).
- *
- * nbatch is always a power of 2; we increase it only by doubling it. This
- * effectively adds one more bit to the top of the batchno.
- */
-void
-ExecHashGetBucketAndBatch(HashJoinTable hashtable,
- uint32 hashvalue,
- int *bucketno,
- int *batchno)
-{
- uint32 nbuckets = (uint32) hashtable->nbuckets;
- uint32 nbatch = (uint32) hashtable->nbatch;
-
- if (nbatch > 1)
- {
- /* we can do MOD by masking, DIV by shifting */
- *bucketno = hashvalue & (nbuckets - 1);
- *batchno = (hashvalue >> hashtable->log2_nbuckets) & (nbatch - 1);
- }
- else
- {
- *bucketno = hashvalue & (nbuckets - 1);
- *batchno = 0;
- }
-}
-
-/*
- * ExecScanHashBucket
- * scan a hash bucket for matches to the current outer tuple
- *
- * The current outer tuple must be stored in econtext->ecxt_outertuple.
- *
- * 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
-ExecScanHashBucket(HashJoinState *hjstate,
- ExprContext *econtext)
-{
- ExprState *hjclauses = hjstate->hashclauses;
- HashJoinTable hashtable = hjstate->hj_HashTable;
- HashJoinTuple hashTuple = hjstate->hj_CurTuple;
- uint32 hashvalue = hjstate->hj_CurHashValue;
-
- /*
- * 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 the tuple hashed to a skew bucket then scan the skew bucket
- * otherwise scan the standard hashtable bucket.
- */
- if (hashTuple != NULL)
- hashTuple = hashTuple->next.unshared;
- else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
- hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
- else
- hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
-
- while (hashTuple != NULL)
- {
- if (hashTuple->hashvalue == hashvalue)
- {
- TupleTableSlot *inntuple;
-
- /* insert hashtable's tuple into exec slot so ExecQual sees it */
- inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
- hjstate->hj_HashTupleSlot,
- false); /* do not pfree */
- econtext->ecxt_innertuple = inntuple;
-
- if (ExecQualAndReset(hjclauses, econtext))
- {
- hjstate->hj_CurTuple = hashTuple;
- return true;
- }
- }
-
- hashTuple = hashTuple->next.unshared;
- }
-
- /*
- * no match
- */
- return false;
-}
-
-/*
- * ExecParallelScanHashBucket
- * scan a hash bucket for matches to the current outer tuple
- *
- * The current outer tuple must be stored in econtext->ecxt_outertuple.
- *
- * 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
-ExecParallelScanHashBucket(HashJoinState *hjstate,
- ExprContext *econtext)
-{
- ExprState *hjclauses = hjstate->hashclauses;
- HashJoinTable hashtable = hjstate->hj_HashTable;
- HashJoinTuple hashTuple = hjstate->hj_CurTuple;
- uint32 hashvalue = hjstate->hj_CurHashValue;
-
- /*
- * 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 = ExecParallelHashNextTuple(hashtable, hashTuple);
- else
- hashTuple = ExecParallelHashFirstTuple(hashtable,
- hjstate->hj_CurBucketNo);
-
- while (hashTuple != NULL)
- {
- if (hashTuple->hashvalue == hashvalue)
- {
- TupleTableSlot *inntuple;
-
- /* insert hashtable's tuple into exec slot so ExecQual sees it */
- inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
- hjstate->hj_HashTupleSlot,
- false); /* do not pfree */
- econtext->ecxt_innertuple = inntuple;
-
- if (ExecQualAndReset(hjclauses, econtext))
- {
- hjstate->hj_CurTuple = hashTuple;
- return true;
- }
- }
-
- hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
- }
-
- /*
- * no match
- */
- return false;
-}
-
-/*
- * ExecPrepHashTableForUnmatched
- * set up for a series of ExecScanHashTableForUnmatched calls
- */
-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;
- hjstate->hj_CurTuple = NULL;
-}
-
-/*
- * ExecScanHashTableForUnmatched
- * scan the hash table for unmatched inner tuples
- *
- * 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
-ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
-{
- HashJoinTable hashtable = hjstate->hj_HashTable;
- HashJoinTuple hashTuple = hjstate->hj_CurTuple;
-
- for (;;)
- {
- /*
- * 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)
- {
- int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
-
- hashTuple = hashtable->skewBucket[j]->tuples;
- hjstate->hj_CurSkewBucketNo++;
- }
- else
- break; /* finished all buckets */
-
- while (hashTuple != NULL)
- {
- if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
- {
- TupleTableSlot *inntuple;
-
- /* insert hashtable's tuple into exec slot */
- inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
- hjstate->hj_HashTupleSlot,
- false); /* do not pfree */
- econtext->ecxt_innertuple = inntuple;
-
- /*
- * 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;
- }
-
- hashTuple = hashTuple->next.unshared;
- }
-
- /* allow this loop to be cancellable */
- CHECK_FOR_INTERRUPTS();
- }
-
- /*
- * no more unmatched tuples
- */
- return false;
-}
-
-/*
- * ExecHashTableReset
- *
- * reset hash table header for new batch
- */
-void
-ExecHashTableReset(HashJoinTable hashtable)
-{
- MemoryContext oldcxt;
- int nbuckets = hashtable->nbuckets;
-
- /*
- * Release all the hash buckets and tuples acquired in the prior pass, and
- * reinitialize the context for a new pass.
- */
- MemoryContextReset(hashtable->batchCxt);
- oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
-
- /* Reallocate and reinitialize the hash bucket headers. */
- hashtable->buckets.unshared = (HashJoinTuple *)
- palloc0(nbuckets * sizeof(HashJoinTuple));
-
- hashtable->spaceUsed = 0;
-
- MemoryContextSwitchTo(oldcxt);
-
- /* Forget the chunks (the memory was freed by the context reset above). */
- hashtable->chunks = NULL;
-}
-
-/*
- * ExecHashTableResetMatchFlags
- * Clear all the HeapTupleHeaderHasMatch flags in the table
- */
-void
-ExecHashTableResetMatchFlags(HashJoinTable hashtable)
-{
- HashJoinTuple tuple;
- int i;
-
- /* Reset all flags in the main table ... */
- for (i = 0; i < hashtable->nbuckets; i++)
- {
- for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
- tuple = tuple->next.unshared)
- HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
- }
-
- /* ... and the same for the skew buckets, if any */
- for (i = 0; i < hashtable->nSkewBuckets; i++)
- {
- int j = hashtable->skewBucketNums[i];
- HashSkewBucket *skewBucket = hashtable->skewBucket[j];
-
- for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
- HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
- }
-}
-
-
-void
-ExecReScanHash(HashState *node)
-{
- /*
- * if chgParam of subnode is not null then plan will be re-scanned by
- * first ExecProcNode.
- */
- if (node->ps.lefttree->chgParam == NULL)
- ExecReScan(node->ps.lefttree);
-}
-
-
-/*
- * ExecHashBuildSkewHash
- *
- * Set up for skew optimization if we can identify the most common values
- * (MCVs) of the outer relation's join key. We make a skew hash bucket
- * for the hash value of each MCV, up to the number of slots allowed
- * based on available memory.
- */
-static void
-ExecHashBuildSkewHash(HashJoinTable hashtable, Hash *node, int mcvsToUse)
-{
- HeapTupleData *statsTuple;
- AttStatsSlot sslot;
-
- /* Do nothing if planner didn't identify the outer relation's join key */
- if (!OidIsValid(node->skewTable))
- return;
- /* Also, do nothing if we don't have room for at least one skew bucket */
- if (mcvsToUse <= 0)
- return;
-
- /*
- * Try to find the MCV statistics for the outer relation's join key.
- */
- statsTuple = SearchSysCache3(STATRELATTINH,
- ObjectIdGetDatum(node->skewTable),
- Int16GetDatum(node->skewColumn),
- BoolGetDatum(node->skewInherit));
- if (!HeapTupleIsValid(statsTuple))
- return;
-
- if (get_attstatsslot(&sslot, statsTuple,
- STATISTIC_KIND_MCV, InvalidOid,
- ATTSTATSSLOT_VALUES | ATTSTATSSLOT_NUMBERS))
- {
- double frac;
- int nbuckets;
- FmgrInfo *hashfunctions;
- int i;
-
- if (mcvsToUse > sslot.nvalues)
- mcvsToUse = sslot.nvalues;
-
- /*
- * Calculate the expected fraction of outer relation that will
- * participate in the skew optimization. If this isn't at least
- * SKEW_MIN_OUTER_FRACTION, don't use skew optimization.
- */
- frac = 0;
- for (i = 0; i < mcvsToUse; i++)
- frac += sslot.numbers[i];
- if (frac < SKEW_MIN_OUTER_FRACTION)
- {
- free_attstatsslot(&sslot);
- ReleaseSysCache(statsTuple);
- return;
- }
-
- /*
- * Okay, set up the skew hashtable.
- *
- * skewBucket[] is an open addressing hashtable with a power of 2 size
- * that is greater than the number of MCV values. (This ensures there
- * will be at least one null entry, so searches will always
- * terminate.)
- *
- * Note: this code could fail if mcvsToUse exceeds INT_MAX/8 or
- * MaxAllocSize/sizeof(void *)/8, but that is not currently possible
- * since we limit pg_statistic entries to much less than that.
- */
- nbuckets = 2;
- while (nbuckets <= mcvsToUse)
- nbuckets <<= 1;
- /* use two more bits just to help avoid collisions */
- nbuckets <<= 2;
-
- hashtable->skewEnabled = true;
- hashtable->skewBucketLen = nbuckets;
-
- /*
- * We allocate the bucket memory in the hashtable's batch context. It
- * is only needed during the first batch, and this ensures it will be
- * automatically removed once the first batch is done.
- */
- hashtable->skewBucket = (HashSkewBucket **)
- MemoryContextAllocZero(hashtable->batchCxt,
- nbuckets * sizeof(HashSkewBucket *));
- hashtable->skewBucketNums = (int *)
- MemoryContextAllocZero(hashtable->batchCxt,
- mcvsToUse * sizeof(int));
-
- hashtable->spaceUsed += nbuckets * sizeof(HashSkewBucket *)
- + mcvsToUse * sizeof(int);
- hashtable->spaceUsedSkew += nbuckets * sizeof(HashSkewBucket *)
- + mcvsToUse * sizeof(int);
- if (hashtable->spaceUsed > hashtable->spacePeak)
- hashtable->spacePeak = hashtable->spaceUsed;
-
- /*
- * Create a skew bucket for each MCV hash value.
- *
- * Note: it is very important that we create the buckets in order of
- * decreasing MCV frequency. If we have to remove some buckets, they
- * must be removed in reverse order of creation (see notes in
- * ExecHashRemoveNextSkewBucket) and we want the least common MCVs to
- * be removed first.
- */
- hashfunctions = hashtable->outer_hashfunctions;
-
- for (i = 0; i < mcvsToUse; i++)
- {
- uint32 hashvalue;
- int bucket;
-
- hashvalue = DatumGetUInt32(FunctionCall1Coll(&hashfunctions[0],
- hashtable->collations[0],
- sslot.values[i]));
-
- /*
- * While we have not hit a hole in the hashtable and have not hit
- * the desired bucket, we have collided with some previous hash
- * value, so try the next bucket location. NB: this code must
- * match ExecHashGetSkewBucket.
- */
- bucket = hashvalue & (nbuckets - 1);
- while (hashtable->skewBucket[bucket] != NULL &&
- hashtable->skewBucket[bucket]->hashvalue != hashvalue)
- bucket = (bucket + 1) & (nbuckets - 1);
-
- /*
- * If we found an existing bucket with the same hashvalue, leave
- * it alone. It's okay for two MCVs to share a hashvalue.
- */
- if (hashtable->skewBucket[bucket] != NULL)
- continue;
-
- /* Okay, create a new skew bucket for this hashvalue. */
- hashtable->skewBucket[bucket] = (HashSkewBucket *)
- MemoryContextAlloc(hashtable->batchCxt,
- sizeof(HashSkewBucket));
- hashtable->skewBucket[bucket]->hashvalue = hashvalue;
- hashtable->skewBucket[bucket]->tuples = NULL;
- hashtable->skewBucketNums[hashtable->nSkewBuckets] = bucket;
- hashtable->nSkewBuckets++;
- hashtable->spaceUsed += SKEW_BUCKET_OVERHEAD;
- hashtable->spaceUsedSkew += SKEW_BUCKET_OVERHEAD;
- if (hashtable->spaceUsed > hashtable->spacePeak)
- hashtable->spacePeak = hashtable->spaceUsed;
- }
-
- free_attstatsslot(&sslot);
- }
-
- ReleaseSysCache(statsTuple);
-}
-
-/*
- * ExecHashGetSkewBucket
- *
- * Returns the index of the skew bucket for this hashvalue,
- * or INVALID_SKEW_BUCKET_NO if the hashvalue is not
- * associated with any active skew bucket.
- */
-int
-ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue)
-{
- int bucket;
-
- /*
- * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
- * particular, this happens after the initial batch is done).
- */
- if (!hashtable->skewEnabled)
- return INVALID_SKEW_BUCKET_NO;
-
- /*
- * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
- */
- bucket = hashvalue & (hashtable->skewBucketLen - 1);
-
- /*
- * While we have not hit a hole in the hashtable and have not hit the
- * desired bucket, we have collided with some other hash value, so try the
- * next bucket location.
- */
- while (hashtable->skewBucket[bucket] != NULL &&
- hashtable->skewBucket[bucket]->hashvalue != hashvalue)
- bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
-
- /*
- * Found the desired bucket?
- */
- if (hashtable->skewBucket[bucket] != NULL)
- return bucket;
-
- /*
- * There must not be any hashtable entry for this hash value.
- */
- return INVALID_SKEW_BUCKET_NO;
-}
-
-/*
- * ExecHashSkewTableInsert
- *
- * Insert a tuple into the skew hashtable.
- *
- * This should generally match up with the current-batch case in
- * ExecHashTableInsert.
- */
-static void
-ExecHashSkewTableInsert(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue,
- int bucketNumber)
-{
- bool shouldFree;
- MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
- HashJoinTuple hashTuple;
- int hashTupleSize;
-
- /* Create the HashJoinTuple */
- hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
- hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
- hashTupleSize);
- hashTuple->hashvalue = hashvalue;
- memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
- HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
-
- /* Push it onto the front of the skew bucket's list */
- hashTuple->next.unshared = hashtable->skewBucket[bucketNumber]->tuples;
- hashtable->skewBucket[bucketNumber]->tuples = hashTuple;
- Assert(hashTuple != hashTuple->next.unshared);
-
- /* Account for space used, and back off if we've used too much */
- hashtable->spaceUsed += hashTupleSize;
- hashtable->spaceUsedSkew += hashTupleSize;
- if (hashtable->spaceUsed > hashtable->spacePeak)
- hashtable->spacePeak = hashtable->spaceUsed;
- while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
- ExecHashRemoveNextSkewBucket(hashtable);
-
- /* Check we are not over the total spaceAllowed, either */
- if (hashtable->spaceUsed > hashtable->spaceAllowed)
- ExecHashIncreaseNumBatches(hashtable);
-
- if (shouldFree)
- heap_free_minimal_tuple(tuple);
-}
-
-/*
- * ExecHashRemoveNextSkewBucket
- *
- * Remove the least valuable skew bucket by pushing its tuples into
- * the main hash table.
- */
-static void
-ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
-{
- int bucketToRemove;
- HashSkewBucket *bucket;
- uint32 hashvalue;
- int bucketno;
- int batchno;
- HashJoinTuple hashTuple;
-
- /* Locate the bucket to remove */
- bucketToRemove = hashtable->skewBucketNums[hashtable->nSkewBuckets - 1];
- bucket = hashtable->skewBucket[bucketToRemove];
-
- /*
- * Calculate which bucket and batch the tuples belong to in the main
- * hashtable. They all have the same hash value, so it's the same for all
- * of them. Also note that it's not possible for nbatch to increase while
- * we are processing the tuples.
- */
- hashvalue = bucket->hashvalue;
- ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
-
- /* Process all tuples in the bucket */
- hashTuple = bucket->tuples;
- while (hashTuple != NULL)
- {
- HashJoinTuple nextHashTuple = hashTuple->next.unshared;
- MinimalTuple tuple;
- Size tupleSize;
-
- /*
- * This code must agree with ExecHashTableInsert. We do not use
- * ExecHashTableInsert directly as ExecHashTableInsert expects a
- * TupleTableSlot while we already have HashJoinTuples.
- */
- tuple = HJTUPLE_MINTUPLE(hashTuple);
- tupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
-
- /* Decide whether to put the tuple in the hash table or a temp file */
- if (batchno == hashtable->curbatch)
- {
- /* Move the tuple to the main hash table */
- HashJoinTuple copyTuple;
-
- /*
- * We must copy the tuple into the dense storage, else it will not
- * be found by, eg, ExecHashIncreaseNumBatches.
- */
- copyTuple = (HashJoinTuple) dense_alloc(hashtable, tupleSize);
- memcpy(copyTuple, hashTuple, tupleSize);
- pfree(hashTuple);
-
- copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
- hashtable->buckets.unshared[bucketno] = copyTuple;
-
- /* We have reduced skew space, but overall space doesn't change */
- hashtable->spaceUsedSkew -= tupleSize;
- }
- else
- {
- /* Put the tuple into a temp file for later batches */
- Assert(batchno > hashtable->curbatch);
- ExecHashJoinSaveTuple(tuple, hashvalue,
- &hashtable->innerBatchFile[batchno]);
- pfree(hashTuple);
- hashtable->spaceUsed -= tupleSize;
- hashtable->spaceUsedSkew -= tupleSize;
- }
-
- hashTuple = nextHashTuple;
-
- /* allow this loop to be cancellable */
- CHECK_FOR_INTERRUPTS();
- }
-
- /*
- * Free the bucket struct itself and reset the hashtable entry to NULL.
- *
- * NOTE: this is not nearly as simple as it looks on the surface, because
- * of the possibility of collisions in the hashtable. Suppose that hash
- * values A and B collide at a particular hashtable entry, and that A was
- * entered first so B gets shifted to a different table entry. If we were
- * to remove A first then ExecHashGetSkewBucket would mistakenly start
- * reporting that B is not in the hashtable, because it would hit the NULL
- * before finding B. However, we always remove entries in the reverse
- * order of creation, so this failure cannot happen.
- */
- hashtable->skewBucket[bucketToRemove] = NULL;
- hashtable->nSkewBuckets--;
- pfree(bucket);
- hashtable->spaceUsed -= SKEW_BUCKET_OVERHEAD;
- hashtable->spaceUsedSkew -= SKEW_BUCKET_OVERHEAD;
-
- /*
- * If we have removed all skew buckets then give up on skew optimization.
- * Release the arrays since they aren't useful any more.
- */
- if (hashtable->nSkewBuckets == 0)
- {
- hashtable->skewEnabled = false;
- pfree(hashtable->skewBucket);
- pfree(hashtable->skewBucketNums);
- hashtable->skewBucket = NULL;
- hashtable->skewBucketNums = NULL;
- hashtable->spaceUsed -= hashtable->spaceUsedSkew;
- hashtable->spaceUsedSkew = 0;
- }
-}
-
-/*
- * Reserve space in the DSM segment for instrumentation data.
- */
-void
-ExecHashEstimate(HashState *node, ParallelContext *pcxt)
-{
- size_t size;
-
- /* don't need this if not instrumenting or no workers */
- if (!node->ps.instrument || pcxt->nworkers == 0)
- return;
-
- size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
- size = add_size(size, offsetof(SharedHashInfo, hinstrument));
- shm_toc_estimate_chunk(&pcxt->estimator, size);
- shm_toc_estimate_keys(&pcxt->estimator, 1);
-}
-
-/*
- * Set up a space in the DSM for all workers to record instrumentation data
- * about their hash table.
- */
-void
-ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt)
-{
- size_t size;
-
- /* don't need this if not instrumenting or no workers */
- if (!node->ps.instrument || pcxt->nworkers == 0)
- return;
-
- size = offsetof(SharedHashInfo, hinstrument) +
- pcxt->nworkers * sizeof(HashInstrumentation);
- node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
- memset(node->shared_info, 0, size);
- node->shared_info->num_workers = pcxt->nworkers;
- shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id,
- node->shared_info);
-}
-
-/*
- * Locate the DSM space for hash table instrumentation data that we'll write
- * to at shutdown time.
- */
-void
-ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt)
-{
- SharedHashInfo *shared_info;
-
- /* don't need this if not instrumenting */
- if (!node->ps.instrument)
- return;
-
- shared_info = (SharedHashInfo *)
- shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false);
- node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
-}
-
-/*
- * Copy instrumentation data from this worker's hash table (if it built one)
- * to DSM memory so the leader can retrieve it. This must be done in an
- * ExecShutdownHash() rather than ExecEndHash() because the latter runs after
- * we've detached from the DSM segment.
- */
-void
-ExecShutdownHash(HashState *node)
-{
- if (node->hinstrument && node->hashtable)
- ExecHashGetInstrumentation(node->hinstrument, node->hashtable);
-}
-
-/*
- * Retrieve instrumentation data from workers before the DSM segment is
- * detached, so that EXPLAIN can access it.
- */
-void
-ExecHashRetrieveInstrumentation(HashState *node)
-{
- SharedHashInfo *shared_info = node->shared_info;
- size_t size;
-
- if (shared_info == NULL)
- return;
-
- /* Replace node->shared_info with a copy in backend-local memory. */
- size = offsetof(SharedHashInfo, hinstrument) +
- shared_info->num_workers * sizeof(HashInstrumentation);
- node->shared_info = palloc(size);
- memcpy(node->shared_info, shared_info, size);
-}
-
-/*
- * Copy the instrumentation data from 'hashtable' into a HashInstrumentation
- * struct.
- */
-void
-ExecHashGetInstrumentation(HashInstrumentation *instrument,
- HashJoinTable hashtable)
-{
- instrument->nbuckets = hashtable->nbuckets;
- instrument->nbuckets_original = hashtable->nbuckets_original;
- instrument->nbatch = hashtable->nbatch;
- instrument->nbatch_original = hashtable->nbatch_original;
- instrument->space_peak = hashtable->spacePeak;
-}
-
-/*
- * Allocate 'size' bytes from the currently active HashMemoryChunk
- */
-static void *
-dense_alloc(HashJoinTable hashtable, Size size)
-{
- HashMemoryChunk newChunk;
- char *ptr;
-
- /* just in case the size is not already aligned properly */
- size = MAXALIGN(size);
-
- /*
- * If tuple size is larger than threshold, allocate a separate chunk.
- */
- if (size > HASH_CHUNK_THRESHOLD)
- {
- /* allocate new chunk and put it at the beginning of the list */
- newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
- HASH_CHUNK_HEADER_SIZE + size);
- newChunk->maxlen = size;
- newChunk->used = size;
- newChunk->ntuples = 1;
-
- /*
- * Add this chunk to the list after the first existing chunk, so that
- * we don't lose the remaining space in the "current" chunk.
- */
- if (hashtable->chunks != NULL)
- {
- newChunk->next = hashtable->chunks->next;
- hashtable->chunks->next.unshared = newChunk;
- }
- else
- {
- newChunk->next.unshared = hashtable->chunks;
- hashtable->chunks = newChunk;
- }
-
- return HASH_CHUNK_DATA(newChunk);
- }
-
- /*
- * See if we have enough space for it in the current chunk (if any). If
- * not, allocate a fresh chunk.
- */
- if ((hashtable->chunks == NULL) ||
- (hashtable->chunks->maxlen - hashtable->chunks->used) < size)
- {
- /* allocate new chunk and put it at the beginning of the list */
- newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
- HASH_CHUNK_HEADER_SIZE + HASH_CHUNK_SIZE);
-
- newChunk->maxlen = HASH_CHUNK_SIZE;
- newChunk->used = size;
- newChunk->ntuples = 1;
-
- newChunk->next.unshared = hashtable->chunks;
- hashtable->chunks = newChunk;
-
- return HASH_CHUNK_DATA(newChunk);
- }
-
- /* There is enough space in the current chunk, let's add the tuple */
- ptr = HASH_CHUNK_DATA(hashtable->chunks) + hashtable->chunks->used;
- hashtable->chunks->used += size;
- hashtable->chunks->ntuples += 1;
-
- /* return pointer to the start of the tuple memory */
- return ptr;
-}
-
-/*
- * Allocate space for a tuple in shared dense storage. This is equivalent to
- * dense_alloc but for Parallel Hash using shared memory.
- *
- * While loading a tuple into shared memory, we might run out of memory and
- * decide to repartition, or determine that the load factor is too high and
- * decide to expand the bucket array, or discover that another participant has
- * commanded us to help do that. Return NULL if number of buckets or batches
- * has changed, indicating that the caller must retry (considering the
- * possibility that the tuple no longer belongs in the same batch).
- */
-static HashJoinTuple
-ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size,
- dsa_pointer *shared)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- dsa_pointer chunk_shared;
- HashMemoryChunk chunk;
- Size chunk_size;
- HashJoinTuple result;
- int curbatch = hashtable->curbatch;
-
- size = MAXALIGN(size);
-
- /*
- * Fast path: if there is enough space in this backend's current chunk,
- * then we can allocate without any locking.
- */
- chunk = hashtable->current_chunk;
- if (chunk != NULL &&
- size <= HASH_CHUNK_THRESHOLD &&
- chunk->maxlen - chunk->used >= size)
- {
-
- chunk_shared = hashtable->current_chunk_shared;
- Assert(chunk == dsa_get_address(hashtable->area, chunk_shared));
- *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE + chunk->used;
- result = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + chunk->used);
- chunk->used += size;
-
- Assert(chunk->used <= chunk->maxlen);
- Assert(result == dsa_get_address(hashtable->area, *shared));
-
- return result;
- }
-
- /* Slow path: try to allocate a new chunk. */
- LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
-
- /*
- * Check if we need to help increase the number of buckets or batches.
- */
- if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
- pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
- {
- ParallelHashGrowth growth = pstate->growth;
-
- hashtable->current_chunk = NULL;
- LWLockRelease(&pstate->lock);
-
- /* Another participant has commanded us to help grow. */
- if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
- ExecParallelHashIncreaseNumBatches(hashtable);
- else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
- ExecParallelHashIncreaseNumBuckets(hashtable);
-
- /* The caller must retry. */
- return NULL;
- }
-
- /* Oversized tuples get their own chunk. */
- if (size > HASH_CHUNK_THRESHOLD)
- chunk_size = size + HASH_CHUNK_HEADER_SIZE;
- else
- chunk_size = HASH_CHUNK_SIZE;
-
- /* Check if it's time to grow batches or buckets. */
- if (pstate->growth != PHJ_GROWTH_DISABLED)
- {
- Assert(curbatch == 0);
- Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
-
- /*
- * Check if our space limit would be exceeded. To avoid choking on
- * very large tuples or very low work_mem setting, we'll always allow
- * each backend to allocate at least one chunk.
- */
- if (hashtable->batches[0].at_least_one_chunk &&
- hashtable->batches[0].shared->size +
- chunk_size > pstate->space_allowed)
- {
- pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
- hashtable->batches[0].shared->space_exhausted = true;
- LWLockRelease(&pstate->lock);
-
- return NULL;
- }
-
- /* Check if our load factor limit would be exceeded. */
- if (hashtable->nbatch == 1)
- {
- hashtable->batches[0].shared->ntuples += hashtable->batches[0].ntuples;
- hashtable->batches[0].ntuples = 0;
- /* Guard against integer overflow and alloc size overflow */
- if (hashtable->batches[0].shared->ntuples + 1 >
- hashtable->nbuckets * NTUP_PER_BUCKET &&
- hashtable->nbuckets < (INT_MAX / 2) &&
- hashtable->nbuckets * 2 <=
- MaxAllocSize / sizeof(dsa_pointer_atomic))
- {
- pstate->growth = PHJ_GROWTH_NEED_MORE_BUCKETS;
- LWLockRelease(&pstate->lock);
-
- return NULL;
- }
- }
- }
-
- /* We are cleared to allocate a new chunk. */
- chunk_shared = dsa_allocate(hashtable->area, chunk_size);
- hashtable->batches[curbatch].shared->size += chunk_size;
- hashtable->batches[curbatch].at_least_one_chunk = true;
-
- /* Set up the chunk. */
- chunk = (HashMemoryChunk) dsa_get_address(hashtable->area, chunk_shared);
- *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE;
- chunk->maxlen = chunk_size - HASH_CHUNK_HEADER_SIZE;
- chunk->used = size;
-
- /*
- * Push it onto the list of chunks, so that it can be found if we need to
- * increase the number of buckets or batches (batch 0 only) and later for
- * freeing the memory (all batches).
- */
- chunk->next.shared = hashtable->batches[curbatch].shared->chunks;
- hashtable->batches[curbatch].shared->chunks = chunk_shared;
-
- if (size <= HASH_CHUNK_THRESHOLD)
- {
- /*
- * Make this the current chunk so that we can use the fast path to
- * fill the rest of it up in future calls.
- */
- hashtable->current_chunk = chunk;
- hashtable->current_chunk_shared = chunk_shared;
- }
- LWLockRelease(&pstate->lock);
-
- Assert(HASH_CHUNK_DATA(chunk) == dsa_get_address(hashtable->area, *shared));
- result = (HashJoinTuple) HASH_CHUNK_DATA(chunk);
-
- return result;
-}
-
-/*
- * One backend needs to set up the shared batch state including tuplestores.
- * Other backends will ensure they have correctly configured accessors by
- * called ExecParallelHashEnsureBatchAccessors().
- */
-static void
-ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- ParallelHashJoinBatch *batches;
- MemoryContext oldcxt;
- int i;
-
- Assert(hashtable->batches == NULL);
-
- /* Allocate space. */
- pstate->batches =
- dsa_allocate0(hashtable->area,
- EstimateParallelHashJoinBatch(hashtable) * nbatch);
- pstate->nbatch = nbatch;
- batches = dsa_get_address(hashtable->area, pstate->batches);
-
- /* Use hash join memory context. */
- oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
-
- /* Allocate this backend's accessor array. */
- hashtable->nbatch = nbatch;
- hashtable->batches = (ParallelHashJoinBatchAccessor *)
- palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
-
- /* Set up the shared state, tuplestores and backend-local accessors. */
- for (i = 0; i < hashtable->nbatch; ++i)
- {
- ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
- ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
- char name[MAXPGPATH];
-
- /*
- * All members of shared were zero-initialized. We just need to set
- * up the Barrier.
- */
- BarrierInit(&shared->batch_barrier, 0);
- if (i == 0)
- {
- /* Batch 0 doesn't need to be loaded. */
- BarrierAttach(&shared->batch_barrier);
- while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING)
- BarrierArriveAndWait(&shared->batch_barrier, 0);
- BarrierDetach(&shared->batch_barrier);
- }
-
- /* Initialize accessor state. All members were zero-initialized. */
- accessor->shared = shared;
-
- /* Initialize the shared tuplestores. */
- snprintf(name, sizeof(name), "i%dof%d", i, hashtable->nbatch);
- accessor->inner_tuples =
- sts_initialize(ParallelHashJoinBatchInner(shared),
- pstate->nparticipants,
- ParallelWorkerNumber + 1,
- sizeof(uint32),
- SHARED_TUPLESTORE_SINGLE_PASS,
- &pstate->fileset,
- name);
- snprintf(name, sizeof(name), "o%dof%d", i, hashtable->nbatch);
- accessor->outer_tuples =
- sts_initialize(ParallelHashJoinBatchOuter(shared,
- pstate->nparticipants),
- pstate->nparticipants,
- ParallelWorkerNumber + 1,
- sizeof(uint32),
- SHARED_TUPLESTORE_SINGLE_PASS,
- &pstate->fileset,
- name);
- }
-
- MemoryContextSwitchTo(oldcxt);
-}
-
-/*
- * Free the current set of ParallelHashJoinBatchAccessor objects.
- */
-static void
-ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
-{
- int i;
-
- for (i = 0; i < hashtable->nbatch; ++i)
- {
- /* Make sure no files are left open. */
- sts_end_write(hashtable->batches[i].inner_tuples);
- sts_end_write(hashtable->batches[i].outer_tuples);
- sts_end_parallel_scan(hashtable->batches[i].inner_tuples);
- sts_end_parallel_scan(hashtable->batches[i].outer_tuples);
- }
- pfree(hashtable->batches);
- hashtable->batches = NULL;
-}
-
-/*
- * Make sure this backend has up-to-date accessors for the current set of
- * batches.
- */
-static void
-ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- ParallelHashJoinBatch *batches;
- MemoryContext oldcxt;
- int i;
-
- if (hashtable->batches != NULL)
- {
- if (hashtable->nbatch == pstate->nbatch)
- return;
- ExecParallelHashCloseBatchAccessors(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.
- */
- if (!DsaPointerIsValid(pstate->batches))
- return;
-
- /* Use hash join memory context. */
- oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
-
- /* Allocate this backend's accessor array. */
- hashtable->nbatch = pstate->nbatch;
- hashtable->batches = (ParallelHashJoinBatchAccessor *)
- palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
-
- /* Find the base of the pseudo-array of ParallelHashJoinBatch objects. */
- batches = (ParallelHashJoinBatch *)
- dsa_get_address(hashtable->area, pstate->batches);
-
- /* Set up the accessor array and attach to the tuplestores. */
- for (i = 0; i < hashtable->nbatch; ++i)
- {
- ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
- ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
-
- accessor->shared = shared;
- accessor->preallocated = 0;
- accessor->done = false;
- accessor->inner_tuples =
- sts_attach(ParallelHashJoinBatchInner(shared),
- ParallelWorkerNumber + 1,
- &pstate->fileset);
- accessor->outer_tuples =
- sts_attach(ParallelHashJoinBatchOuter(shared,
- pstate->nparticipants),
- ParallelWorkerNumber + 1,
- &pstate->fileset);
- }
-
- MemoryContextSwitchTo(oldcxt);
-}
-
-/*
- * Allocate an empty shared memory hash table for a given batch.
- */
-void
-ExecParallelHashTableAlloc(HashJoinTable hashtable, int batchno)
-{
- ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
- dsa_pointer_atomic *buckets;
- int nbuckets = hashtable->parallel_state->nbuckets;
- int i;
-
- batch->buckets =
- dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
- buckets = (dsa_pointer_atomic *)
- dsa_get_address(hashtable->area, batch->buckets);
- for (i = 0; i < nbuckets; ++i)
- dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
-}
-
-/*
- * If we are currently attached to a shared hash join batch, detach. If we
- * are last to detach, clean up.
- */
-void
-ExecHashTableDetachBatch(HashJoinTable hashtable)
-{
- if (hashtable->parallel_state != NULL &&
- hashtable->curbatch >= 0)
- {
- int curbatch = hashtable->curbatch;
- ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
-
- /* 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))
- {
- /*
- * 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_DONE);
-
- /* Free shared chunks and buckets. */
- while (DsaPointerIsValid(batch->chunks))
- {
- HashMemoryChunk chunk =
- dsa_get_address(hashtable->area, batch->chunks);
- dsa_pointer next = chunk->next.shared;
-
- dsa_free(hashtable->area, batch->chunks);
- batch->chunks = next;
- }
- if (DsaPointerIsValid(batch->buckets))
- {
- dsa_free(hashtable->area, batch->buckets);
- batch->buckets = InvalidDsaPointer;
- }
- }
-
- /*
- * 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);
-
- /* Remember that we are not attached to a batch. */
- hashtable->curbatch = -1;
- }
-}
-
-/*
- * Detach from all shared resources. If we are last to detach, clean up.
- */
-void
-ExecHashTableDetach(HashJoinTable hashtable)
-{
- if (hashtable->parallel_state)
- {
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- int i;
-
- /* Make sure any temporary files are closed. */
- if (hashtable->batches)
- {
- for (i = 0; i < hashtable->nbatch; ++i)
- {
- sts_end_write(hashtable->batches[i].inner_tuples);
- sts_end_write(hashtable->batches[i].outer_tuples);
- sts_end_parallel_scan(hashtable->batches[i].inner_tuples);
- sts_end_parallel_scan(hashtable->batches[i].outer_tuples);
- }
- }
-
- /* If we're last to detach, clean up shared memory. */
- if (BarrierDetach(&pstate->build_barrier))
- {
- if (DsaPointerIsValid(pstate->batches))
- {
- dsa_free(hashtable->area, pstate->batches);
- pstate->batches = InvalidDsaPointer;
- }
- }
-
- hashtable->parallel_state = NULL;
- }
-}
-
-/*
- * Get the first tuple in a given bucket identified by number.
- */
-static inline HashJoinTuple
-ExecParallelHashFirstTuple(HashJoinTable hashtable, int bucketno)
-{
- HashJoinTuple tuple;
- dsa_pointer p;
-
- Assert(hashtable->parallel_state);
- p = dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
- tuple = (HashJoinTuple) dsa_get_address(hashtable->area, p);
-
- return tuple;
-}
-
-/*
- * Get the next tuple in the same bucket as 'tuple'.
- */
-static inline HashJoinTuple
-ExecParallelHashNextTuple(HashJoinTable hashtable, HashJoinTuple tuple)
-{
- HashJoinTuple next;
-
- Assert(hashtable->parallel_state);
- next = (HashJoinTuple) dsa_get_address(hashtable->area, tuple->next.shared);
-
- return next;
-}
-
-/*
- * Insert a tuple at the front of a chain of tuples in DSA memory atomically.
- */
-static inline void
-ExecParallelHashPushTuple(dsa_pointer_atomic *head,
- HashJoinTuple tuple,
- dsa_pointer tuple_shared)
-{
- for (;;)
- {
- tuple->next.shared = dsa_pointer_atomic_read(head);
- if (dsa_pointer_atomic_compare_exchange(head,
- &tuple->next.shared,
- tuple_shared))
- break;
- }
-}
-
-/*
- * Prepare to work on a given batch.
- */
-void
-ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno)
-{
- Assert(hashtable->batches[batchno].shared->buckets != InvalidDsaPointer);
-
- hashtable->curbatch = batchno;
- hashtable->buckets.shared = (dsa_pointer_atomic *)
- dsa_get_address(hashtable->area,
- hashtable->batches[batchno].shared->buckets);
- hashtable->nbuckets = hashtable->parallel_state->nbuckets;
- hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
- hashtable->current_chunk = NULL;
- hashtable->current_chunk_shared = InvalidDsaPointer;
- hashtable->batches[batchno].at_least_one_chunk = false;
-}
-
-/*
- * Take the next available chunk from the queue of chunks being worked on in
- * parallel. Return NULL if there are none left. Otherwise return a pointer
- * to the chunk, and set *shared to the DSA pointer to the chunk.
- */
-static HashMemoryChunk
-ExecParallelHashPopChunkQueue(HashJoinTable hashtable, dsa_pointer *shared)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- HashMemoryChunk chunk;
-
- LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
- if (DsaPointerIsValid(pstate->chunk_work_queue))
- {
- *shared = pstate->chunk_work_queue;
- chunk = (HashMemoryChunk)
- dsa_get_address(hashtable->area, *shared);
- pstate->chunk_work_queue = chunk->next.shared;
- }
- else
- chunk = NULL;
- LWLockRelease(&pstate->lock);
-
- return chunk;
-}
-
-/*
- * Increase the space preallocated in this backend for a given inner batch by
- * at least a given amount. This allows us to track whether a given batch
- * would fit in memory when loaded back in. Also increase the number of
- * batches or buckets if required.
- *
- * This maintains a running estimation of how much space will be taken when we
- * load the batch back into memory by simulating the way chunks will be handed
- * out to workers. It's not perfectly accurate because the tuples will be
- * packed into memory chunks differently by ExecParallelHashTupleAlloc(), but
- * it should be pretty close. It tends to overestimate by a fraction of a
- * chunk per worker since all workers gang up to preallocate during hashing,
- * but workers tend to reload batches alone if there are enough to go around,
- * leaving fewer partially filled chunks. This effect is bounded by
- * nparticipants.
- *
- * Return false if the number of batches or buckets has changed, and the
- * caller should reconsider which batch a given tuple now belongs in and call
- * again.
- */
-static bool
-ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size)
-{
- ParallelHashJoinState *pstate = hashtable->parallel_state;
- ParallelHashJoinBatchAccessor *batch = &hashtable->batches[batchno];
- size_t want = Max(size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);
-
- Assert(batchno > 0);
- Assert(batchno < hashtable->nbatch);
- Assert(size == MAXALIGN(size));
-
- LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
-
- /* Has another participant commanded us to help grow? */
- if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
- pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
- {
- ParallelHashGrowth growth = pstate->growth;
-
- LWLockRelease(&pstate->lock);
- if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
- ExecParallelHashIncreaseNumBatches(hashtable);
- else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
- ExecParallelHashIncreaseNumBuckets(hashtable);
-
- return false;
- }
-
- if (pstate->growth != PHJ_GROWTH_DISABLED &&
- batch->at_least_one_chunk &&
- (batch->shared->estimated_size + want + HASH_CHUNK_HEADER_SIZE
- > pstate->space_allowed))
- {
- /*
- * We have determined that this batch would exceed the space budget if
- * loaded into memory. Command all participants to help repartition.
- */
- batch->shared->space_exhausted = true;
- pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
- LWLockRelease(&pstate->lock);
-
- return false;
- }
-
- batch->at_least_one_chunk = true;
- batch->shared->estimated_size += want + HASH_CHUNK_HEADER_SIZE;
- batch->preallocated = want;
- LWLockRelease(&pstate->lock);
-
- return true;
-}
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
index ec37558c127..5b566ba85ab 100644
--- a/src/backend/executor/nodeHashjoin.c
+++ b/src/backend/executor/nodeHashjoin.c
@@ -106,16 +106,25 @@
#include "postgres.h"
+#include <math.h>
+#include <limits.h>
+
#include "access/htup_details.h"
#include "access/parallel.h"
+#include "catalog/pg_statistic.h"
+#include "commands/tablespace.h"
+#include "executor/execdebug.h"
#include "executor/executor.h"
#include "executor/hashjoin.h"
-#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "port/atomics.h"
+#include "utils/dynahash.h"
#include "utils/memutils.h"
+#include "utils/lsyscache.h"
#include "utils/sharedtuplestore.h"
+#include "utils/syscache.h"
/*
@@ -143,10 +152,87 @@ static TupleTableSlot *ExecHashJoinGetSavedTuple(HashJoinState *hjstate,
BufFile *file,
uint32 *hashvalue,
TupleTableSlot *tupleSlot);
+static void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
+ BufFile **fileptr);
static bool ExecHashJoinNewBatch(HashJoinState *hjstate);
static bool ExecParallelHashJoinNewBatch(HashJoinState *hjstate);
static void ExecParallelHashJoinPartitionOuter(HashJoinState *node);
+static void ExecHashIncreaseNumBatches(HashJoinTable hashtable);
+static void ExecHashIncreaseNumBuckets(HashJoinTable hashtable);
+static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable);
+static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable);
+static void ExecHashBuildSkewHash(HashJoinTable hashtable, HashJoin *node,
+ int mcvsToUse);
+static void ExecHashSkewTableInsert(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue,
+ int bucketNumber);
+static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable);
+
+static void *dense_alloc(HashJoinTable hashtable, Size size);
+static HashJoinTuple ExecParallelHashTupleAlloc(HashJoinTable hashtable,
+ size_t size,
+ dsa_pointer *shared);
+static void ExecHashBuildPrivate(HashJoinState *node);
+static void ExecHashBuildParallel(HashJoinState *node);
+static inline HashJoinTuple ExecParallelHashFirstTuple(HashJoinTable table,
+ int bucketno);
+static inline HashJoinTuple ExecParallelHashNextTuple(HashJoinTable table,
+ HashJoinTuple tuple);
+static inline void ExecParallelHashPushTuple(dsa_pointer_atomic *head,
+ HashJoinTuple tuple,
+ dsa_pointer tuple_shared);
+static void ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch);
+static void ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable);
+static void ExecParallelHashRepartitionFirst(HashJoinTable hashtable);
+static void ExecParallelHashRepartitionRest(HashJoinTable hashtable);
+static HashMemoryChunk ExecParallelHashPopChunkQueue(HashJoinTable table,
+ dsa_pointer *shared);
+static bool ExecParallelHashTuplePrealloc(HashJoinTable hashtable,
+ int batchno,
+ size_t size);
+static void ExecParallelHashMergeCounters(HashJoinTable hashtable);
+static void ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable);
+
+static HashJoinTable ExecHashTableCreate(HashJoinState *state, List *hashOperators, List *hashCollations,
+ bool keepNulls);
+static void ExecParallelHashTableAlloc(HashJoinTable hashtable,
+ int batchno);
+static void ExecHashTableDestroy(HashJoinTable hashtable);
+static void ExecHashTableDetach(HashJoinTable hashtable);
+static void ExecHashTableDetachBatch(HashJoinTable hashtable);
+static void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable,
+ int batchno);
+
+static void ExecHashTableInsert(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue);
+static void ExecParallelHashTableInsert(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue);
+static void ExecParallelHashTableInsertCurrentBatch(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue);
+static bool ExecHashGetHashValue(HashJoinTable hashtable,
+ ExprContext *econtext,
+ List *hashkeys,
+ bool outer_tuple,
+ bool keep_nulls,
+ uint32 *hashvalue);
+static void ExecHashGetBucketAndBatch(HashJoinTable hashtable,
+ uint32 hashvalue,
+ int *bucketno,
+ int *batchno);
+static bool ExecScanHashBucket(HashJoinState *hjstate, ExprContext *econtext);
+static bool ExecParallelScanHashBucket(HashJoinState *hjstate, ExprContext *econtext);
+static void ExecPrepHashTableForUnmatched(HashJoinState *hjstate);
+static bool ExecScanHashTableForUnmatched(HashJoinState *hjstate,
+ ExprContext *econtext);
+static void ExecHashTableReset(HashJoinTable hashtable);
+static void ExecHashTableResetMatchFlags(HashJoinTable hashtable);
+static int ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue);
+
/* ----------------------------------------------------------------
* ExecHashJoinImpl
@@ -166,7 +252,6 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
{
HashJoinState *node = castNode(HashJoinState, pstate);
PlanState *outerNode;
- HashState *hashNode;
ExprState *joinqual;
ExprState *otherqual;
ExprContext *econtext;
@@ -181,11 +266,10 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
*/
joinqual = node->js.joinqual;
otherqual = node->js.ps.qual;
- hashNode = (HashState *) innerPlanState(node);
outerNode = outerPlanState(node);
hashtable = node->hj_HashTable;
econtext = node->js.ps.ps_ExprContext;
- parallel_state = hashNode->parallel_state;
+ parallel_state = node->parallel_state;
/*
* Reset per-tuple memory context to free any expression evaluation
@@ -256,7 +340,8 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
node->hj_FirstOuterTupleSlot = NULL;
}
else if (HJ_FILL_OUTER(node) ||
- (outerNode->plan->startup_cost < hashNode->ps.plan->total_cost &&
+ // FIXME: this isn't correct, needs to include the cost of building the hashtable?
+ (outerNode->plan->startup_cost < innerPlanState(node)->plan->total_cost &&
!node->hj_OuterNotEmpty))
{
node->hj_FirstOuterTupleSlot = ExecProcNode(outerNode);
@@ -276,7 +361,7 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
* whoever gets here first will create the hash table and any
* later arrivals will merely attach to it.
*/
- hashtable = ExecHashTableCreate(hashNode,
+ hashtable = ExecHashTableCreate(node,
node->hj_HashOperators,
node->hj_Collations,
HJ_FILL_INNER(node));
@@ -287,8 +372,13 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
* Parallel Hash, then we'll try to help hashing unless we
* arrived too late.
*/
- hashNode->hashtable = hashtable;
- (void) MultiExecProcNode((PlanState *) hashNode);
+ node->hashtable = hashtable;
+
+ /* fill the hash table */
+ if (node->parallel_state != NULL)
+ ExecHashBuildParallel(node);
+ else
+ ExecHashBuildPrivate(node);
/*
* If the inner relation is completely empty, and we're not
@@ -599,10 +689,13 @@ ExecInitHashJoin(HashJoin *node, EState *estate, int eflags)
{
HashJoinState *hjstate;
Plan *outerNode;
- Hash *hashNode;
+ Plan *innerNode;
TupleDesc outerDesc,
innerDesc;
- const TupleTableSlotOps *ops;
+ const TupleTableSlotOps *innerOps;
+ const TupleTableSlotOps *outerOps;
+ bool innerOpsFixed;
+ bool outerOpsFixed;
/* check for unsupported flags */
Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
@@ -637,12 +730,26 @@ ExecInitHashJoin(HashJoin *node, EState *estate, int eflags)
* clear if this would be a win or not.
*/
outerNode = outerPlan(node);
- hashNode = (Hash *) innerPlan(node);
+ innerNode = innerPlan(node);
outerPlanState(hjstate) = ExecInitNode(outerNode, estate, eflags);
outerDesc = ExecGetResultType(outerPlanState(hjstate));
- innerPlanState(hjstate) = ExecInitNode((Plan *) hashNode, estate, eflags);
+ outerOps = ExecGetResultSlotOps(outerPlanState(hjstate), &outerOpsFixed);
+ innerPlanState(hjstate) = ExecInitNode(innerNode, estate, eflags);
innerDesc = ExecGetResultType(innerPlanState(hjstate));
+ innerOps = ExecGetResultSlotOps(innerPlanState(hjstate), &innerOpsFixed);
+
+ /*
+ * For the majority of expressions the inner tuple will come from the hash
+ * table, which stores minimal tuples.
+ */
+ hjstate->hj_HashTupleSlot =
+ ExecAllocTableSlot(&estate->es_tupleTable,
+ innerDesc,
+ &TTSOpsMinimalTuple);
+ hjstate->js.ps.inneropsset = true;
+ hjstate->js.ps.innerops = &TTSOpsMinimalTuple;
+ hjstate->js.ps.inneropsfixed = true;
/*
* Initialize result slot, type and projection.
@@ -653,9 +760,8 @@ ExecInitHashJoin(HashJoin *node, EState *estate, int eflags)
/*
* tuple table initialization
*/
- ops = ExecGetResultSlotOps(outerPlanState(hjstate), NULL);
hjstate->hj_OuterTupleSlot = ExecInitExtraTupleSlot(estate, outerDesc,
- ops);
+ outerOps);
/*
* detect whether we need only consider the first matching inner tuple
@@ -689,20 +795,6 @@ ExecInitHashJoin(HashJoin *node, EState *estate, int eflags)
(int) node->join.jointype);
}
- /*
- * now for some voodoo. our temporary tuple slot is actually the result
- * tuple slot of the Hash node (which is our inner plan). we can do this
- * because Hash nodes don't return tuples via ExecProcNode() -- instead
- * the hash join node uses ExecScanHashBucket() to get at the contents of
- * the hash table. -cim 6/9/91
- */
- {
- HashState *hashstate = (HashState *) innerPlanState(hjstate);
- TupleTableSlot *slot = hashstate->ps.ps_ResultTupleSlot;
-
- hjstate->hj_HashTupleSlot = slot;
- }
-
/*
* initialize child expressions
*/
@@ -724,7 +816,21 @@ ExecInitHashJoin(HashJoin *node, EState *estate, int eflags)
hjstate->hj_CurSkewBucketNo = INVALID_SKEW_BUCKET_NO;
hjstate->hj_CurTuple = NULL;
- hjstate->hj_OuterHashKeys = ExecInitExprList(node->hashkeys,
+ hjstate->hj_OuterHashKeys = ExecInitExprList(node->hashkeys_outer,
+ (PlanState *) hjstate);
+
+ hjstate->js.ps.inneropsset = true;
+ hjstate->js.ps.innerops = &TTSOpsMinimalTuple;
+ hjstate->js.ps.inneropsfixed = true;
+
+ /*
+ *
+ */
+ hjstate->js.ps.inneropsset = true;
+ hjstate->js.ps.innerops = innerOps;
+ hjstate->js.ps.inneropsfixed = innerOpsFixed;
+
+ hjstate->hj_InnerHashKeys = ExecInitExprList(node->hashkeys_inner,
(PlanState *) hjstate);
hjstate->hj_HashOperators = node->hashoperators;
hjstate->hj_Collations = node->hashcollations;
@@ -1195,7 +1301,7 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
* context, not in a shorter-lived context; else the temp file buffers
* will get messed up.
*/
-void
+static void
ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
BufFile **fileptr)
{
@@ -1360,6 +1466,31 @@ ExecShutdownHashJoin(HashJoinState *node)
ExecHashTableDetachBatch(node->hj_HashTable);
ExecHashTableDetach(node->hj_HashTable);
}
+
+
+ /*
+ * Copy instrumentation data from this worker's hash table (if it built one)
+ * to DSM memory so the leader can retrieve it. This must be done in an
+ * ExecShutdownHash() rather than ExecEndHash() because the latter runs after
+ * we've detached from the DSM segment.
+ */
+ if (node->hinstrument && node->hashtable)
+ ExecHashJoinGetInstrumentation(node->hinstrument, node->hashtable);
+}
+
+/*
+ * Copy the instrumentation data from 'hashtable' into a HashInstrumentation
+ * struct.
+ */
+void
+ExecHashJoinGetInstrumentation(HashInstrumentation *instrument,
+ HashJoinTable hashtable)
+{
+ instrument->nbuckets = hashtable->nbuckets;
+ instrument->nbuckets_original = hashtable->nbuckets_original;
+ instrument->nbatch = hashtable->nbatch;
+ instrument->nbatch_original = hashtable->nbatch_original;
+ instrument->space_peak = hashtable->spacePeak;
}
static void
@@ -1411,17 +1542,51 @@ ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate)
void
ExecHashJoinEstimate(HashJoinState *state, ParallelContext *pcxt)
{
- shm_toc_estimate_chunk(&pcxt->estimator, sizeof(ParallelHashJoinState));
- shm_toc_estimate_keys(&pcxt->estimator, 1);
+ size_t size;
+
+ /* Even when not parallel-aware, for EXPLAIN ANALYZE */
+ if (state->js.ps.instrument || pcxt->nworkers > 0)
+ {
+ size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation));
+ size = add_size(size, offsetof(SharedHashInfo, hinstrument));
+ shm_toc_estimate_chunk(&pcxt->estimator, size);
+ shm_toc_estimate_keys(&pcxt->estimator, 1);
+ }
+
+ if (state->js.ps.plan->parallel_aware)
+ {
+ shm_toc_estimate_chunk(&pcxt->estimator, sizeof(ParallelHashJoinState));
+ shm_toc_estimate_keys(&pcxt->estimator, 1);
+ }
}
void
ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
{
int plan_node_id = state->js.ps.plan->plan_node_id;
- HashState *hashNode;
ParallelHashJoinState *pstate;
+ /* even when not parallel-aware, for EXPLAIN ANALYZE */
+ /*
+ * For hashtable, but don't need this if not instrumenting or no workers.
+ */
+ if (state->js.ps.instrument || pcxt->nworkers > 0)
+ {
+ size_t size;
+
+ size = offsetof(SharedHashInfo, hinstrument) +
+ pcxt->nworkers * sizeof(HashInstrumentation);
+ state->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size);
+ memset(state->shared_info, 0, size);
+ state->shared_info->num_workers = pcxt->nworkers;
+ // FIXME: Hack -
+ shm_toc_insert(pcxt->toc, -state->js.ps.plan->plan_node_id,
+ state->shared_info);
+ }
+
+ if (!state->js.ps.plan->parallel_aware)
+ return;
+
/*
* Disable shared hash table mode if we failed to create a real DSM
* segment, because that means that we don't have a DSA area to work with.
@@ -1462,9 +1627,9 @@ ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcxt)
/* Set up the space we'll use for shared temporary files. */
SharedFileSetInit(&pstate->fileset, pcxt->seg);
- /* Initialize the shared state in the hash node. */
- hashNode = (HashState *) innerPlanState(state);
- hashNode->parallel_state = pstate;
+ state->parallel_state = pstate;
+
+
}
/* ----------------------------------------------------------------
@@ -1510,17 +1675,3065 @@ void
ExecHashJoinInitializeWorker(HashJoinState *state,
ParallelWorkerContext *pwcxt)
{
- HashState *hashNode;
int plan_node_id = state->js.ps.plan->plan_node_id;
- ParallelHashJoinState *pstate =
- shm_toc_lookup(pwcxt->toc, plan_node_id, false);
- /* Attach to the space for shared temporary files. */
- SharedFileSetAttach(&pstate->fileset, pwcxt->seg);
+ /* don't need this if not instrumenting */
+ if (state->js.ps.instrument)
+ {
+ SharedHashInfo *shared_info;
- /* Attach to the shared state in the hash node. */
- hashNode = (HashState *) innerPlanState(state);
- hashNode->parallel_state = pstate;
+ // FIXME: Hack -
+ shared_info = (SharedHashInfo *)
+ shm_toc_lookup(pwcxt->toc, -state->js.ps.plan->plan_node_id, false);
+ state->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber];
+ }
- ExecSetExecProcNode(&state->js.ps, ExecParallelHashJoin);
+ if (state->js.ps.plan->parallel_aware)
+ {
+ ParallelHashJoinState *pstate =
+ shm_toc_lookup(pwcxt->toc, plan_node_id, false);
+
+ /* Attach to the space for shared temporary files. */
+ SharedFileSetAttach(&pstate->fileset, pwcxt->seg);
+
+ /* Attach to the shared state */
+ state->parallel_state = pstate;
+
+ ExecSetExecProcNode(&state->js.ps, ExecParallelHashJoin);
+ }
+}
+
+/*
+ * Retrieve instrumentation data from workers before the DSM segment is
+ * detached, so that EXPLAIN can access it.
+ */
+void
+ExecHashJoinRetrieveInstrumentation(HashJoinState *node)
+{
+ SharedHashInfo *shared_info = node->shared_info;
+ size_t size;
+
+ if (shared_info == NULL)
+ return;
+
+ /* Replace node->shared_info with a copy in backend-local memory. */
+ size = offsetof(SharedHashInfo, hinstrument) +
+ shared_info->num_workers * sizeof(HashInstrumentation);
+ node->shared_info = palloc(size);
+ memcpy(node->shared_info, shared_info, size);
+}
+
+/* ----------------------------------------------------------------
+ * ExecHashBuildPrivate
+ *
+ * parallel-oblivious version, building a backend-private
+ * hash table and (if necessary) batch files.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecHashBuildPrivate(HashJoinState *node)
+{
+ PlanState *innerNode;
+ List *hashkeys;
+ HashJoinTable hashtable;
+ TupleTableSlot *slot;
+ ExprContext *econtext;
+ uint32 hashvalue;
+
+ /*
+ * get state info from node
+ */
+ innerNode = innerPlanState(node);
+ hashtable = node->hashtable;
+
+ /*
+ * set expression context
+ */
+ hashkeys = node->hj_InnerHashKeys;
+ econtext = node->js.ps.ps_ExprContext;
+
+ /*
+ * Get all tuples from the node below the Hash node and insert into the
+ * hash table (or temp files).
+ */
+ for (;;)
+ {
+ slot = ExecProcNode(innerNode);
+ if (TupIsNull(slot))
+ break;
+ /* We have to compute the hash value */
+ econtext->ecxt_innertuple = slot;
+ if (ExecHashGetHashValue(hashtable, econtext, hashkeys,
+ false, hashtable->keepNulls,
+ &hashvalue))
+ {
+ int bucketNumber;
+
+ bucketNumber = ExecHashGetSkewBucket(hashtable, hashvalue);
+ if (bucketNumber != INVALID_SKEW_BUCKET_NO)
+ {
+ /* It's a skew tuple, so put it into that hash table */
+ ExecHashSkewTableInsert(hashtable, slot, hashvalue,
+ bucketNumber);
+ hashtable->skewTuples += 1;
+ }
+ else
+ {
+ /* Not subject to skew optimization, so insert normally */
+ ExecHashTableInsert(hashtable, slot, hashvalue);
+ }
+ hashtable->totalTuples += 1;
+ }
+ }
+
+ /* resize the hash table if needed (NTUP_PER_BUCKET exceeded) */
+ if (hashtable->nbuckets != hashtable->nbuckets_optimal)
+ ExecHashIncreaseNumBuckets(hashtable);
+
+ /* Account for the buckets in spaceUsed (reported in EXPLAIN ANALYZE) */
+ hashtable->spaceUsed += hashtable->nbuckets * sizeof(HashJoinTuple);
+ if (hashtable->spaceUsed > hashtable->spacePeak)
+ hashtable->spacePeak = hashtable->spaceUsed;
+
+ hashtable->partialTuples = hashtable->totalTuples;
+}
+
+/* ----------------------------------------------------------------
+ * ExecHashBuildParallel
+ *
+ * parallel-aware version, building a shared hash table and
+ * (if necessary) batch files using the combined effort of
+ * a set of co-operating backends.
+ * ----------------------------------------------------------------
+ */
+static void
+ExecHashBuildParallel(HashJoinState *node)
+{
+ ParallelHashJoinState *pstate;
+ PlanState *innerNode;
+ List *hashkeys;
+ HashJoinTable hashtable;
+ TupleTableSlot *slot;
+ ExprContext *econtext;
+ uint32 hashvalue;
+ Barrier *build_barrier;
+ int i;
+
+ /*
+ * get state info from node
+ */
+ innerNode = innerPlanState(node);
+ hashtable = node->hashtable;
+
+ /*
+ * set expression context
+ */
+ hashkeys = node->hj_InnerHashKeys;
+ econtext = node->js.ps.ps_ExprContext;
+
+ /*
+ * Synchronize the parallel hash table build. At this stage we know that
+ * the shared hash table has been or is being set up by
+ * ExecHashTableCreate(), but we don't know if our peers have returned
+ * from there or are here in ExecHashBuildParallel(), and if so how far
+ * through they are. To find out, we check the build_barrier phase then
+ * and jump to the right step in the build algorithm.
+ */
+ pstate = hashtable->parallel_state;
+ build_barrier = &pstate->build_barrier;
+ Assert(BarrierPhase(build_barrier) >= PHJ_BUILD_ALLOCATING);
+ 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.
+ */
+ BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ALLOCATING);
+ /* Fall through. */
+
+ 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);
+ ExecParallelHashTableSetCurrentBatch(hashtable, 0);
+ for (;;)
+ {
+ slot = ExecProcNode(innerNode);
+ if (TupIsNull(slot))
+ break;
+ econtext->ecxt_innertuple = slot;
+ if (ExecHashGetHashValue(hashtable, econtext, hashkeys,
+ false, hashtable->keepNulls,
+ &hashvalue))
+ ExecParallelHashTableInsert(hashtable, slot, hashvalue);
+ hashtable->partialTuples++;
+ }
+
+ /*
+ * Make sure that any tuples we wrote to disk are visible to
+ * others before anyone tries to load them.
+ */
+ for (i = 0; i < hashtable->nbatch; ++i)
+ sts_end_write(hashtable->batches[i].inner_tuples);
+
+ /*
+ * Update shared counters. We need an accurate total tuple count
+ * to control the empty table optimization.
+ */
+ ExecParallelHashMergeCounters(hashtable);
+
+ BarrierDetach(&pstate->grow_buckets_barrier);
+ BarrierDetach(&pstate->grow_batches_barrier);
+
+ /*
+ * Wait for everyone to finish building and flushing files and
+ * counters.
+ */
+ if (BarrierArriveAndWait(build_barrier,
+ WAIT_EVENT_HASH_BUILD_HASHING_INNER))
+ {
+ /*
+ * Elect one backend to disable any further growth. Batches
+ * are now fixed. While building them we made sure they'd fit
+ * in our memory budget when we load them back in later (or we
+ * tried to do that and gave up because we detected extreme
+ * skew).
+ */
+ pstate->growth = PHJ_GROWTH_DISABLED;
+ }
+ }
+
+ /*
+ * We're not yet attached to a batch. We all agree on the dimensions and
+ * number of inner tuples (for the empty table optimization).
+ */
+ hashtable->curbatch = -1;
+ hashtable->nbuckets = pstate->nbuckets;
+ hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
+ hashtable->totalTuples = pstate->total_tuples;
+ 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
+ * there already).
+ */
+ Assert(BarrierPhase(build_barrier) == PHJ_BUILD_HASHING_OUTER ||
+ BarrierPhase(build_barrier) == PHJ_BUILD_DONE);
+}
+
+/* ----------------------------------------------------------------
+ * ExecHashTableCreate
+ *
+ * create an empty hashtable data structure for hashjoin.
+ * ----------------------------------------------------------------
+ */
+static HashJoinTable
+ExecHashTableCreate(HashJoinState *state, List *hashOperators, List *hashCollations, bool keepNulls)
+{
+ HashJoin *node;
+ HashJoinTable hashtable;
+ Plan *innerNode;
+ size_t space_allowed;
+ int nbuckets;
+ int nbatch;
+ double rows;
+ int num_skew_mcvs;
+ int log2_nbuckets;
+ int nkeys;
+ int i;
+ ListCell *ho;
+ ListCell *hc;
+ MemoryContext oldcxt;
+
+ /*
+ * Get information about the size of the relation to be hashed (i.e. the
+ * inner relation of the hashjoin). Compute the appropriate size of the
+ * hash table.
+ */
+ node = (HashJoin *) state->js.ps.plan;
+ innerNode = innerPlan(node);
+
+ /*
+ * If this is shared hash table with a partial plan, then we can't use
+ * outerNode->plan_rows to estimate its size. We need an estimate of the
+ * total number of rows across all copies of the partial plan.
+ */
+ rows = node->join.plan.parallel_aware ? node->inner_rows_total : innerNode->plan_rows;
+
+ ExecChooseHashTableSize(rows, innerNode->plan_width,
+ OidIsValid(node->skewTable),
+ state->parallel_state != NULL,
+ state->parallel_state != NULL ?
+ state->parallel_state->nparticipants - 1 : 0,
+ &space_allowed,
+ &nbuckets, &nbatch, &num_skew_mcvs);
+
+ /* nbuckets must be a power of 2 */
+ log2_nbuckets = my_log2(nbuckets);
+ Assert(nbuckets == (1 << log2_nbuckets));
+
+ /*
+ * Initialize the hash table control block.
+ *
+ * The hashtable control block is just palloc'd from the executor's
+ * per-query memory context. Everything else should be kept inside the
+ * subsidiary hashCxt or batchCxt.
+ */
+ hashtable = (HashJoinTable) palloc(sizeof(HashJoinTableData));
+ hashtable->nbuckets = nbuckets;
+ hashtable->nbuckets_original = nbuckets;
+ hashtable->nbuckets_optimal = nbuckets;
+ hashtable->log2_nbuckets = log2_nbuckets;
+ hashtable->log2_nbuckets_optimal = log2_nbuckets;
+ hashtable->buckets.unshared = NULL;
+ hashtable->keepNulls = keepNulls;
+ hashtable->skewEnabled = false;
+ hashtable->skewBucket = NULL;
+ hashtable->skewBucketLen = 0;
+ hashtable->nSkewBuckets = 0;
+ hashtable->skewBucketNums = NULL;
+ hashtable->nbatch = nbatch;
+ hashtable->curbatch = 0;
+ hashtable->nbatch_original = nbatch;
+ hashtable->nbatch_outstart = nbatch;
+ hashtable->growEnabled = true;
+ hashtable->totalTuples = 0;
+ hashtable->partialTuples = 0;
+ hashtable->skewTuples = 0;
+ hashtable->innerBatchFile = NULL;
+ hashtable->outerBatchFile = NULL;
+ hashtable->spaceUsed = 0;
+ hashtable->spacePeak = 0;
+ hashtable->spaceAllowed = space_allowed;
+ hashtable->spaceUsedSkew = 0;
+ hashtable->spaceAllowedSkew =
+ hashtable->spaceAllowed * SKEW_WORK_MEM_PERCENT / 100;
+ hashtable->chunks = NULL;
+ hashtable->current_chunk = NULL;
+ hashtable->parallel_state = state->parallel_state;
+ hashtable->area = state->js.ps.state->es_query_dsa;
+ hashtable->batches = NULL;
+
+#ifdef HJDEBUG
+ printf("Hashjoin %p: initial nbatch = %d, nbuckets = %d\n",
+ hashtable, nbatch, nbuckets);
+#endif
+
+ /*
+ * Create temporary memory contexts in which to keep the hashtable working
+ * storage. See notes in executor/hashjoin.h.
+ */
+ hashtable->hashCxt = AllocSetContextCreate(CurrentMemoryContext,
+ "HashTableContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ hashtable->batchCxt = AllocSetContextCreate(hashtable->hashCxt,
+ "HashBatchContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ /* Allocate data that will live for the life of the hashjoin */
+
+ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
+
+ /*
+ * Get info about the hash functions to be used for each hash key. Also
+ * remember whether the join operators are strict.
+ */
+ nkeys = list_length(hashOperators);
+ hashtable->outer_hashfunctions =
+ (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
+ hashtable->inner_hashfunctions =
+ (FmgrInfo *) palloc(nkeys * sizeof(FmgrInfo));
+ hashtable->hashStrict = (bool *) palloc(nkeys * sizeof(bool));
+ hashtable->collations = (Oid *) palloc(nkeys * sizeof(Oid));
+ i = 0;
+ forboth(ho, hashOperators, hc, hashCollations)
+ {
+ Oid hashop = lfirst_oid(ho);
+ Oid left_hashfn;
+ Oid right_hashfn;
+
+ if (!get_op_hash_functions(hashop, &left_hashfn, &right_hashfn))
+ elog(ERROR, "could not find hash function for hash operator %u",
+ hashop);
+ fmgr_info(left_hashfn, &hashtable->outer_hashfunctions[i]);
+ fmgr_info(right_hashfn, &hashtable->inner_hashfunctions[i]);
+ hashtable->hashStrict[i] = op_strict(hashop);
+ hashtable->collations[i] = lfirst_oid(hc);
+ i++;
+ }
+
+ if (nbatch > 1 && hashtable->parallel_state == NULL)
+ {
+ /*
+ * allocate and initialize the file arrays in hashCxt (not needed for
+ * parallel case which uses shared tuplestores instead of raw files)
+ */
+ hashtable->innerBatchFile = (BufFile **)
+ palloc0(nbatch * sizeof(BufFile *));
+ hashtable->outerBatchFile = (BufFile **)
+ palloc0(nbatch * sizeof(BufFile *));
+ /* The files will not be opened until needed... */
+ /* ... but make sure we have temp tablespaces established for them */
+ PrepareTempTablespaces();
+ }
+
+ MemoryContextSwitchTo(oldcxt);
+
+ if (hashtable->parallel_state)
+ {
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ Barrier *build_barrier;
+
+ /*
+ * 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
+ * and then loaded while hashing (the standard hybrid hash join
+ * algorithm), and we'll coordinate that using build_barrier.
+ */
+ build_barrier = &pstate->build_barrier;
+ BarrierAttach(build_barrier);
+
+ /*
+ * So far we have no idea whether there are any other participants,
+ * and if so, what phase they are working on. The only thing we care
+ * about at this point is whether someone has already created the
+ * 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 &&
+ BarrierArriveAndWait(build_barrier, WAIT_EVENT_HASH_BUILD_ELECTING))
+ {
+ pstate->nbatch = nbatch;
+ pstate->space_allowed = space_allowed;
+ pstate->growth = PHJ_GROWTH_OK;
+
+ /* Set up the shared state for coordinating batches. */
+ ExecParallelHashJoinSetUpBatches(hashtable, nbatch);
+
+ /*
+ * Allocate batch 0's hash table up front so we can load it
+ * directly while hashing.
+ */
+ pstate->nbuckets = nbuckets;
+ ExecParallelHashTableAlloc(hashtable, 0);
+ }
+
+ /*
+ * The next Parallel Hash synchronization point is in
+ * ExecHashBuildParallel(), which will progress it all the way to
+ * PHJ_BUILD_DONE. The caller must not return control from this
+ * executor node between now and then.
+ */
+ }
+ else
+ {
+ /*
+ * Prepare context for the first-scan space allocations; allocate the
+ * hashbucket array therein, and set each bucket "empty".
+ */
+ MemoryContextSwitchTo(hashtable->batchCxt);
+
+ hashtable->buckets.unshared = (HashJoinTuple *)
+ palloc0(nbuckets * sizeof(HashJoinTuple));
+
+ /*
+ * Set up for skew optimization, if possible and there's a need for
+ * more than one batch. (In a one-batch join, there's no point in
+ * it.)
+ */
+ if (nbatch > 1)
+ ExecHashBuildSkewHash(hashtable, node, num_skew_mcvs);
+
+ MemoryContextSwitchTo(oldcxt);
+ }
+
+ return hashtable;
+}
+
+
+/*
+ * Compute appropriate size for hashtable given the estimated size of the
+ * relation to be hashed (number of rows and average row width).
+ *
+ * This is exported so that the planner's costsize.c can use it.
+ */
+
+/* Target bucket loading (tuples per bucket) */
+#define NTUP_PER_BUCKET 1
+
+void
+ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
+ bool try_combined_work_mem,
+ int parallel_workers,
+ size_t *space_allowed,
+ int *numbuckets,
+ int *numbatches,
+ int *num_skew_mcvs)
+{
+ int tupsize;
+ double inner_rel_bytes;
+ long bucket_bytes;
+ long hash_table_bytes;
+ long skew_table_bytes;
+ long max_pointers;
+ long mppow2;
+ int nbatch = 1;
+ int nbuckets;
+ double dbuckets;
+
+ /* Force a plausible relation size if no info */
+ if (ntuples <= 0.0)
+ ntuples = 1000.0;
+
+ /*
+ * Estimate tupsize based on footprint of tuple in hashtable... note this
+ * does not allow for any palloc overhead. The manipulations of spaceUsed
+ * don't count palloc overhead either.
+ */
+ tupsize = HJTUPLE_OVERHEAD +
+ MAXALIGN(SizeofMinimalTupleHeader) +
+ MAXALIGN(tupwidth);
+ inner_rel_bytes = ntuples * tupsize;
+
+ /*
+ * Target in-memory hashtable size is work_mem kilobytes.
+ */
+ hash_table_bytes = work_mem * 1024L;
+
+ /*
+ * 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.
+ */
+ if (try_combined_work_mem)
+ hash_table_bytes += hash_table_bytes * parallel_workers;
+
+ *space_allowed = hash_table_bytes;
+
+ /*
+ * If skew optimization is possible, estimate the number of skew buckets
+ * that will fit in the memory allowed, and decrement the assumed space
+ * available for the main hash table accordingly.
+ *
+ * We make the optimistic assumption that each skew bucket will contain
+ * one inner-relation tuple. If that turns out to be low, we will recover
+ * at runtime by reducing the number of skew buckets.
+ *
+ * hashtable->skewBucket will have up to 8 times as many HashSkewBucket
+ * pointers as the number of MCVs we allow, since ExecHashBuildSkewHash
+ * will round up to the next power of 2 and then multiply by 4 to reduce
+ * collisions.
+ */
+ if (useskew)
+ {
+ skew_table_bytes = hash_table_bytes * SKEW_WORK_MEM_PERCENT / 100;
+
+ /*----------
+ * Divisor is:
+ * size of a hash tuple +
+ * worst-case size of skewBucket[] per MCV +
+ * size of skewBucketNums[] entry +
+ * size of skew bucket struct itself
+ *----------
+ */
+ *num_skew_mcvs = skew_table_bytes / (tupsize +
+ (8 * sizeof(HashSkewBucket *)) +
+ sizeof(int) +
+ SKEW_BUCKET_OVERHEAD);
+ if (*num_skew_mcvs > 0)
+ hash_table_bytes -= skew_table_bytes;
+ }
+ else
+ *num_skew_mcvs = 0;
+
+ /*
+ * Set nbuckets to achieve an average bucket load of NTUP_PER_BUCKET when
+ * memory is filled, assuming a single batch; but limit the value so that
+ * the pointer arrays we'll try to allocate do not exceed work_mem nor
+ * MaxAllocSize.
+ *
+ * Note that both nbuckets and nbatch must be powers of 2 to make
+ * ExecHashGetBucketAndBatch fast.
+ */
+ max_pointers = *space_allowed / sizeof(HashJoinTuple);
+ max_pointers = Min(max_pointers, MaxAllocSize / sizeof(HashJoinTuple));
+ /* If max_pointers isn't a power of 2, must round it down to one */
+ mppow2 = 1L << my_log2(max_pointers);
+ if (max_pointers != mppow2)
+ max_pointers = mppow2 / 2;
+
+ /* Also ensure we avoid integer overflow in nbatch and nbuckets */
+ /* (this step is redundant given the current value of MaxAllocSize) */
+ max_pointers = Min(max_pointers, INT_MAX / 2);
+
+ dbuckets = ceil(ntuples / NTUP_PER_BUCKET);
+ dbuckets = Min(dbuckets, max_pointers);
+ nbuckets = (int) dbuckets;
+ /* don't let nbuckets be really small, though ... */
+ nbuckets = Max(nbuckets, 1024);
+ /* ... and force it to be a power of 2. */
+ nbuckets = 1 << my_log2(nbuckets);
+
+ /*
+ * If there's not enough space to store the projected number of tuples and
+ * the required bucket headers, we will need multiple batches.
+ */
+ bucket_bytes = sizeof(HashJoinTuple) * nbuckets;
+ if (inner_rel_bytes + bucket_bytes > hash_table_bytes)
+ {
+ /* We'll need multiple batches */
+ long lbuckets;
+ double dbatch;
+ int minbatch;
+ long bucket_size;
+
+ /*
+ * If Parallel Hash with combined work_mem would still need multiple
+ * batches, we'll have to fall back to regular work_mem budget.
+ */
+ if (try_combined_work_mem)
+ {
+ ExecChooseHashTableSize(ntuples, tupwidth, useskew,
+ false, parallel_workers,
+ space_allowed,
+ numbuckets,
+ numbatches,
+ num_skew_mcvs);
+ return;
+ }
+
+ /*
+ * Estimate the number of buckets we'll want to have when work_mem is
+ * entirely full. Each bucket will contain a bucket pointer plus
+ * NTUP_PER_BUCKET tuples, whose projected size already includes
+ * overhead for the hash code, pointer to the next tuple, etc.
+ */
+ bucket_size = (tupsize * NTUP_PER_BUCKET + sizeof(HashJoinTuple));
+ lbuckets = 1L << my_log2(hash_table_bytes / bucket_size);
+ lbuckets = Min(lbuckets, max_pointers);
+ nbuckets = (int) lbuckets;
+ nbuckets = 1 << my_log2(nbuckets);
+ bucket_bytes = nbuckets * sizeof(HashJoinTuple);
+
+ /*
+ * Buckets are simple pointers to hashjoin tuples, while tupsize
+ * includes the pointer, hash code, and MinimalTupleData. So buckets
+ * should never really exceed 25% of work_mem (even for
+ * NTUP_PER_BUCKET=1); except maybe for work_mem values that are not
+ * 2^N bytes, where we might get more because of doubling. So let's
+ * look for 50% here.
+ */
+ Assert(bucket_bytes <= hash_table_bytes / 2);
+
+ /* Calculate required number of batches. */
+ dbatch = ceil(inner_rel_bytes / (hash_table_bytes - bucket_bytes));
+ dbatch = Min(dbatch, max_pointers);
+ minbatch = (int) dbatch;
+ nbatch = 2;
+ while (nbatch < minbatch)
+ nbatch <<= 1;
+ }
+
+ Assert(nbuckets > 0);
+ Assert(nbatch > 0);
+
+ *numbuckets = nbuckets;
+ *numbatches = nbatch;
+}
+
+
+/* ----------------------------------------------------------------
+ * ExecHashTableDestroy
+ *
+ * destroy a hash table
+ * ----------------------------------------------------------------
+ */
+static void
+ExecHashTableDestroy(HashJoinTable hashtable)
+{
+ int i;
+
+ /*
+ * Make sure all the temp files are closed. We skip batch 0, since it
+ * can't have any temp files (and the arrays might not even exist if
+ * nbatch is only 1). Parallel hash joins don't use these files.
+ */
+ if (hashtable->innerBatchFile != NULL)
+ {
+ for (i = 1; i < hashtable->nbatch; i++)
+ {
+ if (hashtable->innerBatchFile[i])
+ BufFileClose(hashtable->innerBatchFile[i]);
+ if (hashtable->outerBatchFile[i])
+ BufFileClose(hashtable->outerBatchFile[i]);
+ }
+ }
+
+ /* Release working memory (batchCxt is a child, so it goes away too) */
+ MemoryContextDelete(hashtable->hashCxt);
+
+ /* And drop the control block */
+ pfree(hashtable);
+}
+
+
+/*
+ * ExecHashIncreaseNumBatches
+ * increase the original number of batches in order to reduce
+ * current memory consumption
+ */
+static void
+ExecHashIncreaseNumBatches(HashJoinTable hashtable)
+{
+ int oldnbatch = hashtable->nbatch;
+ int curbatch = hashtable->curbatch;
+ int nbatch;
+ MemoryContext oldcxt;
+ long ninmemory;
+ long nfreed;
+ HashMemoryChunk oldchunks;
+
+ /* do nothing if we've decided to shut off growth */
+ if (!hashtable->growEnabled)
+ return;
+
+ /* safety check to avoid overflow */
+ if (oldnbatch > Min(INT_MAX / 2, MaxAllocSize / (sizeof(void *) * 2)))
+ return;
+
+ nbatch = oldnbatch * 2;
+ Assert(nbatch > 1);
+
+#ifdef HJDEBUG
+ printf("Hashjoin %p: increasing nbatch to %d because space = %zu\n",
+ hashtable, nbatch, hashtable->spaceUsed);
+#endif
+
+ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
+
+ if (hashtable->innerBatchFile == NULL)
+ {
+ /* we had no file arrays before */
+ hashtable->innerBatchFile = (BufFile **)
+ palloc0(nbatch * sizeof(BufFile *));
+ hashtable->outerBatchFile = (BufFile **)
+ palloc0(nbatch * sizeof(BufFile *));
+ /* time to establish the temp tablespaces, too */
+ PrepareTempTablespaces();
+ }
+ else
+ {
+ /* enlarge arrays and zero out added entries */
+ hashtable->innerBatchFile = (BufFile **)
+ repalloc(hashtable->innerBatchFile, nbatch * sizeof(BufFile *));
+ hashtable->outerBatchFile = (BufFile **)
+ repalloc(hashtable->outerBatchFile, nbatch * sizeof(BufFile *));
+ MemSet(hashtable->innerBatchFile + oldnbatch, 0,
+ (nbatch - oldnbatch) * sizeof(BufFile *));
+ MemSet(hashtable->outerBatchFile + oldnbatch, 0,
+ (nbatch - oldnbatch) * sizeof(BufFile *));
+ }
+
+ MemoryContextSwitchTo(oldcxt);
+
+ hashtable->nbatch = nbatch;
+
+ /*
+ * Scan through the existing hash table entries and dump out any that are
+ * no longer of the current batch.
+ */
+ ninmemory = nfreed = 0;
+
+ /* If know we need to resize nbuckets, we can do it while rebatching. */
+ if (hashtable->nbuckets_optimal != hashtable->nbuckets)
+ {
+ /* we never decrease the number of buckets */
+ Assert(hashtable->nbuckets_optimal > hashtable->nbuckets);
+
+ hashtable->nbuckets = hashtable->nbuckets_optimal;
+ hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
+
+ hashtable->buckets.unshared =
+ repalloc(hashtable->buckets.unshared,
+ sizeof(HashJoinTuple) * hashtable->nbuckets);
+ }
+
+ /*
+ * We will scan through the chunks directly, so that we can reset the
+ * buckets now and not have to keep track which tuples in the buckets have
+ * already been processed. We will free the old chunks as we go.
+ */
+ memset(hashtable->buckets.unshared, 0,
+ sizeof(HashJoinTuple) * hashtable->nbuckets);
+ oldchunks = hashtable->chunks;
+ hashtable->chunks = NULL;
+
+ /* so, let's scan through the old chunks, and all tuples in each chunk */
+ while (oldchunks != NULL)
+ {
+ HashMemoryChunk nextchunk = oldchunks->next.unshared;
+
+ /* position within the buffer (up to oldchunks->used) */
+ size_t idx = 0;
+
+ /* process all tuples stored in this chunk (and then free it) */
+ while (idx < oldchunks->used)
+ {
+ HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(oldchunks) + idx);
+ MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
+ int hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len);
+ int bucketno;
+ int batchno;
+
+ ninmemory++;
+ ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
+ &bucketno, &batchno);
+
+ if (batchno == curbatch)
+ {
+ /* keep tuple in memory - copy it into the new chunk */
+ HashJoinTuple copyTuple;
+
+ copyTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
+ memcpy(copyTuple, hashTuple, hashTupleSize);
+
+ /* and add it back to the appropriate bucket */
+ copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
+ hashtable->buckets.unshared[bucketno] = copyTuple;
+ }
+ else
+ {
+ /* dump it out */
+ Assert(batchno > curbatch);
+ ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple),
+ hashTuple->hashvalue,
+ &hashtable->innerBatchFile[batchno]);
+
+ hashtable->spaceUsed -= hashTupleSize;
+ nfreed++;
+ }
+
+ /* next tuple in this chunk */
+ idx += MAXALIGN(hashTupleSize);
+
+ /* allow this loop to be cancellable */
+ CHECK_FOR_INTERRUPTS();
+ }
+
+ /* we're done with this chunk - free it and proceed to the next one */
+ pfree(oldchunks);
+ oldchunks = nextchunk;
+ }
+
+#ifdef HJDEBUG
+ printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n",
+ hashtable, nfreed, ninmemory, hashtable->spaceUsed);
+#endif
+
+ /*
+ * If we dumped out either all or none of the tuples in the table, disable
+ * further expansion of nbatch. This situation implies that we have
+ * enough tuples of identical hashvalues to overflow spaceAllowed.
+ * Increasing nbatch will not fix it since there's no way to subdivide the
+ * group any more finely. We have to just gut it out and hope the server
+ * has enough RAM.
+ */
+ if (nfreed == 0 || nfreed == ninmemory)
+ {
+ hashtable->growEnabled = false;
+#ifdef HJDEBUG
+ printf("Hashjoin %p: disabling further increase of nbatch\n",
+ hashtable);
+#endif
+ }
+}
+
+/*
+ * ExecParallelHashIncreaseNumBatches
+ * Every participant attached to grow_batches_barrier must run this
+ * function when it observes growth == PHJ_GROWTH_NEED_MORE_BATCHES.
+ */
+static void
+ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ int i;
+
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
+
+ /*
+ * It's unlikely, but we need to be prepared for new participants to show
+ * up while we're in the middle of this operation so we need to switch on
+ * barrier phase here.
+ */
+ switch (PHJ_GROW_BATCHES_PHASE(BarrierPhase(&pstate->grow_batches_barrier)))
+ {
+ case PHJ_GROW_BATCHES_ELECTING:
+
+ /*
+ * Elect one participant to prepare to grow the number of batches.
+ * This involves reallocating or resetting the buckets of batch 0
+ * in preparation for all participants to begin repartitioning the
+ * tuples.
+ */
+ if (BarrierArriveAndWait(&pstate->grow_batches_barrier,
+ WAIT_EVENT_HASH_GROW_BATCHES_ELECTING))
+ {
+ dsa_pointer_atomic *buckets;
+ ParallelHashJoinBatch *old_batch0;
+ int new_nbatch;
+ int i;
+
+ /* Move the old batch out of the way. */
+ old_batch0 = hashtable->batches[0].shared;
+ pstate->old_batches = pstate->batches;
+ pstate->old_nbatch = hashtable->nbatch;
+ pstate->batches = InvalidDsaPointer;
+
+ /* Free this backend's old accessors. */
+ ExecParallelHashCloseBatchAccessors(hashtable);
+
+ /* Figure out how many batches to use. */
+ if (hashtable->nbatch == 1)
+ {
+ /*
+ * We are going from single-batch to multi-batch. We need
+ * to switch from one large combined memory budget to the
+ * regular work_mem budget.
+ */
+ pstate->space_allowed = work_mem * 1024L;
+
+ /*
+ * The combined work_mem of all participants wasn't
+ * enough. Therefore one batch per participant would be
+ * approximately equivalent and would probably also be
+ * insufficient. So try two batches per participant,
+ * rounded up to a power of two.
+ */
+ new_nbatch = 1 << my_log2(pstate->nparticipants * 2);
+ }
+ else
+ {
+ /*
+ * We were already multi-batched. Try doubling the number
+ * of batches.
+ */
+ new_nbatch = hashtable->nbatch * 2;
+ }
+
+ /* Allocate new larger generation of batches. */
+ Assert(hashtable->nbatch == pstate->nbatch);
+ ExecParallelHashJoinSetUpBatches(hashtable, new_nbatch);
+ Assert(hashtable->nbatch == pstate->nbatch);
+
+ /* Replace or recycle batch 0's bucket array. */
+ if (pstate->old_nbatch == 1)
+ {
+ double dtuples;
+ double dbuckets;
+ int new_nbuckets;
+
+ /*
+ * We probably also need a smaller bucket array. How many
+ * tuples do we expect per batch, assuming we have only
+ * half of them so far? Normally we don't need to change
+ * the bucket array's size, because the size of each batch
+ * stays the same as we add more batches, but in this
+ * special case we move from a large batch to many smaller
+ * batches and it would be wasteful to keep the large
+ * array.
+ */
+ dtuples = (old_batch0->ntuples * 2.0) / new_nbatch;
+ dbuckets = ceil(dtuples / NTUP_PER_BUCKET);
+ dbuckets = Min(dbuckets,
+ MaxAllocSize / sizeof(dsa_pointer_atomic));
+ new_nbuckets = (int) dbuckets;
+ new_nbuckets = Max(new_nbuckets, 1024);
+ new_nbuckets = 1 << my_log2(new_nbuckets);
+ dsa_free(hashtable->area, old_batch0->buckets);
+ hashtable->batches[0].shared->buckets =
+ dsa_allocate(hashtable->area,
+ sizeof(dsa_pointer_atomic) * new_nbuckets);
+ buckets = (dsa_pointer_atomic *)
+ dsa_get_address(hashtable->area,
+ hashtable->batches[0].shared->buckets);
+ for (i = 0; i < new_nbuckets; ++i)
+ dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
+ pstate->nbuckets = new_nbuckets;
+ }
+ else
+ {
+ /* Recycle the existing bucket array. */
+ hashtable->batches[0].shared->buckets = old_batch0->buckets;
+ buckets = (dsa_pointer_atomic *)
+ dsa_get_address(hashtable->area, old_batch0->buckets);
+ for (i = 0; i < hashtable->nbuckets; ++i)
+ dsa_pointer_atomic_write(&buckets[i], InvalidDsaPointer);
+ }
+
+ /* Move all chunks to the work queue for parallel processing. */
+ pstate->chunk_work_queue = old_batch0->chunks;
+
+ /* Disable further growth temporarily while we're growing. */
+ pstate->growth = PHJ_GROWTH_DISABLED;
+ }
+ else
+ {
+ /* All other participants just flush their tuples to disk. */
+ ExecParallelHashCloseBatchAccessors(hashtable);
+ }
+ /* Fall through. */
+
+ case PHJ_GROW_BATCHES_ALLOCATING:
+ /* Wait for the above to be finished. */
+ BarrierArriveAndWait(&pstate->grow_batches_barrier,
+ WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATING);
+ /* Fall through. */
+
+ case PHJ_GROW_BATCHES_REPARTITIONING:
+ /* Make sure that we have the current dimensions and buckets. */
+ ExecParallelHashEnsureBatchAccessors(hashtable);
+ ExecParallelHashTableSetCurrentBatch(hashtable, 0);
+ /* Then partition, flush counters. */
+ ExecParallelHashRepartitionFirst(hashtable);
+ ExecParallelHashRepartitionRest(hashtable);
+ ExecParallelHashMergeCounters(hashtable);
+ /* Wait for the above to be finished. */
+ BarrierArriveAndWait(&pstate->grow_batches_barrier,
+ WAIT_EVENT_HASH_GROW_BATCHES_REPARTITIONING);
+ /* Fall through. */
+
+ case PHJ_GROW_BATCHES_DECIDING:
+
+ /*
+ * Elect one participant to clean up and decide whether further
+ * repartitioning is needed, or should be disabled because it's
+ * not helping.
+ */
+ if (BarrierArriveAndWait(&pstate->grow_batches_barrier,
+ WAIT_EVENT_HASH_GROW_BATCHES_DECIDING))
+ {
+ bool space_exhausted = false;
+ bool extreme_skew_detected = false;
+
+ /* Make sure that we have the current dimensions and buckets. */
+ ExecParallelHashEnsureBatchAccessors(hashtable);
+ ExecParallelHashTableSetCurrentBatch(hashtable, 0);
+
+ /* Are any of the new generation of batches exhausted? */
+ for (i = 0; i < hashtable->nbatch; ++i)
+ {
+ ParallelHashJoinBatch *batch = hashtable->batches[i].shared;
+
+ if (batch->space_exhausted ||
+ batch->estimated_size > pstate->space_allowed)
+ {
+ int parent;
+
+ space_exhausted = true;
+
+ /*
+ * Did this batch receive ALL of the tuples from its
+ * parent batch? That would indicate that further
+ * repartitioning isn't going to help (the hash values
+ * are probably all the same).
+ */
+ parent = i % pstate->old_nbatch;
+ if (batch->ntuples == hashtable->batches[parent].shared->old_ntuples)
+ extreme_skew_detected = true;
+ }
+ }
+
+ /* Don't keep growing if it's not helping or we'd overflow. */
+ if (extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2)
+ pstate->growth = PHJ_GROWTH_DISABLED;
+ else if (space_exhausted)
+ pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
+ else
+ pstate->growth = PHJ_GROWTH_OK;
+
+ /* Free the old batches in shared memory. */
+ dsa_free(hashtable->area, pstate->old_batches);
+ pstate->old_batches = InvalidDsaPointer;
+ }
+ /* Fall through. */
+
+ case PHJ_GROW_BATCHES_FINISHING:
+ /* Wait for the above to complete. */
+ BarrierArriveAndWait(&pstate->grow_batches_barrier,
+ WAIT_EVENT_HASH_GROW_BATCHES_FINISHING);
+ }
+}
+
+/*
+ * Repartition the tuples currently loaded into memory for inner batch 0
+ * because the number of batches has been increased. Some tuples are retained
+ * in memory and some are written out to a later batch.
+ */
+static void
+ExecParallelHashRepartitionFirst(HashJoinTable hashtable)
+{
+ dsa_pointer chunk_shared;
+ HashMemoryChunk chunk;
+
+ Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);
+
+ while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_shared)))
+ {
+ size_t idx = 0;
+
+ /* Repartition all tuples in this chunk. */
+ while (idx < chunk->used)
+ {
+ HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
+ MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple);
+ HashJoinTuple copyTuple;
+ dsa_pointer shared;
+ int bucketno;
+ int batchno;
+
+ ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
+ &bucketno, &batchno);
+
+ Assert(batchno < hashtable->nbatch);
+ if (batchno == 0)
+ {
+ /* It still belongs in batch 0. Copy to a new chunk. */
+ copyTuple =
+ ExecParallelHashTupleAlloc(hashtable,
+ HJTUPLE_OVERHEAD + tuple->t_len,
+ &shared);
+ copyTuple->hashvalue = hashTuple->hashvalue;
+ memcpy(HJTUPLE_MINTUPLE(copyTuple), tuple, tuple->t_len);
+ ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
+ copyTuple, shared);
+ }
+ else
+ {
+ size_t tuple_size =
+ MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
+
+ /* It belongs in a later batch. */
+ hashtable->batches[batchno].estimated_size += tuple_size;
+ sts_puttuple(hashtable->batches[batchno].inner_tuples,
+ &hashTuple->hashvalue, tuple);
+ }
+
+ /* Count this tuple. */
+ ++hashtable->batches[0].old_ntuples;
+ ++hashtable->batches[batchno].ntuples;
+
+ idx += MAXALIGN(HJTUPLE_OVERHEAD +
+ HJTUPLE_MINTUPLE(hashTuple)->t_len);
+ }
+
+ /* Free this chunk. */
+ dsa_free(hashtable->area, chunk_shared);
+
+ CHECK_FOR_INTERRUPTS();
+ }
+}
+
+/*
+ * Help repartition inner batches 1..n.
+ */
+static void
+ExecParallelHashRepartitionRest(HashJoinTable hashtable)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ int old_nbatch = pstate->old_nbatch;
+ SharedTuplestoreAccessor **old_inner_tuples;
+ ParallelHashJoinBatch *old_batches;
+ int i;
+
+ /* Get our hands on the previous generation of batches. */
+ old_batches = (ParallelHashJoinBatch *)
+ dsa_get_address(hashtable->area, pstate->old_batches);
+ old_inner_tuples = palloc0(sizeof(SharedTuplestoreAccessor *) * old_nbatch);
+ for (i = 1; i < old_nbatch; ++i)
+ {
+ ParallelHashJoinBatch *shared =
+ NthParallelHashJoinBatch(old_batches, i);
+
+ old_inner_tuples[i] = sts_attach(ParallelHashJoinBatchInner(shared),
+ ParallelWorkerNumber + 1,
+ &pstate->fileset);
+ }
+
+ /* Join in the effort to repartition them. */
+ for (i = 1; i < old_nbatch; ++i)
+ {
+ MinimalTuple tuple;
+ uint32 hashvalue;
+
+ /* Scan one partition from the previous generation. */
+ sts_begin_parallel_scan(old_inner_tuples[i]);
+ while ((tuple = sts_parallel_scan_next(old_inner_tuples[i], &hashvalue)))
+ {
+ size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
+ int bucketno;
+ int batchno;
+
+ /* Decide which partition it goes to in the new generation. */
+ ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno,
+ &batchno);
+
+ hashtable->batches[batchno].estimated_size += tuple_size;
+ ++hashtable->batches[batchno].ntuples;
+ ++hashtable->batches[i].old_ntuples;
+
+ /* Store the tuple its new batch. */
+ sts_puttuple(hashtable->batches[batchno].inner_tuples,
+ &hashvalue, tuple);
+
+ CHECK_FOR_INTERRUPTS();
+ }
+ sts_end_parallel_scan(old_inner_tuples[i]);
+ }
+
+ pfree(old_inner_tuples);
+}
+
+/*
+ * Transfer the backend-local per-batch counters to the shared totals.
+ */
+static void
+ExecParallelHashMergeCounters(HashJoinTable hashtable)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ int i;
+
+ LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
+ pstate->total_tuples = 0;
+ for (i = 0; i < hashtable->nbatch; ++i)
+ {
+ ParallelHashJoinBatchAccessor *batch = &hashtable->batches[i];
+
+ batch->shared->size += batch->size;
+ batch->shared->estimated_size += batch->estimated_size;
+ batch->shared->ntuples += batch->ntuples;
+ batch->shared->old_ntuples += batch->old_ntuples;
+ batch->size = 0;
+ batch->estimated_size = 0;
+ batch->ntuples = 0;
+ batch->old_ntuples = 0;
+ pstate->total_tuples += batch->shared->ntuples;
+ }
+ LWLockRelease(&pstate->lock);
+}
+
+/*
+ * ExecHashIncreaseNumBuckets
+ * increase the original number of buckets in order to reduce
+ * number of tuples per bucket
+ */
+static void
+ExecHashIncreaseNumBuckets(HashJoinTable hashtable)
+{
+ HashMemoryChunk chunk;
+
+ /* do nothing if not an increase (it's called increase for a reason) */
+ if (hashtable->nbuckets >= hashtable->nbuckets_optimal)
+ return;
+
+#ifdef HJDEBUG
+ printf("Hashjoin %p: increasing nbuckets %d => %d\n",
+ hashtable, hashtable->nbuckets, hashtable->nbuckets_optimal);
+#endif
+
+ hashtable->nbuckets = hashtable->nbuckets_optimal;
+ hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;
+
+ Assert(hashtable->nbuckets > 1);
+ Assert(hashtable->nbuckets <= (INT_MAX / 2));
+ Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets));
+
+ /*
+ * Just reallocate the proper number of buckets - we don't need to walk
+ * through them - we can walk the dense-allocated chunks (just like in
+ * ExecHashIncreaseNumBatches, but without all the copying into new
+ * chunks)
+ */
+ hashtable->buckets.unshared =
+ (HashJoinTuple *) repalloc(hashtable->buckets.unshared,
+ hashtable->nbuckets * sizeof(HashJoinTuple));
+
+ memset(hashtable->buckets.unshared, 0,
+ hashtable->nbuckets * sizeof(HashJoinTuple));
+
+ /* scan through all tuples in all chunks to rebuild the hash table */
+ for (chunk = hashtable->chunks; chunk != NULL; chunk = chunk->next.unshared)
+ {
+ /* process all tuples stored in this chunk */
+ size_t idx = 0;
+
+ while (idx < chunk->used)
+ {
+ HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
+ int bucketno;
+ int batchno;
+
+ ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
+ &bucketno, &batchno);
+
+ /* add the tuple to the proper bucket */
+ hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
+ hashtable->buckets.unshared[bucketno] = hashTuple;
+
+ /* advance index past the tuple */
+ idx += MAXALIGN(HJTUPLE_OVERHEAD +
+ HJTUPLE_MINTUPLE(hashTuple)->t_len);
+ }
+
+ /* allow this loop to be cancellable */
+ CHECK_FOR_INTERRUPTS();
+ }
+}
+
+static void
+ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ int i;
+ HashMemoryChunk chunk;
+ dsa_pointer chunk_s;
+
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
+
+ /*
+ * It's unlikely, but we need to be prepared for new participants to show
+ * up while we're in the middle of this operation so we need to switch on
+ * barrier phase here.
+ */
+ switch (PHJ_GROW_BUCKETS_PHASE(BarrierPhase(&pstate->grow_buckets_barrier)))
+ {
+ case PHJ_GROW_BUCKETS_ELECTING:
+ /* Elect one participant to prepare to increase nbuckets. */
+ if (BarrierArriveAndWait(&pstate->grow_buckets_barrier,
+ WAIT_EVENT_HASH_GROW_BUCKETS_ELECTING))
+ {
+ size_t size;
+ dsa_pointer_atomic *buckets;
+
+ /* Double the size of the bucket array. */
+ pstate->nbuckets *= 2;
+ size = pstate->nbuckets * sizeof(dsa_pointer_atomic);
+ hashtable->batches[0].shared->size += size / 2;
+ dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
+ hashtable->batches[0].shared->buckets =
+ dsa_allocate(hashtable->area, size);
+ buckets = (dsa_pointer_atomic *)
+ dsa_get_address(hashtable->area,
+ hashtable->batches[0].shared->buckets);
+ for (i = 0; i < pstate->nbuckets; ++i)
+ dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
+
+ /* Put the chunk list onto the work queue. */
+ pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;
+
+ /* Clear the flag. */
+ pstate->growth = PHJ_GROWTH_OK;
+ }
+ /* Fall through. */
+
+ case PHJ_GROW_BUCKETS_ALLOCATING:
+ /* Wait for the above to complete. */
+ BarrierArriveAndWait(&pstate->grow_buckets_barrier,
+ WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATING);
+ /* Fall through. */
+
+ case PHJ_GROW_BUCKETS_REINSERTING:
+ /* Reinsert all tuples into the hash table. */
+ ExecParallelHashEnsureBatchAccessors(hashtable);
+ ExecParallelHashTableSetCurrentBatch(hashtable, 0);
+ while ((chunk = ExecParallelHashPopChunkQueue(hashtable, &chunk_s)))
+ {
+ size_t idx = 0;
+
+ while (idx < chunk->used)
+ {
+ HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + idx);
+ dsa_pointer shared = chunk_s + HASH_CHUNK_HEADER_SIZE + idx;
+ int bucketno;
+ int batchno;
+
+ ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue,
+ &bucketno, &batchno);
+ Assert(batchno == 0);
+
+ /* add the tuple to the proper bucket */
+ ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
+ hashTuple, shared);
+
+ /* advance index past the tuple */
+ idx += MAXALIGN(HJTUPLE_OVERHEAD +
+ HJTUPLE_MINTUPLE(hashTuple)->t_len);
+ }
+
+ /* allow this loop to be cancellable */
+ CHECK_FOR_INTERRUPTS();
+ }
+ BarrierArriveAndWait(&pstate->grow_buckets_barrier,
+ WAIT_EVENT_HASH_GROW_BUCKETS_REINSERTING);
+ }
+}
+
+/*
+ * ExecHashTableInsert
+ * insert a tuple into the hash table depending on the hash value
+ * it may just go to a temp file for later batches
+ *
+ * Note: the passed TupleTableSlot may contain a regular, minimal, or virtual
+ * tuple; the minimal case in particular is certain to happen while reloading
+ * tuples from batch files. We could save some cycles in the regular-tuple
+ * case by not forcing the slot contents into minimal form; not clear if it's
+ * worth the messiness required.
+ */
+static void
+ExecHashTableInsert(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue)
+{
+ bool shouldFree;
+ MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
+ int bucketno;
+ int batchno;
+
+ ExecHashGetBucketAndBatch(hashtable, hashvalue,
+ &bucketno, &batchno);
+
+ /*
+ * decide whether to put the tuple in the hash table or a temp file
+ */
+ if (batchno == hashtable->curbatch)
+ {
+ /*
+ * put the tuple in hash table
+ */
+ HashJoinTuple hashTuple;
+ int hashTupleSize;
+ double ntuples = (hashtable->totalTuples - hashtable->skewTuples);
+
+ /* Create the HashJoinTuple */
+ hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
+ hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize);
+
+ hashTuple->hashvalue = hashvalue;
+ memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
+
+ /*
+ * We always reset the tuple-matched flag on insertion. This is okay
+ * even when reloading a tuple from a batch file, since the tuple
+ * could not possibly have been matched to an outer tuple before it
+ * went into the batch file.
+ */
+ HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
+
+ /* Push it onto the front of the bucket's list */
+ hashTuple->next.unshared = hashtable->buckets.unshared[bucketno];
+ hashtable->buckets.unshared[bucketno] = hashTuple;
+
+ /*
+ * Increase the (optimal) number of buckets if we just exceeded the
+ * NTUP_PER_BUCKET threshold, but only when there's still a single
+ * batch.
+ */
+ if (hashtable->nbatch == 1 &&
+ ntuples > (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
+ {
+ /* Guard against integer overflow and alloc size overflow */
+ if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
+ hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
+ {
+ hashtable->nbuckets_optimal *= 2;
+ hashtable->log2_nbuckets_optimal += 1;
+ }
+ }
+
+ /* Account for space used, and back off if we've used too much */
+ hashtable->spaceUsed += hashTupleSize;
+ if (hashtable->spaceUsed > hashtable->spacePeak)
+ hashtable->spacePeak = hashtable->spaceUsed;
+ if (hashtable->spaceUsed +
+ hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
+ > hashtable->spaceAllowed)
+ ExecHashIncreaseNumBatches(hashtable);
+ }
+ else
+ {
+ /*
+ * put the tuple into a temp file for later batches
+ */
+ Assert(batchno > hashtable->curbatch);
+ ExecHashJoinSaveTuple(tuple,
+ hashvalue,
+ &hashtable->innerBatchFile[batchno]);
+ }
+
+ if (shouldFree)
+ heap_free_minimal_tuple(tuple);
+}
+
+/*
+ * ExecParallelHashTableInsert
+ * insert a tuple into a shared hash table or shared batch tuplestore
+ */
+static void
+ExecParallelHashTableInsert(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue)
+{
+ bool shouldFree;
+ MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
+ dsa_pointer shared;
+ int bucketno;
+ int batchno;
+
+retry:
+ ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
+
+ if (batchno == 0)
+ {
+ HashJoinTuple hashTuple;
+
+ /* Try to load it into memory. */
+ Assert(BarrierPhase(&hashtable->parallel_state->build_barrier) ==
+ PHJ_BUILD_HASHING_INNER);
+ hashTuple = ExecParallelHashTupleAlloc(hashtable,
+ HJTUPLE_OVERHEAD + tuple->t_len,
+ &shared);
+ if (hashTuple == NULL)
+ goto retry;
+
+ /* Store the hash value in the HashJoinTuple header. */
+ hashTuple->hashvalue = hashvalue;
+ memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
+
+ /* Push it onto the front of the bucket's list */
+ ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
+ hashTuple, shared);
+ }
+ else
+ {
+ size_t tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + tuple->t_len);
+
+ Assert(batchno > 0);
+
+ /* Try to preallocate space in the batch if necessary. */
+ if (hashtable->batches[batchno].preallocated < tuple_size)
+ {
+ if (!ExecParallelHashTuplePrealloc(hashtable, batchno, tuple_size))
+ goto retry;
+ }
+
+ Assert(hashtable->batches[batchno].preallocated >= tuple_size);
+ hashtable->batches[batchno].preallocated -= tuple_size;
+ sts_puttuple(hashtable->batches[batchno].inner_tuples, &hashvalue,
+ tuple);
+ }
+ ++hashtable->batches[batchno].ntuples;
+
+ if (shouldFree)
+ heap_free_minimal_tuple(tuple);
+}
+
+/*
+ * Insert a tuple into the current hash table. Unlike
+ * ExecParallelHashTableInsert, this version is not prepared to send the tuple
+ * to other batches or to run out of memory, and should only be called with
+ * tuples that belong in the current batch once growth has been disabled.
+ */
+static void
+ExecParallelHashTableInsertCurrentBatch(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue)
+{
+ bool shouldFree;
+ MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
+ HashJoinTuple hashTuple;
+ dsa_pointer shared;
+ int batchno;
+ int bucketno;
+
+ ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
+ Assert(batchno == hashtable->curbatch);
+ hashTuple = ExecParallelHashTupleAlloc(hashtable,
+ HJTUPLE_OVERHEAD + tuple->t_len,
+ &shared);
+ hashTuple->hashvalue = hashvalue;
+ memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
+ HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
+ ExecParallelHashPushTuple(&hashtable->buckets.shared[bucketno],
+ hashTuple, shared);
+
+ if (shouldFree)
+ heap_free_minimal_tuple(tuple);
+}
+
+/*
+ * ExecHashGetHashValue
+ * Compute the hash value for a tuple
+ *
+ * The tuple to be tested must be in econtext->ecxt_outertuple (thus Vars in
+ * the hashkeys expressions need to have OUTER_VAR as varno). If outer_tuple
+ * is false (meaning it's the HashJoin's inner node, Hash), econtext,
+ * hashkeys, and slot need to be from Hash, with hashkeys/slot referencing and
+ * being suitable for tuples from the node below the Hash. Conversely, if
+ * outer_tuple is true, econtext is from HashJoin, and hashkeys/slot need to
+ * be appropriate for tuples from HashJoin's outer node.
+ *
+ * A true result means the tuple's hash value has been successfully computed
+ * and stored at *hashvalue. A false result means the tuple cannot match
+ * because it contains a null attribute, and hence it should be discarded
+ * immediately. (If keep_nulls is true then false is never returned.)
+ */
+static bool
+ExecHashGetHashValue(HashJoinTable hashtable,
+ ExprContext *econtext,
+ List *hashkeys,
+ bool outer_tuple,
+ bool keep_nulls,
+ uint32 *hashvalue)
+{
+ uint32 hashkey = 0;
+ FmgrInfo *hashfunctions;
+ ListCell *hk;
+ int i = 0;
+ MemoryContext oldContext;
+
+ /*
+ * We reset the eval context each time to reclaim any memory leaked in the
+ * hashkey expressions.
+ */
+ ResetExprContext(econtext);
+
+ oldContext = MemoryContextSwitchTo(econtext->ecxt_per_tuple_memory);
+
+ if (outer_tuple)
+ hashfunctions = hashtable->outer_hashfunctions;
+ else
+ hashfunctions = hashtable->inner_hashfunctions;
+
+ foreach(hk, hashkeys)
+ {
+ ExprState *keyexpr = (ExprState *) lfirst(hk);
+ Datum keyval;
+ bool isNull;
+
+ /* rotate hashkey left 1 bit at each step */
+ hashkey = (hashkey << 1) | ((hashkey & 0x80000000) ? 1 : 0);
+
+ /*
+ * Get the join attribute value of the tuple
+ */
+ keyval = ExecEvalExpr(keyexpr, econtext, &isNull);
+
+ /*
+ * If the attribute is NULL, and the join operator is strict, then
+ * this tuple cannot pass the join qual so we can reject it
+ * immediately (unless we're scanning the outside of an outer join, in
+ * which case we must not reject it). Otherwise we act like the
+ * hashcode of NULL is zero (this will support operators that act like
+ * IS NOT DISTINCT, though not any more-random behavior). We treat
+ * the hash support function as strict even if the operator is not.
+ *
+ * Note: currently, all hashjoinable operators must be strict since
+ * the hash index AM assumes that. However, it takes so little extra
+ * code here to allow non-strict that we may as well do it.
+ */
+ if (isNull)
+ {
+ if (hashtable->hashStrict[i] && !keep_nulls)
+ {
+ MemoryContextSwitchTo(oldContext);
+ return false; /* cannot match */
+ }
+ /* else, leave hashkey unmodified, equivalent to hashcode 0 */
+ }
+ else
+ {
+ /* Compute the hash function */
+ uint32 hkey;
+
+ hkey = DatumGetUInt32(FunctionCall1Coll(&hashfunctions[i], hashtable->collations[i], keyval));
+ hashkey ^= hkey;
+ }
+
+ i++;
+ }
+
+ MemoryContextSwitchTo(oldContext);
+
+ *hashvalue = hashkey;
+ return true;
+}
+
+/*
+ * ExecHashGetBucketAndBatch
+ * Determine the bucket number and batch number for a hash value
+ *
+ * Note: on-the-fly increases of nbatch must not change the bucket number
+ * for a given hash code (since we don't move tuples to different hash
+ * chains), and must only cause the batch number to remain the same or
+ * increase. Our algorithm is
+ * bucketno = hashvalue MOD nbuckets
+ * batchno = (hashvalue DIV nbuckets) MOD nbatch
+ * where nbuckets and nbatch are both expected to be powers of 2, so we can
+ * do the computations by shifting and masking. (This assumes that all hash
+ * functions are good about randomizing all their output bits, else we are
+ * likely to have very skewed bucket or batch occupancy.)
+ *
+ * nbuckets and log2_nbuckets may change while nbatch == 1 because of dynamic
+ * bucket count growth. Once we start batching, the value is fixed and does
+ * not change over the course of the join (making it possible to compute batch
+ * number the way we do here).
+ *
+ * nbatch is always a power of 2; we increase it only by doubling it. This
+ * effectively adds one more bit to the top of the batchno.
+ */
+static void
+ExecHashGetBucketAndBatch(HashJoinTable hashtable,
+ uint32 hashvalue,
+ int *bucketno,
+ int *batchno)
+{
+ uint32 nbuckets = (uint32) hashtable->nbuckets;
+ uint32 nbatch = (uint32) hashtable->nbatch;
+
+ if (nbatch > 1)
+ {
+ /* we can do MOD by masking, DIV by shifting */
+ *bucketno = hashvalue & (nbuckets - 1);
+ *batchno = (hashvalue >> hashtable->log2_nbuckets) & (nbatch - 1);
+ }
+ else
+ {
+ *bucketno = hashvalue & (nbuckets - 1);
+ *batchno = 0;
+ }
+}
+
+/*
+ * ExecScanHashBucket
+ * scan a hash bucket for matches to the current outer tuple
+ *
+ * The current outer tuple must be stored in econtext->ecxt_outertuple.
+ *
+ * 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
+ExecScanHashBucket(HashJoinState *hjstate,
+ ExprContext *econtext)
+{
+ ExprState *hjclauses = hjstate->hashclauses;
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+ HashJoinTuple hashTuple = hjstate->hj_CurTuple;
+ uint32 hashvalue = hjstate->hj_CurHashValue;
+
+ /*
+ * 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 the tuple hashed to a skew bucket then scan the skew bucket
+ * otherwise scan the standard hashtable bucket.
+ */
+ if (hashTuple != NULL)
+ hashTuple = hashTuple->next.unshared;
+ else if (hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
+ hashTuple = hashtable->skewBucket[hjstate->hj_CurSkewBucketNo]->tuples;
+ else
+ hashTuple = hashtable->buckets.unshared[hjstate->hj_CurBucketNo];
+
+ while (hashTuple != NULL)
+ {
+ if (hashTuple->hashvalue == hashvalue)
+ {
+ TupleTableSlot *inntuple;
+
+ /* insert hashtable's tuple into exec slot so ExecQual sees it */
+ inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ hjstate->hj_HashTupleSlot,
+ false); /* do not pfree */
+ econtext->ecxt_innertuple = inntuple;
+
+ if (ExecQualAndReset(hjclauses, econtext))
+ {
+ hjstate->hj_CurTuple = hashTuple;
+ return true;
+ }
+ }
+
+ hashTuple = hashTuple->next.unshared;
+ }
+
+ /*
+ * no match
+ */
+ return false;
+}
+
+/*
+ * ExecParallelScanHashBucket
+ * scan a hash bucket for matches to the current outer tuple
+ *
+ * The current outer tuple must be stored in econtext->ecxt_outertuple.
+ *
+ * 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.
+ */
+static bool
+ExecParallelScanHashBucket(HashJoinState *hjstate,
+ ExprContext *econtext)
+{
+ ExprState *hjclauses = hjstate->hashclauses;
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+ HashJoinTuple hashTuple = hjstate->hj_CurTuple;
+ uint32 hashvalue = hjstate->hj_CurHashValue;
+
+ /*
+ * 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 = ExecParallelHashNextTuple(hashtable, hashTuple);
+ else
+ hashTuple = ExecParallelHashFirstTuple(hashtable,
+ hjstate->hj_CurBucketNo);
+
+ while (hashTuple != NULL)
+ {
+ if (hashTuple->hashvalue == hashvalue)
+ {
+ TupleTableSlot *inntuple;
+
+ /* insert hashtable's tuple into exec slot so ExecQual sees it */
+ inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ hjstate->hj_HashTupleSlot,
+ false); /* do not pfree */
+ econtext->ecxt_innertuple = inntuple;
+
+ if (ExecQualAndReset(hjclauses, econtext))
+ {
+ hjstate->hj_CurTuple = hashTuple;
+ return true;
+ }
+ }
+
+ hashTuple = ExecParallelHashNextTuple(hashtable, hashTuple);
+ }
+
+ /*
+ * no match
+ */
+ return false;
+}
+
+/*
+ * ExecPrepHashTableForUnmatched
+ * set up for a series of ExecScanHashTableForUnmatched calls
+ */
+static 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;
+ hjstate->hj_CurTuple = NULL;
+}
+
+/*
+ * ExecScanHashTableForUnmatched
+ * scan the hash table for unmatched inner tuples
+ *
+ * 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.
+ */
+static bool
+ExecScanHashTableForUnmatched(HashJoinState *hjstate, ExprContext *econtext)
+{
+ HashJoinTable hashtable = hjstate->hj_HashTable;
+ HashJoinTuple hashTuple = hjstate->hj_CurTuple;
+
+ for (;;)
+ {
+ /*
+ * 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)
+ {
+ int j = hashtable->skewBucketNums[hjstate->hj_CurSkewBucketNo];
+
+ hashTuple = hashtable->skewBucket[j]->tuples;
+ hjstate->hj_CurSkewBucketNo++;
+ }
+ else
+ break; /* finished all buckets */
+
+ while (hashTuple != NULL)
+ {
+ if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(hashTuple)))
+ {
+ TupleTableSlot *inntuple;
+
+ /* insert hashtable's tuple into exec slot */
+ inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(hashTuple),
+ hjstate->hj_HashTupleSlot,
+ false); /* do not pfree */
+ econtext->ecxt_innertuple = inntuple;
+
+ /*
+ * 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;
+ }
+
+ hashTuple = hashTuple->next.unshared;
+ }
+
+ /* allow this loop to be cancellable */
+ CHECK_FOR_INTERRUPTS();
+ }
+
+ /*
+ * no more unmatched tuples
+ */
+ return false;
+}
+
+/*
+ * ExecHashTableReset
+ *
+ * reset hash table header for new batch
+ */
+static void
+ExecHashTableReset(HashJoinTable hashtable)
+{
+ MemoryContext oldcxt;
+ int nbuckets = hashtable->nbuckets;
+
+ /*
+ * Release all the hash buckets and tuples acquired in the prior pass, and
+ * reinitialize the context for a new pass.
+ */
+ MemoryContextReset(hashtable->batchCxt);
+ oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);
+
+ /* Reallocate and reinitialize the hash bucket headers. */
+ hashtable->buckets.unshared = (HashJoinTuple *)
+ palloc0(nbuckets * sizeof(HashJoinTuple));
+
+ hashtable->spaceUsed = 0;
+
+ MemoryContextSwitchTo(oldcxt);
+
+ /* Forget the chunks (the memory was freed by the context reset above). */
+ hashtable->chunks = NULL;
+}
+
+/*
+ * ExecHashTableResetMatchFlags
+ * Clear all the HeapTupleHeaderHasMatch flags in the table
+ */
+static void
+ExecHashTableResetMatchFlags(HashJoinTable hashtable)
+{
+ HashJoinTuple tuple;
+ int i;
+
+ /* Reset all flags in the main table ... */
+ for (i = 0; i < hashtable->nbuckets; i++)
+ {
+ for (tuple = hashtable->buckets.unshared[i]; tuple != NULL;
+ tuple = tuple->next.unshared)
+ HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
+ }
+
+ /* ... and the same for the skew buckets, if any */
+ for (i = 0; i < hashtable->nSkewBuckets; i++)
+ {
+ int j = hashtable->skewBucketNums[i];
+ HashSkewBucket *skewBucket = hashtable->skewBucket[j];
+
+ for (tuple = skewBucket->tuples; tuple != NULL; tuple = tuple->next.unshared)
+ HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(tuple));
+ }
+}
+
+/*
+ * ExecHashBuildSkewHash
+ *
+ * Set up for skew optimization if we can identify the most common values
+ * (MCVs) of the outer relation's join key. We make a skew hash bucket
+ * for the hash value of each MCV, up to the number of slots allowed
+ * based on available memory.
+ */
+static void
+ExecHashBuildSkewHash(HashJoinTable hashtable, HashJoin *node, int mcvsToUse)
+{
+ HeapTupleData *statsTuple;
+ AttStatsSlot sslot;
+
+ /* Do nothing if planner didn't identify the outer relation's join key */
+ if (!OidIsValid(node->skewTable))
+ return;
+ /* Also, do nothing if we don't have room for at least one skew bucket */
+ if (mcvsToUse <= 0)
+ return;
+
+ /*
+ * Try to find the MCV statistics for the outer relation's join key.
+ */
+ statsTuple = SearchSysCache3(STATRELATTINH,
+ ObjectIdGetDatum(node->skewTable),
+ Int16GetDatum(node->skewColumn),
+ BoolGetDatum(node->skewInherit));
+ if (!HeapTupleIsValid(statsTuple))
+ return;
+
+ if (get_attstatsslot(&sslot, statsTuple,
+ STATISTIC_KIND_MCV, InvalidOid,
+ ATTSTATSSLOT_VALUES | ATTSTATSSLOT_NUMBERS))
+ {
+ double frac;
+ int nbuckets;
+ FmgrInfo *hashfunctions;
+ int i;
+
+ if (mcvsToUse > sslot.nvalues)
+ mcvsToUse = sslot.nvalues;
+
+ /*
+ * Calculate the expected fraction of outer relation that will
+ * participate in the skew optimization. If this isn't at least
+ * SKEW_MIN_OUTER_FRACTION, don't use skew optimization.
+ */
+ frac = 0;
+ for (i = 0; i < mcvsToUse; i++)
+ frac += sslot.numbers[i];
+ if (frac < SKEW_MIN_OUTER_FRACTION)
+ {
+ free_attstatsslot(&sslot);
+ ReleaseSysCache(statsTuple);
+ return;
+ }
+
+ /*
+ * Okay, set up the skew hashtable.
+ *
+ * skewBucket[] is an open addressing hashtable with a power of 2 size
+ * that is greater than the number of MCV values. (This ensures there
+ * will be at least one null entry, so searches will always
+ * terminate.)
+ *
+ * Note: this code could fail if mcvsToUse exceeds INT_MAX/8 or
+ * MaxAllocSize/sizeof(void *)/8, but that is not currently possible
+ * since we limit pg_statistic entries to much less than that.
+ */
+ nbuckets = 2;
+ while (nbuckets <= mcvsToUse)
+ nbuckets <<= 1;
+ /* use two more bits just to help avoid collisions */
+ nbuckets <<= 2;
+
+ hashtable->skewEnabled = true;
+ hashtable->skewBucketLen = nbuckets;
+
+ /*
+ * We allocate the bucket memory in the hashtable's batch context. It
+ * is only needed during the first batch, and this ensures it will be
+ * automatically removed once the first batch is done.
+ */
+ hashtable->skewBucket = (HashSkewBucket **)
+ MemoryContextAllocZero(hashtable->batchCxt,
+ nbuckets * sizeof(HashSkewBucket *));
+ hashtable->skewBucketNums = (int *)
+ MemoryContextAllocZero(hashtable->batchCxt,
+ mcvsToUse * sizeof(int));
+
+ hashtable->spaceUsed += nbuckets * sizeof(HashSkewBucket *)
+ + mcvsToUse * sizeof(int);
+ hashtable->spaceUsedSkew += nbuckets * sizeof(HashSkewBucket *)
+ + mcvsToUse * sizeof(int);
+ if (hashtable->spaceUsed > hashtable->spacePeak)
+ hashtable->spacePeak = hashtable->spaceUsed;
+
+ /*
+ * Create a skew bucket for each MCV hash value.
+ *
+ * Note: it is very important that we create the buckets in order of
+ * decreasing MCV frequency. If we have to remove some buckets, they
+ * must be removed in reverse order of creation (see notes in
+ * ExecHashRemoveNextSkewBucket) and we want the least common MCVs to
+ * be removed first.
+ */
+ hashfunctions = hashtable->outer_hashfunctions;
+
+ for (i = 0; i < mcvsToUse; i++)
+ {
+ uint32 hashvalue;
+ int bucket;
+
+ hashvalue = DatumGetUInt32(FunctionCall1Coll(&hashfunctions[0],
+ hashtable->collations[0],
+ sslot.values[i]));
+
+ /*
+ * While we have not hit a hole in the hashtable and have not hit
+ * the desired bucket, we have collided with some previous hash
+ * value, so try the next bucket location. NB: this code must
+ * match ExecHashGetSkewBucket.
+ */
+ bucket = hashvalue & (nbuckets - 1);
+ while (hashtable->skewBucket[bucket] != NULL &&
+ hashtable->skewBucket[bucket]->hashvalue != hashvalue)
+ bucket = (bucket + 1) & (nbuckets - 1);
+
+ /*
+ * If we found an existing bucket with the same hashvalue, leave
+ * it alone. It's okay for two MCVs to share a hashvalue.
+ */
+ if (hashtable->skewBucket[bucket] != NULL)
+ continue;
+
+ /* Okay, create a new skew bucket for this hashvalue. */
+ hashtable->skewBucket[bucket] = (HashSkewBucket *)
+ MemoryContextAlloc(hashtable->batchCxt,
+ sizeof(HashSkewBucket));
+ hashtable->skewBucket[bucket]->hashvalue = hashvalue;
+ hashtable->skewBucket[bucket]->tuples = NULL;
+ hashtable->skewBucketNums[hashtable->nSkewBuckets] = bucket;
+ hashtable->nSkewBuckets++;
+ hashtable->spaceUsed += SKEW_BUCKET_OVERHEAD;
+ hashtable->spaceUsedSkew += SKEW_BUCKET_OVERHEAD;
+ if (hashtable->spaceUsed > hashtable->spacePeak)
+ hashtable->spacePeak = hashtable->spaceUsed;
+ }
+
+ free_attstatsslot(&sslot);
+ }
+
+ ReleaseSysCache(statsTuple);
+}
+
+/*
+ * ExecHashGetSkewBucket
+ *
+ * Returns the index of the skew bucket for this hashvalue,
+ * or INVALID_SKEW_BUCKET_NO if the hashvalue is not
+ * associated with any active skew bucket.
+ */
+static int
+ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue)
+{
+ int bucket;
+
+ /*
+ * Always return INVALID_SKEW_BUCKET_NO if not doing skew optimization (in
+ * particular, this happens after the initial batch is done).
+ */
+ if (!hashtable->skewEnabled)
+ return INVALID_SKEW_BUCKET_NO;
+
+ /*
+ * Since skewBucketLen is a power of 2, we can do a modulo by ANDing.
+ */
+ bucket = hashvalue & (hashtable->skewBucketLen - 1);
+
+ /*
+ * While we have not hit a hole in the hashtable and have not hit the
+ * desired bucket, we have collided with some other hash value, so try the
+ * next bucket location.
+ */
+ while (hashtable->skewBucket[bucket] != NULL &&
+ hashtable->skewBucket[bucket]->hashvalue != hashvalue)
+ bucket = (bucket + 1) & (hashtable->skewBucketLen - 1);
+
+ /*
+ * Found the desired bucket?
+ */
+ if (hashtable->skewBucket[bucket] != NULL)
+ return bucket;
+
+ /*
+ * There must not be any hashtable entry for this hash value.
+ */
+ return INVALID_SKEW_BUCKET_NO;
+}
+
+/*
+ * ExecHashSkewTableInsert
+ *
+ * Insert a tuple into the skew hashtable.
+ *
+ * This should generally match up with the current-batch case in
+ * ExecHashTableInsert.
+ */
+static void
+ExecHashSkewTableInsert(HashJoinTable hashtable,
+ TupleTableSlot *slot,
+ uint32 hashvalue,
+ int bucketNumber)
+{
+ bool shouldFree;
+ MinimalTuple tuple = ExecFetchSlotMinimalTuple(slot, &shouldFree);
+ HashJoinTuple hashTuple;
+ int hashTupleSize;
+
+ /* Create the HashJoinTuple */
+ hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
+ hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
+ hashTupleSize);
+ hashTuple->hashvalue = hashvalue;
+ memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len);
+ HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(hashTuple));
+
+ /* Push it onto the front of the skew bucket's list */
+ hashTuple->next.unshared = hashtable->skewBucket[bucketNumber]->tuples;
+ hashtable->skewBucket[bucketNumber]->tuples = hashTuple;
+ Assert(hashTuple != hashTuple->next.unshared);
+
+ /* Account for space used, and back off if we've used too much */
+ hashtable->spaceUsed += hashTupleSize;
+ hashtable->spaceUsedSkew += hashTupleSize;
+ if (hashtable->spaceUsed > hashtable->spacePeak)
+ hashtable->spacePeak = hashtable->spaceUsed;
+ while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
+ ExecHashRemoveNextSkewBucket(hashtable);
+
+ /* Check we are not over the total spaceAllowed, either */
+ if (hashtable->spaceUsed > hashtable->spaceAllowed)
+ ExecHashIncreaseNumBatches(hashtable);
+
+ if (shouldFree)
+ heap_free_minimal_tuple(tuple);
+}
+
+/*
+ * ExecHashRemoveNextSkewBucket
+ *
+ * Remove the least valuable skew bucket by pushing its tuples into
+ * the main hash table.
+ */
+static void
+ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
+{
+ int bucketToRemove;
+ HashSkewBucket *bucket;
+ uint32 hashvalue;
+ int bucketno;
+ int batchno;
+ HashJoinTuple hashTuple;
+
+ /* Locate the bucket to remove */
+ bucketToRemove = hashtable->skewBucketNums[hashtable->nSkewBuckets - 1];
+ bucket = hashtable->skewBucket[bucketToRemove];
+
+ /*
+ * Calculate which bucket and batch the tuples belong to in the main
+ * hashtable. They all have the same hash value, so it's the same for all
+ * of them. Also note that it's not possible for nbatch to increase while
+ * we are processing the tuples.
+ */
+ hashvalue = bucket->hashvalue;
+ ExecHashGetBucketAndBatch(hashtable, hashvalue, &bucketno, &batchno);
+
+ /* Process all tuples in the bucket */
+ hashTuple = bucket->tuples;
+ while (hashTuple != NULL)
+ {
+ HashJoinTuple nextHashTuple = hashTuple->next.unshared;
+ MinimalTuple tuple;
+ Size tupleSize;
+
+ /*
+ * This code must agree with ExecHashTableInsert. We do not use
+ * ExecHashTableInsert directly as ExecHashTableInsert expects a
+ * TupleTableSlot while we already have HashJoinTuples.
+ */
+ tuple = HJTUPLE_MINTUPLE(hashTuple);
+ tupleSize = HJTUPLE_OVERHEAD + tuple->t_len;
+
+ /* Decide whether to put the tuple in the hash table or a temp file */
+ if (batchno == hashtable->curbatch)
+ {
+ /* Move the tuple to the main hash table */
+ HashJoinTuple copyTuple;
+
+ /*
+ * We must copy the tuple into the dense storage, else it will not
+ * be found by, eg, ExecHashIncreaseNumBatches.
+ */
+ copyTuple = (HashJoinTuple) dense_alloc(hashtable, tupleSize);
+ memcpy(copyTuple, hashTuple, tupleSize);
+ pfree(hashTuple);
+
+ copyTuple->next.unshared = hashtable->buckets.unshared[bucketno];
+ hashtable->buckets.unshared[bucketno] = copyTuple;
+
+ /* We have reduced skew space, but overall space doesn't change */
+ hashtable->spaceUsedSkew -= tupleSize;
+ }
+ else
+ {
+ /* Put the tuple into a temp file for later batches */
+ Assert(batchno > hashtable->curbatch);
+ ExecHashJoinSaveTuple(tuple, hashvalue,
+ &hashtable->innerBatchFile[batchno]);
+ pfree(hashTuple);
+ hashtable->spaceUsed -= tupleSize;
+ hashtable->spaceUsedSkew -= tupleSize;
+ }
+
+ hashTuple = nextHashTuple;
+
+ /* allow this loop to be cancellable */
+ CHECK_FOR_INTERRUPTS();
+ }
+
+ /*
+ * Free the bucket struct itself and reset the hashtable entry to NULL.
+ *
+ * NOTE: this is not nearly as simple as it looks on the surface, because
+ * of the possibility of collisions in the hashtable. Suppose that hash
+ * values A and B collide at a particular hashtable entry, and that A was
+ * entered first so B gets shifted to a different table entry. If we were
+ * to remove A first then ExecHashGetSkewBucket would mistakenly start
+ * reporting that B is not in the hashtable, because it would hit the NULL
+ * before finding B. However, we always remove entries in the reverse
+ * order of creation, so this failure cannot happen.
+ */
+ hashtable->skewBucket[bucketToRemove] = NULL;
+ hashtable->nSkewBuckets--;
+ pfree(bucket);
+ hashtable->spaceUsed -= SKEW_BUCKET_OVERHEAD;
+ hashtable->spaceUsedSkew -= SKEW_BUCKET_OVERHEAD;
+
+ /*
+ * If we have removed all skew buckets then give up on skew optimization.
+ * Release the arrays since they aren't useful any more.
+ */
+ if (hashtable->nSkewBuckets == 0)
+ {
+ hashtable->skewEnabled = false;
+ pfree(hashtable->skewBucket);
+ pfree(hashtable->skewBucketNums);
+ hashtable->skewBucket = NULL;
+ hashtable->skewBucketNums = NULL;
+ hashtable->spaceUsed -= hashtable->spaceUsedSkew;
+ hashtable->spaceUsedSkew = 0;
+ }
+}
+
+/*
+ * Allocate 'size' bytes from the currently active HashMemoryChunk
+ */
+static void *
+dense_alloc(HashJoinTable hashtable, Size size)
+{
+ HashMemoryChunk newChunk;
+ char *ptr;
+
+ /* just in case the size is not already aligned properly */
+ size = MAXALIGN(size);
+
+ /*
+ * If tuple size is larger than threshold, allocate a separate chunk.
+ */
+ if (size > HASH_CHUNK_THRESHOLD)
+ {
+ /* allocate new chunk and put it at the beginning of the list */
+ newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
+ HASH_CHUNK_HEADER_SIZE + size);
+ newChunk->maxlen = size;
+ newChunk->used = size;
+ newChunk->ntuples = 1;
+
+ /*
+ * Add this chunk to the list after the first existing chunk, so that
+ * we don't lose the remaining space in the "current" chunk.
+ */
+ if (hashtable->chunks != NULL)
+ {
+ newChunk->next = hashtable->chunks->next;
+ hashtable->chunks->next.unshared = newChunk;
+ }
+ else
+ {
+ newChunk->next.unshared = hashtable->chunks;
+ hashtable->chunks = newChunk;
+ }
+
+ return HASH_CHUNK_DATA(newChunk);
+ }
+
+ /*
+ * See if we have enough space for it in the current chunk (if any). If
+ * not, allocate a fresh chunk.
+ */
+ if ((hashtable->chunks == NULL) ||
+ (hashtable->chunks->maxlen - hashtable->chunks->used) < size)
+ {
+ /* allocate new chunk and put it at the beginning of the list */
+ newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
+ HASH_CHUNK_HEADER_SIZE + HASH_CHUNK_SIZE);
+
+ newChunk->maxlen = HASH_CHUNK_SIZE;
+ newChunk->used = size;
+ newChunk->ntuples = 1;
+
+ newChunk->next.unshared = hashtable->chunks;
+ hashtable->chunks = newChunk;
+
+ return HASH_CHUNK_DATA(newChunk);
+ }
+
+ /* There is enough space in the current chunk, let's add the tuple */
+ ptr = HASH_CHUNK_DATA(hashtable->chunks) + hashtable->chunks->used;
+ hashtable->chunks->used += size;
+ hashtable->chunks->ntuples += 1;
+
+ /* return pointer to the start of the tuple memory */
+ return ptr;
+}
+
+/*
+ * Allocate space for a tuple in shared dense storage. This is equivalent to
+ * dense_alloc but for Parallel Hash using shared memory.
+ *
+ * While loading a tuple into shared memory, we might run out of memory and
+ * decide to repartition, or determine that the load factor is too high and
+ * decide to expand the bucket array, or discover that another participant has
+ * commanded us to help do that. Return NULL if number of buckets or batches
+ * has changed, indicating that the caller must retry (considering the
+ * possibility that the tuple no longer belongs in the same batch).
+ */
+static HashJoinTuple
+ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t size,
+ dsa_pointer *shared)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ dsa_pointer chunk_shared;
+ HashMemoryChunk chunk;
+ Size chunk_size;
+ HashJoinTuple result;
+ int curbatch = hashtable->curbatch;
+
+ size = MAXALIGN(size);
+
+ /*
+ * Fast path: if there is enough space in this backend's current chunk,
+ * then we can allocate without any locking.
+ */
+ chunk = hashtable->current_chunk;
+ if (chunk != NULL &&
+ size <= HASH_CHUNK_THRESHOLD &&
+ chunk->maxlen - chunk->used >= size)
+ {
+
+ chunk_shared = hashtable->current_chunk_shared;
+ Assert(chunk == dsa_get_address(hashtable->area, chunk_shared));
+ *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE + chunk->used;
+ result = (HashJoinTuple) (HASH_CHUNK_DATA(chunk) + chunk->used);
+ chunk->used += size;
+
+ Assert(chunk->used <= chunk->maxlen);
+ Assert(result == dsa_get_address(hashtable->area, *shared));
+
+ return result;
+ }
+
+ /* Slow path: try to allocate a new chunk. */
+ LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
+
+ /*
+ * Check if we need to help increase the number of buckets or batches.
+ */
+ if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
+ pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
+ {
+ ParallelHashGrowth growth = pstate->growth;
+
+ hashtable->current_chunk = NULL;
+ LWLockRelease(&pstate->lock);
+
+ /* Another participant has commanded us to help grow. */
+ if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
+ ExecParallelHashIncreaseNumBatches(hashtable);
+ else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
+ ExecParallelHashIncreaseNumBuckets(hashtable);
+
+ /* The caller must retry. */
+ return NULL;
+ }
+
+ /* Oversized tuples get their own chunk. */
+ if (size > HASH_CHUNK_THRESHOLD)
+ chunk_size = size + HASH_CHUNK_HEADER_SIZE;
+ else
+ chunk_size = HASH_CHUNK_SIZE;
+
+ /* Check if it's time to grow batches or buckets. */
+ if (pstate->growth != PHJ_GROWTH_DISABLED)
+ {
+ Assert(curbatch == 0);
+ Assert(BarrierPhase(&pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);
+
+ /*
+ * Check if our space limit would be exceeded. To avoid choking on
+ * very large tuples or very low work_mem setting, we'll always allow
+ * each backend to allocate at least one chunk.
+ */
+ if (hashtable->batches[0].at_least_one_chunk &&
+ hashtable->batches[0].shared->size +
+ chunk_size > pstate->space_allowed)
+ {
+ pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
+ hashtable->batches[0].shared->space_exhausted = true;
+ LWLockRelease(&pstate->lock);
+
+ return NULL;
+ }
+
+ /* Check if our load factor limit would be exceeded. */
+ if (hashtable->nbatch == 1)
+ {
+ hashtable->batches[0].shared->ntuples += hashtable->batches[0].ntuples;
+ hashtable->batches[0].ntuples = 0;
+ /* Guard against integer overflow and alloc size overflow */
+ if (hashtable->batches[0].shared->ntuples + 1 >
+ hashtable->nbuckets * NTUP_PER_BUCKET &&
+ hashtable->nbuckets < (INT_MAX / 2) &&
+ hashtable->nbuckets * 2 <=
+ MaxAllocSize / sizeof(dsa_pointer_atomic))
+ {
+ pstate->growth = PHJ_GROWTH_NEED_MORE_BUCKETS;
+ LWLockRelease(&pstate->lock);
+
+ return NULL;
+ }
+ }
+ }
+
+ /* We are cleared to allocate a new chunk. */
+ chunk_shared = dsa_allocate(hashtable->area, chunk_size);
+ hashtable->batches[curbatch].shared->size += chunk_size;
+ hashtable->batches[curbatch].at_least_one_chunk = true;
+
+ /* Set up the chunk. */
+ chunk = (HashMemoryChunk) dsa_get_address(hashtable->area, chunk_shared);
+ *shared = chunk_shared + HASH_CHUNK_HEADER_SIZE;
+ chunk->maxlen = chunk_size - HASH_CHUNK_HEADER_SIZE;
+ chunk->used = size;
+
+ /*
+ * Push it onto the list of chunks, so that it can be found if we need to
+ * increase the number of buckets or batches (batch 0 only) and later for
+ * freeing the memory (all batches).
+ */
+ chunk->next.shared = hashtable->batches[curbatch].shared->chunks;
+ hashtable->batches[curbatch].shared->chunks = chunk_shared;
+
+ if (size <= HASH_CHUNK_THRESHOLD)
+ {
+ /*
+ * Make this the current chunk so that we can use the fast path to
+ * fill the rest of it up in future calls.
+ */
+ hashtable->current_chunk = chunk;
+ hashtable->current_chunk_shared = chunk_shared;
+ }
+ LWLockRelease(&pstate->lock);
+
+ Assert(HASH_CHUNK_DATA(chunk) == dsa_get_address(hashtable->area, *shared));
+ result = (HashJoinTuple) HASH_CHUNK_DATA(chunk);
+
+ return result;
+}
+
+/*
+ * One backend needs to set up the shared batch state including tuplestores.
+ * Other backends will ensure they have correctly configured accessors by
+ * called ExecParallelHashEnsureBatchAccessors().
+ */
+static void
+ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ ParallelHashJoinBatch *batches;
+ MemoryContext oldcxt;
+ int i;
+
+ Assert(hashtable->batches == NULL);
+
+ /* Allocate space. */
+ pstate->batches =
+ dsa_allocate0(hashtable->area,
+ EstimateParallelHashJoinBatch(hashtable) * nbatch);
+ pstate->nbatch = nbatch;
+ batches = dsa_get_address(hashtable->area, pstate->batches);
+
+ /* Use hash join memory context. */
+ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
+
+ /* Allocate this backend's accessor array. */
+ hashtable->nbatch = nbatch;
+ hashtable->batches = (ParallelHashJoinBatchAccessor *)
+ palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
+
+ /* Set up the shared state, tuplestores and backend-local accessors. */
+ for (i = 0; i < hashtable->nbatch; ++i)
+ {
+ ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
+ ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
+ char name[MAXPGPATH];
+
+ /*
+ * All members of shared were zero-initialized. We just need to set
+ * up the Barrier.
+ */
+ BarrierInit(&shared->batch_barrier, 0);
+ if (i == 0)
+ {
+ /* Batch 0 doesn't need to be loaded. */
+ BarrierAttach(&shared->batch_barrier);
+ while (BarrierPhase(&shared->batch_barrier) < PHJ_BATCH_PROBING)
+ BarrierArriveAndWait(&shared->batch_barrier, 0);
+ BarrierDetach(&shared->batch_barrier);
+ }
+
+ /* Initialize accessor state. All members were zero-initialized. */
+ accessor->shared = shared;
+
+ /* Initialize the shared tuplestores. */
+ snprintf(name, sizeof(name), "i%dof%d", i, hashtable->nbatch);
+ accessor->inner_tuples =
+ sts_initialize(ParallelHashJoinBatchInner(shared),
+ pstate->nparticipants,
+ ParallelWorkerNumber + 1,
+ sizeof(uint32),
+ SHARED_TUPLESTORE_SINGLE_PASS,
+ &pstate->fileset,
+ name);
+ snprintf(name, sizeof(name), "o%dof%d", i, hashtable->nbatch);
+ accessor->outer_tuples =
+ sts_initialize(ParallelHashJoinBatchOuter(shared,
+ pstate->nparticipants),
+ pstate->nparticipants,
+ ParallelWorkerNumber + 1,
+ sizeof(uint32),
+ SHARED_TUPLESTORE_SINGLE_PASS,
+ &pstate->fileset,
+ name);
+ }
+
+ MemoryContextSwitchTo(oldcxt);
+}
+
+/*
+ * Free the current set of ParallelHashJoinBatchAccessor objects.
+ */
+static void
+ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
+{
+ int i;
+
+ for (i = 0; i < hashtable->nbatch; ++i)
+ {
+ /* Make sure no files are left open. */
+ sts_end_write(hashtable->batches[i].inner_tuples);
+ sts_end_write(hashtable->batches[i].outer_tuples);
+ sts_end_parallel_scan(hashtable->batches[i].inner_tuples);
+ sts_end_parallel_scan(hashtable->batches[i].outer_tuples);
+ }
+ pfree(hashtable->batches);
+ hashtable->batches = NULL;
+}
+
+/*
+ * Make sure this backend has up-to-date accessors for the current set of
+ * batches.
+ */
+static void
+ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ ParallelHashJoinBatch *batches;
+ MemoryContext oldcxt;
+ int i;
+
+ if (hashtable->batches != NULL)
+ {
+ if (hashtable->nbatch == pstate->nbatch)
+ return;
+ ExecParallelHashCloseBatchAccessors(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.
+ */
+ if (!DsaPointerIsValid(pstate->batches))
+ return;
+
+ /* Use hash join memory context. */
+ oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
+
+ /* Allocate this backend's accessor array. */
+ hashtable->nbatch = pstate->nbatch;
+ hashtable->batches = (ParallelHashJoinBatchAccessor *)
+ palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);
+
+ /* Find the base of the pseudo-array of ParallelHashJoinBatch objects. */
+ batches = (ParallelHashJoinBatch *)
+ dsa_get_address(hashtable->area, pstate->batches);
+
+ /* Set up the accessor array and attach to the tuplestores. */
+ for (i = 0; i < hashtable->nbatch; ++i)
+ {
+ ParallelHashJoinBatchAccessor *accessor = &hashtable->batches[i];
+ ParallelHashJoinBatch *shared = NthParallelHashJoinBatch(batches, i);
+
+ accessor->shared = shared;
+ accessor->preallocated = 0;
+ accessor->done = false;
+ accessor->inner_tuples =
+ sts_attach(ParallelHashJoinBatchInner(shared),
+ ParallelWorkerNumber + 1,
+ &pstate->fileset);
+ accessor->outer_tuples =
+ sts_attach(ParallelHashJoinBatchOuter(shared,
+ pstate->nparticipants),
+ ParallelWorkerNumber + 1,
+ &pstate->fileset);
+ }
+
+ MemoryContextSwitchTo(oldcxt);
+}
+
+/*
+ * Allocate an empty shared memory hash table for a given batch.
+ */
+static void
+ExecParallelHashTableAlloc(HashJoinTable hashtable, int batchno)
+{
+ ParallelHashJoinBatch *batch = hashtable->batches[batchno].shared;
+ dsa_pointer_atomic *buckets;
+ int nbuckets = hashtable->parallel_state->nbuckets;
+ int i;
+
+ batch->buckets =
+ dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * nbuckets);
+ buckets = (dsa_pointer_atomic *)
+ dsa_get_address(hashtable->area, batch->buckets);
+ for (i = 0; i < nbuckets; ++i)
+ dsa_pointer_atomic_init(&buckets[i], InvalidDsaPointer);
+}
+
+/*
+ * If we are currently attached to a shared hash join batch, detach. If we
+ * are last to detach, clean up.
+ */
+static void
+ExecHashTableDetachBatch(HashJoinTable hashtable)
+{
+ if (hashtable->parallel_state != NULL &&
+ hashtable->curbatch >= 0)
+ {
+ int curbatch = hashtable->curbatch;
+ ParallelHashJoinBatch *batch = hashtable->batches[curbatch].shared;
+
+ /* 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))
+ {
+ /*
+ * 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_DONE);
+
+ /* Free shared chunks and buckets. */
+ while (DsaPointerIsValid(batch->chunks))
+ {
+ HashMemoryChunk chunk =
+ dsa_get_address(hashtable->area, batch->chunks);
+ dsa_pointer next = chunk->next.shared;
+
+ dsa_free(hashtable->area, batch->chunks);
+ batch->chunks = next;
+ }
+ if (DsaPointerIsValid(batch->buckets))
+ {
+ dsa_free(hashtable->area, batch->buckets);
+ batch->buckets = InvalidDsaPointer;
+ }
+ }
+
+ /*
+ * 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);
+
+ /* Remember that we are not attached to a batch. */
+ hashtable->curbatch = -1;
+ }
+}
+
+/*
+ * Detach from all shared resources. If we are last to detach, clean up.
+ */
+void
+ExecHashTableDetach(HashJoinTable hashtable)
+{
+ if (hashtable->parallel_state)
+ {
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ int i;
+
+ /* Make sure any temporary files are closed. */
+ if (hashtable->batches)
+ {
+ for (i = 0; i < hashtable->nbatch; ++i)
+ {
+ sts_end_write(hashtable->batches[i].inner_tuples);
+ sts_end_write(hashtable->batches[i].outer_tuples);
+ sts_end_parallel_scan(hashtable->batches[i].inner_tuples);
+ sts_end_parallel_scan(hashtable->batches[i].outer_tuples);
+ }
+ }
+
+ /* If we're last to detach, clean up shared memory. */
+ if (BarrierDetach(&pstate->build_barrier))
+ {
+ if (DsaPointerIsValid(pstate->batches))
+ {
+ dsa_free(hashtable->area, pstate->batches);
+ pstate->batches = InvalidDsaPointer;
+ }
+ }
+
+ hashtable->parallel_state = NULL;
+ }
+}
+
+/*
+ * Get the first tuple in a given bucket identified by number.
+ */
+static inline HashJoinTuple
+ExecParallelHashFirstTuple(HashJoinTable hashtable, int bucketno)
+{
+ HashJoinTuple tuple;
+ dsa_pointer p;
+
+ Assert(hashtable->parallel_state);
+ p = dsa_pointer_atomic_read(&hashtable->buckets.shared[bucketno]);
+ tuple = (HashJoinTuple) dsa_get_address(hashtable->area, p);
+
+ return tuple;
+}
+
+/*
+ * Get the next tuple in the same bucket as 'tuple'.
+ */
+static inline HashJoinTuple
+ExecParallelHashNextTuple(HashJoinTable hashtable, HashJoinTuple tuple)
+{
+ HashJoinTuple next;
+
+ Assert(hashtable->parallel_state);
+ next = (HashJoinTuple) dsa_get_address(hashtable->area, tuple->next.shared);
+
+ return next;
+}
+
+/*
+ * Insert a tuple at the front of a chain of tuples in DSA memory atomically.
+ */
+static inline void
+ExecParallelHashPushTuple(dsa_pointer_atomic *head,
+ HashJoinTuple tuple,
+ dsa_pointer tuple_shared)
+{
+ for (;;)
+ {
+ tuple->next.shared = dsa_pointer_atomic_read(head);
+ if (dsa_pointer_atomic_compare_exchange(head,
+ &tuple->next.shared,
+ tuple_shared))
+ break;
+ }
+}
+
+/*
+ * Prepare to work on a given batch.
+ */
+static void
+ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno)
+{
+ Assert(hashtable->batches[batchno].shared->buckets != InvalidDsaPointer);
+
+ hashtable->curbatch = batchno;
+ hashtable->buckets.shared = (dsa_pointer_atomic *)
+ dsa_get_address(hashtable->area,
+ hashtable->batches[batchno].shared->buckets);
+ hashtable->nbuckets = hashtable->parallel_state->nbuckets;
+ hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
+ hashtable->current_chunk = NULL;
+ hashtable->current_chunk_shared = InvalidDsaPointer;
+ hashtable->batches[batchno].at_least_one_chunk = false;
+}
+
+/*
+ * Take the next available chunk from the queue of chunks being worked on in
+ * parallel. Return NULL if there are none left. Otherwise return a pointer
+ * to the chunk, and set *shared to the DSA pointer to the chunk.
+ */
+static HashMemoryChunk
+ExecParallelHashPopChunkQueue(HashJoinTable hashtable, dsa_pointer *shared)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ HashMemoryChunk chunk;
+
+ LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
+ if (DsaPointerIsValid(pstate->chunk_work_queue))
+ {
+ *shared = pstate->chunk_work_queue;
+ chunk = (HashMemoryChunk)
+ dsa_get_address(hashtable->area, *shared);
+ pstate->chunk_work_queue = chunk->next.shared;
+ }
+ else
+ chunk = NULL;
+ LWLockRelease(&pstate->lock);
+
+ return chunk;
+}
+
+/*
+ * Increase the space preallocated in this backend for a given inner batch by
+ * at least a given amount. This allows us to track whether a given batch
+ * would fit in memory when loaded back in. Also increase the number of
+ * batches or buckets if required.
+ *
+ * This maintains a running estimation of how much space will be taken when we
+ * load the batch back into memory by simulating the way chunks will be handed
+ * out to workers. It's not perfectly accurate because the tuples will be
+ * packed into memory chunks differently by ExecParallelHashTupleAlloc(), but
+ * it should be pretty close. It tends to overestimate by a fraction of a
+ * chunk per worker since all workers gang up to preallocate during hashing,
+ * but workers tend to reload batches alone if there are enough to go around,
+ * leaving fewer partially filled chunks. This effect is bounded by
+ * nparticipants.
+ *
+ * Return false if the number of batches or buckets has changed, and the
+ * caller should reconsider which batch a given tuple now belongs in and call
+ * again.
+ */
+static bool
+ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int batchno, size_t size)
+{
+ ParallelHashJoinState *pstate = hashtable->parallel_state;
+ ParallelHashJoinBatchAccessor *batch = &hashtable->batches[batchno];
+ size_t want = Max(size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);
+
+ Assert(batchno > 0);
+ Assert(batchno < hashtable->nbatch);
+ Assert(size == MAXALIGN(size));
+
+ LWLockAcquire(&pstate->lock, LW_EXCLUSIVE);
+
+ /* Has another participant commanded us to help grow? */
+ if (pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
+ pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
+ {
+ ParallelHashGrowth growth = pstate->growth;
+
+ LWLockRelease(&pstate->lock);
+ if (growth == PHJ_GROWTH_NEED_MORE_BATCHES)
+ ExecParallelHashIncreaseNumBatches(hashtable);
+ else if (growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
+ ExecParallelHashIncreaseNumBuckets(hashtable);
+
+ return false;
+ }
+
+ if (pstate->growth != PHJ_GROWTH_DISABLED &&
+ batch->at_least_one_chunk &&
+ (batch->shared->estimated_size + want + HASH_CHUNK_HEADER_SIZE
+ > pstate->space_allowed))
+ {
+ /*
+ * We have determined that this batch would exceed the space budget if
+ * loaded into memory. Command all participants to help repartition.
+ */
+ batch->shared->space_exhausted = true;
+ pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
+ LWLockRelease(&pstate->lock);
+
+ return false;
+ }
+
+ batch->at_least_one_chunk = true;
+ batch->shared->estimated_size += want + HASH_CHUNK_HEADER_SIZE;
+ batch->preallocated = want;
+ LWLockRelease(&pstate->lock);
+
+ return true;
}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 3432bb921dd..c862e8726fa 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -901,7 +901,15 @@ _copyHashJoin(const HashJoin *from)
COPY_NODE_FIELD(hashclauses);
COPY_NODE_FIELD(hashoperators);
COPY_NODE_FIELD(hashcollations);
- COPY_NODE_FIELD(hashkeys);
+ COPY_NODE_FIELD(hashkeys_outer);
+ COPY_NODE_FIELD(hashkeys_inner);
+
+ COPY_SCALAR_FIELD(skewTable);
+ COPY_SCALAR_FIELD(skewColumn);
+ COPY_SCALAR_FIELD(skewInherit);
+ COPY_SCALAR_FIELD(inner_rows_total);
+
+ COPY_NODE_FIELD(inner_tlist);
return newnode;
}
@@ -1053,31 +1061,6 @@ _copyUnique(const Unique *from)
return newnode;
}
-/*
- * _copyHash
- */
-static Hash *
-_copyHash(const Hash *from)
-{
- Hash *newnode = makeNode(Hash);
-
- /*
- * copy node superclass fields
- */
- CopyPlanFields((const Plan *) from, (Plan *) newnode);
-
- /*
- * copy remainder of node
- */
- COPY_NODE_FIELD(hashkeys);
- COPY_SCALAR_FIELD(skewTable);
- COPY_SCALAR_FIELD(skewColumn);
- COPY_SCALAR_FIELD(skewInherit);
- COPY_SCALAR_FIELD(rows_total);
-
- return newnode;
-}
-
/*
* _copySetOp
*/
@@ -4886,9 +4869,6 @@ copyObjectImpl(const void *from)
case T_Unique:
retval = _copyUnique(from);
break;
- case T_Hash:
- retval = _copyHash(from);
- break;
case T_SetOp:
retval = _copySetOp(from);
break;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index b0dcd02ff68..56ce5abea4f 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -763,7 +763,15 @@ _outHashJoin(StringInfo str, const HashJoin *node)
WRITE_NODE_FIELD(hashclauses);
WRITE_NODE_FIELD(hashoperators);
WRITE_NODE_FIELD(hashcollations);
- WRITE_NODE_FIELD(hashkeys);
+ WRITE_NODE_FIELD(hashkeys_outer);
+ WRITE_NODE_FIELD(hashkeys_inner);
+
+ WRITE_OID_FIELD(skewTable);
+ WRITE_INT_FIELD(skewColumn);
+ WRITE_BOOL_FIELD(skewInherit);
+ WRITE_FLOAT_FIELD(inner_rows_total, "%.0f");
+
+ WRITE_NODE_FIELD(inner_tlist);
}
static void
@@ -859,20 +867,6 @@ _outUnique(StringInfo str, const Unique *node)
WRITE_OID_ARRAY(uniqCollations, node->numCols);
}
-static void
-_outHash(StringInfo str, const Hash *node)
-{
- WRITE_NODE_TYPE("HASH");
-
- _outPlanInfo(str, (const Plan *) node);
-
- WRITE_NODE_FIELD(hashkeys);
- WRITE_OID_FIELD(skewTable);
- WRITE_INT_FIELD(skewColumn);
- WRITE_BOOL_FIELD(skewInherit);
- WRITE_FLOAT_FIELD(rows_total, "%.0f");
-}
-
static void
_outSetOp(StringInfo str, const SetOp *node)
{
@@ -3774,9 +3768,6 @@ outNode(StringInfo str, const void *obj)
case T_Unique:
_outUnique(str, obj);
break;
- case T_Hash:
- _outHash(str, obj);
- break;
case T_SetOp:
_outSetOp(str, obj);
break;
diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c
index 764e3bb90c9..07adfde0212 100644
--- a/src/backend/nodes/readfuncs.c
+++ b/src/backend/nodes/readfuncs.c
@@ -2098,7 +2098,14 @@ _readHashJoin(void)
READ_NODE_FIELD(hashclauses);
READ_NODE_FIELD(hashoperators);
READ_NODE_FIELD(hashcollations);
- READ_NODE_FIELD(hashkeys);
+ READ_NODE_FIELD(hashkeys_outer);
+ READ_NODE_FIELD(hashkeys_inner);
+ READ_OID_FIELD(skewTable);
+ READ_INT_FIELD(skewColumn);
+ READ_BOOL_FIELD(skewInherit);
+ READ_FLOAT_FIELD(inner_rows_total);
+
+ READ_NODE_FIELD(inner_tlist);
READ_DONE();
}
@@ -2267,25 +2274,6 @@ _readGatherMerge(void)
READ_DONE();
}
-/*
- * _readHash
- */
-static Hash *
-_readHash(void)
-{
- READ_LOCALS(Hash);
-
- ReadCommonPlan(&local_node->plan);
-
- READ_NODE_FIELD(hashkeys);
- READ_OID_FIELD(skewTable);
- READ_INT_FIELD(skewColumn);
- READ_BOOL_FIELD(skewInherit);
- READ_FLOAT_FIELD(rows_total);
-
- READ_DONE();
-}
-
/*
* _readSetOp
*/
@@ -2777,8 +2765,6 @@ parseNodeString(void)
return_value = _readGather();
else if (MATCH("GATHERMERGE", 11))
return_value = _readGatherMerge();
- else if (MATCH("HASH", 4))
- return_value = _readHash();
else if (MATCH("SETOP", 5))
return_value = _readSetOp();
else if (MATCH("LOCKROWS", 8))
diff --git a/src/backend/optimizer/path/costsize.c b/src/backend/optimizer/path/costsize.c
index c5f65934859..da30f29f94f 100644
--- a/src/backend/optimizer/path/costsize.c
+++ b/src/backend/optimizer/path/costsize.c
@@ -77,7 +77,7 @@
#include "access/htup_details.h"
#include "access/tsmapi.h"
#include "executor/executor.h"
-#include "executor/nodeHash.h"
+#include "executor/nodeHashjoin.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "nodes/nodeFuncs.h"
diff --git a/src/backend/optimizer/plan/createplan.c b/src/backend/optimizer/plan/createplan.c
index 0c036209f09..e45e0a2e253 100644
--- a/src/backend/optimizer/plan/createplan.c
+++ b/src/backend/optimizer/plan/createplan.c
@@ -223,14 +223,12 @@ static HashJoin *make_hashjoin(List *tlist,
List *joinclauses, List *otherclauses,
List *hashclauses,
List *hashoperators, List *hashcollations,
- List *hashkeys,
+ List *hashkeys_outer, List *hashkeys_inner,
+ Oid skewTable,
+ AttrNumber skewColumn,
+ bool skewInherit,
Plan *lefttree, Plan *righttree,
JoinType jointype, bool inner_unique);
-static Hash *make_hash(Plan *lefttree,
- List *hashkeys,
- Oid skewTable,
- AttrNumber skewColumn,
- bool skewInherit);
static MergeJoin *make_mergejoin(List *tlist,
List *joinclauses, List *otherclauses,
List *mergeclauses,
@@ -4376,7 +4374,6 @@ create_hashjoin_plan(PlannerInfo *root,
HashPath *best_path)
{
HashJoin *join_plan;
- Hash *hash_plan;
Plan *outer_plan;
Plan *inner_plan;
List *tlist = build_path_tlist(root, &best_path->jpath.path);
@@ -4501,31 +4498,8 @@ create_hashjoin_plan(PlannerInfo *root,
}
/*
- * Build the hash node and hash join node.
+ * Build the hash join node.
*/
- hash_plan = make_hash(inner_plan,
- inner_hashkeys,
- skewTable,
- skewColumn,
- skewInherit);
-
- /*
- * Set Hash node's startup & total costs equal to total cost of input
- * plan; this only affects EXPLAIN display not decisions.
- */
- copy_plan_costsize(&hash_plan->plan, inner_plan);
- hash_plan->plan.startup_cost = hash_plan->plan.total_cost;
-
- /*
- * If parallel-aware, the executor will also need an estimate of the total
- * number of rows expected from all participants so that it can size the
- * shared hash table.
- */
- if (best_path->jpath.path.parallel_aware)
- {
- hash_plan->plan.parallel_aware = true;
- hash_plan->rows_total = best_path->inner_rows_total;
- }
join_plan = make_hashjoin(tlist,
joinclauses,
@@ -4534,11 +4508,26 @@ create_hashjoin_plan(PlannerInfo *root,
hashoperators,
hashcollations,
outer_hashkeys,
+ inner_hashkeys,
+ skewTable,
+ skewColumn,
+ skewInherit,
outer_plan,
- (Plan *) hash_plan,
+ (Plan *) inner_plan,
best_path->jpath.jointype,
best_path->jpath.inner_unique);
+ /*
+ * If parallel-aware, the executor will also need an estimate of the total
+ * number of rows expected from all participants so that it can size the
+ * shared hash table.
+ */
+ if (best_path->jpath.path.parallel_aware)
+ {
+ join_plan->join.plan.parallel_aware = true;
+ join_plan->inner_rows_total = best_path->inner_rows_total;
+ }
+
copy_generic_path_info(&join_plan->join.plan, &best_path->jpath.path);
return join_plan;
@@ -5577,7 +5566,11 @@ make_hashjoin(List *tlist,
List *hashclauses,
List *hashoperators,
List *hashcollations,
- List *hashkeys,
+ List *outer_hashkeys,
+ List *inner_hashkeys,
+ Oid skewTable,
+ AttrNumber skewColumn,
+ bool skewInherit,
Plan *lefttree,
Plan *righttree,
JoinType jointype,
@@ -5593,34 +5586,16 @@ make_hashjoin(List *tlist,
node->hashclauses = hashclauses;
node->hashoperators = hashoperators;
node->hashcollations = hashcollations;
- node->hashkeys = hashkeys;
- node->join.jointype = jointype;
- node->join.inner_unique = inner_unique;
- node->join.joinqual = joinclauses;
-
- return node;
-}
-
-static Hash *
-make_hash(Plan *lefttree,
- List *hashkeys,
- Oid skewTable,
- AttrNumber skewColumn,
- bool skewInherit)
-{
- Hash *node = makeNode(Hash);
- Plan *plan = &node->plan;
-
- plan->targetlist = lefttree->targetlist;
- plan->qual = NIL;
- plan->lefttree = lefttree;
- plan->righttree = NULL;
-
- node->hashkeys = hashkeys;
+ node->hashkeys_outer = outer_hashkeys;
+ node->hashkeys_inner = inner_hashkeys;
node->skewTable = skewTable;
node->skewColumn = skewColumn;
node->skewInherit = skewInherit;
+ node->join.jointype = jointype;
+ node->join.inner_unique = inner_unique;
+ node->join.joinqual = joinclauses;
+ node->inner_tlist = righttree->targetlist;
return node;
}
@@ -6763,7 +6738,6 @@ is_projection_capable_path(Path *path)
/* Most plan types can project, so just list the ones that can't */
switch (path->pathtype)
{
- case T_Hash:
case T_Material:
case T_Sort:
case T_Unique:
@@ -6807,7 +6781,6 @@ is_projection_capable_plan(Plan *plan)
/* Most plan types can project, so just list the ones that can't */
switch (nodeTag(plan))
{
- case T_Hash:
case T_Material:
case T_Sort:
case T_Unique:
diff --git a/src/backend/optimizer/plan/setrefs.c b/src/backend/optimizer/plan/setrefs.c
index 566ee96da8c..9e63b3a773e 100644
--- a/src/backend/optimizer/plan/setrefs.c
+++ b/src/backend/optimizer/plan/setrefs.c
@@ -107,7 +107,6 @@ static Plan *set_append_references(PlannerInfo *root,
static Plan *set_mergeappend_references(PlannerInfo *root,
MergeAppend *mplan,
int rtoffset);
-static void set_hash_references(PlannerInfo *root, Plan *plan, int rtoffset);
static Node *fix_scan_expr(PlannerInfo *root, Node *node, int rtoffset);
static Node *fix_scan_expr_mutator(Node *node, fix_scan_expr_context *context);
static bool fix_scan_expr_walker(Node *node, fix_scan_expr_context *context);
@@ -646,10 +645,6 @@ set_plan_refs(PlannerInfo *root, Plan *plan, int rtoffset)
}
break;
- case T_Hash:
- set_hash_references(root, plan, rtoffset);
- break;
-
case T_Material:
case T_Sort:
case T_Unique:
@@ -1423,37 +1418,6 @@ set_mergeappend_references(PlannerInfo *root,
return (Plan *) mplan;
}
-/*
- * set_hash_references
- * Do set_plan_references processing on a Hash node
- */
-static void
-set_hash_references(PlannerInfo *root, Plan *plan, int rtoffset)
-{
- Hash *hplan = (Hash *) plan;
- Plan *outer_plan = plan->lefttree;
- indexed_tlist *outer_itlist;
-
- /*
- * Hash's hashkeys are used when feeding tuples into the hashtable,
- * therefore have them reference Hash's outer plan (which itself is the
- * inner plan of the HashJoin).
- */
- outer_itlist = build_tlist_index(outer_plan->targetlist);
- hplan->hashkeys = (List *)
- fix_upper_expr(root,
- (Node *) hplan->hashkeys,
- outer_itlist,
- OUTER_VAR,
- rtoffset);
-
- /* Hash doesn't project */
- set_dummy_tlist_references(plan, rtoffset);
-
- /* Hash nodes don't have their own quals */
- Assert(plan->qual == NIL);
-}
-
/*
* copyVar
* Copy a Var node.
@@ -1789,15 +1753,23 @@ set_join_references(PlannerInfo *root, Join *join, int rtoffset)
(Index) 0,
rtoffset);
- /*
- * HashJoin's hashkeys are used to look for matching tuples from its
- * outer plan (not the Hash node!) in the hashtable.
- */
- hj->hashkeys = (List *) fix_upper_expr(root,
- (Node *) hj->hashkeys,
- outer_itlist,
- OUTER_VAR,
- rtoffset);
+ hj->hashkeys_outer = (List *) fix_upper_expr(root,
+ (Node *) hj->hashkeys_outer,
+ outer_itlist,
+ OUTER_VAR,
+ rtoffset);
+
+ hj->hashkeys_inner = (List *) fix_upper_expr(root,
+ (Node *) hj->hashkeys_inner,
+ inner_itlist,
+ INNER_VAR,
+ rtoffset);
+
+ hj->inner_tlist = (List *) fix_upper_expr(root,
+ (Node *) hj->inner_tlist,
+ inner_itlist,
+ INNER_VAR,
+ rtoffset);
}
/*
diff --git a/src/backend/optimizer/plan/subselect.c b/src/backend/optimizer/plan/subselect.c
index 48b62a55de8..d1316faf183 100644
--- a/src/backend/optimizer/plan/subselect.c
+++ b/src/backend/optimizer/plan/subselect.c
@@ -2683,7 +2683,6 @@ finalize_plan(PlannerInfo *root, Plan *plan,
break;
case T_ProjectSet:
- case T_Hash:
case T_Material:
case T_Sort:
case T_Unique:
diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h
deleted file mode 100644
index fc80f03aa8d..00000000000
--- a/src/include/executor/nodeHash.h
+++ /dev/null
@@ -1,79 +0,0 @@
-/*-------------------------------------------------------------------------
- *
- * nodeHash.h
- * prototypes for nodeHash.c
- *
- *
- * Portions Copyright (c) 1996-2019, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- * src/include/executor/nodeHash.h
- *
- *-------------------------------------------------------------------------
- */
-#ifndef NODEHASH_H
-#define NODEHASH_H
-
-#include "access/parallel.h"
-#include "nodes/execnodes.h"
-
-struct SharedHashJoinBatch;
-
-extern HashState *ExecInitHash(Hash *node, EState *estate, int eflags);
-extern Node *MultiExecHash(HashState *node);
-extern void ExecEndHash(HashState *node);
-extern void ExecReScanHash(HashState *node);
-
-extern HashJoinTable ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
- bool keepNulls);
-extern void ExecParallelHashTableAlloc(HashJoinTable hashtable,
- int batchno);
-extern void ExecHashTableDestroy(HashJoinTable hashtable);
-extern void ExecHashTableDetach(HashJoinTable hashtable);
-extern void ExecHashTableDetachBatch(HashJoinTable hashtable);
-extern void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable,
- int batchno);
-
-extern void ExecHashTableInsert(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue);
-extern void ExecParallelHashTableInsert(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue);
-extern void ExecParallelHashTableInsertCurrentBatch(HashJoinTable hashtable,
- TupleTableSlot *slot,
- uint32 hashvalue);
-extern bool ExecHashGetHashValue(HashJoinTable hashtable,
- ExprContext *econtext,
- List *hashkeys,
- bool outer_tuple,
- bool keep_nulls,
- uint32 *hashvalue);
-extern void ExecHashGetBucketAndBatch(HashJoinTable hashtable,
- uint32 hashvalue,
- int *bucketno,
- int *batchno);
-extern bool ExecScanHashBucket(HashJoinState *hjstate, ExprContext *econtext);
-extern bool ExecParallelScanHashBucket(HashJoinState *hjstate, ExprContext *econtext);
-extern void ExecPrepHashTableForUnmatched(HashJoinState *hjstate);
-extern bool ExecScanHashTableForUnmatched(HashJoinState *hjstate,
- ExprContext *econtext);
-extern void ExecHashTableReset(HashJoinTable hashtable);
-extern void ExecHashTableResetMatchFlags(HashJoinTable hashtable);
-extern void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
- bool try_combined_work_mem,
- int parallel_workers,
- size_t *space_allowed,
- int *numbuckets,
- int *numbatches,
- int *num_skew_mcvs);
-extern int ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue);
-extern void ExecHashEstimate(HashState *node, ParallelContext *pcxt);
-extern void ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt);
-extern void ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt);
-extern void ExecHashRetrieveInstrumentation(HashState *node);
-extern void ExecShutdownHash(HashState *node);
-extern void ExecHashGetInstrumentation(HashInstrumentation *instrument,
- HashJoinTable hashtable);
-
-#endif /* NODEHASH_H */
diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h
index 1752b3b208d..5ae2b535754 100644
--- a/src/include/executor/nodeHashjoin.h
+++ b/src/include/executor/nodeHashjoin.h
@@ -27,8 +27,15 @@ extern void ExecHashJoinInitializeDSM(HashJoinState *state, ParallelContext *pcx
extern void ExecHashJoinReInitializeDSM(HashJoinState *state, ParallelContext *pcxt);
extern void ExecHashJoinInitializeWorker(HashJoinState *state,
ParallelWorkerContext *pwcxt);
-
-extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
- BufFile **fileptr);
+extern void ExecHashJoinRetrieveInstrumentation(HashJoinState *node);
+extern void ExecHashJoinGetInstrumentation(HashInstrumentation *instrument,
+ HashJoinTable hashtable);
+extern void ExecChooseHashTableSize(double ntuples, int tupwidth, bool useskew,
+ bool try_combined_work_mem,
+ int parallel_workers,
+ size_t *space_allowed,
+ int *numbuckets,
+ int *numbatches,
+ int *num_skew_mcvs);
#endif /* NODEHASHJOIN_H */
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 44f76082e99..2e13f5a25b2 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -1924,6 +1924,30 @@ typedef struct MergeJoinState
* ----------------
*/
+
+/* ----------------
+ * Values displayed by EXPLAIN ANALYZE
+ * ----------------
+ */
+typedef struct HashInstrumentation
+{
+ int nbuckets; /* number of buckets at end of execution */
+ int nbuckets_original; /* planned number of buckets */
+ int nbatch; /* number of batches at end of execution */
+ int nbatch_original; /* planned number of batches */
+ size_t space_peak; /* speak memory usage in bytes */
+} HashInstrumentation;
+
+/* ----------------
+ * Shared memory container for per-worker hash information
+ * ----------------
+ */
+typedef struct SharedHashInfo
+{
+ int num_workers;
+ HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER];
+} SharedHashInfo;
+
/* these structs are defined in executor/hashjoin.h: */
typedef struct HashJoinTupleData *HashJoinTuple;
typedef struct HashJoinTableData *HashJoinTable;
@@ -1933,6 +1957,7 @@ typedef struct HashJoinState
JoinState js; /* its first field is NodeTag */
ExprState *hashclauses;
List *hj_OuterHashKeys; /* list of ExprState nodes */
+ List *hj_InnerHashKeys; /* list of ExprState nodes */
List *hj_HashOperators; /* list of operator OIDs */
List *hj_Collations;
HashJoinTable hj_HashTable;
@@ -1948,6 +1973,14 @@ typedef struct HashJoinState
int hj_JoinState;
bool hj_MatchedOuter;
bool hj_OuterNotEmpty;
+
+ HashJoinTable hashtable; /* hash table for the hashjoin */
+
+ SharedHashInfo *shared_info; /* one entry per worker */
+ HashInstrumentation *hinstrument; /* this worker's entry */
+
+ /* Parallel hash state. */
+ struct ParallelHashJoinState *parallel_state;
} HashJoinState;
@@ -2239,46 +2272,6 @@ typedef struct GatherMergeState
struct binaryheap *gm_heap; /* binary heap of slot indices */
} GatherMergeState;
-/* ----------------
- * Values displayed by EXPLAIN ANALYZE
- * ----------------
- */
-typedef struct HashInstrumentation
-{
- int nbuckets; /* number of buckets at end of execution */
- int nbuckets_original; /* planned number of buckets */
- int nbatch; /* number of batches at end of execution */
- int nbatch_original; /* planned number of batches */
- size_t space_peak; /* speak memory usage in bytes */
-} HashInstrumentation;
-
-/* ----------------
- * Shared memory container for per-worker hash information
- * ----------------
- */
-typedef struct SharedHashInfo
-{
- int num_workers;
- HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER];
-} SharedHashInfo;
-
-/* ----------------
- * HashState information
- * ----------------
- */
-typedef struct HashState
-{
- PlanState ps; /* its first field is NodeTag */
- HashJoinTable hashtable; /* hash table for the hashjoin */
- List *hashkeys; /* list of ExprState nodes */
-
- SharedHashInfo *shared_info; /* one entry per worker */
- HashInstrumentation *hinstrument; /* this worker's entry */
-
- /* Parallel hash state. */
- struct ParallelHashJoinState *parallel_state;
-} HashState;
-
/* ----------------
* SetOpState information
*
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index bce2d59b0db..17fd211c0b6 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -80,7 +80,6 @@ typedef enum NodeTag
T_Unique,
T_Gather,
T_GatherMerge,
- T_Hash,
T_SetOp,
T_LockRows,
T_Limit,
@@ -136,7 +135,6 @@ typedef enum NodeTag
T_UniqueState,
T_GatherState,
T_GatherMergeState,
- T_HashState,
T_SetOpState,
T_LockRowsState,
T_LimitState,
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 8e6594e3551..40e42716ec9 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -744,7 +744,27 @@ typedef struct HashJoin
* List of expressions to be hashed for tuples from the outer plan, to
* perform lookups in the hashtable over the inner plan.
*/
- List *hashkeys;
+ List *hashkeys_outer;
+
+ /*
+ * List of expressions to be hashed for tuples from inner plan, needed to
+ * put them into the hashtable.
+ */
+ List *hashkeys_inner; /* hash keys for the hashjoin condition */
+
+ /*
+ * If the executor is supposed to try to apply skew join optimization,
+ * then skewTable/skewColumn/skewInherit identify the outer relation's
+ * join key column, from which the relevant MCV statistics can be fetched.
+ */
+ Oid skewTable; /* outer join key's table OID, or InvalidOid */
+ AttrNumber skewColumn; /* outer join key's column #, or zero */
+ bool skewInherit; /* is outer join rel an inheritance tree? */
+
+ double inner_rows_total; /* estimate total inner rows if parallel_aware */
+
+ /* XXX: explain hack */
+ List *inner_tlist;
} HashJoin;
/* ----------------
@@ -896,30 +916,6 @@ typedef struct GatherMerge
* at gather merge or one of it's child node */
} GatherMerge;
-/* ----------------
- * hash build node
- *
- * If the executor is supposed to try to apply skew join optimization, then
- * skewTable/skewColumn/skewInherit identify the outer relation's join key
- * column, from which the relevant MCV statistics can be fetched.
- * ----------------
- */
-typedef struct Hash
-{
- Plan plan;
-
- /*
- * List of expressions to be hashed for tuples from Hash's outer plan,
- * needed to put them into the hashtable.
- */
- List *hashkeys; /* hash keys for the hashjoin condition */
- Oid skewTable; /* outer join key's table OID, or InvalidOid */
- AttrNumber skewColumn; /* outer join key's column #, or zero */
- bool skewInherit; /* is outer join rel an inheritance tree? */
- /* all other info is in the parent HashJoin node */
- double rows_total; /* estimate total rows if parallel_aware */
-} Hash;
-
/* ----------------
* setop node
* ----------------
diff --git a/src/test/regress/expected/join_hash.out b/src/test/regress/expected/join_hash.out
index 3a91c144a27..fd16886f301 100644
--- a/src/test/regress/expected/join_hash.out
+++ b/src/test/regress/expected/join_hash.out
@@ -9,7 +9,7 @@ set local enable_hashjoin = on;
-- general we can't make assertions about how many batches (or
-- buckets) will be required because it can vary, but we can in some
-- special cases and we can check for growth.
-create or replace function find_hash(node json)
+create or replace function find_hash_join(node json)
returns json language plpgsql
as
$$
@@ -17,12 +17,12 @@ declare
x json;
child json;
begin
- if node->>'Node Type' = 'Hash' then
+ if node->>'Node Type' = 'Hash Join' then
return node;
else
for child in select json_array_elements(node->'Plans')
loop
- x := find_hash(child);
+ x := find_hash_join(child);
if x is not null then
return x;
end if;
@@ -42,7 +42,7 @@ begin
for whole_plan in
execute 'explain (analyze, format ''json'') ' || query
loop
- hash_node := find_hash(json_extract_path(whole_plan, '0', 'Plan'));
+ hash_node := find_hash_join(json_extract_path(whole_plan, '0', 'Plan'));
original := hash_node->>'Original Hash Batches';
final := hash_node->>'Hash Batches';
return next;
@@ -933,16 +933,16 @@ WHERE
SubPlan 5
-> Result
Output: (hjtest_2.c * 5)
- SubPlan 1
- -> Result
- Output: 1
- One-Time Filter: (hjtest_2.id = 1)
- SubPlan 3
- -> Result
- Output: (hjtest_2.c * 5)
+ SubPlan 1
+ -> Result
+ Output: 1
+ One-Time Filter: (hjtest_2.id = 1)
SubPlan 2
-> Result
Output: (hjtest_1.b * 5)
+ SubPlan 3
+ -> Result
+ Output: (hjtest_2.c * 5)
(28 rows)
SELECT hjtest_1.a a1, hjtest_2.a a2,hjtest_1.tableoid::regclass t1, hjtest_2.tableoid::regclass t2
@@ -987,9 +987,6 @@ WHERE
SubPlan 4
-> Result
Output: (hjtest_1.b * 5)
- SubPlan 2
- -> Result
- Output: (hjtest_1.b * 5)
SubPlan 1
-> Result
Output: 1
@@ -997,6 +994,9 @@ WHERE
SubPlan 3
-> Result
Output: (hjtest_2.c * 5)
+ SubPlan 2
+ -> Result
+ Output: (hjtest_1.b * 5)
(28 rows)
SELECT hjtest_1.a a1, hjtest_2.a a2,hjtest_1.tableoid::regclass t1, hjtest_2.tableoid::regclass t2
diff --git a/src/test/regress/sql/join_hash.sql b/src/test/regress/sql/join_hash.sql
index 68c1a8c7b65..868a839719e 100644
--- a/src/test/regress/sql/join_hash.sql
+++ b/src/test/regress/sql/join_hash.sql
@@ -12,7 +12,7 @@ set local enable_hashjoin = on;
-- general we can't make assertions about how many batches (or
-- buckets) will be required because it can vary, but we can in some
-- special cases and we can check for growth.
-create or replace function find_hash(node json)
+create or replace function find_hash_join(node json)
returns json language plpgsql
as
$$
@@ -20,12 +20,12 @@ declare
x json;
child json;
begin
- if node->>'Node Type' = 'Hash' then
+ if node->>'Node Type' = 'Hash Join' then
return node;
else
for child in select json_array_elements(node->'Plans')
loop
- x := find_hash(child);
+ x := find_hash_join(child);
if x is not null then
return x;
end if;
@@ -45,7 +45,7 @@ begin
for whole_plan in
execute 'explain (analyze, format ''json'') ' || query
loop
- hash_node := find_hash(json_extract_path(whole_plan, '0', 'Plan'));
+ hash_node := find_hash_join(json_extract_path(whole_plan, '0', 'Plan'));
original := hash_node->>'Original Hash Batches';
final := hash_node->>'Hash Batches';
return next;
--
2.23.0.385.gbc12974a89
--jgzjztp6euershzl--
view thread (2+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH] Merge Hash nodes into their associated HashJoin node.
In-Reply-To: <no-message-id-201911@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox