diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 589b2f1..d5acfb1 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -47,6 +47,7 @@ static void ExecHashSkewTableInsert(HashJoinTable hashtable, int bucketNumber); static void ExecHashRemoveNextSkewBucket(HashJoinTable hashtable); +static char * dense_alloc(HashJoinTable hashtable, int tupleSize); /* ---------------------------------------------------------------- * ExecHash @@ -294,6 +295,8 @@ ExecHashTableCreate(Hash *node, List *hashOperators, bool keepNulls) hashtable->spaceAllowedSkew = hashtable->spaceAllowed * SKEW_WORK_MEM_PERCENT / 100; + hashtable->chunks = NULL; + /* * Get info about the hash functions to be used for each hash key. Also * remember whether the join operators are strict. @@ -556,10 +559,10 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) int oldnbatch = hashtable->nbatch; int curbatch = hashtable->curbatch; int nbatch; - int i; MemoryContext oldcxt; long ninmemory; long nfreed; + MemoryChunk oldchunks = hashtable->chunks; /* do nothing if we've decided to shut off growth */ if (!hashtable->growEnabled) @@ -612,51 +615,71 @@ ExecHashIncreaseNumBatches(HashJoinTable hashtable) */ ninmemory = nfreed = 0; - for (i = 0; i < hashtable->nbuckets; i++) + /* + * We will scan through the chunks directly, so we can reset the buckets + * and copy/insert the tuples from scratch. So set all buckets to NULL. + * And throw away the old chunks - we have a pointer in oldchunks, and + * we'll free them once we've copied all the tuples. + */ + memset(hashtable->buckets, 0, sizeof(void*) * hashtable->nbuckets); + hashtable->chunks = NULL; + + /* so, let's scan through the old chunks, and all tuples in each chunk */ + while (oldchunks != NULL) { - HashJoinTuple prevtuple; - HashJoinTuple tuple; - prevtuple = NULL; - tuple = hashtable->buckets[i]; + /* pointer to the next memory chunk (may be NULL for the last chunk) */ + MemoryChunk nextchunk = oldchunks->next; + + /* position within the buffer (up to oldchunks->used) */ + size_t idx = 0; - while (tuple != NULL) + /* process all tuples stored in this chunk (and then free it) */ + while (idx < oldchunks->used) { - /* save link in case we delete */ - HashJoinTuple nexttuple = tuple->next; - int bucketno; - int batchno; + + /* get the hashjoin tuple and mintuple stored right after it */ + HashJoinTuple hashTuple = (HashJoinTuple)(oldchunks->data + idx); + MinimalTuple tuple = HJTUPLE_MINTUPLE(hashTuple); + + int hashTupleSize = (HJTUPLE_OVERHEAD + tuple->t_len); + + int bucketno; + int batchno; ninmemory++; - ExecHashGetBucketAndBatch(hashtable, tuple->hashvalue, + ExecHashGetBucketAndBatch(hashtable, hashTuple->hashvalue, &bucketno, &batchno); - Assert(bucketno == i); + if (batchno == curbatch) { - /* keep tuple */ - prevtuple = tuple; + /* keep tuple - this means we need to copy it into the new chunks */ + HashJoinTuple copyTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize); + memcpy(copyTuple, hashTuple, hashTupleSize); + + /* course add it to the apropriate bucket */ + copyTuple->next = hashtable->buckets[bucketno]; + hashtable->buckets[bucketno] = copyTuple; } else { /* dump it out */ Assert(batchno > curbatch); - ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(tuple), - tuple->hashvalue, + ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(hashTuple), + hashTuple->hashvalue, &hashtable->innerBatchFile[batchno]); - /* and remove from hash table */ - if (prevtuple) - prevtuple->next = nexttuple; - else - hashtable->buckets[i] = nexttuple; - /* prevtuple doesn't change */ - hashtable->spaceUsed -= - HJTUPLE_OVERHEAD + HJTUPLE_MINTUPLE(tuple)->t_len; - pfree(tuple); + + hashtable->spaceUsed -= hashTupleSize; nfreed++; } - tuple = nexttuple; + /* bytes occupied in memory HJ tuple overhead + actual tuple length */ + idx += MAXALIGN(hashTupleSize); } + + /* so, we're done with this chunk - free it and proceed to the next one */ + pfree(oldchunks); + oldchunks = nextchunk; } #ifdef HJDEBUG @@ -717,8 +740,8 @@ ExecHashTableInsert(HashJoinTable hashtable, /* Create the HashJoinTuple */ hashTupleSize = HJTUPLE_OVERHEAD + tuple->t_len; - hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt, - hashTupleSize); + hashTuple = (HashJoinTuple) dense_alloc(hashtable, hashTupleSize); + hashTuple->hashvalue = hashvalue; memcpy(HJTUPLE_MINTUPLE(hashTuple), tuple, tuple->t_len); @@ -1068,6 +1091,9 @@ ExecHashTableReset(HashJoinTable hashtable) hashtable->spaceUsed = 0; MemoryContextSwitchTo(oldcxt); + + /* Forget the chunks (the memory was freed by the context reset above). */ + hashtable->chunks = NULL; } /* @@ -1318,6 +1344,69 @@ ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 hashvalue) return INVALID_SKEW_BUCKET_NO; } +static char * +dense_alloc(HashJoinTable hashtable, int size) +{ + + /* just in case the size is not already aligned properly */ + size = MAXALIGN(size); + + /* if tuple size is larger than of 1/8 of chunk size, allocate a separate chunk */ + if (size > HASH_CHUNK_THRESHOLD) { + + /* allocate new chunk and put it at the beginning of the list */ + MemoryChunk newChunk + = (MemoryChunk)MemoryContextAlloc(hashtable->batchCxt, + offsetof(MemoryChunkData, data) + size); + + newChunk->maxlen = size; + newChunk->used = 0; + newChunk->ntuples = 0; + + /* If there already is a chunk, add if after it so we can still use the space in it */ + if (hashtable->chunks != NULL) { + newChunk->next = hashtable->chunks->next; + hashtable->chunks->next = newChunk; + } else { + newChunk->next = hashtable->chunks; + hashtable->chunks = newChunk; + } + + newChunk->used += size; + newChunk->ntuples += 1; + + return newChunk->data; + + } + + /* + * If it's within the size limit, see if we have enough space for it in the + * current chunk (if there's one). 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 */ + MemoryChunk newChunk + = (MemoryChunk)MemoryContextAlloc(hashtable->batchCxt, + offsetof(MemoryChunkData, data) + HASH_CHUNK_SIZE); + + newChunk->maxlen = HASH_CHUNK_SIZE; + newChunk->used = 0; + newChunk->ntuples = 0; + + newChunk->next = hashtable->chunks; + hashtable->chunks = newChunk; + } + + /* OK, we have enough space in the chunk, let's add the tuple */ + hashtable->chunks->used += size; + hashtable->chunks->ntuples += 1; + + /* allocate pointer to the start of the tuple memory */ + return hashtable->chunks->data + (hashtable->chunks->used - size); +} + + /* * ExecHashSkewTableInsert * diff --git a/src/include/executor/hashjoin.h b/src/include/executor/hashjoin.h index 3beae40..166bb7a 100644 --- a/src/include/executor/hashjoin.h +++ b/src/include/executor/hashjoin.h @@ -102,6 +102,23 @@ typedef struct HashSkewBucket #define SKEW_WORK_MEM_PERCENT 2 #define SKEW_MIN_OUTER_FRACTION 0.01 +/* used for dense allocation of tuples */ +#define HASH_CHUNK_SIZE (32*1024L) +#define HASH_CHUNK_THRESHOLD (HASH_CHUNK_SIZE/4) + +typedef struct MemoryChunkData +{ + int ntuples; /* number of tuples stored in this chunk */ + size_t maxlen; /* length of the buffer */ + size_t used; /* number of chunk bytes already used */ + + struct MemoryChunkData *next; /* pointer to the next chunk (linked list) */ + + char data[1]; /* buffer allocated at the end */ +} MemoryChunkData; + +typedef struct MemoryChunkData* MemoryChunk; + typedef struct HashJoinTableData { @@ -157,6 +174,9 @@ typedef struct HashJoinTableData MemoryContext hashCxt; /* context for whole-hash-join storage */ MemoryContext batchCxt; /* context for this-batch-only storage */ + + /* used for dense allocation of tuples (into linked chunks) */ + MemoryChunk chunks; /* one list for the whole batch */ } HashJoinTableData; #endif /* HASHJOIN_H */