From c71b1821f3acd958ee24125556fb72aa7ecc9c65 Mon Sep 17 00:00:00 2001 From: Shihao Date: Sun, 20 Sep 2026 16:17:58 -0400 Subject: [PATCH v1] Collapse hash join batches when the inner side turns out to fit The number of batches and buckets is chosen before execution, from the planner's estimate of the inner side. When that estimate is far too high we get many batches and a large bucket array. The relation would have fit in memory all along. Every batch transition then clears the whole bucket array, and that cost has nothing to do with how much data there really is. Raising work_mem makes this worse. The initial batch count is capped by the memory budget, so a larger budget starts from a larger count, and the balancing loop added by a1b4f289bee then trades batches for buckets at a fixed product. With a good estimate every batch is used and the trade is a win. With a large overestimate the number of non empty batches collapses towards the real row count, so only the bucket array grows. Once the inner side has been read we know its real size, so redo the sizing decision with the real row count. If it now comes out at one batch, rebuild the hash table that way and read the spilled tuples back. Changing nbuckets is safe here because we are moving to a single batch. The batch number then no longer comes from the bits above log2_nbuckets. This is the counterpart of ExecHashIncreaseNumBatches, which handles the opposite estimation error. Reported-by: iany Discussion: https://postgr.es/m/19708-bca71f8de0d45605@postgresql.org --- src/backend/executor/nodeHash.c | 135 ++++++++++++++++++++++++++++ src/backend/executor/nodeHashjoin.c | 78 ++++++++++++++++ src/include/executor/nodeHash.h | 1 + 3 files changed, 214 insertions(+) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 8825bb6fa23..5b12f4de4de 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -1759,6 +1759,141 @@ ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable) } } +/* + * ExecHashUnbatch + * collapse a batched hash table back into a single batch + * + * nbatch and nbuckets are picked before execution starts, from the planner's + * estimate of the inner side. When that estimate is much too high we end up + * with many batches and a large bucket array for a relation that would have + * fit in memory all along. That costs one full clear of the bucket array per + * batch, plus two temp files per batch, and neither of those costs has + * anything to do with how much data there actually is. + * + * We have no way to fix that up front, but once the inner side has been read + * we know its real size, so redo the sizing decision with the real row count. + * If it now comes out at a single batch, rebuild the table that way. This is + * the counterpart of ExecHashIncreaseNumBatches, which handles the opposite + * error. + * + * Changing nbuckets is only safe here because we are going to nbatch = 1: with + * a single batch ExecHashGetBucketAndBatch stops deriving the batch number + * from the bits above log2_nbuckets, so moving that boundary cannot strand a + * tuple in the wrong batch. + * + * Only the tuples already in memory are rehashed here. The caller must load + * back whatever was spilled to the batch files, and close them. + * + * Returns false, leaving the hash table untouched, if the real size still + * needs more than one batch. + */ +bool +ExecHashUnbatch(HashJoinTable hashtable, int tupwidth) +{ + size_t space_allowed; + int nbuckets; + int nbatch; + int num_skew_mcvs; + HashMemoryChunk oldchunks; + MemoryContext oldcxt; + + Assert(hashtable->nbatch > 1); + Assert(hashtable->parallel_state == NULL); + Assert(hashtable->curbatch == 0); + + /* + * Skew tuples live outside the main bucket array and only mean anything + * while we are batching, so leave those joins alone. + */ + if (hashtable->skewEnabled) + return false; + + /* Redo the sizing decision, this time with the row count we measured. */ + ExecChooseHashTableSize(hashtable->totalTuples, tupwidth, + false, /* no skew table in a single-batch join */ + false, /* not parallel */ + 0, + &space_allowed, + &nbuckets, &nbatch, &num_skew_mcvs); + + if (nbatch != 1) + return false; + + /* + * Keep nbuckets_original and nbatch_original as they were: EXPLAIN + * reports them next to the current values, which is how the shrink + * becomes visible. + */ + hashtable->nbatch = 1; + hashtable->nbuckets = nbuckets; + hashtable->nbuckets_optimal = nbuckets; + hashtable->log2_nbuckets = pg_ceil_log2_32(nbuckets); + hashtable->log2_nbuckets_optimal = hashtable->log2_nbuckets; + hashtable->spaceAllowed = space_allowed; + hashtable->spaceAllowedSkew = space_allowed * SKEW_HASH_MEM_PERCENT / 100; + + Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets)); + + /* + * Rebuild the bucket array at the new size, then rehash everything that + * is in memory into it. As in ExecHashIncreaseNumBatches we walk the + * dense-allocated chunks rather than the buckets, so we don't have to + * keep track of which tuples have already been moved; the tuples are + * copied into fresh chunks and the old ones freed as we go. + */ + oldchunks = hashtable->chunks; + hashtable->chunks = NULL; + hashtable->spaceUsed = 0; + + pfree(hashtable->buckets.unshared); + oldcxt = MemoryContextSwitchTo(hashtable->batchCxt); + hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets); + MemoryContextSwitchTo(oldcxt); + + while (oldchunks != NULL) + { + HashMemoryChunk nextchunk = oldchunks->next.unshared; + size_t idx = 0; + + while (idx < oldchunks->used) + { + HashJoinTuple hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(oldchunks) + idx); + int hashTupleSize = (HJTUPLE_OVERHEAD + + HJTUPLE_MINTUPLE(hashTuple)->t_len); + HashJoinTuple copyTuple; + int bucketno; + int batchno; + + ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue, + &bucketno, &batchno); + Assert(batchno == 0); + + copyTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize); + memcpy(copyTuple, hashTuple, hashTupleSize); + + copyTuple->next.unshared = hashtable->buckets.unshared[bucketno]; + hashtable->buckets.unshared[bucketno] = copyTuple; + + hashtable->spaceUsed += hashTupleSize; + + idx += MAXALIGN(hashTupleSize); + + CHECK_FOR_INTERRUPTS(); + } + + pfree(oldchunks); + oldchunks = nextchunk; + } + +#ifdef HJDEBUG + printf("Hashjoin %p: unbatched %d batches into 1, nbuckets %d => %d\n", + hashtable, hashtable->nbatch_original, + hashtable->nbuckets_original, hashtable->nbuckets); +#endif + + return true; +} + /* * ExecHashTableInsert * insert a tuple into the hash table depending on the hash value diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 202dd866251..596a323c652 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -204,6 +204,7 @@ static TupleTableSlot *ExecHashJoinGetSavedTuple(HashJoinState *hjstate, uint32 *hashvalue, TupleTableSlot *tupleSlot); static bool ExecHashJoinNewBatch(HashJoinState *hjstate); +static void ExecHashJoinUnbatch(HashJoinState *hjstate, HashState *hashNode); static bool ExecParallelHashJoinNewBatch(HashJoinState *hjstate); static void ExecParallelHashJoinPartitionOuter(HashJoinState *hjstate); @@ -375,6 +376,17 @@ ExecHashJoinImpl(PlanState *pstate, bool parallel) return NULL; } + /* + * The batch count was chosen from the planner's estimate of + * the inner side. Now that we have actually read it we know + * how big it really is, so if it would have fit in memory all + * along, collapse the batches before we touch the outer side. + * That saves a bucket-array clear per batch, and saves + * spilling the outer side at all. + */ + if (!parallel && hashtable->nbatch > 1) + ExecHashJoinUnbatch(node, hashNode); + /* * need to remember whether nbatch has increased since we * began scanning the outer relation @@ -1601,6 +1613,72 @@ ExecHashJoinSaveTuple(MinimalTuple tuple, uint32 hashvalue, BufFileWrite(file, tuple, tuple->t_len); } +/* + * ExecHashJoinUnbatch + * collapse the batches once we know the inner side is small enough + * + * ExecHashUnbatch decides whether this is worth doing and rebuilds the bucket + * array for a single batch; what is left for us is to read back the tuples + * that were spilled and get rid of the batch files. + * + * plan_width is only an estimate, so the tuples we read back can turn out to + * need more memory than the sizing decision assumed. In that case + * ExecHashTableInsert starts batching again underneath us. To keep that safe + * we detach the old file array first: a new one is then allocated for the new + * batches, and the tuples we have not read yet are still reachable through our + * own pointer and get redistributed as they are inserted. + */ +static void +ExecHashJoinUnbatch(HashJoinState *hjstate, HashState *hashNode) +{ + HashJoinTable hashtable = hjstate->hj_HashTable; + int oldnbatch = hashtable->nbatch; + Plan *innerPlan = outerPlan((Hash *) hashNode->ps.plan); + BufFile **oldInnerFiles; + BufFile **oldOuterFiles; + int i; + + if (!ExecHashUnbatch(hashtable, innerPlan->plan_width)) + return; + + Assert(hashtable->nbatch == 1); + + oldInnerFiles = hashtable->innerBatchFile; + oldOuterFiles = hashtable->outerBatchFile; + hashtable->innerBatchFile = NULL; + hashtable->outerBatchFile = NULL; + + for (i = 1; i < oldnbatch; i++) + { + BufFile *innerFile = oldInnerFiles[i]; + TupleTableSlot *slot; + uint32 hashvalue; + + /* The outer side has not been scanned yet, so it has no files. */ + Assert(oldOuterFiles[i] == NULL); + + if (innerFile == NULL) + continue; + + if (BufFileSeek(innerFile, 0, 0, SEEK_SET)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not rewind hash-join temporary file"))); + + while ((slot = ExecHashJoinGetSavedTuple(hjstate, + innerFile, + &hashvalue, + hjstate->hj_HashTupleSlot))) + ExecHashTableInsert(hashtable, slot, hashvalue); + + BufFileClose(innerFile); + oldInnerFiles[i] = NULL; + } + + pfree(oldInnerFiles); + pfree(oldOuterFiles); +} + /* * ExecHashJoinGetSavedTuple * read the next tuple from a batch file. Return NULL if no more. diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h index 9ff493b627a..9ae5516c916 100644 --- a/src/include/executor/nodeHash.h +++ b/src/include/executor/nodeHash.h @@ -33,6 +33,7 @@ extern void ExecHashTableDetachBatch(HashJoinTable hashtable); extern void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int batchno); +extern bool ExecHashUnbatch(HashJoinTable hashtable, int tupwidth); extern void ExecHashTableInsert(HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue); -- 2.37.1 (Apple Git-137.1)