From fcabae7d9105325135be3b39054efa26ad899c53 Mon Sep 17 00:00:00 2001 From: TatsuyaKawata Date: Sat, 14 Nov 2026 18:17:21 +0900 Subject: [PATCH v1 1/2] Link tuples into large private hash join tables in groups With a large bucket array, linking each tuple into its bucket usually misses the cache, and the per-tuple work in between keeps the CPU from overlapping those misses. So prefetch the bucket header, queue the tuple, and link HJ_DEFER_GROUP_SIZE queued tuples at a time in a short loop. This is used for private hash tables with at least 65536 buckets, while building and when relinking tuples after the number of buckets or batches has grown. --- src/backend/executor/nodeHash.c | 110 ++++++++++++++++++++++++++-- src/backend/executor/nodeHashjoin.c | 3 + src/include/c.h | 10 +++ src/include/executor/hashjoin.h | 10 +++ src/include/executor/nodeHash.h | 1 + 5 files changed, 128 insertions(+), 6 deletions(-) diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 8825bb6fa23..354b22e8517 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -45,6 +45,9 @@ static void ExecHashIncreaseNumBatches(HashJoinTable hashtable); static void ExecHashIncreaseNumBuckets(HashJoinTable hashtable); +static void ExecHashSetDeferInserts(HashJoinTable hashtable); +static inline void ExecHashDeferInsert(HashJoinTable hashtable, + HashJoinTuple hashTuple, int bucketno); static void ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable); static void ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable); static void ExecHashBuildSkewHash(HashState *hashstate, @@ -209,6 +212,9 @@ MultiExecPrivateHash(HashState *node) /* else we can discard the tuple immediately */ } + /* link any tuples still queued by ExecHashTableInsert into the buckets */ + ExecHashFlushDeferredInserts(hashtable); + /* resize the hash table if needed (NTUP_PER_BUCKET exceeded) */ if (hashtable->nbuckets != hashtable->nbuckets_optimal) ExecHashIncreaseNumBuckets(hashtable); @@ -516,6 +522,8 @@ ExecHashTableCreate(HashState *state) * subsidiary hashCxt, batchCxt or spillCxt. */ hashtable = palloc_object(HashJoinTableData); + hashtable->defer_inserts = false; + hashtable->ndeferred = 0; hashtable->nbuckets = nbuckets; hashtable->nbuckets_original = nbuckets; hashtable->nbuckets_optimal = nbuckets; @@ -653,6 +661,7 @@ ExecHashTableCreate(HashState *state) MemoryContextSwitchTo(hashtable->batchCxt); hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets); + ExecHashSetDeferInserts(hashtable); /* * Set up for skew optimization, if possible and there's a need for @@ -1061,6 +1070,9 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) long nfreed; HashMemoryChunk oldchunks; + /* the bucket lists must be complete before we rebuild them below */ + ExecHashFlushDeferredInserts(hashtable); + /* do nothing if we've decided to shut off growth */ if (!hashtable->growEnabled) return; @@ -1121,6 +1133,7 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) hashtable->buckets.unshared = repalloc_array(hashtable->buckets.unshared, HashJoinTuple, hashtable->nbuckets); + ExecHashSetDeferInserts(hashtable); } /* @@ -1163,8 +1176,13 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) memcpy(copyTuple, hashTuple, hashTupleSize); /* and add it back to the appropriate bucket */ - copyTuple->next.unshared = hashtable->buckets.unshared[bucketno]; - hashtable->buckets.unshared[bucketno] = copyTuple; + if (hashtable->defer_inserts) + ExecHashDeferInsert(hashtable, copyTuple, bucketno); + else + { + copyTuple->next.unshared = hashtable->buckets.unshared[bucketno]; + hashtable->buckets.unshared[bucketno] = copyTuple; + } } else { @@ -1191,6 +1209,9 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) oldchunks = nextchunk; } + /* link the tuples still queued by ExecHashDeferInsert */ + ExecHashFlushDeferredInserts(hashtable); + #ifdef HJDEBUG printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n", hashtable, nfreed, ninmemory, hashtable->spaceUsed); @@ -1613,6 +1634,9 @@ ExecHashIncreaseNumBuckets(HashJoinTable hashtable) { HashMemoryChunk chunk; + /* our caller has linked any queued tuples */ + Assert(hashtable->ndeferred == 0); + /* do nothing if not an increase (it's called increase for a reason) */ if (hashtable->nbuckets >= hashtable->nbuckets_optimal) return; @@ -1641,6 +1665,7 @@ ExecHashIncreaseNumBuckets(HashJoinTable hashtable) memset(hashtable->buckets.unshared, 0, hashtable->nbuckets * sizeof(HashJoinTuple)); + ExecHashSetDeferInserts(hashtable); /* scan through all tuples in all chunks to rebuild the hash table */ for (chunk = hashtable->chunks; chunk != NULL; chunk = chunk->next.unshared) @@ -1658,8 +1683,13 @@ ExecHashIncreaseNumBuckets(HashJoinTable hashtable) &bucketno, &batchno); /* add the tuple to the proper bucket */ - hashTuple->next.unshared = hashtable->buckets.unshared[bucketno]; - hashtable->buckets.unshared[bucketno] = hashTuple; + if (hashtable->defer_inserts) + ExecHashDeferInsert(hashtable, hashTuple, bucketno); + else + { + hashTuple->next.unshared = hashtable->buckets.unshared[bucketno]; + hashtable->buckets.unshared[bucketno] = hashTuple; + } /* advance index past the tuple */ idx += MAXALIGN(HJTUPLE_OVERHEAD + @@ -1669,6 +1699,9 @@ ExecHashIncreaseNumBuckets(HashJoinTable hashtable) /* allow this loop to be cancellable */ CHECK_FOR_INTERRUPTS(); } + + /* link the tuples still queued by ExecHashDeferInsert */ + ExecHashFlushDeferredInserts(hashtable); } static void @@ -1759,6 +1792,64 @@ ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable) } } +/* + * ExecHashSetDeferInserts + * decide whether tuples should be linked into the private bucket array + * via ExecHashDeferInsert + */ +static void +ExecHashSetDeferInserts(HashJoinTable hashtable) +{ + Assert(hashtable->ndeferred == 0); + hashtable->defer_inserts = + (hashtable->parallel_state == NULL && + hashtable->nbuckets >= HJ_DEFER_MIN_NBUCKETS); +} + +/* + * ExecHashDeferInsert + * queue a tuple for insertion into the given bucket + * + * Callers must use ExecHashFlushDeferredInserts() before reading the bucket + * array. + */ +static pg_always_inline void +ExecHashDeferInsert(HashJoinTable hashtable, HashJoinTuple hashTuple, + int bucketno) +{ + int n = hashtable->ndeferred; + + pg_prefetch_mem(&hashtable->buckets.unshared[bucketno]); + + hashtable->deferred_tuples[n] = hashTuple; + hashtable->deferred_bucketnos[n] = bucketno; + hashtable->ndeferred = ++n; + + if (n == HJ_DEFER_GROUP_SIZE) + ExecHashFlushDeferredInserts(hashtable); +} + +/* + * ExecHashFlushDeferredInserts + * push all tuples queued by ExecHashDeferInsert onto their buckets + */ +void +ExecHashFlushDeferredInserts(HashJoinTable hashtable) +{ + HashJoinTuple *buckets = hashtable->buckets.unshared; + int n = hashtable->ndeferred; + + for (int i = 0; i < n; i++) + { + HashJoinTuple hashTuple = hashtable->deferred_tuples[i]; + int bucketno = hashtable->deferred_bucketnos[i]; + + hashTuple->next.unshared = buckets[bucketno]; + buckets[bucketno] = hashTuple; + } + hashtable->ndeferred = 0; +} + /* * ExecHashTableInsert * insert a tuple into the hash table depending on the hash value @@ -1810,8 +1901,13 @@ ExecHashTableInsert(HashJoinTable hashtable, 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; + if (hashtable->defer_inserts) + ExecHashDeferInsert(hashtable, hashTuple, bucketno); + else + { + hashTuple->next.unshared = hashtable->buckets.unshared[bucketno]; + hashtable->buckets.unshared[bucketno] = hashTuple; + } /* * Increase the (optimal) number of buckets if we just exceeded the @@ -2362,6 +2458,8 @@ ExecHashTableReset(HashJoinTable hashtable) MemoryContextReset(hashtable->batchCxt); oldcxt = MemoryContextSwitchTo(hashtable->batchCxt); + Assert(hashtable->ndeferred == 0); + /* Reallocate and reinitialize the hash bucket headers. */ hashtable->buckets.unshared = palloc0_array(HashJoinTuple, nbuckets); diff --git a/src/backend/executor/nodeHashjoin.c b/src/backend/executor/nodeHashjoin.c index 5df351ac5b6..53b72624b26 100644 --- a/src/backend/executor/nodeHashjoin.c +++ b/src/backend/executor/nodeHashjoin.c @@ -1406,6 +1406,9 @@ ExecHashJoinNewBatch(HashJoinState *hjstate) ExecHashTableInsert(hashtable, slot, hashvalue); } + /* link any tuples still queued by ExecHashTableInsert */ + ExecHashFlushDeferredInserts(hashtable); + /* * after we build the hash table, the inner batch file is no longer * needed diff --git a/src/include/c.h b/src/include/c.h index b285755e8c4..91c7f785534 100644 --- a/src/include/c.h +++ b/src/include/c.h @@ -491,6 +491,16 @@ extern "C++" #define pg_unreachable() abort() #endif +/* + * Hint that the cache line containing addr will be read soon. This is only a + * hint, and does nothing on compilers without a builtin for it. + */ +#if defined(__GNUC__) +#define pg_prefetch_mem(addr) __builtin_prefetch(addr) +#else +#define pg_prefetch_mem(addr) ((void) (addr)) +#endif + /* * Define a compiler-independent macro for determining if an expression is a * compile-time integer const. We don't define this macro to return 0 when diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index e77971ed0d1..f4a0812a71f 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -307,6 +307,10 @@ typedef struct ParallelHashJoinState #define PHJ_GROW_BUCKETS_REINSERT 2 #define PHJ_GROW_BUCKETS_PHASE(n) ((n) % 3) /* circular phases */ +/* Deferred insertion into the bucket array; see ExecHashDeferInsert() */ +#define HJ_DEFER_GROUP_SIZE 32 +#define HJ_DEFER_MIN_NBUCKETS 65536 /* 512kB of bucket headers */ + typedef struct HashJoinTableData { int nbuckets; /* # buckets in the in-memory hash table */ @@ -379,6 +383,12 @@ typedef struct HashJoinTableData ParallelHashJoinState *parallel_state; ParallelHashJoinBatchAccessor *batches; dsa_pointer current_chunk_shared; + + /* tuples waiting to be linked into buckets; see ExecHashDeferInsert() */ + bool defer_inserts; /* use deferred insertion? */ + int ndeferred; /* number of queued tuples */ + HashJoinTuple deferred_tuples[HJ_DEFER_GROUP_SIZE]; + int deferred_bucketnos[HJ_DEFER_GROUP_SIZE]; } HashJoinTableData; #endif /* HASHJOIN_H */ diff --git a/src/include/executor/nodeHash.h b/src/include/executor/nodeHash.h index 9ff493b627a..1e873f882ff 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 void ExecHashFlushDeferredInserts(HashJoinTable hashtable); extern void ExecHashTableInsert(HashJoinTable hashtable, TupleTableSlot *slot, uint32 hashvalue); -- 2.34.1