Hi,
Thanks for your review!
On Sat, 13 May 2023 23:47:53 +0200
Tomas Vondra <tomas.von...@enterprisedb.com> wrote:
> Thanks for the patches. A couple mostly minor comments, to complement
> Melanie's review:
>
> 0001
>
> I'm not really sure about calling this "hybrid hash-join". What does it
> even mean? The new comments simply describe the existing batching, and
> how we increment number of batches, etc.
Unless I'm wrong, I believed it comes from the "Hybrid hash join algorithm" as
described here (+ see pdf in this page ref):
https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
I added the ref in the v7 documentation to avoid futur confusion, is it ok?
> 0002
>
> I wouldn't call the new ExecHashJoinSaveTuple parameter spillcxt but
> just something generic (e.g. cxt). Yes, we're passing spillCxt, but from
> the function's POV it's just a pointer.
changed in v7.
> I also wouldn't move the ExecHashJoinSaveTuple comment inside - it just
> needs to be reworded that we're expecting the context to be with the
> right lifespan. The function comment is the right place to document such
> expectations, people are less likely to read the function body.
moved and reworded in v7.
> The modified comment in hashjoin.h has a some alignment issues.
I see no alignment issue. I suppose what bother you might be the spaces
before spillCxt and batchCxt to show they are childs of hashCxt? Should I
remove them?
Regards,
>From 997a82a0ee9eda1774b5210401b2d9bf281318da Mon Sep 17 00:00:00 2001
From: Melanie Plageman <melanieplage...@gmail.com>
Date: Thu, 30 Apr 2020 07:16:28 -0700
Subject: [PATCH 1/3] Describe hybrid hash join implementation
This is just a draft to spark conversation on what a good comment might
be like in this file on how the hybrid hash join algorithm is
implemented in Postgres. I'm pretty sure this is the accepted term for
this algorithm https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
---
src/backend/executor/nodeHashjoin.c | 41 +++++++++++++++++++++++++++++
1 file changed, 41 insertions(+)
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
index 0a3f32f731..a39164c15d 100644
--- a/src/backend/executor/nodeHashjoin.c
+++ b/src/backend/executor/nodeHashjoin.c
@@ -10,6 +10,47 @@
* IDENTIFICATION
* src/backend/executor/nodeHashjoin.c
*
+ * HYBRID HASH JOIN
+ *
+ * This is based on the hybrid hash join algorythm described shortly in the
+ * following page, and in length in the pdf in page's notes:
+ *
+ * https://en.wikipedia.org/wiki/Hash_join#Hybrid_hash_join
+ *
+ * If the inner side tuples of a hash join do not fit in memory, the hash join
+ * can be executed in multiple batches.
+ *
+ * If the statistics on the inner side relation are accurate, planner chooses a
+ * multi-batch strategy and estimates the number of batches.
+ *
+ * The query executor measures the real size of the hashtable and increases the
+ * number of batches if the hashtable grows too large.
+ *
+ * The number of batches is always a power of two, so an increase in the number
+ * of batches doubles it.
+ *
+ * Serial hash join measures batch size lazily -- waiting until it is loading a
+ * batch to determine if it will fit in memory. While inserting tuples into the
+ * hashtable, serial hash join will, if that tuple were to exceed work_mem,
+ * dump out the hashtable and reassign them either to other batch files or the
+ * current batch resident in the hashtable.
+ *
+ * Parallel hash join, on the other hand, completes all changes to the number
+ * of batches during the build phase. If it increases the number of batches, it
+ * dumps out all the tuples from all batches and reassigns them to entirely new
+ * batch files. Then it checks every batch to ensure it will fit in the space
+ * budget for the query.
+ *
+ * In both parallel and serial hash join, the executor currently makes a best
+ * effort. If a particular batch will not fit in memory, it tries doubling the
+ * number of batches. If after a batch increase, there is a batch which
+ * retained all or none of its tuples, the executor disables growth in the
+ * number of batches globally. After growth is disabled, all batches that would
+ * have previously triggered an increase in the number of batches instead
+ * exceed the space allowed.
+ *
+ * TODO: should we discuss that tuples can only spill forward?
+ *
* PARALLELISM
*
* Hash joins can participate in parallel query execution in several ways. A
--
2.40.1
>From 52e989d5eb6405e86e0d460dffbfaca292bc4274 Mon Sep 17 00:00:00 2001
From: Jehan-Guillaume de Rorthais <j...@dalibo.com>
Date: Mon, 27 Mar 2023 15:54:39 +0200
Subject: [PATCH 2/3] Allocate hash batches related BufFile in a dedicated
context
---
src/backend/executor/nodeHash.c | 43 ++++++++++++++++-------
src/backend/executor/nodeHashjoin.c | 22 ++++++++----
src/backend/utils/sort/sharedtuplestore.c | 8 +++++
src/include/executor/hashjoin.h | 30 ++++++++++------
src/include/executor/nodeHashjoin.h | 2 +-
5 files changed, 75 insertions(+), 30 deletions(-)
diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c
index 5fd1c5553b..e529f50a3e 100644
--- a/src/backend/executor/nodeHash.c
+++ b/src/backend/executor/nodeHash.c
@@ -484,7 +484,7 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
*
* 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.
+ * subsidiary hashCxt, batchCxt or spillCxt.
*/
hashtable = palloc_object(HashJoinTableData);
hashtable->nbuckets = nbuckets;
@@ -538,6 +538,10 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
"HashBatchContext",
ALLOCSET_DEFAULT_SIZES);
+ hashtable->spillCxt = AllocSetContextCreate(hashtable->hashCxt,
+ "HashSpillContext",
+ ALLOCSET_DEFAULT_SIZES);
+
/* Allocate data that will live for the life of the hashjoin */
oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
@@ -570,12 +574,19 @@ ExecHashTableCreate(HashState *state, List *hashOperators, List *hashCollations,
if (nbatch > 1 && hashtable->parallel_state == NULL)
{
+ MemoryContext oldctx;
+
/*
* allocate and initialize the file arrays in hashCxt (not needed for
* parallel case which uses shared tuplestores instead of raw files)
*/
+ oldctx = MemoryContextSwitchTo(hashtable->spillCxt);
+
hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch);
hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch);
+
+ MemoryContextSwitchTo(oldctx);
+
/* The files will not be opened until needed... */
/* ... but make sure we have temp tablespaces established for them */
PrepareTempTablespaces();
@@ -913,7 +924,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable)
int oldnbatch = hashtable->nbatch;
int curbatch = hashtable->curbatch;
int nbatch;
- MemoryContext oldcxt;
long ninmemory;
long nfreed;
HashMemoryChunk oldchunks;
@@ -934,13 +944,16 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable)
hashtable, nbatch, hashtable->spaceUsed);
#endif
- oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
-
if (hashtable->innerBatchFile == NULL)
{
+ MemoryContext oldcxt = MemoryContextSwitchTo(hashtable->spillCxt);
+
/* we had no file arrays before */
hashtable->innerBatchFile = palloc0_array(BufFile *, nbatch);
hashtable->outerBatchFile = palloc0_array(BufFile *, nbatch);
+
+ MemoryContextSwitchTo(oldcxt);
+
/* time to establish the temp tablespaces, too */
PrepareTempTablespaces();
}
@@ -951,8 +964,6 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable)
hashtable->outerBatchFile = repalloc0_array(hashtable->outerBatchFile, BufFile *, oldnbatch, nbatch);
}
- MemoryContextSwitchTo(oldcxt);
-
hashtable->nbatch = nbatch;
/*
@@ -1024,7 +1035,8 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable)
Assert(batchno > curbatch);
ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple),
hashTuple->hashvalue,
- &hashtable->innerBatchFile[batchno]);
+ &hashtable->innerBatchFile[batchno],
+ hashtable->spillCxt);
hashtable->spaceUsed -= hashTupleSize;
nfreed++;
@@ -1683,7 +1695,8 @@ ExecHashTableInsert(HashJoinTable hashtable,
Assert(batchno > hashtable->curbatch);
ExecHashJoinSaveTuple(tuple,
hashvalue,
- &hashtable->innerBatchFile[batchno]);
+ &hashtable->innerBatchFile[batchno],
+ hashtable->spillCxt);
}
if (shouldFree)
@@ -2664,7 +2677,8 @@ ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
/* Put the tuple into a temp file for later batches */
Assert(batchno > hashtable->curbatch);
ExecHashJoinSaveTuple(tuple, hashvalue,
- &hashtable->innerBatchFile[batchno]);
+ &hashtable->innerBatchFile[batchno],
+ hashtable->spillCxt);
pfree(hashTuple);
hashtable->spaceUsed -= tupleSize;
hashtable->spaceUsedSkew -= tupleSize;
@@ -3093,8 +3107,11 @@ ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int nbatch)
pstate->nbatch = nbatch;
batches = dsa_get_address(hashtable->area, pstate->batches);
- /* Use hash join memory context. */
- oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
+ /*
+ * Use hash join spill memory context to allocate accessors and their
+ * buffers
+ */
+ oldcxt = MemoryContextSwitchTo(hashtable->spillCxt);
/* Allocate this backend's accessor array. */
hashtable->nbatch = nbatch;
@@ -3196,8 +3213,8 @@ ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
*/
Assert(DsaPointerIsValid(pstate->batches));
- /* Use hash join memory context. */
- oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);
+ /* Use hash join spill memory context to allocate accessors */
+ oldcxt = MemoryContextSwitchTo(hashtable->spillCxt);
/* Allocate this backend's accessor array. */
hashtable->nbatch = pstate->nbatch;
diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c
index a39164c15d..8618e0df40 100644
--- a/src/backend/executor/nodeHashjoin.c
+++ b/src/backend/executor/nodeHashjoin.c
@@ -491,7 +491,8 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel)
Assert(parallel_state == NULL);
Assert(batchno > hashtable->curbatch);
ExecHashJoinSaveTuple(mintuple, hashvalue,
- &hashtable->outerBatchFile[batchno]);
+ &hashtable->outerBatchFile[batchno],
+ hashtable->spillCxt);
if (shouldFree)
heap_free_minimal_tuple(mintuple);
@@ -1313,21 +1314,30 @@ ExecParallelHashJoinNewBatch(HashJoinState *hjstate)
* The data recorded in the file for each tuple is its hash value,
* then the tuple in MinimalTuple format.
*
- * Note: it is important always to call this in the regular executor
- * context, not in a shorter-lived context; else the temp file buffers
- * will get messed up.
+ * If this is the first write to the batch file, this function first
+ * create it. The associated BufFile buffer is allocated in the given
+ * context. It is important to always give the HashSpillContext
+ * context. First to avoid a shorter-lived context, else the temp file
+ * buffers will get messed up. Second, for a better accounting of the
+ * spilling memory consumption.
+ *
*/
void
ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
- BufFile **fileptr)
+ BufFile **fileptr, MemoryContext cxt)
{
BufFile *file = *fileptr;
if (file == NULL)
{
- /* First write to this batch file, so open it. */
+ MemoryContext oldctx;
+
+ oldctx = MemoryContextSwitchTo(cxt);
+
file = BufFileCreateTemp(false);
*fileptr = file;
+
+ MemoryContextSwitchTo(oldctx);
}
BufFileWrite(file, &hashvalue, sizeof(uint32));
diff --git a/src/backend/utils/sort/sharedtuplestore.c b/src/backend/utils/sort/sharedtuplestore.c
index 0831249159..236be65f22 100644
--- a/src/backend/utils/sort/sharedtuplestore.c
+++ b/src/backend/utils/sort/sharedtuplestore.c
@@ -308,11 +308,15 @@ sts_puttuple(SharedTuplestoreAccessor *accessor, void *meta_data,
{
SharedTuplestoreParticipant *participant;
char name[MAXPGPATH];
+ MemoryContext oldcxt;
/* Create one. Only this backend will write into it. */
sts_filename(name, accessor, accessor->participant);
+
+ oldcxt = MemoryContextSwitchTo(accessor->context);
accessor->write_file =
BufFileCreateFileSet(&accessor->fileset->fs, name);
+ MemoryContextSwitchTo(oldcxt);
/* Set up the shared state for this backend's file. */
participant = &accessor->sts->participants[accessor->participant];
@@ -527,11 +531,15 @@ sts_parallel_scan_next(SharedTuplestoreAccessor *accessor, void *meta_data)
if (accessor->read_file == NULL)
{
char name[MAXPGPATH];
+ MemoryContext oldcxt;
sts_filename(name, accessor, accessor->read_participant);
+
+ oldcxt = MemoryContextSwitchTo(accessor->context);
accessor->read_file =
BufFileOpenFileSet(&accessor->fileset->fs, name, O_RDONLY,
false);
+ MemoryContextSwitchTo(oldcxt);
}
/* Seek and load the chunk header. */
diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h
index 8ee59d2c71..09113f0192 100644
--- a/src/include/executor/hashjoin.h
+++ b/src/include/executor/hashjoin.h
@@ -23,22 +23,31 @@
/* ----------------------------------------------------------------
* hash-join hash table structures
*
- * Each active hashjoin has a HashJoinTable control block, which is
- * palloc'd in the executor's per-query context. All other storage needed
- * for the hashjoin is kept in private memory contexts, two for each hashjoin.
- * This makes it easy and fast to release the storage when we don't need it
- * anymore. (Exception: data associated with the temp files lives in the
- * per-query context too, since we always call buffile.c in that context.)
+ * Each active hashjoin has a HashJoinTable structure, which is
+ * palloc'd in the executor's per-query context. Other storages needed for
+ * each hashjoin is kept in child contexts, three for each hashjoin:
+ * - HashTableContext (hashCxt): the parent hash table storage context
+ * - HashSpillContext (spillCxt): storage for temp files buffers
+ * - HashBatchContext (batchCxt): storage for a batch in serial hash join
*
* The hashtable contexts are made children of the per-query context, ensuring
* that they will be discarded at end of statement even if the join is
* aborted early by an error. (Likewise, any temporary files we make will
* be cleaned up by the virtual file manager in event of an error.)
*
- * Storage that should live through the entire join is allocated from the
- * "hashCxt", while storage that is only wanted for the current batch is
- * allocated in the "batchCxt". By resetting the batchCxt at the end of
- * each batch, we free all the per-batch storage reliably and without tedium.
+ * The "hashCxt" context makes it easy and fast to release the storage when we
+ * don't need it anymore. Storage that should live through the entire join is
+ * allocated from the "hashCxt" (mainly hash's meta datas).
+ *
+ * Data associated with temp files lives in the "spillCxt" context which lives
+ * during the entire join as temp files might need to survives batches. These
+ * files are explicitly destroyed by calling BufFileClose() when the code is
+ * done with them. The aim of this context is to help accounting the memory
+ * allocations dedicated to temp files and their buffers.
+ *
+ * Finaly, storage that is only wanted for the current batch is allocated in
+ * the "batchCxt". By resetting the batchCxt at the end of each batch, we free
+ * all the per-batch storage reliably and without tedium.
*
* During first scan of inner relation, we get its tuples from executor.
* If nbatch > 1 then tuples that don't belong in first batch get saved
@@ -350,6 +359,7 @@ typedef struct HashJoinTableData
MemoryContext hashCxt; /* context for whole-hash-join storage */
MemoryContext batchCxt; /* context for this-batch-only storage */
+ MemoryContext spillCxt; /* context for spilling to temp files */
/* used for dense allocation of tuples (into linked chunks) */
HashMemoryChunk chunks; /* one list for the whole batch */
diff --git a/src/include/executor/nodeHashjoin.h b/src/include/executor/nodeHashjoin.h
index d367070883..958df42240 100644
--- a/src/include/executor/nodeHashjoin.h
+++ b/src/include/executor/nodeHashjoin.h
@@ -29,6 +29,6 @@ extern void ExecHashJoinInitializeWorker(HashJoinState *state,
ParallelWorkerContext *pwcxt);
extern void ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue,
- BufFile **fileptr);
+ BufFile **fileptr, MemoryContext spillcxt);
#endif /* NODEHASHJOIN_H */
--
2.40.1
