hashjoin.h

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * hashjoin.h
 *    internal structures for hash joins
 *
 *
 * Portions Copyright (c) 1996-2021, 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_DONE                  4

/* 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 */