123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363 |
- /*-------------------------------------------------------------------------
- *
- * hashjoin.h
- * internal structures for hash joins
- *
- *
- * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
- * Portions Copyright (c) 1994, Regents of the University of California
- *
- * src/include/executor/hashjoin.h
- *
- *-------------------------------------------------------------------------
- */
- #ifndef HASHJOIN_H
- #define HASHJOIN_H
- #include "nodes/execnodes.h"
- #include "port/atomics.h"
- #include "storage/barrier.h"
- #include "storage/buffile.h"
- #include "storage/lwlock.h"
- /* ----------------------------------------------------------------
- * 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.)
- *
- * 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.
- *
- * 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
- * into inner-batch temp files. The same statements apply for the
- * first scan of the outer relation, except we write tuples to outer-batch
- * temp files. After finishing the first scan, we do the following for
- * each remaining batch:
- * 1. Read tuples from inner batch file, load into hash buckets.
- * 2. Read tuples from outer batch file, match to hash buckets and output.
- *
- * It is possible to increase nbatch on the fly if the in-memory hash table
- * gets too big. The hash-value-to-batch computation is arranged so that this
- * can only cause a tuple to go into a later batch than previously thought,
- * never into an earlier batch. When we increase nbatch, we rescan the hash
- * table and dump out any tuples that are now of a later batch to the correct
- * inner batch file. Subsequently, while reading either inner or outer batch
- * files, we might find tuples that no longer belong to the current batch;
- * if so, we just dump them out to the correct batch file.
- * ----------------------------------------------------------------
- */
- /* these are in nodes/execnodes.h: */
- /* typedef struct HashJoinTupleData *HashJoinTuple; */
- /* typedef struct HashJoinTableData *HashJoinTable; */
- typedef struct HashJoinTupleData
- {
- /* link to next tuple in same bucket */
- union
- {
- struct HashJoinTupleData *unshared;
- dsa_pointer shared;
- } next;
- uint32 hashvalue; /* tuple's hash code */
- /* Tuple data, in MinimalTuple format, follows on a MAXALIGN boundary */
- } HashJoinTupleData;
- #define HJTUPLE_OVERHEAD MAXALIGN(sizeof(HashJoinTupleData))
- #define HJTUPLE_MINTUPLE(hjtup) \
- ((MinimalTuple) ((char *) (hjtup) + HJTUPLE_OVERHEAD))
- /*
- * If the outer relation's distribution is sufficiently nonuniform, we attempt
- * to optimize the join by treating the hash values corresponding to the outer
- * relation's MCVs specially. Inner relation tuples matching these hash
- * values go into the "skew" hashtable instead of the main hashtable, and
- * outer relation tuples with these hash values are matched against that
- * table instead of the main one. Thus, tuples with these hash values are
- * effectively handled as part of the first batch and will never go to disk.
- * The skew hashtable is limited to SKEW_HASH_MEM_PERCENT of the total memory
- * allowed for the join; while building the hashtables, we decrease the number
- * of MCVs being specially treated if needed to stay under this limit.
- *
- * Note: you might wonder why we look at the outer relation stats for this,
- * rather than the inner. One reason is that the outer relation is typically
- * bigger, so we get more I/O savings by optimizing for its most common values.
- * Also, for similarly-sized relations, the planner prefers to put the more
- * uniformly distributed relation on the inside, so we're more likely to find
- * interesting skew in the outer relation.
- */
- typedef struct HashSkewBucket
- {
- uint32 hashvalue; /* common hash value */
- HashJoinTuple tuples; /* linked list of inner-relation tuples */
- } HashSkewBucket;
- #define SKEW_BUCKET_OVERHEAD MAXALIGN(sizeof(HashSkewBucket))
- #define INVALID_SKEW_BUCKET_NO (-1)
- #define SKEW_HASH_MEM_PERCENT 2
- #define SKEW_MIN_OUTER_FRACTION 0.01
- /*
- * To reduce palloc overhead, the HashJoinTuples for the current batch are
- * packed in 32kB buffers instead of pallocing each tuple individually.
- */
- typedef struct HashMemoryChunkData
- {
- int ntuples; /* number of tuples stored in this chunk */
- size_t maxlen; /* size of the chunk's tuple buffer */
- size_t used; /* number of buffer bytes already used */
- /* pointer to the next chunk (linked list) */
- union
- {
- struct HashMemoryChunkData *unshared;
- dsa_pointer shared;
- } next;
- /*
- * The chunk's tuple buffer starts after the HashMemoryChunkData struct,
- * at offset HASH_CHUNK_HEADER_SIZE (which must be maxaligned). Note that
- * that offset is not included in "maxlen" or "used".
- */
- } HashMemoryChunkData;
- typedef struct HashMemoryChunkData *HashMemoryChunk;
- #define HASH_CHUNK_SIZE (32 * 1024L)
- #define HASH_CHUNK_HEADER_SIZE MAXALIGN(sizeof(HashMemoryChunkData))
- #define HASH_CHUNK_DATA(hc) (((char *) (hc)) + HASH_CHUNK_HEADER_SIZE)
- /* tuples exceeding HASH_CHUNK_THRESHOLD bytes are put in their own chunk */
- #define HASH_CHUNK_THRESHOLD (HASH_CHUNK_SIZE / 4)
- /*
- * For each batch of a Parallel Hash Join, we have a ParallelHashJoinBatch
- * object in shared memory to coordinate access to it. Since they are
- * followed by variable-sized objects, they are arranged in contiguous memory
- * but not accessed directly as an array.
- */
- typedef struct ParallelHashJoinBatch
- {
- dsa_pointer buckets; /* array of hash table buckets */
- Barrier batch_barrier; /* synchronization for joining this batch */
- dsa_pointer chunks; /* chunks of tuples loaded */
- size_t size; /* size of buckets + chunks in memory */
- size_t estimated_size; /* size of buckets + chunks while writing */
- size_t ntuples; /* number of tuples loaded */
- size_t old_ntuples; /* number of tuples before repartitioning */
- bool space_exhausted;
- /*
- * Variable-sized SharedTuplestore objects follow this struct in memory.
- * See the accessor macros below.
- */
- } ParallelHashJoinBatch;
- /* Accessor for inner batch tuplestore following a ParallelHashJoinBatch. */
- #define ParallelHashJoinBatchInner(batch) \
- ((SharedTuplestore *) \
- ((char *) (batch) + MAXALIGN(sizeof(ParallelHashJoinBatch))))
- /* Accessor for outer batch tuplestore following a ParallelHashJoinBatch. */
- #define ParallelHashJoinBatchOuter(batch, nparticipants) \
- ((SharedTuplestore *) \
- ((char *) ParallelHashJoinBatchInner(batch) + \
- MAXALIGN(sts_estimate(nparticipants))))
- /* Total size of a ParallelHashJoinBatch and tuplestores. */
- #define EstimateParallelHashJoinBatch(hashtable) \
- (MAXALIGN(sizeof(ParallelHashJoinBatch)) + \
- MAXALIGN(sts_estimate((hashtable)->parallel_state->nparticipants)) * 2)
- /* Accessor for the nth ParallelHashJoinBatch given the base. */
- #define NthParallelHashJoinBatch(base, n) \
- ((ParallelHashJoinBatch *) \
- ((char *) (base) + \
- EstimateParallelHashJoinBatch(hashtable) * (n)))
- /*
- * Each backend requires a small amount of per-batch state to interact with
- * each ParallelHashJoinBatch.
- */
- typedef struct ParallelHashJoinBatchAccessor
- {
- ParallelHashJoinBatch *shared; /* pointer to shared state */
- /* Per-backend partial counters to reduce contention. */
- size_t preallocated; /* pre-allocated space for this backend */
- size_t ntuples; /* number of tuples */
- size_t size; /* size of partition in memory */
- size_t estimated_size; /* size of partition on disk */
- size_t old_ntuples; /* how many tuples before repartitioning? */
- bool at_least_one_chunk; /* has this backend allocated a chunk? */
- bool done; /* flag to remember that a batch is done */
- SharedTuplestoreAccessor *inner_tuples;
- SharedTuplestoreAccessor *outer_tuples;
- } ParallelHashJoinBatchAccessor;
- /*
- * While hashing the inner relation, any participant might determine that it's
- * time to increase the number of buckets to reduce the load factor or batches
- * to reduce the memory size. This is indicated by setting the growth flag to
- * these values.
- */
- typedef enum ParallelHashGrowth
- {
- /* The current dimensions are sufficient. */
- PHJ_GROWTH_OK,
- /* The load factor is too high, so we need to add buckets. */
- PHJ_GROWTH_NEED_MORE_BUCKETS,
- /* The memory budget would be exhausted, so we need to repartition. */
- PHJ_GROWTH_NEED_MORE_BATCHES,
- /* Repartitioning didn't help last time, so don't try to do that again. */
- PHJ_GROWTH_DISABLED
- } ParallelHashGrowth;
- /*
- * The shared state used to coordinate a Parallel Hash Join. This is stored
- * in the DSM segment.
- */
- typedef struct ParallelHashJoinState
- {
- dsa_pointer batches; /* array of ParallelHashJoinBatch */
- dsa_pointer old_batches; /* previous generation during repartition */
- int nbatch; /* number of batches now */
- int old_nbatch; /* previous number of batches */
- int nbuckets; /* number of buckets */
- ParallelHashGrowth growth; /* control batch/bucket growth */
- dsa_pointer chunk_work_queue; /* chunk work queue */
- int nparticipants;
- size_t space_allowed;
- size_t total_tuples; /* total number of inner tuples */
- LWLock lock; /* lock protecting the above */
- Barrier build_barrier; /* synchronization for the build phases */
- Barrier grow_batches_barrier;
- Barrier grow_buckets_barrier;
- pg_atomic_uint32 distributor; /* counter for load balancing */
- SharedFileSet fileset; /* space for shared temporary files */
- } ParallelHashJoinState;
- /* The phases for building batches, used by build_barrier. */
- #define PHJ_BUILD_ELECTING 0
- #define PHJ_BUILD_ALLOCATING 1
- #define PHJ_BUILD_HASHING_INNER 2
- #define PHJ_BUILD_HASHING_OUTER 3
- #define PHJ_BUILD_RUNNING 4
- #define PHJ_BUILD_DONE 5
- /* The phases for probing each batch, used by for batch_barrier. */
- #define PHJ_BATCH_ELECTING 0
- #define PHJ_BATCH_ALLOCATING 1
- #define PHJ_BATCH_LOADING 2
- #define PHJ_BATCH_PROBING 3
- #define PHJ_BATCH_DONE 4
- /* The phases of batch growth while hashing, for grow_batches_barrier. */
- #define PHJ_GROW_BATCHES_ELECTING 0
- #define PHJ_GROW_BATCHES_ALLOCATING 1
- #define PHJ_GROW_BATCHES_REPARTITIONING 2
- #define PHJ_GROW_BATCHES_DECIDING 3
- #define PHJ_GROW_BATCHES_FINISHING 4
- #define PHJ_GROW_BATCHES_PHASE(n) ((n) % 5) /* circular phases */
- /* The phases of bucket growth while hashing, for grow_buckets_barrier. */
- #define PHJ_GROW_BUCKETS_ELECTING 0
- #define PHJ_GROW_BUCKETS_ALLOCATING 1
- #define PHJ_GROW_BUCKETS_REINSERTING 2
- #define PHJ_GROW_BUCKETS_PHASE(n) ((n) % 3) /* circular phases */
- typedef struct HashJoinTableData
- {
- int nbuckets; /* # buckets in the in-memory hash table */
- int log2_nbuckets; /* its log2 (nbuckets must be a power of 2) */
- int nbuckets_original; /* # buckets when starting the first hash */
- int nbuckets_optimal; /* optimal # buckets (per batch) */
- int log2_nbuckets_optimal; /* log2(nbuckets_optimal) */
- /* buckets[i] is head of list of tuples in i'th in-memory bucket */
- union
- {
- /* unshared array is per-batch storage, as are all the tuples */
- struct HashJoinTupleData **unshared;
- /* shared array is per-query DSA area, as are all the tuples */
- dsa_pointer_atomic *shared;
- } buckets;
- bool keepNulls; /* true to store unmatchable NULL tuples */
- bool skewEnabled; /* are we using skew optimization? */
- HashSkewBucket **skewBucket; /* hashtable of skew buckets */
- int skewBucketLen; /* size of skewBucket array (a power of 2!) */
- int nSkewBuckets; /* number of active skew buckets */
- int *skewBucketNums; /* array indexes of active skew buckets */
- int nbatch; /* number of batches */
- int curbatch; /* current batch #; 0 during 1st pass */
- int nbatch_original; /* nbatch when we started inner scan */
- int nbatch_outstart; /* nbatch when we started outer scan */
- bool growEnabled; /* flag to shut off nbatch increases */
- double totalTuples; /* # tuples obtained from inner plan */
- double partialTuples; /* # tuples obtained from inner plan by me */
- double skewTuples; /* # tuples inserted into skew tuples */
- /*
- * These arrays are allocated for the life of the hash join, but only if
- * nbatch > 1. A file is opened only when we first write a tuple into it
- * (otherwise its pointer remains NULL). Note that the zero'th array
- * elements never get used, since we will process rather than dump out any
- * tuples of batch zero.
- */
- BufFile **innerBatchFile; /* buffered virtual temp file per batch */
- BufFile **outerBatchFile; /* buffered virtual temp file per batch */
- /*
- * Info about the datatype-specific hash functions for the datatypes being
- * hashed. These are arrays of the same length as the number of hash join
- * clauses (hash keys).
- */
- FmgrInfo *outer_hashfunctions; /* lookup data for hash functions */
- FmgrInfo *inner_hashfunctions; /* lookup data for hash functions */
- bool *hashStrict; /* is each hash join operator strict? */
- Oid *collations;
- Size spaceUsed; /* memory space currently used by tuples */
- Size spaceAllowed; /* upper limit for space used */
- Size spacePeak; /* peak space used */
- Size spaceUsedSkew; /* skew hash table's current space usage */
- Size spaceAllowedSkew; /* upper limit for skew hashtable */
- 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) */
- HashMemoryChunk chunks; /* one list for the whole batch */
- /* Shared and private state for Parallel Hash. */
- HashMemoryChunk current_chunk; /* this backend's current chunk */
- dsa_area *area; /* DSA area to allocate memory from */
- ParallelHashJoinState *parallel_state;
- ParallelHashJoinBatchAccessor *batches;
- dsa_pointer current_chunk_shared;
- } HashJoinTableData;
- #endif /* HASHJOIN_H */
|