LCOV - code coverage report
Current view: top level - src/backend/executor - nodeAgg.c (source / functions) Hit Total Coverage
Test: PostgreSQL 19devel Lines: 1427 1507 94.7 %
Date: 2025-10-10 10:17:52 Functions: 57 58 98.3 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * nodeAgg.c
       4             :  *    Routines to handle aggregate nodes.
       5             :  *
       6             :  *    ExecAgg normally evaluates each aggregate in the following steps:
       7             :  *
       8             :  *       transvalue = initcond
       9             :  *       foreach input_tuple do
      10             :  *          transvalue = transfunc(transvalue, input_value(s))
      11             :  *       result = finalfunc(transvalue, direct_argument(s))
      12             :  *
      13             :  *    If a finalfunc is not supplied then the result is just the ending
      14             :  *    value of transvalue.
      15             :  *
      16             :  *    Other behaviors can be selected by the "aggsplit" mode, which exists
      17             :  *    to support partial aggregation.  It is possible to:
      18             :  *    * Skip running the finalfunc, so that the output is always the
      19             :  *    final transvalue state.
      20             :  *    * Substitute the combinefunc for the transfunc, so that transvalue
      21             :  *    states (propagated up from a child partial-aggregation step) are merged
      22             :  *    rather than processing raw input rows.  (The statements below about
      23             :  *    the transfunc apply equally to the combinefunc, when it's selected.)
      24             :  *    * Apply the serializefunc to the output values (this only makes sense
      25             :  *    when skipping the finalfunc, since the serializefunc works on the
      26             :  *    transvalue data type).
      27             :  *    * Apply the deserializefunc to the input values (this only makes sense
      28             :  *    when using the combinefunc, for similar reasons).
      29             :  *    It is the planner's responsibility to connect up Agg nodes using these
      30             :  *    alternate behaviors in a way that makes sense, with partial aggregation
      31             :  *    results being fed to nodes that expect them.
      32             :  *
      33             :  *    If a normal aggregate call specifies DISTINCT or ORDER BY, we sort the
      34             :  *    input tuples and eliminate duplicates (if required) before performing
      35             :  *    the above-depicted process.  (However, we don't do that for ordered-set
      36             :  *    aggregates; their "ORDER BY" inputs are ordinary aggregate arguments
      37             :  *    so far as this module is concerned.)  Note that partial aggregation
      38             :  *    is not supported in these cases, since we couldn't ensure global
      39             :  *    ordering or distinctness of the inputs.
      40             :  *
      41             :  *    If transfunc is marked "strict" in pg_proc and initcond is NULL,
      42             :  *    then the first non-NULL input_value is assigned directly to transvalue,
      43             :  *    and transfunc isn't applied until the second non-NULL input_value.
      44             :  *    The agg's first input type and transtype must be the same in this case!
      45             :  *
      46             :  *    If transfunc is marked "strict" then NULL input_values are skipped,
      47             :  *    keeping the previous transvalue.  If transfunc is not strict then it
      48             :  *    is called for every input tuple and must deal with NULL initcond
      49             :  *    or NULL input_values for itself.
      50             :  *
      51             :  *    If finalfunc is marked "strict" then it is not called when the
      52             :  *    ending transvalue is NULL, instead a NULL result is created
      53             :  *    automatically (this is just the usual handling of strict functions,
      54             :  *    of course).  A non-strict finalfunc can make its own choice of
      55             :  *    what to return for a NULL ending transvalue.
      56             :  *
      57             :  *    Ordered-set aggregates are treated specially in one other way: we
      58             :  *    evaluate any "direct" arguments and pass them to the finalfunc along
      59             :  *    with the transition value.
      60             :  *
      61             :  *    A finalfunc can have additional arguments beyond the transvalue and
      62             :  *    any "direct" arguments, corresponding to the input arguments of the
      63             :  *    aggregate.  These are always just passed as NULL.  Such arguments may be
      64             :  *    needed to allow resolution of a polymorphic aggregate's result type.
      65             :  *
      66             :  *    We compute aggregate input expressions and run the transition functions
      67             :  *    in a temporary econtext (aggstate->tmpcontext).  This is reset at least
      68             :  *    once per input tuple, so when the transvalue datatype is
      69             :  *    pass-by-reference, we have to be careful to copy it into a longer-lived
      70             :  *    memory context, and free the prior value to avoid memory leakage.  We
      71             :  *    store transvalues in another set of econtexts, aggstate->aggcontexts
      72             :  *    (one per grouping set, see below), which are also used for the hashtable
      73             :  *    structures in AGG_HASHED mode.  These econtexts are rescanned, not just
      74             :  *    reset, at group boundaries so that aggregate transition functions can
      75             :  *    register shutdown callbacks via AggRegisterCallback.
      76             :  *
      77             :  *    The node's regular econtext (aggstate->ss.ps.ps_ExprContext) is used to
      78             :  *    run finalize functions and compute the output tuple; this context can be
      79             :  *    reset once per output tuple.
      80             :  *
      81             :  *    The executor's AggState node is passed as the fmgr "context" value in
      82             :  *    all transfunc and finalfunc calls.  It is not recommended that the
      83             :  *    transition functions look at the AggState node directly, but they can
      84             :  *    use AggCheckCallContext() to verify that they are being called by
      85             :  *    nodeAgg.c (and not as ordinary SQL functions).  The main reason a
      86             :  *    transition function might want to know this is so that it can avoid
      87             :  *    palloc'ing a fixed-size pass-by-ref transition value on every call:
      88             :  *    it can instead just scribble on and return its left input.  Ordinarily
      89             :  *    it is completely forbidden for functions to modify pass-by-ref inputs,
      90             :  *    but in the aggregate case we know the left input is either the initial
      91             :  *    transition value or a previous function result, and in either case its
      92             :  *    value need not be preserved.  See int8inc() for an example.  Notice that
      93             :  *    the EEOP_AGG_PLAIN_TRANS step is coded to avoid a data copy step when
      94             :  *    the previous transition value pointer is returned.  It is also possible
      95             :  *    to avoid repeated data copying when the transition value is an expanded
      96             :  *    object: to do that, the transition function must take care to return
      97             :  *    an expanded object that is in a child context of the memory context
      98             :  *    returned by AggCheckCallContext().  Also, some transition functions want
      99             :  *    to store working state in addition to the nominal transition value; they
     100             :  *    can use the memory context returned by AggCheckCallContext() to do that.
     101             :  *
     102             :  *    Note: AggCheckCallContext() is available as of PostgreSQL 9.0.  The
     103             :  *    AggState is available as context in earlier releases (back to 8.1),
     104             :  *    but direct examination of the node is needed to use it before 9.0.
     105             :  *
     106             :  *    As of 9.4, aggregate transition functions can also use AggGetAggref()
     107             :  *    to get hold of the Aggref expression node for their aggregate call.
     108             :  *    This is mainly intended for ordered-set aggregates, which are not
     109             :  *    supported as window functions.  (A regular aggregate function would
     110             :  *    need some fallback logic to use this, since there's no Aggref node
     111             :  *    for a window function.)
     112             :  *
     113             :  *    Grouping sets:
     114             :  *
     115             :  *    A list of grouping sets which is structurally equivalent to a ROLLUP
     116             :  *    clause (e.g. (a,b,c), (a,b), (a)) can be processed in a single pass over
     117             :  *    ordered data.  We do this by keeping a separate set of transition values
     118             :  *    for each grouping set being concurrently processed; for each input tuple
     119             :  *    we update them all, and on group boundaries we reset those states
     120             :  *    (starting at the front of the list) whose grouping values have changed
     121             :  *    (the list of grouping sets is ordered from most specific to least
     122             :  *    specific).
     123             :  *
     124             :  *    Where more complex grouping sets are used, we break them down into
     125             :  *    "phases", where each phase has a different sort order (except phase 0
     126             :  *    which is reserved for hashing).  During each phase but the last, the
     127             :  *    input tuples are additionally stored in a tuplesort which is keyed to the
     128             :  *    next phase's sort order; during each phase but the first, the input
     129             :  *    tuples are drawn from the previously sorted data.  (The sorting of the
     130             :  *    data for the first phase is handled by the planner, as it might be
     131             :  *    satisfied by underlying nodes.)
     132             :  *
     133             :  *    Hashing can be mixed with sorted grouping.  To do this, we have an
     134             :  *    AGG_MIXED strategy that populates the hashtables during the first sorted
     135             :  *    phase, and switches to reading them out after completing all sort phases.
     136             :  *    We can also support AGG_HASHED with multiple hash tables and no sorting
     137             :  *    at all.
     138             :  *
     139             :  *    From the perspective of aggregate transition and final functions, the
     140             :  *    only issue regarding grouping sets is this: a single call site (flinfo)
     141             :  *    of an aggregate function may be used for updating several different
     142             :  *    transition values in turn. So the function must not cache in the flinfo
     143             :  *    anything which logically belongs as part of the transition value (most
     144             :  *    importantly, the memory context in which the transition value exists).
     145             :  *    The support API functions (AggCheckCallContext, AggRegisterCallback) are
     146             :  *    sensitive to the grouping set for which the aggregate function is
     147             :  *    currently being called.
     148             :  *
     149             :  *    Plan structure:
     150             :  *
     151             :  *    What we get from the planner is actually one "real" Agg node which is
     152             :  *    part of the plan tree proper, but which optionally has an additional list
     153             :  *    of Agg nodes hung off the side via the "chain" field.  This is because an
     154             :  *    Agg node happens to be a convenient representation of all the data we
     155             :  *    need for grouping sets.
     156             :  *
     157             :  *    For many purposes, we treat the "real" node as if it were just the first
     158             :  *    node in the chain.  The chain must be ordered such that hashed entries
     159             :  *    come before sorted/plain entries; the real node is marked AGG_MIXED if
     160             :  *    there are both types present (in which case the real node describes one
     161             :  *    of the hashed groupings, other AGG_HASHED nodes may optionally follow in
     162             :  *    the chain, followed in turn by AGG_SORTED or (one) AGG_PLAIN node).  If
     163             :  *    the real node is marked AGG_HASHED or AGG_SORTED, then all the chained
     164             :  *    nodes must be of the same type; if it is AGG_PLAIN, there can be no
     165             :  *    chained nodes.
     166             :  *
     167             :  *    We collect all hashed nodes into a single "phase", numbered 0, and create
     168             :  *    a sorted phase (numbered 1..n) for each AGG_SORTED or AGG_PLAIN node.
     169             :  *    Phase 0 is allocated even if there are no hashes, but remains unused in
     170             :  *    that case.
     171             :  *
     172             :  *    AGG_HASHED nodes actually refer to only a single grouping set each,
     173             :  *    because for each hashed grouping we need a separate grpColIdx and
     174             :  *    numGroups estimate.  AGG_SORTED nodes represent a "rollup", a list of
     175             :  *    grouping sets that share a sort order.  Each AGG_SORTED node other than
     176             :  *    the first one has an associated Sort node which describes the sort order
     177             :  *    to be used; the first sorted node takes its input from the outer subtree,
     178             :  *    which the planner has already arranged to provide ordered data.
     179             :  *
     180             :  *    Memory and ExprContext usage:
     181             :  *
     182             :  *    Because we're accumulating aggregate values across input rows, we need to
     183             :  *    use more memory contexts than just simple input/output tuple contexts.
     184             :  *    In fact, for a rollup, we need a separate context for each grouping set
     185             :  *    so that we can reset the inner (finer-grained) aggregates on their group
     186             :  *    boundaries while continuing to accumulate values for outer
     187             :  *    (coarser-grained) groupings.  On top of this, we might be simultaneously
     188             :  *    populating hashtables; however, we only need one context for all the
     189             :  *    hashtables.
     190             :  *
     191             :  *    So we create an array, aggcontexts, with an ExprContext for each grouping
     192             :  *    set in the largest rollup that we're going to process, and use the
     193             :  *    per-tuple memory context of those ExprContexts to store the aggregate
     194             :  *    transition values.  hashcontext is the single context created to support
     195             :  *    all hash tables.
     196             :  *
     197             :  *    Spilling To Disk
     198             :  *
     199             :  *    When performing hash aggregation, if the hash table memory exceeds the
     200             :  *    limit (see hash_agg_check_limits()), we enter "spill mode". In spill
     201             :  *    mode, we advance the transition states only for groups already in the
     202             :  *    hash table. For tuples that would need to create a new hash table
     203             :  *    entries (and initialize new transition states), we instead spill them to
     204             :  *    disk to be processed later. The tuples are spilled in a partitioned
     205             :  *    manner, so that subsequent batches are smaller and less likely to exceed
     206             :  *    hash_mem (if a batch does exceed hash_mem, it must be spilled
     207             :  *    recursively).
     208             :  *
     209             :  *    Spilled data is written to logical tapes. These provide better control
     210             :  *    over memory usage, disk space, and the number of files than if we were
     211             :  *    to use a BufFile for each spill.  We don't know the number of tapes needed
     212             :  *    at the start of the algorithm (because it can recurse), so a tape set is
     213             :  *    allocated at the beginning, and individual tapes are created as needed.
     214             :  *    As a particular tape is read, logtape.c recycles its disk space. When a
     215             :  *    tape is read to completion, it is destroyed entirely.
     216             :  *
     217             :  *    Tapes' buffers can take up substantial memory when many tapes are open at
     218             :  *    once. We only need one tape open at a time in read mode (using a buffer
     219             :  *    that's a multiple of BLCKSZ); but we need one tape open in write mode (each
     220             :  *    requiring a buffer of size BLCKSZ) for each partition.
     221             :  *
     222             :  *    Note that it's possible for transition states to start small but then
     223             :  *    grow very large; for instance in the case of ARRAY_AGG. In such cases,
     224             :  *    it's still possible to significantly exceed hash_mem. We try to avoid
     225             :  *    this situation by estimating what will fit in the available memory, and
     226             :  *    imposing a limit on the number of groups separately from the amount of
     227             :  *    memory consumed.
     228             :  *
     229             :  *    Transition / Combine function invocation:
     230             :  *
     231             :  *    For performance reasons transition functions, including combine
     232             :  *    functions, aren't invoked one-by-one from nodeAgg.c after computing
     233             :  *    arguments using the expression evaluation engine. Instead
     234             :  *    ExecBuildAggTrans() builds one large expression that does both argument
     235             :  *    evaluation and transition function invocation. That avoids performance
     236             :  *    issues due to repeated uses of expression evaluation, complications due
     237             :  *    to filter expressions having to be evaluated early, and allows to JIT
     238             :  *    the entire expression into one native function.
     239             :  *
     240             :  * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
     241             :  * Portions Copyright (c) 1994, Regents of the University of California
     242             :  *
     243             :  * IDENTIFICATION
     244             :  *    src/backend/executor/nodeAgg.c
     245             :  *
     246             :  *-------------------------------------------------------------------------
     247             :  */
     248             : 
     249             : #include "postgres.h"
     250             : 
     251             : #include "access/htup_details.h"
     252             : #include "access/parallel.h"
     253             : #include "catalog/objectaccess.h"
     254             : #include "catalog/pg_aggregate.h"
     255             : #include "catalog/pg_proc.h"
     256             : #include "catalog/pg_type.h"
     257             : #include "common/hashfn.h"
     258             : #include "executor/execExpr.h"
     259             : #include "executor/executor.h"
     260             : #include "executor/nodeAgg.h"
     261             : #include "lib/hyperloglog.h"
     262             : #include "miscadmin.h"
     263             : #include "nodes/nodeFuncs.h"
     264             : #include "optimizer/optimizer.h"
     265             : #include "parser/parse_agg.h"
     266             : #include "parser/parse_coerce.h"
     267             : #include "utils/acl.h"
     268             : #include "utils/builtins.h"
     269             : #include "utils/datum.h"
     270             : #include "utils/expandeddatum.h"
     271             : #include "utils/injection_point.h"
     272             : #include "utils/logtape.h"
     273             : #include "utils/lsyscache.h"
     274             : #include "utils/memutils.h"
     275             : #include "utils/memutils_memorychunk.h"
     276             : #include "utils/syscache.h"
     277             : #include "utils/tuplesort.h"
     278             : 
     279             : /*
     280             :  * Control how many partitions are created when spilling HashAgg to
     281             :  * disk.
     282             :  *
     283             :  * HASHAGG_PARTITION_FACTOR is multiplied by the estimated number of
     284             :  * partitions needed such that each partition will fit in memory. The factor
     285             :  * is set higher than one because there's not a high cost to having a few too
     286             :  * many partitions, and it makes it less likely that a partition will need to
     287             :  * be spilled recursively. Another benefit of having more, smaller partitions
     288             :  * is that small hash tables may perform better than large ones due to memory
     289             :  * caching effects.
     290             :  *
     291             :  * We also specify a min and max number of partitions per spill. Too few might
     292             :  * mean a lot of wasted I/O from repeated spilling of the same tuples. Too
     293             :  * many will result in lots of memory wasted buffering the spill files (which
     294             :  * could instead be spent on a larger hash table).
     295             :  */
     296             : #define HASHAGG_PARTITION_FACTOR 1.50
     297             : #define HASHAGG_MIN_PARTITIONS 4
     298             : #define HASHAGG_MAX_PARTITIONS 1024
     299             : 
     300             : /*
     301             :  * For reading from tapes, the buffer size must be a multiple of
     302             :  * BLCKSZ. Larger values help when reading from multiple tapes concurrently,
     303             :  * but that doesn't happen in HashAgg, so we simply use BLCKSZ. Writing to a
     304             :  * tape always uses a buffer of size BLCKSZ.
     305             :  */
     306             : #define HASHAGG_READ_BUFFER_SIZE BLCKSZ
     307             : #define HASHAGG_WRITE_BUFFER_SIZE BLCKSZ
     308             : 
     309             : /*
     310             :  * HyperLogLog is used for estimating the cardinality of the spilled tuples in
     311             :  * a given partition. 5 bits corresponds to a size of about 32 bytes and a
     312             :  * worst-case error of around 18%. That's effective enough to choose a
     313             :  * reasonable number of partitions when recursing.
     314             :  */
     315             : #define HASHAGG_HLL_BIT_WIDTH 5
     316             : 
     317             : /*
     318             :  * Assume the palloc overhead always uses sizeof(MemoryChunk) bytes.
     319             :  */
     320             : #define CHUNKHDRSZ sizeof(MemoryChunk)
     321             : 
     322             : /*
     323             :  * Represents partitioned spill data for a single hashtable. Contains the
     324             :  * necessary information to route tuples to the correct partition, and to
     325             :  * transform the spilled data into new batches.
     326             :  *
     327             :  * The high bits are used for partition selection (when recursing, we ignore
     328             :  * the bits that have already been used for partition selection at an earlier
     329             :  * level).
     330             :  */
     331             : typedef struct HashAggSpill
     332             : {
     333             :     int         npartitions;    /* number of partitions */
     334             :     LogicalTape **partitions;   /* spill partition tapes */
     335             :     int64      *ntuples;        /* number of tuples in each partition */
     336             :     uint32      mask;           /* mask to find partition from hash value */
     337             :     int         shift;          /* after masking, shift by this amount */
     338             :     hyperLogLogState *hll_card; /* cardinality estimate for contents */
     339             : } HashAggSpill;
     340             : 
     341             : /*
     342             :  * Represents work to be done for one pass of hash aggregation (with only one
     343             :  * grouping set).
     344             :  *
     345             :  * Also tracks the bits of the hash already used for partition selection by
     346             :  * earlier iterations, so that this batch can use new bits. If all bits have
     347             :  * already been used, no partitioning will be done (any spilled data will go
     348             :  * to a single output tape).
     349             :  */
     350             : typedef struct HashAggBatch
     351             : {
     352             :     int         setno;          /* grouping set */
     353             :     int         used_bits;      /* number of bits of hash already used */
     354             :     LogicalTape *input_tape;    /* input partition tape */
     355             :     int64       input_tuples;   /* number of tuples in this batch */
     356             :     double      input_card;     /* estimated group cardinality */
     357             : } HashAggBatch;
     358             : 
     359             : /* used to find referenced colnos */
     360             : typedef struct FindColsContext
     361             : {
     362             :     bool        is_aggref;      /* is under an aggref */
     363             :     Bitmapset  *aggregated;     /* column references under an aggref */
     364             :     Bitmapset  *unaggregated;   /* other column references */
     365             : } FindColsContext;
     366             : 
     367             : static void select_current_set(AggState *aggstate, int setno, bool is_hash);
     368             : static void initialize_phase(AggState *aggstate, int newphase);
     369             : static TupleTableSlot *fetch_input_tuple(AggState *aggstate);
     370             : static void initialize_aggregates(AggState *aggstate,
     371             :                                   AggStatePerGroup *pergroups,
     372             :                                   int numReset);
     373             : static void advance_transition_function(AggState *aggstate,
     374             :                                         AggStatePerTrans pertrans,
     375             :                                         AggStatePerGroup pergroupstate);
     376             : static void advance_aggregates(AggState *aggstate);
     377             : static void process_ordered_aggregate_single(AggState *aggstate,
     378             :                                              AggStatePerTrans pertrans,
     379             :                                              AggStatePerGroup pergroupstate);
     380             : static void process_ordered_aggregate_multi(AggState *aggstate,
     381             :                                             AggStatePerTrans pertrans,
     382             :                                             AggStatePerGroup pergroupstate);
     383             : static void finalize_aggregate(AggState *aggstate,
     384             :                                AggStatePerAgg peragg,
     385             :                                AggStatePerGroup pergroupstate,
     386             :                                Datum *resultVal, bool *resultIsNull);
     387             : static void finalize_partialaggregate(AggState *aggstate,
     388             :                                       AggStatePerAgg peragg,
     389             :                                       AggStatePerGroup pergroupstate,
     390             :                                       Datum *resultVal, bool *resultIsNull);
     391             : static inline void prepare_hash_slot(AggStatePerHash perhash,
     392             :                                      TupleTableSlot *inputslot,
     393             :                                      TupleTableSlot *hashslot);
     394             : static void prepare_projection_slot(AggState *aggstate,
     395             :                                     TupleTableSlot *slot,
     396             :                                     int currentSet);
     397             : static void finalize_aggregates(AggState *aggstate,
     398             :                                 AggStatePerAgg peraggs,
     399             :                                 AggStatePerGroup pergroup);
     400             : static TupleTableSlot *project_aggregates(AggState *aggstate);
     401             : static void find_cols(AggState *aggstate, Bitmapset **aggregated,
     402             :                       Bitmapset **unaggregated);
     403             : static bool find_cols_walker(Node *node, FindColsContext *context);
     404             : static void build_hash_tables(AggState *aggstate);
     405             : static void build_hash_table(AggState *aggstate, int setno, long nbuckets);
     406             : static void hashagg_recompile_expressions(AggState *aggstate, bool minslot,
     407             :                                           bool nullcheck);
     408             : static void hash_create_memory(AggState *aggstate);
     409             : static long hash_choose_num_buckets(double hashentrysize,
     410             :                                     long ngroups, Size memory);
     411             : static int  hash_choose_num_partitions(double input_groups,
     412             :                                        double hashentrysize,
     413             :                                        int used_bits,
     414             :                                        int *log2_npartitions);
     415             : static void initialize_hash_entry(AggState *aggstate,
     416             :                                   TupleHashTable hashtable,
     417             :                                   TupleHashEntry entry);
     418             : static void lookup_hash_entries(AggState *aggstate);
     419             : static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
     420             : static void agg_fill_hash_table(AggState *aggstate);
     421             : static bool agg_refill_hash_table(AggState *aggstate);
     422             : static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
     423             : static TupleTableSlot *agg_retrieve_hash_table_in_memory(AggState *aggstate);
     424             : static void hash_agg_check_limits(AggState *aggstate);
     425             : static void hash_agg_enter_spill_mode(AggState *aggstate);
     426             : static void hash_agg_update_metrics(AggState *aggstate, bool from_tape,
     427             :                                     int npartitions);
     428             : static void hashagg_finish_initial_spills(AggState *aggstate);
     429             : static void hashagg_reset_spill_state(AggState *aggstate);
     430             : static HashAggBatch *hashagg_batch_new(LogicalTape *input_tape, int setno,
     431             :                                        int64 input_tuples, double input_card,
     432             :                                        int used_bits);
     433             : static MinimalTuple hashagg_batch_read(HashAggBatch *batch, uint32 *hashp);
     434             : static void hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset,
     435             :                                int used_bits, double input_groups,
     436             :                                double hashentrysize);
     437             : static Size hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill,
     438             :                                 TupleTableSlot *inputslot, uint32 hash);
     439             : static void hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill,
     440             :                                  int setno);
     441             : static Datum GetAggInitVal(Datum textInitVal, Oid transtype);
     442             : static void build_pertrans_for_aggref(AggStatePerTrans pertrans,
     443             :                                       AggState *aggstate, EState *estate,
     444             :                                       Aggref *aggref, Oid transfn_oid,
     445             :                                       Oid aggtranstype, Oid aggserialfn,
     446             :                                       Oid aggdeserialfn, Datum initValue,
     447             :                                       bool initValueIsNull, Oid *inputTypes,
     448             :                                       int numArguments);
     449             : 
     450             : 
     451             : /*
     452             :  * Select the current grouping set; affects current_set and
     453             :  * curaggcontext.
     454             :  */
     455             : static void
     456     7901936 : select_current_set(AggState *aggstate, int setno, bool is_hash)
     457             : {
     458             :     /*
     459             :      * When changing this, also adapt ExecAggPlainTransByVal() and
     460             :      * ExecAggPlainTransByRef().
     461             :      */
     462     7901936 :     if (is_hash)
     463     7228762 :         aggstate->curaggcontext = aggstate->hashcontext;
     464             :     else
     465      673174 :         aggstate->curaggcontext = aggstate->aggcontexts[setno];
     466             : 
     467     7901936 :     aggstate->current_set = setno;
     468     7901936 : }
     469             : 
     470             : /*
     471             :  * Switch to phase "newphase", which must either be 0 or 1 (to reset) or
     472             :  * current_phase + 1. Juggle the tuplesorts accordingly.
     473             :  *
     474             :  * Phase 0 is for hashing, which we currently handle last in the AGG_MIXED
     475             :  * case, so when entering phase 0, all we need to do is drop open sorts.
     476             :  */
     477             : static void
     478       89324 : initialize_phase(AggState *aggstate, int newphase)
     479             : {
     480             :     Assert(newphase <= 1 || newphase == aggstate->current_phase + 1);
     481             : 
     482             :     /*
     483             :      * Whatever the previous state, we're now done with whatever input
     484             :      * tuplesort was in use.
     485             :      */
     486       89324 :     if (aggstate->sort_in)
     487             :     {
     488          42 :         tuplesort_end(aggstate->sort_in);
     489          42 :         aggstate->sort_in = NULL;
     490             :     }
     491             : 
     492       89324 :     if (newphase <= 1)
     493             :     {
     494             :         /*
     495             :          * Discard any existing output tuplesort.
     496             :          */
     497       89126 :         if (aggstate->sort_out)
     498             :         {
     499           6 :             tuplesort_end(aggstate->sort_out);
     500           6 :             aggstate->sort_out = NULL;
     501             :         }
     502             :     }
     503             :     else
     504             :     {
     505             :         /*
     506             :          * The old output tuplesort becomes the new input one, and this is the
     507             :          * right time to actually sort it.
     508             :          */
     509         198 :         aggstate->sort_in = aggstate->sort_out;
     510         198 :         aggstate->sort_out = NULL;
     511             :         Assert(aggstate->sort_in);
     512         198 :         tuplesort_performsort(aggstate->sort_in);
     513             :     }
     514             : 
     515             :     /*
     516             :      * If this isn't the last phase, we need to sort appropriately for the
     517             :      * next phase in sequence.
     518             :      */
     519       89324 :     if (newphase > 0 && newphase < aggstate->numphases - 1)
     520             :     {
     521         246 :         Sort       *sortnode = aggstate->phases[newphase + 1].sortnode;
     522         246 :         PlanState  *outerNode = outerPlanState(aggstate);
     523         246 :         TupleDesc   tupDesc = ExecGetResultType(outerNode);
     524             : 
     525         246 :         aggstate->sort_out = tuplesort_begin_heap(tupDesc,
     526             :                                                   sortnode->numCols,
     527             :                                                   sortnode->sortColIdx,
     528             :                                                   sortnode->sortOperators,
     529             :                                                   sortnode->collations,
     530             :                                                   sortnode->nullsFirst,
     531             :                                                   work_mem,
     532             :                                                   NULL, TUPLESORT_NONE);
     533             :     }
     534             : 
     535       89324 :     aggstate->current_phase = newphase;
     536       89324 :     aggstate->phase = &aggstate->phases[newphase];
     537       89324 : }
     538             : 
     539             : /*
     540             :  * Fetch a tuple from either the outer plan (for phase 1) or from the sorter
     541             :  * populated by the previous phase.  Copy it to the sorter for the next phase
     542             :  * if any.
     543             :  *
     544             :  * Callers cannot rely on memory for tuple in returned slot remaining valid
     545             :  * past any subsequently fetched tuple.
     546             :  */
     547             : static TupleTableSlot *
     548    28626184 : fetch_input_tuple(AggState *aggstate)
     549             : {
     550             :     TupleTableSlot *slot;
     551             : 
     552    28626184 :     if (aggstate->sort_in)
     553             :     {
     554             :         /* make sure we check for interrupts in either path through here */
     555      294894 :         CHECK_FOR_INTERRUPTS();
     556      294894 :         if (!tuplesort_gettupleslot(aggstate->sort_in, true, false,
     557             :                                     aggstate->sort_slot, NULL))
     558         198 :             return NULL;
     559      294696 :         slot = aggstate->sort_slot;
     560             :     }
     561             :     else
     562    28331290 :         slot = ExecProcNode(outerPlanState(aggstate));
     563             : 
     564    28625896 :     if (!TupIsNull(slot) && aggstate->sort_out)
     565      294696 :         tuplesort_puttupleslot(aggstate->sort_out, slot);
     566             : 
     567    28625896 :     return slot;
     568             : }
     569             : 
     570             : /*
     571             :  * (Re)Initialize an individual aggregate.
     572             :  *
     573             :  * This function handles only one grouping set, already set in
     574             :  * aggstate->current_set.
     575             :  *
     576             :  * When called, CurrentMemoryContext should be the per-query context.
     577             :  */
     578             : static void
     579     1132824 : initialize_aggregate(AggState *aggstate, AggStatePerTrans pertrans,
     580             :                      AggStatePerGroup pergroupstate)
     581             : {
     582             :     /*
     583             :      * Start a fresh sort operation for each DISTINCT/ORDER BY aggregate.
     584             :      */
     585     1132824 :     if (pertrans->aggsortrequired)
     586             :     {
     587             :         /*
     588             :          * In case of rescan, maybe there could be an uncompleted sort
     589             :          * operation?  Clean it up if so.
     590             :          */
     591       53842 :         if (pertrans->sortstates[aggstate->current_set])
     592           0 :             tuplesort_end(pertrans->sortstates[aggstate->current_set]);
     593             : 
     594             : 
     595             :         /*
     596             :          * We use a plain Datum sorter when there's a single input column;
     597             :          * otherwise sort the full tuple.  (See comments for
     598             :          * process_ordered_aggregate_single.)
     599             :          */
     600       53842 :         if (pertrans->numInputs == 1)
     601             :         {
     602       53758 :             Form_pg_attribute attr = TupleDescAttr(pertrans->sortdesc, 0);
     603             : 
     604       53758 :             pertrans->sortstates[aggstate->current_set] =
     605       53758 :                 tuplesort_begin_datum(attr->atttypid,
     606       53758 :                                       pertrans->sortOperators[0],
     607       53758 :                                       pertrans->sortCollations[0],
     608       53758 :                                       pertrans->sortNullsFirst[0],
     609             :                                       work_mem, NULL, TUPLESORT_NONE);
     610             :         }
     611             :         else
     612          84 :             pertrans->sortstates[aggstate->current_set] =
     613          84 :                 tuplesort_begin_heap(pertrans->sortdesc,
     614             :                                      pertrans->numSortCols,
     615             :                                      pertrans->sortColIdx,
     616             :                                      pertrans->sortOperators,
     617             :                                      pertrans->sortCollations,
     618             :                                      pertrans->sortNullsFirst,
     619             :                                      work_mem, NULL, TUPLESORT_NONE);
     620             :     }
     621             : 
     622             :     /*
     623             :      * (Re)set transValue to the initial value.
     624             :      *
     625             :      * Note that when the initial value is pass-by-ref, we must copy it (into
     626             :      * the aggcontext) since we will pfree the transValue later.
     627             :      */
     628     1132824 :     if (pertrans->initValueIsNull)
     629      597740 :         pergroupstate->transValue = pertrans->initValue;
     630             :     else
     631             :     {
     632             :         MemoryContext oldContext;
     633             : 
     634      535084 :         oldContext = MemoryContextSwitchTo(aggstate->curaggcontext->ecxt_per_tuple_memory);
     635     1070168 :         pergroupstate->transValue = datumCopy(pertrans->initValue,
     636      535084 :                                               pertrans->transtypeByVal,
     637      535084 :                                               pertrans->transtypeLen);
     638      535084 :         MemoryContextSwitchTo(oldContext);
     639             :     }
     640     1132824 :     pergroupstate->transValueIsNull = pertrans->initValueIsNull;
     641             : 
     642             :     /*
     643             :      * If the initial value for the transition state doesn't exist in the
     644             :      * pg_aggregate table then we will let the first non-NULL value returned
     645             :      * from the outer procNode become the initial value. (This is useful for
     646             :      * aggregates like max() and min().) The noTransValue flag signals that we
     647             :      * still need to do this.
     648             :      */
     649     1132824 :     pergroupstate->noTransValue = pertrans->initValueIsNull;
     650     1132824 : }
     651             : 
     652             : /*
     653             :  * Initialize all aggregate transition states for a new group of input values.
     654             :  *
     655             :  * If there are multiple grouping sets, we initialize only the first numReset
     656             :  * of them (the grouping sets are ordered so that the most specific one, which
     657             :  * is reset most often, is first). As a convenience, if numReset is 0, we
     658             :  * reinitialize all sets.
     659             :  *
     660             :  * NB: This cannot be used for hash aggregates, as for those the grouping set
     661             :  * number has to be specified from further up.
     662             :  *
     663             :  * When called, CurrentMemoryContext should be the per-query context.
     664             :  */
     665             : static void
     666      301864 : initialize_aggregates(AggState *aggstate,
     667             :                       AggStatePerGroup *pergroups,
     668             :                       int numReset)
     669             : {
     670             :     int         transno;
     671      301864 :     int         numGroupingSets = Max(aggstate->phase->numsets, 1);
     672      301864 :     int         setno = 0;
     673      301864 :     int         numTrans = aggstate->numtrans;
     674      301864 :     AggStatePerTrans transstates = aggstate->pertrans;
     675             : 
     676      301864 :     if (numReset == 0)
     677           0 :         numReset = numGroupingSets;
     678             : 
     679      617890 :     for (setno = 0; setno < numReset; setno++)
     680             :     {
     681      316026 :         AggStatePerGroup pergroup = pergroups[setno];
     682             : 
     683      316026 :         select_current_set(aggstate, setno, false);
     684             : 
     685      984106 :         for (transno = 0; transno < numTrans; transno++)
     686             :         {
     687      668080 :             AggStatePerTrans pertrans = &transstates[transno];
     688      668080 :             AggStatePerGroup pergroupstate = &pergroup[transno];
     689             : 
     690      668080 :             initialize_aggregate(aggstate, pertrans, pergroupstate);
     691             :         }
     692             :     }
     693      301864 : }
     694             : 
     695             : /*
     696             :  * Given new input value(s), advance the transition function of one aggregate
     697             :  * state within one grouping set only (already set in aggstate->current_set)
     698             :  *
     699             :  * The new values (and null flags) have been preloaded into argument positions
     700             :  * 1 and up in pertrans->transfn_fcinfo, so that we needn't copy them again to
     701             :  * pass to the transition function.  We also expect that the static fields of
     702             :  * the fcinfo are already initialized; that was done by ExecInitAgg().
     703             :  *
     704             :  * It doesn't matter which memory context this is called in.
     705             :  */
     706             : static void
     707      724218 : advance_transition_function(AggState *aggstate,
     708             :                             AggStatePerTrans pertrans,
     709             :                             AggStatePerGroup pergroupstate)
     710             : {
     711      724218 :     FunctionCallInfo fcinfo = pertrans->transfn_fcinfo;
     712             :     MemoryContext oldContext;
     713             :     Datum       newVal;
     714             : 
     715      724218 :     if (pertrans->transfn.fn_strict)
     716             :     {
     717             :         /*
     718             :          * For a strict transfn, nothing happens when there's a NULL input; we
     719             :          * just keep the prior transValue.
     720             :          */
     721      225000 :         int         numTransInputs = pertrans->numTransInputs;
     722             :         int         i;
     723             : 
     724      450000 :         for (i = 1; i <= numTransInputs; i++)
     725             :         {
     726      225000 :             if (fcinfo->args[i].isnull)
     727           0 :                 return;
     728             :         }
     729      225000 :         if (pergroupstate->noTransValue)
     730             :         {
     731             :             /*
     732             :              * transValue has not been initialized. This is the first non-NULL
     733             :              * input value. We use it as the initial value for transValue. (We
     734             :              * already checked that the agg's input type is binary-compatible
     735             :              * with its transtype, so straight copy here is OK.)
     736             :              *
     737             :              * We must copy the datum into aggcontext if it is pass-by-ref. We
     738             :              * do not need to pfree the old transValue, since it's NULL.
     739             :              */
     740           0 :             oldContext = MemoryContextSwitchTo(aggstate->curaggcontext->ecxt_per_tuple_memory);
     741           0 :             pergroupstate->transValue = datumCopy(fcinfo->args[1].value,
     742           0 :                                                   pertrans->transtypeByVal,
     743           0 :                                                   pertrans->transtypeLen);
     744           0 :             pergroupstate->transValueIsNull = false;
     745           0 :             pergroupstate->noTransValue = false;
     746           0 :             MemoryContextSwitchTo(oldContext);
     747           0 :             return;
     748             :         }
     749      225000 :         if (pergroupstate->transValueIsNull)
     750             :         {
     751             :             /*
     752             :              * Don't call a strict function with NULL inputs.  Note it is
     753             :              * possible to get here despite the above tests, if the transfn is
     754             :              * strict *and* returned a NULL on a prior cycle. If that happens
     755             :              * we will propagate the NULL all the way to the end.
     756             :              */
     757           0 :             return;
     758             :         }
     759             :     }
     760             : 
     761             :     /* We run the transition functions in per-input-tuple memory context */
     762      724218 :     oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory);
     763             : 
     764             :     /* set up aggstate->curpertrans for AggGetAggref() */
     765      724218 :     aggstate->curpertrans = pertrans;
     766             : 
     767             :     /*
     768             :      * OK to call the transition function
     769             :      */
     770      724218 :     fcinfo->args[0].value = pergroupstate->transValue;
     771      724218 :     fcinfo->args[0].isnull = pergroupstate->transValueIsNull;
     772      724218 :     fcinfo->isnull = false;      /* just in case transfn doesn't set it */
     773             : 
     774      724218 :     newVal = FunctionCallInvoke(fcinfo);
     775             : 
     776      724218 :     aggstate->curpertrans = NULL;
     777             : 
     778             :     /*
     779             :      * If pass-by-ref datatype, must copy the new value into aggcontext and
     780             :      * free the prior transValue.  But if transfn returned a pointer to its
     781             :      * first input, we don't need to do anything.
     782             :      *
     783             :      * It's safe to compare newVal with pergroup->transValue without regard
     784             :      * for either being NULL, because ExecAggCopyTransValue takes care to set
     785             :      * transValue to 0 when NULL. Otherwise we could end up accidentally not
     786             :      * reparenting, when the transValue has the same numerical value as
     787             :      * newValue, despite being NULL.  This is a somewhat hot path, making it
     788             :      * undesirable to instead solve this with another branch for the common
     789             :      * case of the transition function returning its (modified) input
     790             :      * argument.
     791             :      */
     792      724218 :     if (!pertrans->transtypeByVal &&
     793           0 :         DatumGetPointer(newVal) != DatumGetPointer(pergroupstate->transValue))
     794           0 :         newVal = ExecAggCopyTransValue(aggstate, pertrans,
     795           0 :                                        newVal, fcinfo->isnull,
     796             :                                        pergroupstate->transValue,
     797           0 :                                        pergroupstate->transValueIsNull);
     798             : 
     799      724218 :     pergroupstate->transValue = newVal;
     800      724218 :     pergroupstate->transValueIsNull = fcinfo->isnull;
     801             : 
     802      724218 :     MemoryContextSwitchTo(oldContext);
     803             : }
     804             : 
     805             : /*
     806             :  * Advance each aggregate transition state for one input tuple.  The input
     807             :  * tuple has been stored in tmpcontext->ecxt_outertuple, so that it is
     808             :  * accessible to ExecEvalExpr.
     809             :  *
     810             :  * We have two sets of transition states to handle: one for sorted aggregation
     811             :  * and one for hashed; we do them both here, to avoid multiple evaluation of
     812             :  * the inputs.
     813             :  *
     814             :  * When called, CurrentMemoryContext should be the per-query context.
     815             :  */
     816             : static void
     817    29306736 : advance_aggregates(AggState *aggstate)
     818             : {
     819    29306736 :     ExecEvalExprNoReturnSwitchContext(aggstate->phase->evaltrans,
     820             :                                       aggstate->tmpcontext);
     821    29306658 : }
     822             : 
     823             : /*
     824             :  * Run the transition function for a DISTINCT or ORDER BY aggregate
     825             :  * with only one input.  This is called after we have completed
     826             :  * entering all the input values into the sort object.  We complete the
     827             :  * sort, read out the values in sorted order, and run the transition
     828             :  * function on each value (applying DISTINCT if appropriate).
     829             :  *
     830             :  * Note that the strictness of the transition function was checked when
     831             :  * entering the values into the sort, so we don't check it again here;
     832             :  * we just apply standard SQL DISTINCT logic.
     833             :  *
     834             :  * The one-input case is handled separately from the multi-input case
     835             :  * for performance reasons: for single by-value inputs, such as the
     836             :  * common case of count(distinct id), the tuplesort_getdatum code path
     837             :  * is around 300% faster.  (The speedup for by-reference types is less
     838             :  * but still noticeable.)
     839             :  *
     840             :  * This function handles only one grouping set (already set in
     841             :  * aggstate->current_set).
     842             :  *
     843             :  * When called, CurrentMemoryContext should be the per-query context.
     844             :  */
     845             : static void
     846       53758 : process_ordered_aggregate_single(AggState *aggstate,
     847             :                                  AggStatePerTrans pertrans,
     848             :                                  AggStatePerGroup pergroupstate)
     849             : {
     850       53758 :     Datum       oldVal = (Datum) 0;
     851       53758 :     bool        oldIsNull = true;
     852       53758 :     bool        haveOldVal = false;
     853       53758 :     MemoryContext workcontext = aggstate->tmpcontext->ecxt_per_tuple_memory;
     854             :     MemoryContext oldContext;
     855       53758 :     bool        isDistinct = (pertrans->numDistinctCols > 0);
     856       53758 :     Datum       newAbbrevVal = (Datum) 0;
     857       53758 :     Datum       oldAbbrevVal = (Datum) 0;
     858       53758 :     FunctionCallInfo fcinfo = pertrans->transfn_fcinfo;
     859             :     Datum      *newVal;
     860             :     bool       *isNull;
     861             : 
     862             :     Assert(pertrans->numDistinctCols < 2);
     863             : 
     864       53758 :     tuplesort_performsort(pertrans->sortstates[aggstate->current_set]);
     865             : 
     866             :     /* Load the column into argument 1 (arg 0 will be transition value) */
     867       53758 :     newVal = &fcinfo->args[1].value;
     868       53758 :     isNull = &fcinfo->args[1].isnull;
     869             : 
     870             :     /*
     871             :      * Note: if input type is pass-by-ref, the datums returned by the sort are
     872             :      * freshly palloc'd in the per-query context, so we must be careful to
     873             :      * pfree them when they are no longer needed.
     874             :      */
     875             : 
     876      898276 :     while (tuplesort_getdatum(pertrans->sortstates[aggstate->current_set],
     877             :                               true, false, newVal, isNull, &newAbbrevVal))
     878             :     {
     879             :         /*
     880             :          * Clear and select the working context for evaluation of the equality
     881             :          * function and transition function.
     882             :          */
     883      844518 :         MemoryContextReset(workcontext);
     884      844518 :         oldContext = MemoryContextSwitchTo(workcontext);
     885             : 
     886             :         /*
     887             :          * If DISTINCT mode, and not distinct from prior, skip it.
     888             :          */
     889      844518 :         if (isDistinct &&
     890      310454 :             haveOldVal &&
     891           0 :             ((oldIsNull && *isNull) ||
     892      310454 :              (!oldIsNull && !*isNull &&
     893      605948 :               oldAbbrevVal == newAbbrevVal &&
     894      295494 :               DatumGetBool(FunctionCall2Coll(&pertrans->equalfnOne,
     895             :                                              pertrans->aggCollation,
     896             :                                              oldVal, *newVal)))))
     897             :         {
     898      120516 :             MemoryContextSwitchTo(oldContext);
     899      120516 :             continue;
     900             :         }
     901             :         else
     902             :         {
     903      724002 :             advance_transition_function(aggstate, pertrans, pergroupstate);
     904             : 
     905      724002 :             MemoryContextSwitchTo(oldContext);
     906             : 
     907             :             /*
     908             :              * Forget the old value, if any, and remember the new one for
     909             :              * subsequent equality checks.
     910             :              */
     911      724002 :             if (!pertrans->inputtypeByVal)
     912             :             {
     913      525288 :                 if (!oldIsNull)
     914      525108 :                     pfree(DatumGetPointer(oldVal));
     915      525288 :                 if (!*isNull)
     916      525228 :                     oldVal = datumCopy(*newVal, pertrans->inputtypeByVal,
     917      525228 :                                        pertrans->inputtypeLen);
     918             :             }
     919             :             else
     920      198714 :                 oldVal = *newVal;
     921      724002 :             oldAbbrevVal = newAbbrevVal;
     922      724002 :             oldIsNull = *isNull;
     923      724002 :             haveOldVal = true;
     924             :         }
     925             :     }
     926             : 
     927       53758 :     if (!oldIsNull && !pertrans->inputtypeByVal)
     928         120 :         pfree(DatumGetPointer(oldVal));
     929             : 
     930       53758 :     tuplesort_end(pertrans->sortstates[aggstate->current_set]);
     931       53758 :     pertrans->sortstates[aggstate->current_set] = NULL;
     932       53758 : }
     933             : 
     934             : /*
     935             :  * Run the transition function for a DISTINCT or ORDER BY aggregate
     936             :  * with more than one input.  This is called after we have completed
     937             :  * entering all the input values into the sort object.  We complete the
     938             :  * sort, read out the values in sorted order, and run the transition
     939             :  * function on each value (applying DISTINCT if appropriate).
     940             :  *
     941             :  * This function handles only one grouping set (already set in
     942             :  * aggstate->current_set).
     943             :  *
     944             :  * When called, CurrentMemoryContext should be the per-query context.
     945             :  */
     946             : static void
     947          84 : process_ordered_aggregate_multi(AggState *aggstate,
     948             :                                 AggStatePerTrans pertrans,
     949             :                                 AggStatePerGroup pergroupstate)
     950             : {
     951          84 :     ExprContext *tmpcontext = aggstate->tmpcontext;
     952          84 :     FunctionCallInfo fcinfo = pertrans->transfn_fcinfo;
     953          84 :     TupleTableSlot *slot1 = pertrans->sortslot;
     954          84 :     TupleTableSlot *slot2 = pertrans->uniqslot;
     955          84 :     int         numTransInputs = pertrans->numTransInputs;
     956          84 :     int         numDistinctCols = pertrans->numDistinctCols;
     957          84 :     Datum       newAbbrevVal = (Datum) 0;
     958          84 :     Datum       oldAbbrevVal = (Datum) 0;
     959          84 :     bool        haveOldValue = false;
     960          84 :     TupleTableSlot *save = aggstate->tmpcontext->ecxt_outertuple;
     961             :     int         i;
     962             : 
     963          84 :     tuplesort_performsort(pertrans->sortstates[aggstate->current_set]);
     964             : 
     965          84 :     ExecClearTuple(slot1);
     966          84 :     if (slot2)
     967           0 :         ExecClearTuple(slot2);
     968             : 
     969         300 :     while (tuplesort_gettupleslot(pertrans->sortstates[aggstate->current_set],
     970             :                                   true, true, slot1, &newAbbrevVal))
     971             :     {
     972         216 :         CHECK_FOR_INTERRUPTS();
     973             : 
     974         216 :         tmpcontext->ecxt_outertuple = slot1;
     975         216 :         tmpcontext->ecxt_innertuple = slot2;
     976             : 
     977         216 :         if (numDistinctCols == 0 ||
     978           0 :             !haveOldValue ||
     979           0 :             newAbbrevVal != oldAbbrevVal ||
     980           0 :             !ExecQual(pertrans->equalfnMulti, tmpcontext))
     981             :         {
     982             :             /*
     983             :              * Extract the first numTransInputs columns as datums to pass to
     984             :              * the transfn.
     985             :              */
     986         216 :             slot_getsomeattrs(slot1, numTransInputs);
     987             : 
     988             :             /* Load values into fcinfo */
     989             :             /* Start from 1, since the 0th arg will be the transition value */
     990         612 :             for (i = 0; i < numTransInputs; i++)
     991             :             {
     992         396 :                 fcinfo->args[i + 1].value = slot1->tts_values[i];
     993         396 :                 fcinfo->args[i + 1].isnull = slot1->tts_isnull[i];
     994             :             }
     995             : 
     996         216 :             advance_transition_function(aggstate, pertrans, pergroupstate);
     997             : 
     998         216 :             if (numDistinctCols > 0)
     999             :             {
    1000             :                 /* swap the slot pointers to retain the current tuple */
    1001           0 :                 TupleTableSlot *tmpslot = slot2;
    1002             : 
    1003           0 :                 slot2 = slot1;
    1004           0 :                 slot1 = tmpslot;
    1005             :                 /* avoid ExecQual() calls by reusing abbreviated keys */
    1006           0 :                 oldAbbrevVal = newAbbrevVal;
    1007           0 :                 haveOldValue = true;
    1008             :             }
    1009             :         }
    1010             : 
    1011             :         /* Reset context each time */
    1012         216 :         ResetExprContext(tmpcontext);
    1013             : 
    1014         216 :         ExecClearTuple(slot1);
    1015             :     }
    1016             : 
    1017          84 :     if (slot2)
    1018           0 :         ExecClearTuple(slot2);
    1019             : 
    1020          84 :     tuplesort_end(pertrans->sortstates[aggstate->current_set]);
    1021          84 :     pertrans->sortstates[aggstate->current_set] = NULL;
    1022             : 
    1023             :     /* restore previous slot, potentially in use for grouping sets */
    1024          84 :     tmpcontext->ecxt_outertuple = save;
    1025          84 : }
    1026             : 
    1027             : /*
    1028             :  * Compute the final value of one aggregate.
    1029             :  *
    1030             :  * This function handles only one grouping set (already set in
    1031             :  * aggstate->current_set).
    1032             :  *
    1033             :  * The finalfn will be run, and the result delivered, in the
    1034             :  * output-tuple context; caller's CurrentMemoryContext does not matter.
    1035             :  * (But note that in some cases, such as when there is no finalfn, the
    1036             :  * result might be a pointer to or into the agg's transition value.)
    1037             :  *
    1038             :  * The finalfn uses the state as set in the transno.  This also might be
    1039             :  * being used by another aggregate function, so it's important that we do
    1040             :  * nothing destructive here.  Moreover, the aggregate's final value might
    1041             :  * get used in multiple places, so we mustn't return a R/W expanded datum.
    1042             :  */
    1043             : static void
    1044     1119538 : finalize_aggregate(AggState *aggstate,
    1045             :                    AggStatePerAgg peragg,
    1046             :                    AggStatePerGroup pergroupstate,
    1047             :                    Datum *resultVal, bool *resultIsNull)
    1048             : {
    1049     1119538 :     LOCAL_FCINFO(fcinfo, FUNC_MAX_ARGS);
    1050     1119538 :     bool        anynull = false;
    1051             :     MemoryContext oldContext;
    1052             :     int         i;
    1053             :     ListCell   *lc;
    1054     1119538 :     AggStatePerTrans pertrans = &aggstate->pertrans[peragg->transno];
    1055             : 
    1056     1119538 :     oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
    1057             : 
    1058             :     /*
    1059             :      * Evaluate any direct arguments.  We do this even if there's no finalfn
    1060             :      * (which is unlikely anyway), so that side-effects happen as expected.
    1061             :      * The direct arguments go into arg positions 1 and up, leaving position 0
    1062             :      * for the transition state value.
    1063             :      */
    1064     1119538 :     i = 1;
    1065     1120512 :     foreach(lc, peragg->aggdirectargs)
    1066             :     {
    1067         974 :         ExprState  *expr = (ExprState *) lfirst(lc);
    1068             : 
    1069         974 :         fcinfo->args[i].value = ExecEvalExpr(expr,
    1070             :                                              aggstate->ss.ps.ps_ExprContext,
    1071             :                                              &fcinfo->args[i].isnull);
    1072         974 :         anynull |= fcinfo->args[i].isnull;
    1073         974 :         i++;
    1074             :     }
    1075             : 
    1076             :     /*
    1077             :      * Apply the agg's finalfn if one is provided, else return transValue.
    1078             :      */
    1079     1119538 :     if (OidIsValid(peragg->finalfn_oid))
    1080             :     {
    1081      339412 :         int         numFinalArgs = peragg->numFinalArgs;
    1082             : 
    1083             :         /* set up aggstate->curperagg for AggGetAggref() */
    1084      339412 :         aggstate->curperagg = peragg;
    1085             : 
    1086      339412 :         InitFunctionCallInfoData(*fcinfo, &peragg->finalfn,
    1087             :                                  numFinalArgs,
    1088             :                                  pertrans->aggCollation,
    1089             :                                  (Node *) aggstate, NULL);
    1090             : 
    1091             :         /* Fill in the transition state value */
    1092      339412 :         fcinfo->args[0].value =
    1093      339412 :             MakeExpandedObjectReadOnly(pergroupstate->transValue,
    1094             :                                        pergroupstate->transValueIsNull,
    1095             :                                        pertrans->transtypeLen);
    1096      339412 :         fcinfo->args[0].isnull = pergroupstate->transValueIsNull;
    1097      339412 :         anynull |= pergroupstate->transValueIsNull;
    1098             : 
    1099             :         /* Fill any remaining argument positions with nulls */
    1100      492634 :         for (; i < numFinalArgs; i++)
    1101             :         {
    1102      153222 :             fcinfo->args[i].value = (Datum) 0;
    1103      153222 :             fcinfo->args[i].isnull = true;
    1104      153222 :             anynull = true;
    1105             :         }
    1106             : 
    1107      339412 :         if (fcinfo->flinfo->fn_strict && anynull)
    1108             :         {
    1109             :             /* don't call a strict function with NULL inputs */
    1110           0 :             *resultVal = (Datum) 0;
    1111           0 :             *resultIsNull = true;
    1112             :         }
    1113             :         else
    1114             :         {
    1115             :             Datum       result;
    1116             : 
    1117      339412 :             result = FunctionCallInvoke(fcinfo);
    1118      339400 :             *resultIsNull = fcinfo->isnull;
    1119      339400 :             *resultVal = MakeExpandedObjectReadOnly(result,
    1120             :                                                     fcinfo->isnull,
    1121             :                                                     peragg->resulttypeLen);
    1122             :         }
    1123      339400 :         aggstate->curperagg = NULL;
    1124             :     }
    1125             :     else
    1126             :     {
    1127      780126 :         *resultVal =
    1128      780126 :             MakeExpandedObjectReadOnly(pergroupstate->transValue,
    1129             :                                        pergroupstate->transValueIsNull,
    1130             :                                        pertrans->transtypeLen);
    1131      780126 :         *resultIsNull = pergroupstate->transValueIsNull;
    1132             :     }
    1133             : 
    1134     1119526 :     MemoryContextSwitchTo(oldContext);
    1135     1119526 : }
    1136             : 
    1137             : /*
    1138             :  * Compute the output value of one partial aggregate.
    1139             :  *
    1140             :  * The serialization function will be run, and the result delivered, in the
    1141             :  * output-tuple context; caller's CurrentMemoryContext does not matter.
    1142             :  */
    1143             : static void
    1144       16952 : finalize_partialaggregate(AggState *aggstate,
    1145             :                           AggStatePerAgg peragg,
    1146             :                           AggStatePerGroup pergroupstate,
    1147             :                           Datum *resultVal, bool *resultIsNull)
    1148             : {
    1149       16952 :     AggStatePerTrans pertrans = &aggstate->pertrans[peragg->transno];
    1150             :     MemoryContext oldContext;
    1151             : 
    1152       16952 :     oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
    1153             : 
    1154             :     /*
    1155             :      * serialfn_oid will be set if we must serialize the transvalue before
    1156             :      * returning it
    1157             :      */
    1158       16952 :     if (OidIsValid(pertrans->serialfn_oid))
    1159             :     {
    1160             :         /* Don't call a strict serialization function with NULL input. */
    1161         666 :         if (pertrans->serialfn.fn_strict && pergroupstate->transValueIsNull)
    1162             :         {
    1163          80 :             *resultVal = (Datum) 0;
    1164          80 :             *resultIsNull = true;
    1165             :         }
    1166             :         else
    1167             :         {
    1168         586 :             FunctionCallInfo fcinfo = pertrans->serialfn_fcinfo;
    1169             :             Datum       result;
    1170             : 
    1171         586 :             fcinfo->args[0].value =
    1172         586 :                 MakeExpandedObjectReadOnly(pergroupstate->transValue,
    1173             :                                            pergroupstate->transValueIsNull,
    1174             :                                            pertrans->transtypeLen);
    1175         586 :             fcinfo->args[0].isnull = pergroupstate->transValueIsNull;
    1176         586 :             fcinfo->isnull = false;
    1177             : 
    1178         586 :             result = FunctionCallInvoke(fcinfo);
    1179         586 :             *resultIsNull = fcinfo->isnull;
    1180         586 :             *resultVal = MakeExpandedObjectReadOnly(result,
    1181             :                                                     fcinfo->isnull,
    1182             :                                                     peragg->resulttypeLen);
    1183             :         }
    1184             :     }
    1185             :     else
    1186             :     {
    1187       16286 :         *resultVal =
    1188       16286 :             MakeExpandedObjectReadOnly(pergroupstate->transValue,
    1189             :                                        pergroupstate->transValueIsNull,
    1190             :                                        pertrans->transtypeLen);
    1191       16286 :         *resultIsNull = pergroupstate->transValueIsNull;
    1192             :     }
    1193             : 
    1194       16952 :     MemoryContextSwitchTo(oldContext);
    1195       16952 : }
    1196             : 
    1197             : /*
    1198             :  * Extract the attributes that make up the grouping key into the
    1199             :  * hashslot. This is necessary to compute the hash or perform a lookup.
    1200             :  */
    1201             : static inline void
    1202     8266968 : prepare_hash_slot(AggStatePerHash perhash,
    1203             :                   TupleTableSlot *inputslot,
    1204             :                   TupleTableSlot *hashslot)
    1205             : {
    1206             :     int         i;
    1207             : 
    1208             :     /* transfer just the needed columns into hashslot */
    1209     8266968 :     slot_getsomeattrs(inputslot, perhash->largestGrpColIdx);
    1210     8266968 :     ExecClearTuple(hashslot);
    1211             : 
    1212    20481936 :     for (i = 0; i < perhash->numhashGrpCols; i++)
    1213             :     {
    1214    12214968 :         int         varNumber = perhash->hashGrpColIdxInput[i] - 1;
    1215             : 
    1216    12214968 :         hashslot->tts_values[i] = inputslot->tts_values[varNumber];
    1217    12214968 :         hashslot->tts_isnull[i] = inputslot->tts_isnull[varNumber];
    1218             :     }
    1219     8266968 :     ExecStoreVirtualTuple(hashslot);
    1220     8266968 : }
    1221             : 
    1222             : /*
    1223             :  * Prepare to finalize and project based on the specified representative tuple
    1224             :  * slot and grouping set.
    1225             :  *
    1226             :  * In the specified tuple slot, force to null all attributes that should be
    1227             :  * read as null in the context of the current grouping set.  Also stash the
    1228             :  * current group bitmap where GroupingExpr can get at it.
    1229             :  *
    1230             :  * This relies on three conditions:
    1231             :  *
    1232             :  * 1) Nothing is ever going to try and extract the whole tuple from this slot,
    1233             :  * only reference it in evaluations, which will only access individual
    1234             :  * attributes.
    1235             :  *
    1236             :  * 2) No system columns are going to need to be nulled. (If a system column is
    1237             :  * referenced in a group clause, it is actually projected in the outer plan
    1238             :  * tlist.)
    1239             :  *
    1240             :  * 3) Within a given phase, we never need to recover the value of an attribute
    1241             :  * once it has been set to null.
    1242             :  *
    1243             :  * Poking into the slot this way is a bit ugly, but the consensus is that the
    1244             :  * alternative was worse.
    1245             :  */
    1246             : static void
    1247      844782 : prepare_projection_slot(AggState *aggstate, TupleTableSlot *slot, int currentSet)
    1248             : {
    1249      844782 :     if (aggstate->phase->grouped_cols)
    1250             :     {
    1251      558300 :         Bitmapset  *grouped_cols = aggstate->phase->grouped_cols[currentSet];
    1252             : 
    1253      558300 :         aggstate->grouped_cols = grouped_cols;
    1254             : 
    1255      558300 :         if (TTS_EMPTY(slot))
    1256             :         {
    1257             :             /*
    1258             :              * Force all values to be NULL if working on an empty input tuple
    1259             :              * (i.e. an empty grouping set for which no input rows were
    1260             :              * supplied).
    1261             :              */
    1262          48 :             ExecStoreAllNullTuple(slot);
    1263             :         }
    1264      558252 :         else if (aggstate->all_grouped_cols)
    1265             :         {
    1266             :             ListCell   *lc;
    1267             : 
    1268             :             /* all_grouped_cols is arranged in desc order */
    1269      558204 :             slot_getsomeattrs(slot, linitial_int(aggstate->all_grouped_cols));
    1270             : 
    1271     1522858 :             foreach(lc, aggstate->all_grouped_cols)
    1272             :             {
    1273      964654 :                 int         attnum = lfirst_int(lc);
    1274             : 
    1275      964654 :                 if (!bms_is_member(attnum, grouped_cols))
    1276       57826 :                     slot->tts_isnull[attnum - 1] = true;
    1277             :             }
    1278             :         }
    1279             :     }
    1280      844782 : }
    1281             : 
    1282             : /*
    1283             :  * Compute the final value of all aggregates for one group.
    1284             :  *
    1285             :  * This function handles only one grouping set at a time, which the caller must
    1286             :  * have selected.  It's also the caller's responsibility to adjust the supplied
    1287             :  * pergroup parameter to point to the current set's transvalues.
    1288             :  *
    1289             :  * Results are stored in the output econtext aggvalues/aggnulls.
    1290             :  */
    1291             : static void
    1292      844782 : finalize_aggregates(AggState *aggstate,
    1293             :                     AggStatePerAgg peraggs,
    1294             :                     AggStatePerGroup pergroup)
    1295             : {
    1296      844782 :     ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
    1297      844782 :     Datum      *aggvalues = econtext->ecxt_aggvalues;
    1298      844782 :     bool       *aggnulls = econtext->ecxt_aggnulls;
    1299             :     int         aggno;
    1300             : 
    1301             :     /*
    1302             :      * If there were any DISTINCT and/or ORDER BY aggregates, sort their
    1303             :      * inputs and run the transition functions.
    1304             :      */
    1305     1981014 :     for (int transno = 0; transno < aggstate->numtrans; transno++)
    1306             :     {
    1307     1136232 :         AggStatePerTrans pertrans = &aggstate->pertrans[transno];
    1308             :         AggStatePerGroup pergroupstate;
    1309             : 
    1310     1136232 :         pergroupstate = &pergroup[transno];
    1311             : 
    1312     1136232 :         if (pertrans->aggsortrequired)
    1313             :         {
    1314             :             Assert(aggstate->aggstrategy != AGG_HASHED &&
    1315             :                    aggstate->aggstrategy != AGG_MIXED);
    1316             : 
    1317       53842 :             if (pertrans->numInputs == 1)
    1318       53758 :                 process_ordered_aggregate_single(aggstate,
    1319             :                                                  pertrans,
    1320             :                                                  pergroupstate);
    1321             :             else
    1322          84 :                 process_ordered_aggregate_multi(aggstate,
    1323             :                                                 pertrans,
    1324             :                                                 pergroupstate);
    1325             :         }
    1326     1082390 :         else if (pertrans->numDistinctCols > 0 && pertrans->haslast)
    1327             :         {
    1328       18358 :             pertrans->haslast = false;
    1329             : 
    1330       18358 :             if (pertrans->numDistinctCols == 1)
    1331             :             {
    1332       18262 :                 if (!pertrans->inputtypeByVal && !pertrans->lastisnull)
    1333         262 :                     pfree(DatumGetPointer(pertrans->lastdatum));
    1334             : 
    1335       18262 :                 pertrans->lastisnull = false;
    1336       18262 :                 pertrans->lastdatum = (Datum) 0;
    1337             :             }
    1338             :             else
    1339          96 :                 ExecClearTuple(pertrans->uniqslot);
    1340             :         }
    1341             :     }
    1342             : 
    1343             :     /*
    1344             :      * Run the final functions.
    1345             :      */
    1346     1981260 :     for (aggno = 0; aggno < aggstate->numaggs; aggno++)
    1347             :     {
    1348     1136490 :         AggStatePerAgg peragg = &peraggs[aggno];
    1349     1136490 :         int         transno = peragg->transno;
    1350             :         AggStatePerGroup pergroupstate;
    1351             : 
    1352     1136490 :         pergroupstate = &pergroup[transno];
    1353             : 
    1354     1136490 :         if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
    1355       16952 :             finalize_partialaggregate(aggstate, peragg, pergroupstate,
    1356       16952 :                                       &aggvalues[aggno], &aggnulls[aggno]);
    1357             :         else
    1358     1119538 :             finalize_aggregate(aggstate, peragg, pergroupstate,
    1359     1119538 :                                &aggvalues[aggno], &aggnulls[aggno]);
    1360             :     }
    1361      844770 : }
    1362             : 
    1363             : /*
    1364             :  * Project the result of a group (whose aggs have already been calculated by
    1365             :  * finalize_aggregates). Returns the result slot, or NULL if no row is
    1366             :  * projected (suppressed by qual).
    1367             :  */
    1368             : static TupleTableSlot *
    1369      844770 : project_aggregates(AggState *aggstate)
    1370             : {
    1371      844770 :     ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
    1372             : 
    1373             :     /*
    1374             :      * Check the qual (HAVING clause); if the group does not match, ignore it.
    1375             :      */
    1376      844770 :     if (ExecQual(aggstate->ss.ps.qual, econtext))
    1377             :     {
    1378             :         /*
    1379             :          * Form and return projection tuple using the aggregate results and
    1380             :          * the representative input tuple.
    1381             :          */
    1382      738306 :         return ExecProject(aggstate->ss.ps.ps_ProjInfo);
    1383             :     }
    1384             :     else
    1385      106464 :         InstrCountFiltered1(aggstate, 1);
    1386             : 
    1387      106464 :     return NULL;
    1388             : }
    1389             : 
    1390             : /*
    1391             :  * Find input-tuple columns that are needed, dividing them into
    1392             :  * aggregated and unaggregated sets.
    1393             :  */
    1394             : static void
    1395        6710 : find_cols(AggState *aggstate, Bitmapset **aggregated, Bitmapset **unaggregated)
    1396             : {
    1397        6710 :     Agg        *agg = (Agg *) aggstate->ss.ps.plan;
    1398             :     FindColsContext context;
    1399             : 
    1400        6710 :     context.is_aggref = false;
    1401        6710 :     context.aggregated = NULL;
    1402        6710 :     context.unaggregated = NULL;
    1403             : 
    1404             :     /* Examine tlist and quals */
    1405        6710 :     (void) find_cols_walker((Node *) agg->plan.targetlist, &context);
    1406        6710 :     (void) find_cols_walker((Node *) agg->plan.qual, &context);
    1407             : 
    1408             :     /* In some cases, grouping columns will not appear in the tlist */
    1409       16616 :     for (int i = 0; i < agg->numCols; i++)
    1410        9906 :         context.unaggregated = bms_add_member(context.unaggregated,
    1411        9906 :                                               agg->grpColIdx[i]);
    1412             : 
    1413        6710 :     *aggregated = context.aggregated;
    1414        6710 :     *unaggregated = context.unaggregated;
    1415        6710 : }
    1416             : 
    1417             : static bool
    1418       80138 : find_cols_walker(Node *node, FindColsContext *context)
    1419             : {
    1420       80138 :     if (node == NULL)
    1421       14792 :         return false;
    1422       65346 :     if (IsA(node, Var))
    1423             :     {
    1424       17376 :         Var        *var = (Var *) node;
    1425             : 
    1426             :         /* setrefs.c should have set the varno to OUTER_VAR */
    1427             :         Assert(var->varno == OUTER_VAR);
    1428             :         Assert(var->varlevelsup == 0);
    1429       17376 :         if (context->is_aggref)
    1430        6098 :             context->aggregated = bms_add_member(context->aggregated,
    1431        6098 :                                                  var->varattno);
    1432             :         else
    1433       11278 :             context->unaggregated = bms_add_member(context->unaggregated,
    1434       11278 :                                                    var->varattno);
    1435       17376 :         return false;
    1436             :     }
    1437       47970 :     if (IsA(node, Aggref))
    1438             :     {
    1439             :         Assert(!context->is_aggref);
    1440        8582 :         context->is_aggref = true;
    1441        8582 :         expression_tree_walker(node, find_cols_walker, context);
    1442        8582 :         context->is_aggref = false;
    1443        8582 :         return false;
    1444             :     }
    1445       39388 :     return expression_tree_walker(node, find_cols_walker, context);
    1446             : }
    1447             : 
    1448             : /*
    1449             :  * (Re-)initialize the hash table(s) to empty.
    1450             :  *
    1451             :  * To implement hashed aggregation, we need a hashtable that stores a
    1452             :  * representative tuple and an array of AggStatePerGroup structs for each
    1453             :  * distinct set of GROUP BY column values.  We compute the hash key from the
    1454             :  * GROUP BY columns.  The per-group data is allocated in initialize_hash_entry(),
    1455             :  * for each entry.
    1456             :  *
    1457             :  * We have a separate hashtable and associated perhash data structure for each
    1458             :  * grouping set for which we're doing hashing.
    1459             :  *
    1460             :  * The contents of the hash tables always live in the hashcontext's per-tuple
    1461             :  * memory context (there is only one of these for all tables together, since
    1462             :  * they are all reset at the same time).
    1463             :  */
    1464             : static void
    1465       17266 : build_hash_tables(AggState *aggstate)
    1466             : {
    1467             :     int         setno;
    1468             : 
    1469       34876 :     for (setno = 0; setno < aggstate->num_hashes; ++setno)
    1470             :     {
    1471       17610 :         AggStatePerHash perhash = &aggstate->perhash[setno];
    1472             :         long        nbuckets;
    1473             :         Size        memory;
    1474             : 
    1475       17610 :         if (perhash->hashtable != NULL)
    1476             :         {
    1477       12462 :             ResetTupleHashTable(perhash->hashtable);
    1478       12462 :             continue;
    1479             :         }
    1480             : 
    1481             :         Assert(perhash->aggnode->numGroups > 0);
    1482             : 
    1483        5148 :         memory = aggstate->hash_mem_limit / aggstate->num_hashes;
    1484             : 
    1485             :         /* choose reasonable number of buckets per hashtable */
    1486        5148 :         nbuckets = hash_choose_num_buckets(aggstate->hashentrysize,
    1487        5148 :                                            perhash->aggnode->numGroups,
    1488             :                                            memory);
    1489             : 
    1490             : #ifdef USE_INJECTION_POINTS
    1491        5148 :         if (IS_INJECTION_POINT_ATTACHED("hash-aggregate-oversize-table"))
    1492             :         {
    1493           0 :             nbuckets = memory / TupleHashEntrySize();
    1494           0 :             INJECTION_POINT_CACHED("hash-aggregate-oversize-table", NULL);
    1495             :         }
    1496             : #endif
    1497             : 
    1498        5148 :         build_hash_table(aggstate, setno, nbuckets);
    1499             :     }
    1500             : 
    1501       17266 :     aggstate->hash_ngroups_current = 0;
    1502       17266 : }
    1503             : 
    1504             : /*
    1505             :  * Build a single hashtable for this grouping set.
    1506             :  */
    1507             : static void
    1508        5148 : build_hash_table(AggState *aggstate, int setno, long nbuckets)
    1509             : {
    1510        5148 :     AggStatePerHash perhash = &aggstate->perhash[setno];
    1511        5148 :     MemoryContext metacxt = aggstate->hash_metacxt;
    1512        5148 :     MemoryContext tablecxt = aggstate->hash_tablecxt;
    1513        5148 :     MemoryContext tmpcxt = aggstate->tmpcontext->ecxt_per_tuple_memory;
    1514             :     Size        additionalsize;
    1515             : 
    1516             :     Assert(aggstate->aggstrategy == AGG_HASHED ||
    1517             :            aggstate->aggstrategy == AGG_MIXED);
    1518             : 
    1519             :     /*
    1520             :      * Used to make sure initial hash table allocation does not exceed
    1521             :      * hash_mem. Note that the estimate does not include space for
    1522             :      * pass-by-reference transition data values, nor for the representative
    1523             :      * tuple of each group.
    1524             :      */
    1525        5148 :     additionalsize = aggstate->numtrans * sizeof(AggStatePerGroupData);
    1526             : 
    1527       10296 :     perhash->hashtable = BuildTupleHashTable(&aggstate->ss.ps,
    1528        5148 :                                              perhash->hashslot->tts_tupleDescriptor,
    1529        5148 :                                              perhash->hashslot->tts_ops,
    1530             :                                              perhash->numCols,
    1531             :                                              perhash->hashGrpColIdxHash,
    1532        5148 :                                              perhash->eqfuncoids,
    1533             :                                              perhash->hashfunctions,
    1534        5148 :                                              perhash->aggnode->grpCollations,
    1535             :                                              nbuckets,
    1536             :                                              additionalsize,
    1537             :                                              metacxt,
    1538             :                                              tablecxt,
    1539             :                                              tmpcxt,
    1540        5148 :                                              DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit));
    1541        5148 : }
    1542             : 
    1543             : /*
    1544             :  * Compute columns that actually need to be stored in hashtable entries.  The
    1545             :  * incoming tuples from the child plan node will contain grouping columns,
    1546             :  * other columns referenced in our targetlist and qual, columns used to
    1547             :  * compute the aggregate functions, and perhaps just junk columns we don't use
    1548             :  * at all.  Only columns of the first two types need to be stored in the
    1549             :  * hashtable, and getting rid of the others can make the table entries
    1550             :  * significantly smaller.  The hashtable only contains the relevant columns,
    1551             :  * and is packed/unpacked in lookup_hash_entries() / agg_retrieve_hash_table()
    1552             :  * into the format of the normal input descriptor.
    1553             :  *
    1554             :  * Additional columns, in addition to the columns grouped by, come from two
    1555             :  * sources: Firstly functionally dependent columns that we don't need to group
    1556             :  * by themselves, and secondly ctids for row-marks.
    1557             :  *
    1558             :  * To eliminate duplicates, we build a bitmapset of the needed columns, and
    1559             :  * then build an array of the columns included in the hashtable. We might
    1560             :  * still have duplicates if the passed-in grpColIdx has them, which can happen
    1561             :  * in edge cases from semijoins/distinct; these can't always be removed,
    1562             :  * because it's not certain that the duplicate cols will be using the same
    1563             :  * hash function.
    1564             :  *
    1565             :  * Note that the array is preserved over ExecReScanAgg, so we allocate it in
    1566             :  * the per-query context (unlike the hash table itself).
    1567             :  */
    1568             : static void
    1569        6710 : find_hash_columns(AggState *aggstate)
    1570             : {
    1571             :     Bitmapset  *base_colnos;
    1572             :     Bitmapset  *aggregated_colnos;
    1573        6710 :     TupleDesc   scanDesc = aggstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
    1574        6710 :     List       *outerTlist = outerPlanState(aggstate)->plan->targetlist;
    1575        6710 :     int         numHashes = aggstate->num_hashes;
    1576        6710 :     EState     *estate = aggstate->ss.ps.state;
    1577             :     int         j;
    1578             : 
    1579             :     /* Find Vars that will be needed in tlist and qual */
    1580        6710 :     find_cols(aggstate, &aggregated_colnos, &base_colnos);
    1581        6710 :     aggstate->colnos_needed = bms_union(base_colnos, aggregated_colnos);
    1582        6710 :     aggstate->max_colno_needed = 0;
    1583        6710 :     aggstate->all_cols_needed = true;
    1584             : 
    1585       27774 :     for (int i = 0; i < scanDesc->natts; i++)
    1586             :     {
    1587       21064 :         int         colno = i + 1;
    1588             : 
    1589       21064 :         if (bms_is_member(colno, aggstate->colnos_needed))
    1590       14694 :             aggstate->max_colno_needed = colno;
    1591             :         else
    1592        6370 :             aggstate->all_cols_needed = false;
    1593             :     }
    1594             : 
    1595       13934 :     for (j = 0; j < numHashes; ++j)
    1596             :     {
    1597        7224 :         AggStatePerHash perhash = &aggstate->perhash[j];
    1598        7224 :         Bitmapset  *colnos = bms_copy(base_colnos);
    1599        7224 :         AttrNumber *grpColIdx = perhash->aggnode->grpColIdx;
    1600        7224 :         List       *hashTlist = NIL;
    1601             :         TupleDesc   hashDesc;
    1602             :         int         maxCols;
    1603             :         int         i;
    1604             : 
    1605        7224 :         perhash->largestGrpColIdx = 0;
    1606             : 
    1607             :         /*
    1608             :          * If we're doing grouping sets, then some Vars might be referenced in
    1609             :          * tlist/qual for the benefit of other grouping sets, but not needed
    1610             :          * when hashing; i.e. prepare_projection_slot will null them out, so
    1611             :          * there'd be no point storing them.  Use prepare_projection_slot's
    1612             :          * logic to determine which.
    1613             :          */
    1614        7224 :         if (aggstate->phases[0].grouped_cols)
    1615             :         {
    1616        7224 :             Bitmapset  *grouped_cols = aggstate->phases[0].grouped_cols[j];
    1617             :             ListCell   *lc;
    1618             : 
    1619       19096 :             foreach(lc, aggstate->all_grouped_cols)
    1620             :             {
    1621       11872 :                 int         attnum = lfirst_int(lc);
    1622             : 
    1623       11872 :                 if (!bms_is_member(attnum, grouped_cols))
    1624        1344 :                     colnos = bms_del_member(colnos, attnum);
    1625             :             }
    1626             :         }
    1627             : 
    1628             :         /*
    1629             :          * Compute maximum number of input columns accounting for possible
    1630             :          * duplications in the grpColIdx array, which can happen in some edge
    1631             :          * cases where HashAggregate was generated as part of a semijoin or a
    1632             :          * DISTINCT.
    1633             :          */
    1634        7224 :         maxCols = bms_num_members(colnos) + perhash->numCols;
    1635             : 
    1636        7224 :         perhash->hashGrpColIdxInput =
    1637        7224 :             palloc(maxCols * sizeof(AttrNumber));
    1638        7224 :         perhash->hashGrpColIdxHash =
    1639        7224 :             palloc(perhash->numCols * sizeof(AttrNumber));
    1640             : 
    1641             :         /* Add all the grouping columns to colnos */
    1642       17752 :         for (i = 0; i < perhash->numCols; i++)
    1643       10528 :             colnos = bms_add_member(colnos, grpColIdx[i]);
    1644             : 
    1645             :         /*
    1646             :          * First build mapping for columns directly hashed. These are the
    1647             :          * first, because they'll be accessed when computing hash values and
    1648             :          * comparing tuples for exact matches. We also build simple mapping
    1649             :          * for execGrouping, so it knows where to find the to-be-hashed /
    1650             :          * compared columns in the input.
    1651             :          */
    1652       17752 :         for (i = 0; i < perhash->numCols; i++)
    1653             :         {
    1654       10528 :             perhash->hashGrpColIdxInput[i] = grpColIdx[i];
    1655       10528 :             perhash->hashGrpColIdxHash[i] = i + 1;
    1656       10528 :             perhash->numhashGrpCols++;
    1657             :             /* delete already mapped columns */
    1658       10528 :             colnos = bms_del_member(colnos, grpColIdx[i]);
    1659             :         }
    1660             : 
    1661             :         /* and add the remaining columns */
    1662        7224 :         i = -1;
    1663        7874 :         while ((i = bms_next_member(colnos, i)) >= 0)
    1664             :         {
    1665         650 :             perhash->hashGrpColIdxInput[perhash->numhashGrpCols] = i;
    1666         650 :             perhash->numhashGrpCols++;
    1667             :         }
    1668             : 
    1669             :         /* and build a tuple descriptor for the hashtable */
    1670       18402 :         for (i = 0; i < perhash->numhashGrpCols; i++)
    1671             :         {
    1672       11178 :             int         varNumber = perhash->hashGrpColIdxInput[i] - 1;
    1673             : 
    1674       11178 :             hashTlist = lappend(hashTlist, list_nth(outerTlist, varNumber));
    1675       11178 :             perhash->largestGrpColIdx =
    1676       11178 :                 Max(varNumber + 1, perhash->largestGrpColIdx);
    1677             :         }
    1678             : 
    1679        7224 :         hashDesc = ExecTypeFromTL(hashTlist);
    1680             : 
    1681        7224 :         execTuplesHashPrepare(perhash->numCols,
    1682        7224 :                               perhash->aggnode->grpOperators,
    1683             :                               &perhash->eqfuncoids,
    1684             :                               &perhash->hashfunctions);
    1685        7224 :         perhash->hashslot =
    1686        7224 :             ExecAllocTableSlot(&estate->es_tupleTable, hashDesc,
    1687             :                                &TTSOpsMinimalTuple);
    1688             : 
    1689        7224 :         list_free(hashTlist);
    1690        7224 :         bms_free(colnos);
    1691             :     }
    1692             : 
    1693        6710 :     bms_free(base_colnos);
    1694        6710 : }
    1695             : 
    1696             : /*
    1697             :  * Estimate per-hash-table-entry overhead.
    1698             :  */
    1699             : Size
    1700       40752 : hash_agg_entry_size(int numTrans, Size tupleWidth, Size transitionSpace)
    1701             : {
    1702             :     Size        tupleChunkSize;
    1703             :     Size        pergroupChunkSize;
    1704             :     Size        transitionChunkSize;
    1705       40752 :     Size        tupleSize = (MAXALIGN(SizeofMinimalTupleHeader) +
    1706             :                              tupleWidth);
    1707       40752 :     Size        pergroupSize = numTrans * sizeof(AggStatePerGroupData);
    1708             : 
    1709             :     /*
    1710             :      * Entries use the Bump allocator, so the chunk sizes are the same as the
    1711             :      * requested sizes.
    1712             :      */
    1713       40752 :     tupleChunkSize = MAXALIGN(tupleSize);
    1714       40752 :     pergroupChunkSize = pergroupSize;
    1715             : 
    1716             :     /*
    1717             :      * Transition values use AllocSet, which has a chunk header and also uses
    1718             :      * power-of-two allocations.
    1719             :      */
    1720       40752 :     if (transitionSpace > 0)
    1721        5400 :         transitionChunkSize = CHUNKHDRSZ + pg_nextpower2_size_t(transitionSpace);
    1722             :     else
    1723       35352 :         transitionChunkSize = 0;
    1724             : 
    1725             :     return
    1726       40752 :         TupleHashEntrySize() +
    1727       40752 :         tupleChunkSize +
    1728       40752 :         pergroupChunkSize +
    1729             :         transitionChunkSize;
    1730             : }
    1731             : 
    1732             : /*
    1733             :  * hashagg_recompile_expressions()
    1734             :  *
    1735             :  * Identifies the right phase, compiles the right expression given the
    1736             :  * arguments, and then sets phase->evalfunc to that expression.
    1737             :  *
    1738             :  * Different versions of the compiled expression are needed depending on
    1739             :  * whether hash aggregation has spilled or not, and whether it's reading from
    1740             :  * the outer plan or a tape. Before spilling to disk, the expression reads
    1741             :  * from the outer plan and does not need to perform a NULL check. After
    1742             :  * HashAgg begins to spill, new groups will not be created in the hash table,
    1743             :  * and the AggStatePerGroup array may be NULL; therefore we need to add a null
    1744             :  * pointer check to the expression. Then, when reading spilled data from a
    1745             :  * tape, we change the outer slot type to be a fixed minimal tuple slot.
    1746             :  *
    1747             :  * It would be wasteful to recompile every time, so cache the compiled
    1748             :  * expressions in the AggStatePerPhase, and reuse when appropriate.
    1749             :  */
    1750             : static void
    1751       65804 : hashagg_recompile_expressions(AggState *aggstate, bool minslot, bool nullcheck)
    1752             : {
    1753             :     AggStatePerPhase phase;
    1754       65804 :     int         i = minslot ? 1 : 0;
    1755       65804 :     int         j = nullcheck ? 1 : 0;
    1756             : 
    1757             :     Assert(aggstate->aggstrategy == AGG_HASHED ||
    1758             :            aggstate->aggstrategy == AGG_MIXED);
    1759             : 
    1760       65804 :     if (aggstate->aggstrategy == AGG_HASHED)
    1761       13232 :         phase = &aggstate->phases[0];
    1762             :     else                        /* AGG_MIXED */
    1763       52572 :         phase = &aggstate->phases[1];
    1764             : 
    1765       65804 :     if (phase->evaltrans_cache[i][j] == NULL)
    1766             :     {
    1767          88 :         const TupleTableSlotOps *outerops = aggstate->ss.ps.outerops;
    1768          88 :         bool        outerfixed = aggstate->ss.ps.outeropsfixed;
    1769          88 :         bool        dohash = true;
    1770          88 :         bool        dosort = false;
    1771             : 
    1772             :         /*
    1773             :          * If minslot is true, that means we are processing a spilled batch
    1774             :          * (inside agg_refill_hash_table()), and we must not advance the
    1775             :          * sorted grouping sets.
    1776             :          */
    1777          88 :         if (aggstate->aggstrategy == AGG_MIXED && !minslot)
    1778          12 :             dosort = true;
    1779             : 
    1780             :         /* temporarily change the outerops while compiling the expression */
    1781          88 :         if (minslot)
    1782             :         {
    1783          44 :             aggstate->ss.ps.outerops = &TTSOpsMinimalTuple;
    1784          44 :             aggstate->ss.ps.outeropsfixed = true;
    1785             :         }
    1786             : 
    1787          88 :         phase->evaltrans_cache[i][j] = ExecBuildAggTrans(aggstate, phase,
    1788             :                                                          dosort, dohash,
    1789             :                                                          nullcheck);
    1790             : 
    1791             :         /* change back */
    1792          88 :         aggstate->ss.ps.outerops = outerops;
    1793          88 :         aggstate->ss.ps.outeropsfixed = outerfixed;
    1794             :     }
    1795             : 
    1796       65804 :     phase->evaltrans = phase->evaltrans_cache[i][j];
    1797       65804 : }
    1798             : 
    1799             : /*
    1800             :  * Set limits that trigger spilling to avoid exceeding hash_mem. Consider the
    1801             :  * number of partitions we expect to create (if we do spill).
    1802             :  *
    1803             :  * There are two limits: a memory limit, and also an ngroups limit. The
    1804             :  * ngroups limit becomes important when we expect transition values to grow
    1805             :  * substantially larger than the initial value.
    1806             :  */
    1807             : void
    1808       65346 : hash_agg_set_limits(double hashentrysize, double input_groups, int used_bits,
    1809             :                     Size *mem_limit, uint64 *ngroups_limit,
    1810             :                     int *num_partitions)
    1811             : {
    1812             :     int         npartitions;
    1813             :     Size        partition_mem;
    1814       65346 :     Size        hash_mem_limit = get_hash_memory_limit();
    1815             : 
    1816             :     /* if not expected to spill, use all of hash_mem */
    1817       65346 :     if (input_groups * hashentrysize <= hash_mem_limit)
    1818             :     {
    1819       62916 :         if (num_partitions != NULL)
    1820       38306 :             *num_partitions = 0;
    1821       62916 :         *mem_limit = hash_mem_limit;
    1822       62916 :         *ngroups_limit = hash_mem_limit / hashentrysize;
    1823       62916 :         return;
    1824             :     }
    1825             : 
    1826             :     /*
    1827             :      * Calculate expected memory requirements for spilling, which is the size
    1828             :      * of the buffers needed for all the tapes that need to be open at once.
    1829             :      * Then, subtract that from the memory available for holding hash tables.
    1830             :      */
    1831        2430 :     npartitions = hash_choose_num_partitions(input_groups,
    1832             :                                              hashentrysize,
    1833             :                                              used_bits,
    1834             :                                              NULL);
    1835        2430 :     if (num_partitions != NULL)
    1836          96 :         *num_partitions = npartitions;
    1837             : 
    1838        2430 :     partition_mem =
    1839        2430 :         HASHAGG_READ_BUFFER_SIZE +
    1840             :         HASHAGG_WRITE_BUFFER_SIZE * npartitions;
    1841             : 
    1842             :     /*
    1843             :      * Don't set the limit below 3/4 of hash_mem. In that case, we are at the
    1844             :      * minimum number of partitions, so we aren't going to dramatically exceed
    1845             :      * work mem anyway.
    1846             :      */
    1847        2430 :     if (hash_mem_limit > 4 * partition_mem)
    1848           0 :         *mem_limit = hash_mem_limit - partition_mem;
    1849             :     else
    1850        2430 :         *mem_limit = hash_mem_limit * 0.75;
    1851             : 
    1852        2430 :     if (*mem_limit > hashentrysize)
    1853        2430 :         *ngroups_limit = *mem_limit / hashentrysize;
    1854             :     else
    1855           0 :         *ngroups_limit = 1;
    1856             : }
    1857             : 
    1858             : /*
    1859             :  * hash_agg_check_limits
    1860             :  *
    1861             :  * After adding a new group to the hash table, check whether we need to enter
    1862             :  * spill mode. Allocations may happen without adding new groups (for instance,
    1863             :  * if the transition state size grows), so this check is imperfect.
    1864             :  */
    1865             : static void
    1866      526888 : hash_agg_check_limits(AggState *aggstate)
    1867             : {
    1868      526888 :     uint64      ngroups = aggstate->hash_ngroups_current;
    1869      526888 :     Size        meta_mem = MemoryContextMemAllocated(aggstate->hash_metacxt,
    1870             :                                                      true);
    1871      526888 :     Size        entry_mem = MemoryContextMemAllocated(aggstate->hash_tablecxt,
    1872             :                                                       true);
    1873      526888 :     Size        tval_mem = MemoryContextMemAllocated(aggstate->hashcontext->ecxt_per_tuple_memory,
    1874             :                                                      true);
    1875      526888 :     Size        total_mem = meta_mem + entry_mem + tval_mem;
    1876      526888 :     bool        do_spill = false;
    1877             : 
    1878             : #ifdef USE_INJECTION_POINTS
    1879      526888 :     if (ngroups >= 1000)
    1880             :     {
    1881       95650 :         if (IS_INJECTION_POINT_ATTACHED("hash-aggregate-spill-1000"))
    1882             :         {
    1883          10 :             do_spill = true;
    1884          10 :             INJECTION_POINT_CACHED("hash-aggregate-spill-1000", NULL);
    1885             :         }
    1886             :     }
    1887             : #endif
    1888             : 
    1889             :     /*
    1890             :      * Don't spill unless there's at least one group in the hash table so we
    1891             :      * can be sure to make progress even in edge cases.
    1892             :      */
    1893      526888 :     if (aggstate->hash_ngroups_current > 0 &&
    1894      526888 :         (total_mem > aggstate->hash_mem_limit ||
    1895      500476 :          ngroups > aggstate->hash_ngroups_limit))
    1896             :     {
    1897       26448 :         do_spill = true;
    1898             :     }
    1899             : 
    1900      526888 :     if (do_spill)
    1901       26458 :         hash_agg_enter_spill_mode(aggstate);
    1902      526888 : }
    1903             : 
    1904             : /*
    1905             :  * Enter "spill mode", meaning that no new groups are added to any of the hash
    1906             :  * tables. Tuples that would create a new group are instead spilled, and
    1907             :  * processed later.
    1908             :  */
    1909             : static void
    1910       26458 : hash_agg_enter_spill_mode(AggState *aggstate)
    1911             : {
    1912       26458 :     INJECTION_POINT("hash-aggregate-enter-spill-mode", NULL);
    1913       26458 :     aggstate->hash_spill_mode = true;
    1914       26458 :     hashagg_recompile_expressions(aggstate, aggstate->table_filled, true);
    1915             : 
    1916       26458 :     if (!aggstate->hash_ever_spilled)
    1917             :     {
    1918             :         Assert(aggstate->hash_tapeset == NULL);
    1919             :         Assert(aggstate->hash_spills == NULL);
    1920             : 
    1921          62 :         aggstate->hash_ever_spilled = true;
    1922             : 
    1923          62 :         aggstate->hash_tapeset = LogicalTapeSetCreate(true, NULL, -1);
    1924             : 
    1925          62 :         aggstate->hash_spills = palloc(sizeof(HashAggSpill) * aggstate->num_hashes);
    1926             : 
    1927         184 :         for (int setno = 0; setno < aggstate->num_hashes; setno++)
    1928             :         {
    1929         122 :             AggStatePerHash perhash = &aggstate->perhash[setno];
    1930         122 :             HashAggSpill *spill = &aggstate->hash_spills[setno];
    1931             : 
    1932         122 :             hashagg_spill_init(spill, aggstate->hash_tapeset, 0,
    1933         122 :                                perhash->aggnode->numGroups,
    1934             :                                aggstate->hashentrysize);
    1935             :         }
    1936             :     }
    1937       26458 : }
    1938             : 
    1939             : /*
    1940             :  * Update metrics after filling the hash table.
    1941             :  *
    1942             :  * If reading from the outer plan, from_tape should be false; if reading from
    1943             :  * another tape, from_tape should be true.
    1944             :  */
    1945             : static void
    1946       43942 : hash_agg_update_metrics(AggState *aggstate, bool from_tape, int npartitions)
    1947             : {
    1948             :     Size        meta_mem;
    1949             :     Size        entry_mem;
    1950             :     Size        hashkey_mem;
    1951             :     Size        buffer_mem;
    1952             :     Size        total_mem;
    1953             : 
    1954       43942 :     if (aggstate->aggstrategy != AGG_MIXED &&
    1955       17536 :         aggstate->aggstrategy != AGG_HASHED)
    1956           0 :         return;
    1957             : 
    1958             :     /* memory for the hash table itself */
    1959       43942 :     meta_mem = MemoryContextMemAllocated(aggstate->hash_metacxt, true);
    1960             : 
    1961             :     /* memory for hash entries */
    1962       43942 :     entry_mem = MemoryContextMemAllocated(aggstate->hash_tablecxt, true);
    1963             : 
    1964             :     /* memory for byref transition states */
    1965       43942 :     hashkey_mem = MemoryContextMemAllocated(aggstate->hashcontext->ecxt_per_tuple_memory, true);
    1966             : 
    1967             :     /* memory for read/write tape buffers, if spilled */
    1968       43942 :     buffer_mem = npartitions * HASHAGG_WRITE_BUFFER_SIZE;
    1969       43942 :     if (from_tape)
    1970       26944 :         buffer_mem += HASHAGG_READ_BUFFER_SIZE;
    1971             : 
    1972             :     /* update peak mem */
    1973       43942 :     total_mem = meta_mem + entry_mem + hashkey_mem + buffer_mem;
    1974       43942 :     if (total_mem > aggstate->hash_mem_peak)
    1975        4646 :         aggstate->hash_mem_peak = total_mem;
    1976             : 
    1977             :     /* update disk usage */
    1978       43942 :     if (aggstate->hash_tapeset != NULL)
    1979             :     {
    1980       27006 :         uint64      disk_used = LogicalTapeSetBlocks(aggstate->hash_tapeset) * (BLCKSZ / 1024);
    1981             : 
    1982       27006 :         if (aggstate->hash_disk_used < disk_used)
    1983          52 :             aggstate->hash_disk_used = disk_used;
    1984             :     }
    1985             : 
    1986             :     /* update hashentrysize estimate based on contents */
    1987       43942 :     if (aggstate->hash_ngroups_current > 0)
    1988             :     {
    1989       43680 :         aggstate->hashentrysize =
    1990       43680 :             TupleHashEntrySize() +
    1991       43680 :             (hashkey_mem / (double) aggstate->hash_ngroups_current);
    1992             :     }
    1993             : }
    1994             : 
    1995             : /*
    1996             :  * Create memory contexts used for hash aggregation.
    1997             :  */
    1998             : static void
    1999        6710 : hash_create_memory(AggState *aggstate)
    2000             : {
    2001        6710 :     Size        maxBlockSize = ALLOCSET_DEFAULT_MAXSIZE;
    2002             : 
    2003             :     /*
    2004             :      * The hashcontext's per-tuple memory will be used for byref transition
    2005             :      * values and returned by AggCheckCallContext().
    2006             :      */
    2007        6710 :     aggstate->hashcontext = CreateWorkExprContext(aggstate->ss.ps.state);
    2008             : 
    2009             :     /*
    2010             :      * The meta context will be used for the bucket array of
    2011             :      * TupleHashEntryData (or arrays, in the case of grouping sets). As the
    2012             :      * hash table grows, the bucket array will double in size and the old one
    2013             :      * will be freed, so an AllocSet is appropriate. For large bucket arrays,
    2014             :      * the large allocation path will be used, so it's not worth worrying
    2015             :      * about wasting space due to power-of-two allocations.
    2016             :      */
    2017        6710 :     aggstate->hash_metacxt = AllocSetContextCreate(aggstate->ss.ps.state->es_query_cxt,
    2018             :                                                    "HashAgg meta context",
    2019             :                                                    ALLOCSET_DEFAULT_SIZES);
    2020             : 
    2021             :     /*
    2022             :      * The hash entries themselves, which include the grouping key
    2023             :      * (firstTuple) and pergroup data, are stored in the table context. The
    2024             :      * bump allocator can be used because the entries are not freed until the
    2025             :      * entire hash table is reset. The bump allocator is faster for
    2026             :      * allocations and avoids wasting space on the chunk header or
    2027             :      * power-of-two allocations.
    2028             :      *
    2029             :      * Like CreateWorkExprContext(), use smaller sizings for smaller work_mem,
    2030             :      * to avoid large jumps in memory usage.
    2031             :      */
    2032             : 
    2033             :     /*
    2034             :      * Like CreateWorkExprContext(), use smaller sizings for smaller work_mem,
    2035             :      * to avoid large jumps in memory usage.
    2036             :      */
    2037        6710 :     maxBlockSize = pg_prevpower2_size_t(work_mem * (Size) 1024 / 16);
    2038             : 
    2039             :     /* But no bigger than ALLOCSET_DEFAULT_MAXSIZE */
    2040        6710 :     maxBlockSize = Min(maxBlockSize, ALLOCSET_DEFAULT_MAXSIZE);
    2041             : 
    2042             :     /* and no smaller than ALLOCSET_DEFAULT_INITSIZE */
    2043        6710 :     maxBlockSize = Max(maxBlockSize, ALLOCSET_DEFAULT_INITSIZE);
    2044             : 
    2045        6710 :     aggstate->hash_tablecxt = BumpContextCreate(aggstate->ss.ps.state->es_query_cxt,
    2046             :                                                 "HashAgg table context",
    2047             :                                                 ALLOCSET_DEFAULT_MINSIZE,
    2048             :                                                 ALLOCSET_DEFAULT_INITSIZE,
    2049             :                                                 maxBlockSize);
    2050             : 
    2051        6710 : }
    2052             : 
    2053             : /*
    2054             :  * Choose a reasonable number of buckets for the initial hash table size.
    2055             :  */
    2056             : static long
    2057        5148 : hash_choose_num_buckets(double hashentrysize, long ngroups, Size memory)
    2058             : {
    2059             :     long        max_nbuckets;
    2060        5148 :     long        nbuckets = ngroups;
    2061             : 
    2062        5148 :     max_nbuckets = memory / hashentrysize;
    2063             : 
    2064             :     /*
    2065             :      * Underestimating is better than overestimating. Too many buckets crowd
    2066             :      * out space for group keys and transition state values.
    2067             :      */
    2068        5148 :     max_nbuckets >>= 1;
    2069             : 
    2070        5148 :     if (nbuckets > max_nbuckets)
    2071          72 :         nbuckets = max_nbuckets;
    2072             : 
    2073        5148 :     return Max(nbuckets, 1);
    2074             : }
    2075             : 
    2076             : /*
    2077             :  * Determine the number of partitions to create when spilling, which will
    2078             :  * always be a power of two. If log2_npartitions is non-NULL, set
    2079             :  * *log2_npartitions to the log2() of the number of partitions.
    2080             :  */
    2081             : static int
    2082       15052 : hash_choose_num_partitions(double input_groups, double hashentrysize,
    2083             :                            int used_bits, int *log2_npartitions)
    2084             : {
    2085       15052 :     Size        hash_mem_limit = get_hash_memory_limit();
    2086             :     double      partition_limit;
    2087             :     double      mem_wanted;
    2088             :     double      dpartitions;
    2089             :     int         npartitions;
    2090             :     int         partition_bits;
    2091             : 
    2092             :     /*
    2093             :      * Avoid creating so many partitions that the memory requirements of the
    2094             :      * open partition files are greater than 1/4 of hash_mem.
    2095             :      */
    2096       15052 :     partition_limit =
    2097       15052 :         (hash_mem_limit * 0.25 - HASHAGG_READ_BUFFER_SIZE) /
    2098             :         HASHAGG_WRITE_BUFFER_SIZE;
    2099             : 
    2100       15052 :     mem_wanted = HASHAGG_PARTITION_FACTOR * input_groups * hashentrysize;
    2101             : 
    2102             :     /* make enough partitions so that each one is likely to fit in memory */
    2103       15052 :     dpartitions = 1 + (mem_wanted / hash_mem_limit);
    2104             : 
    2105       15052 :     if (dpartitions > partition_limit)
    2106       14988 :         dpartitions = partition_limit;
    2107             : 
    2108       15052 :     if (dpartitions < HASHAGG_MIN_PARTITIONS)
    2109       15052 :         dpartitions = HASHAGG_MIN_PARTITIONS;
    2110       15052 :     if (dpartitions > HASHAGG_MAX_PARTITIONS)
    2111           0 :         dpartitions = HASHAGG_MAX_PARTITIONS;
    2112             : 
    2113             :     /* HASHAGG_MAX_PARTITIONS limit makes this safe */
    2114       15052 :     npartitions = (int) dpartitions;
    2115             : 
    2116             :     /* ceil(log2(npartitions)) */
    2117       15052 :     partition_bits = pg_ceil_log2_32(npartitions);
    2118             : 
    2119             :     /* make sure that we don't exhaust the hash bits */
    2120       15052 :     if (partition_bits + used_bits >= 32)
    2121           0 :         partition_bits = 32 - used_bits;
    2122             : 
    2123       15052 :     if (log2_npartitions != NULL)
    2124       12622 :         *log2_npartitions = partition_bits;
    2125             : 
    2126             :     /* number of partitions will be a power of two */
    2127       15052 :     npartitions = 1 << partition_bits;
    2128             : 
    2129       15052 :     return npartitions;
    2130             : }
    2131             : 
    2132             : /*
    2133             :  * Initialize a freshly-created TupleHashEntry.
    2134             :  */
    2135             : static void
    2136      526888 : initialize_hash_entry(AggState *aggstate, TupleHashTable hashtable,
    2137             :                       TupleHashEntry entry)
    2138             : {
    2139             :     AggStatePerGroup pergroup;
    2140             :     int         transno;
    2141             : 
    2142      526888 :     aggstate->hash_ngroups_current++;
    2143      526888 :     hash_agg_check_limits(aggstate);
    2144             : 
    2145             :     /* no need to allocate or initialize per-group state */
    2146      526888 :     if (aggstate->numtrans == 0)
    2147      211434 :         return;
    2148             : 
    2149      315454 :     pergroup = (AggStatePerGroup) TupleHashEntryGetAdditional(hashtable, entry);
    2150             : 
    2151             :     /*
    2152             :      * Initialize aggregates for new tuple group, lookup_hash_entries()
    2153             :      * already has selected the relevant grouping set.
    2154             :      */
    2155      780198 :     for (transno = 0; transno < aggstate->numtrans; transno++)
    2156             :     {
    2157      464744 :         AggStatePerTrans pertrans = &aggstate->pertrans[transno];
    2158      464744 :         AggStatePerGroup pergroupstate = &pergroup[transno];
    2159             : 
    2160      464744 :         initialize_aggregate(aggstate, pertrans, pergroupstate);
    2161             :     }
    2162             : }
    2163             : 
    2164             : /*
    2165             :  * Look up hash entries for the current tuple in all hashed grouping sets.
    2166             :  *
    2167             :  * Some entries may be left NULL if we are in "spill mode". The same tuple
    2168             :  * will belong to different groups for each grouping set, so may match a group
    2169             :  * already in memory for one set and match a group not in memory for another
    2170             :  * set. When in "spill mode", the tuple will be spilled for each grouping set
    2171             :  * where it doesn't match a group in memory.
    2172             :  *
    2173             :  * NB: It's possible to spill the same tuple for several different grouping
    2174             :  * sets. This may seem wasteful, but it's actually a trade-off: if we spill
    2175             :  * the tuple multiple times for multiple grouping sets, it can be partitioned
    2176             :  * for each grouping set, making the refilling of the hash table very
    2177             :  * efficient.
    2178             :  */
    2179             : static void
    2180     6915740 : lookup_hash_entries(AggState *aggstate)
    2181             : {
    2182     6915740 :     AggStatePerGroup *pergroup = aggstate->hash_pergroup;
    2183     6915740 :     TupleTableSlot *outerslot = aggstate->tmpcontext->ecxt_outertuple;
    2184             :     int         setno;
    2185             : 
    2186    13965932 :     for (setno = 0; setno < aggstate->num_hashes; setno++)
    2187             :     {
    2188     7050192 :         AggStatePerHash perhash = &aggstate->perhash[setno];
    2189     7050192 :         TupleHashTable hashtable = perhash->hashtable;
    2190     7050192 :         TupleTableSlot *hashslot = perhash->hashslot;
    2191             :         TupleHashEntry entry;
    2192             :         uint32      hash;
    2193     7050192 :         bool        isnew = false;
    2194             :         bool       *p_isnew;
    2195             : 
    2196             :         /* if hash table already spilled, don't create new entries */
    2197     7050192 :         p_isnew = aggstate->hash_spill_mode ? NULL : &isnew;
    2198             : 
    2199     7050192 :         select_current_set(aggstate, setno, true);
    2200     7050192 :         prepare_hash_slot(perhash,
    2201             :                           outerslot,
    2202             :                           hashslot);
    2203             : 
    2204     7050192 :         entry = LookupTupleHashEntry(hashtable, hashslot,
    2205             :                                      p_isnew, &hash);
    2206             : 
    2207     7050192 :         if (entry != NULL)
    2208             :         {
    2209     6282956 :             if (isnew)
    2210      371208 :                 initialize_hash_entry(aggstate, hashtable, entry);
    2211     6282956 :             pergroup[setno] = TupleHashEntryGetAdditional(hashtable, entry);
    2212             :         }
    2213             :         else
    2214             :         {
    2215      767236 :             HashAggSpill *spill = &aggstate->hash_spills[setno];
    2216      767236 :             TupleTableSlot *slot = aggstate->tmpcontext->ecxt_outertuple;
    2217             : 
    2218      767236 :             if (spill->partitions == NULL)
    2219           0 :                 hashagg_spill_init(spill, aggstate->hash_tapeset, 0,
    2220           0 :                                    perhash->aggnode->numGroups,
    2221             :                                    aggstate->hashentrysize);
    2222             : 
    2223      767236 :             hashagg_spill_tuple(aggstate, spill, slot, hash);
    2224      767236 :             pergroup[setno] = NULL;
    2225             :         }
    2226             :     }
    2227     6915740 : }
    2228             : 
    2229             : /*
    2230             :  * ExecAgg -
    2231             :  *
    2232             :  *    ExecAgg receives tuples from its outer subplan and aggregates over
    2233             :  *    the appropriate attribute for each aggregate function use (Aggref
    2234             :  *    node) appearing in the targetlist or qual of the node.  The number
    2235             :  *    of tuples to aggregate over depends on whether grouped or plain
    2236             :  *    aggregation is selected.  In grouped aggregation, we produce a result
    2237             :  *    row for each group; in plain aggregation there's a single result row
    2238             :  *    for the whole query.  In either case, the value of each aggregate is
    2239             :  *    stored in the expression context to be used when ExecProject evaluates
    2240             :  *    the result tuple.
    2241             :  */
    2242             : static TupleTableSlot *
    2243      820982 : ExecAgg(PlanState *pstate)
    2244             : {
    2245      820982 :     AggState   *node = castNode(AggState, pstate);
    2246      820982 :     TupleTableSlot *result = NULL;
    2247             : 
    2248      820982 :     CHECK_FOR_INTERRUPTS();
    2249             : 
    2250      820982 :     if (!node->agg_done)
    2251             :     {
    2252             :         /* Dispatch based on strategy */
    2253      757090 :         switch (node->phase->aggstrategy)
    2254             :         {
    2255      483530 :             case AGG_HASHED:
    2256      483530 :                 if (!node->table_filled)
    2257       16854 :                     agg_fill_hash_table(node);
    2258             :                 /* FALLTHROUGH */
    2259             :             case AGG_MIXED:
    2260      510892 :                 result = agg_retrieve_hash_table(node);
    2261      510892 :                 break;
    2262      246198 :             case AGG_PLAIN:
    2263             :             case AGG_SORTED:
    2264      246198 :                 result = agg_retrieve_direct(node);
    2265      246006 :                 break;
    2266             :         }
    2267             : 
    2268      756898 :         if (!TupIsNull(result))
    2269      738294 :             return result;
    2270             :     }
    2271             : 
    2272       82496 :     return NULL;
    2273             : }
    2274             : 
    2275             : /*
    2276             :  * ExecAgg for non-hashed case
    2277             :  */
    2278             : static TupleTableSlot *
    2279      246198 : agg_retrieve_direct(AggState *aggstate)
    2280             : {
    2281      246198 :     Agg        *node = aggstate->phase->aggnode;
    2282             :     ExprContext *econtext;
    2283             :     ExprContext *tmpcontext;
    2284             :     AggStatePerAgg peragg;
    2285             :     AggStatePerGroup *pergroups;
    2286             :     TupleTableSlot *outerslot;
    2287             :     TupleTableSlot *firstSlot;
    2288             :     TupleTableSlot *result;
    2289      246198 :     bool        hasGroupingSets = aggstate->phase->numsets > 0;
    2290      246198 :     int         numGroupingSets = Max(aggstate->phase->numsets, 1);
    2291             :     int         currentSet;
    2292             :     int         nextSetSize;
    2293             :     int         numReset;
    2294             :     int         i;
    2295             : 
    2296             :     /*
    2297             :      * get state info from node
    2298             :      *
    2299             :      * econtext is the per-output-tuple expression context
    2300             :      *
    2301             :      * tmpcontext is the per-input-tuple expression context
    2302             :      */
    2303      246198 :     econtext = aggstate->ss.ps.ps_ExprContext;
    2304      246198 :     tmpcontext = aggstate->tmpcontext;
    2305             : 
    2306      246198 :     peragg = aggstate->peragg;
    2307      246198 :     pergroups = aggstate->pergroups;
    2308      246198 :     firstSlot = aggstate->ss.ss_ScanTupleSlot;
    2309             : 
    2310             :     /*
    2311             :      * We loop retrieving groups until we find one matching
    2312             :      * aggstate->ss.ps.qual
    2313             :      *
    2314             :      * For grouping sets, we have the invariant that aggstate->projected_set
    2315             :      * is either -1 (initial call) or the index (starting from 0) in
    2316             :      * gset_lengths for the group we just completed (either by projecting a
    2317             :      * row or by discarding it in the qual).
    2318             :      */
    2319      317006 :     while (!aggstate->agg_done)
    2320             :     {
    2321             :         /*
    2322             :          * Clear the per-output-tuple context for each group, as well as
    2323             :          * aggcontext (which contains any pass-by-ref transvalues of the old
    2324             :          * group).  Some aggregate functions store working state in child
    2325             :          * contexts; those now get reset automatically without us needing to
    2326             :          * do anything special.
    2327             :          *
    2328             :          * We use ReScanExprContext not just ResetExprContext because we want
    2329             :          * any registered shutdown callbacks to be called.  That allows
    2330             :          * aggregate functions to ensure they've cleaned up any non-memory
    2331             :          * resources.
    2332             :          */
    2333      316796 :         ReScanExprContext(econtext);
    2334             : 
    2335             :         /*
    2336             :          * Determine how many grouping sets need to be reset at this boundary.
    2337             :          */
    2338      316796 :         if (aggstate->projected_set >= 0 &&
    2339      247086 :             aggstate->projected_set < numGroupingSets)
    2340      247080 :             numReset = aggstate->projected_set + 1;
    2341             :         else
    2342       69716 :             numReset = numGroupingSets;
    2343             : 
    2344             :         /*
    2345             :          * numReset can change on a phase boundary, but that's OK; we want to
    2346             :          * reset the contexts used in _this_ phase, and later, after possibly
    2347             :          * changing phase, initialize the right number of aggregates for the
    2348             :          * _new_ phase.
    2349             :          */
    2350             : 
    2351      655846 :         for (i = 0; i < numReset; i++)
    2352             :         {
    2353      339050 :             ReScanExprContext(aggstate->aggcontexts[i]);
    2354             :         }
    2355             : 
    2356             :         /*
    2357             :          * Check if input is complete and there are no more groups to project
    2358             :          * in this phase; move to next phase or mark as done.
    2359             :          */
    2360      316796 :         if (aggstate->input_done == true &&
    2361        1578 :             aggstate->projected_set >= (numGroupingSets - 1))
    2362             :         {
    2363         768 :             if (aggstate->current_phase < aggstate->numphases - 1)
    2364             :             {
    2365         198 :                 initialize_phase(aggstate, aggstate->current_phase + 1);
    2366         198 :                 aggstate->input_done = false;
    2367         198 :                 aggstate->projected_set = -1;
    2368         198 :                 numGroupingSets = Max(aggstate->phase->numsets, 1);
    2369         198 :                 node = aggstate->phase->aggnode;
    2370         198 :                 numReset = numGroupingSets;
    2371             :             }
    2372         570 :             else if (aggstate->aggstrategy == AGG_MIXED)
    2373             :             {
    2374             :                 /*
    2375             :                  * Mixed mode; we've output all the grouped stuff and have
    2376             :                  * full hashtables, so switch to outputting those.
    2377             :                  */
    2378         156 :                 initialize_phase(aggstate, 0);
    2379         156 :                 aggstate->table_filled = true;
    2380         156 :                 ResetTupleHashIterator(aggstate->perhash[0].hashtable,
    2381             :                                        &aggstate->perhash[0].hashiter);
    2382         156 :                 select_current_set(aggstate, 0, true);
    2383         156 :                 return agg_retrieve_hash_table(aggstate);
    2384             :             }
    2385             :             else
    2386             :             {
    2387         414 :                 aggstate->agg_done = true;
    2388         414 :                 break;
    2389             :             }
    2390             :         }
    2391             : 
    2392             :         /*
    2393             :          * Get the number of columns in the next grouping set after the last
    2394             :          * projected one (if any). This is the number of columns to compare to
    2395             :          * see if we reached the boundary of that set too.
    2396             :          */
    2397      316226 :         if (aggstate->projected_set >= 0 &&
    2398      246318 :             aggstate->projected_set < (numGroupingSets - 1))
    2399       27282 :             nextSetSize = aggstate->phase->gset_lengths[aggstate->projected_set + 1];
    2400             :         else
    2401      288944 :             nextSetSize = 0;
    2402             : 
    2403             :         /*----------
    2404             :          * If a subgroup for the current grouping set is present, project it.
    2405             :          *
    2406             :          * We have a new group if:
    2407             :          *  - we're out of input but haven't projected all grouping sets
    2408             :          *    (checked above)
    2409             :          * OR
    2410             :          *    - we already projected a row that wasn't from the last grouping
    2411             :          *      set
    2412             :          *    AND
    2413             :          *    - the next grouping set has at least one grouping column (since
    2414             :          *      empty grouping sets project only once input is exhausted)
    2415             :          *    AND
    2416             :          *    - the previous and pending rows differ on the grouping columns
    2417             :          *      of the next grouping set
    2418             :          *----------
    2419             :          */
    2420      316226 :         tmpcontext->ecxt_innertuple = econtext->ecxt_outertuple;
    2421      316226 :         if (aggstate->input_done ||
    2422      315416 :             (node->aggstrategy != AGG_PLAIN &&
    2423      247326 :              aggstate->projected_set != -1 &&
    2424      245508 :              aggstate->projected_set < (numGroupingSets - 1) &&
    2425       19940 :              nextSetSize > 0 &&
    2426       19940 :              !ExecQualAndReset(aggstate->phase->eqfunctions[nextSetSize - 1],
    2427             :                                tmpcontext)))
    2428             :         {
    2429       14144 :             aggstate->projected_set += 1;
    2430             : 
    2431             :             Assert(aggstate->projected_set < numGroupingSets);
    2432       14144 :             Assert(nextSetSize > 0 || aggstate->input_done);
    2433             :         }
    2434             :         else
    2435             :         {
    2436             :             /*
    2437             :              * We no longer care what group we just projected, the next
    2438             :              * projection will always be the first (or only) grouping set
    2439             :              * (unless the input proves to be empty).
    2440             :              */
    2441      302082 :             aggstate->projected_set = 0;
    2442             : 
    2443             :             /*
    2444             :              * If we don't already have the first tuple of the new group,
    2445             :              * fetch it from the outer plan.
    2446             :              */
    2447      302082 :             if (aggstate->grp_firstTuple == NULL)
    2448             :             {
    2449       69908 :                 outerslot = fetch_input_tuple(aggstate);
    2450       69848 :                 if (!TupIsNull(outerslot))
    2451             :                 {
    2452             :                     /*
    2453             :                      * Make a copy of the first input tuple; we will use this
    2454             :                      * for comparisons (in group mode) and for projection.
    2455             :                      */
    2456       55996 :                     aggstate->grp_firstTuple = ExecCopySlotHeapTuple(outerslot);
    2457             :                 }
    2458             :                 else
    2459             :                 {
    2460             :                     /* outer plan produced no tuples at all */
    2461       13852 :                     if (hasGroupingSets)
    2462             :                     {
    2463             :                         /*
    2464             :                          * If there was no input at all, we need to project
    2465             :                          * rows only if there are grouping sets of size 0.
    2466             :                          * Note that this implies that there can't be any
    2467             :                          * references to ungrouped Vars, which would otherwise
    2468             :                          * cause issues with the empty output slot.
    2469             :                          *
    2470             :                          * XXX: This is no longer true, we currently deal with
    2471             :                          * this in finalize_aggregates().
    2472             :                          */
    2473          54 :                         aggstate->input_done = true;
    2474             : 
    2475          78 :                         while (aggstate->phase->gset_lengths[aggstate->projected_set] > 0)
    2476             :                         {
    2477          30 :                             aggstate->projected_set += 1;
    2478          30 :                             if (aggstate->projected_set >= numGroupingSets)
    2479             :                             {
    2480             :                                 /*
    2481             :                                  * We can't set agg_done here because we might
    2482             :                                  * have more phases to do, even though the
    2483             :                                  * input is empty. So we need to restart the
    2484             :                                  * whole outer loop.
    2485             :                                  */
    2486           6 :                                 break;
    2487             :                             }
    2488             :                         }
    2489             : 
    2490          54 :                         if (aggstate->projected_set >= numGroupingSets)
    2491           6 :                             continue;
    2492             :                     }
    2493             :                     else
    2494             :                     {
    2495       13798 :                         aggstate->agg_done = true;
    2496             :                         /* If we are grouping, we should produce no tuples too */
    2497       13798 :                         if (node->aggstrategy != AGG_PLAIN)
    2498         152 :                             return NULL;
    2499             :                     }
    2500             :                 }
    2501             :             }
    2502             : 
    2503             :             /*
    2504             :              * Initialize working state for a new input tuple group.
    2505             :              */
    2506      301864 :             initialize_aggregates(aggstate, pergroups, numReset);
    2507             : 
    2508      301864 :             if (aggstate->grp_firstTuple != NULL)
    2509             :             {
    2510             :                 /*
    2511             :                  * Store the copied first input tuple in the tuple table slot
    2512             :                  * reserved for it.  The tuple will be deleted when it is
    2513             :                  * cleared from the slot.
    2514             :                  */
    2515      288170 :                 ExecForceStoreHeapTuple(aggstate->grp_firstTuple,
    2516             :                                         firstSlot, true);
    2517      288170 :                 aggstate->grp_firstTuple = NULL; /* don't keep two pointers */
    2518             : 
    2519             :                 /* set up for first advance_aggregates call */
    2520      288170 :                 tmpcontext->ecxt_outertuple = firstSlot;
    2521             : 
    2522             :                 /*
    2523             :                  * Process each outer-plan tuple, and then fetch the next one,
    2524             :                  * until we exhaust the outer plan or cross a group boundary.
    2525             :                  */
    2526             :                 for (;;)
    2527             :                 {
    2528             :                     /*
    2529             :                      * During phase 1 only of a mixed agg, we need to update
    2530             :                      * hashtables as well in advance_aggregates.
    2531             :                      */
    2532    21661822 :                     if (aggstate->aggstrategy == AGG_MIXED &&
    2533       38062 :                         aggstate->current_phase == 1)
    2534             :                     {
    2535       38062 :                         lookup_hash_entries(aggstate);
    2536             :                     }
    2537             : 
    2538             :                     /* Advance the aggregates (or combine functions) */
    2539    21661822 :                     advance_aggregates(aggstate);
    2540             : 
    2541             :                     /* Reset per-input-tuple context after each tuple */
    2542    21661744 :                     ResetExprContext(tmpcontext);
    2543             : 
    2544    21661744 :                     outerslot = fetch_input_tuple(aggstate);
    2545    21661714 :                     if (TupIsNull(outerslot))
    2546             :                     {
    2547             :                         /* no more outer-plan tuples available */
    2548             : 
    2549             :                         /* if we built hash tables, finalize any spills */
    2550       55882 :                         if (aggstate->aggstrategy == AGG_MIXED &&
    2551         144 :                             aggstate->current_phase == 1)
    2552         144 :                             hashagg_finish_initial_spills(aggstate);
    2553             : 
    2554       55882 :                         if (hasGroupingSets)
    2555             :                         {
    2556         714 :                             aggstate->input_done = true;
    2557         714 :                             break;
    2558             :                         }
    2559             :                         else
    2560             :                         {
    2561       55168 :                             aggstate->agg_done = true;
    2562       55168 :                             break;
    2563             :                         }
    2564             :                     }
    2565             :                     /* set up for next advance_aggregates call */
    2566    21605832 :                     tmpcontext->ecxt_outertuple = outerslot;
    2567             : 
    2568             :                     /*
    2569             :                      * If we are grouping, check whether we've crossed a group
    2570             :                      * boundary.
    2571             :                      */
    2572    21605832 :                     if (node->aggstrategy != AGG_PLAIN && node->numCols > 0)
    2573             :                     {
    2574     2467012 :                         tmpcontext->ecxt_innertuple = firstSlot;
    2575     2467012 :                         if (!ExecQual(aggstate->phase->eqfunctions[node->numCols - 1],
    2576             :                                       tmpcontext))
    2577             :                         {
    2578      232180 :                             aggstate->grp_firstTuple = ExecCopySlotHeapTuple(outerslot);
    2579      232180 :                             break;
    2580             :                         }
    2581             :                     }
    2582             :                 }
    2583             :             }
    2584             : 
    2585             :             /*
    2586             :              * Use the representative input tuple for any references to
    2587             :              * non-aggregated input columns in aggregate direct args, the node
    2588             :              * qual, and the tlist.  (If we are not grouping, and there are no
    2589             :              * input rows at all, we will come here with an empty firstSlot
    2590             :              * ... but if not grouping, there can't be any references to
    2591             :              * non-aggregated input columns, so no problem.)
    2592             :              */
    2593      301756 :             econtext->ecxt_outertuple = firstSlot;
    2594             :         }
    2595             : 
    2596             :         Assert(aggstate->projected_set >= 0);
    2597             : 
    2598      315900 :         currentSet = aggstate->projected_set;
    2599             : 
    2600      315900 :         prepare_projection_slot(aggstate, econtext->ecxt_outertuple, currentSet);
    2601             : 
    2602      315900 :         select_current_set(aggstate, currentSet, false);
    2603             : 
    2604      315900 :         finalize_aggregates(aggstate,
    2605             :                             peragg,
    2606      315900 :                             pergroups[currentSet]);
    2607             : 
    2608             :         /*
    2609             :          * If there's no row to project right now, we must continue rather
    2610             :          * than returning a null since there might be more groups.
    2611             :          */
    2612      315888 :         result = project_aggregates(aggstate);
    2613      315876 :         if (result)
    2614      245074 :             return result;
    2615             :     }
    2616             : 
    2617             :     /* No more groups */
    2618         624 :     return NULL;
    2619             : }
    2620             : 
    2621             : /*
    2622             :  * ExecAgg for hashed case: read input and build hash table
    2623             :  */
    2624             : static void
    2625       16854 : agg_fill_hash_table(AggState *aggstate)
    2626             : {
    2627             :     TupleTableSlot *outerslot;
    2628       16854 :     ExprContext *tmpcontext = aggstate->tmpcontext;
    2629             : 
    2630             :     /*
    2631             :      * Process each outer-plan tuple, and then fetch the next one, until we
    2632             :      * exhaust the outer plan.
    2633             :      */
    2634             :     for (;;)
    2635             :     {
    2636     6894532 :         outerslot = fetch_input_tuple(aggstate);
    2637     6894532 :         if (TupIsNull(outerslot))
    2638             :             break;
    2639             : 
    2640             :         /* set up for lookup_hash_entries and advance_aggregates */
    2641     6877678 :         tmpcontext->ecxt_outertuple = outerslot;
    2642             : 
    2643             :         /* Find or build hashtable entries */
    2644     6877678 :         lookup_hash_entries(aggstate);
    2645             : 
    2646             :         /* Advance the aggregates (or combine functions) */
    2647     6877678 :         advance_aggregates(aggstate);
    2648             : 
    2649             :         /*
    2650             :          * Reset per-input-tuple context after each tuple, but note that the
    2651             :          * hash lookups do this too
    2652             :          */
    2653     6877678 :         ResetExprContext(aggstate->tmpcontext);
    2654             :     }
    2655             : 
    2656             :     /* finalize spills, if any */
    2657       16854 :     hashagg_finish_initial_spills(aggstate);
    2658             : 
    2659       16854 :     aggstate->table_filled = true;
    2660             :     /* Initialize to walk the first hash table */
    2661       16854 :     select_current_set(aggstate, 0, true);
    2662       16854 :     ResetTupleHashIterator(aggstate->perhash[0].hashtable,
    2663             :                            &aggstate->perhash[0].hashiter);
    2664       16854 : }
    2665             : 
    2666             : /*
    2667             :  * If any data was spilled during hash aggregation, reset the hash table and
    2668             :  * reprocess one batch of spilled data. After reprocessing a batch, the hash
    2669             :  * table will again contain data, ready to be consumed by
    2670             :  * agg_retrieve_hash_table_in_memory().
    2671             :  *
    2672             :  * Should only be called after all in memory hash table entries have been
    2673             :  * finalized and emitted.
    2674             :  *
    2675             :  * Return false when input is exhausted and there's no more work to be done;
    2676             :  * otherwise return true.
    2677             :  */
    2678             : static bool
    2679       44772 : agg_refill_hash_table(AggState *aggstate)
    2680             : {
    2681             :     HashAggBatch *batch;
    2682             :     AggStatePerHash perhash;
    2683             :     HashAggSpill spill;
    2684       44772 :     LogicalTapeSet *tapeset = aggstate->hash_tapeset;
    2685       44772 :     bool        spill_initialized = false;
    2686             : 
    2687       44772 :     if (aggstate->hash_batches == NIL)
    2688       17828 :         return false;
    2689             : 
    2690             :     /* hash_batches is a stack, with the top item at the end of the list */
    2691       26944 :     batch = llast(aggstate->hash_batches);
    2692       26944 :     aggstate->hash_batches = list_delete_last(aggstate->hash_batches);
    2693             : 
    2694       26944 :     hash_agg_set_limits(aggstate->hashentrysize, batch->input_card,
    2695             :                         batch->used_bits, &aggstate->hash_mem_limit,
    2696             :                         &aggstate->hash_ngroups_limit, NULL);
    2697             : 
    2698             :     /*
    2699             :      * Each batch only processes one grouping set; set the rest to NULL so
    2700             :      * that advance_aggregates() knows to ignore them. We don't touch
    2701             :      * pergroups for sorted grouping sets here, because they will be needed if
    2702             :      * we rescan later. The expressions for sorted grouping sets will not be
    2703             :      * evaluated after we recompile anyway.
    2704             :      */
    2705      207428 :     MemSet(aggstate->hash_pergroup, 0,
    2706             :            sizeof(AggStatePerGroup) * aggstate->num_hashes);
    2707             : 
    2708             :     /* free memory and reset hash tables */
    2709       26944 :     ReScanExprContext(aggstate->hashcontext);
    2710       26944 :     MemoryContextReset(aggstate->hash_tablecxt);
    2711      207428 :     for (int setno = 0; setno < aggstate->num_hashes; setno++)
    2712      180484 :         ResetTupleHashTable(aggstate->perhash[setno].hashtable);
    2713             : 
    2714       26944 :     aggstate->hash_ngroups_current = 0;
    2715             : 
    2716             :     /*
    2717             :      * In AGG_MIXED mode, hash aggregation happens in phase 1 and the output
    2718             :      * happens in phase 0. So, we switch to phase 1 when processing a batch,
    2719             :      * and back to phase 0 after the batch is done.
    2720             :      */
    2721             :     Assert(aggstate->current_phase == 0);
    2722       26944 :     if (aggstate->phase->aggstrategy == AGG_MIXED)
    2723             :     {
    2724       26262 :         aggstate->current_phase = 1;
    2725       26262 :         aggstate->phase = &aggstate->phases[aggstate->current_phase];
    2726             :     }
    2727             : 
    2728       26944 :     select_current_set(aggstate, batch->setno, true);
    2729             : 
    2730       26944 :     perhash = &aggstate->perhash[aggstate->current_set];
    2731             : 
    2732             :     /*
    2733             :      * Spilled tuples are always read back as MinimalTuples, which may be
    2734             :      * different from the outer plan, so recompile the aggregate expressions.
    2735             :      *
    2736             :      * We still need the NULL check, because we are only processing one
    2737             :      * grouping set at a time and the rest will be NULL.
    2738             :      */
    2739       26944 :     hashagg_recompile_expressions(aggstate, true, true);
    2740             : 
    2741       26944 :     INJECTION_POINT("hash-aggregate-process-batch", NULL);
    2742             :     for (;;)
    2743     1216776 :     {
    2744     1243720 :         TupleTableSlot *spillslot = aggstate->hash_spill_rslot;
    2745     1243720 :         TupleTableSlot *hashslot = perhash->hashslot;
    2746     1243720 :         TupleHashTable hashtable = perhash->hashtable;
    2747             :         TupleHashEntry entry;
    2748             :         MinimalTuple tuple;
    2749             :         uint32      hash;
    2750     1243720 :         bool        isnew = false;
    2751     1243720 :         bool       *p_isnew = aggstate->hash_spill_mode ? NULL : &isnew;
    2752             : 
    2753     1243720 :         CHECK_FOR_INTERRUPTS();
    2754             : 
    2755     1243720 :         tuple = hashagg_batch_read(batch, &hash);
    2756     1243720 :         if (tuple == NULL)
    2757       26944 :             break;
    2758             : 
    2759     1216776 :         ExecStoreMinimalTuple(tuple, spillslot, true);
    2760     1216776 :         aggstate->tmpcontext->ecxt_outertuple = spillslot;
    2761             : 
    2762     1216776 :         prepare_hash_slot(perhash,
    2763     1216776 :                           aggstate->tmpcontext->ecxt_outertuple,
    2764             :                           hashslot);
    2765     1216776 :         entry = LookupTupleHashEntryHash(hashtable, hashslot,
    2766             :                                          p_isnew, hash);
    2767             : 
    2768     1216776 :         if (entry != NULL)
    2769             :         {
    2770      767236 :             if (isnew)
    2771      155680 :                 initialize_hash_entry(aggstate, hashtable, entry);
    2772      767236 :             aggstate->hash_pergroup[batch->setno] = TupleHashEntryGetAdditional(hashtable, entry);
    2773      767236 :             advance_aggregates(aggstate);
    2774             :         }
    2775             :         else
    2776             :         {
    2777      449540 :             if (!spill_initialized)
    2778             :             {
    2779             :                 /*
    2780             :                  * Avoid initializing the spill until we actually need it so
    2781             :                  * that we don't assign tapes that will never be used.
    2782             :                  */
    2783       12500 :                 spill_initialized = true;
    2784       12500 :                 hashagg_spill_init(&spill, tapeset, batch->used_bits,
    2785             :                                    batch->input_card, aggstate->hashentrysize);
    2786             :             }
    2787             :             /* no memory for a new group, spill */
    2788      449540 :             hashagg_spill_tuple(aggstate, &spill, spillslot, hash);
    2789             : 
    2790      449540 :             aggstate->hash_pergroup[batch->setno] = NULL;
    2791             :         }
    2792             : 
    2793             :         /*
    2794             :          * Reset per-input-tuple context after each tuple, but note that the
    2795             :          * hash lookups do this too
    2796             :          */
    2797     1216776 :         ResetExprContext(aggstate->tmpcontext);
    2798             :     }
    2799             : 
    2800       26944 :     LogicalTapeClose(batch->input_tape);
    2801             : 
    2802             :     /* change back to phase 0 */
    2803       26944 :     aggstate->current_phase = 0;
    2804       26944 :     aggstate->phase = &aggstate->phases[aggstate->current_phase];
    2805             : 
    2806       26944 :     if (spill_initialized)
    2807             :     {
    2808       12500 :         hashagg_spill_finish(aggstate, &spill, batch->setno);
    2809       12500 :         hash_agg_update_metrics(aggstate, true, spill.npartitions);
    2810             :     }
    2811             :     else
    2812       14444 :         hash_agg_update_metrics(aggstate, true, 0);
    2813             : 
    2814       26944 :     aggstate->hash_spill_mode = false;
    2815             : 
    2816             :     /* prepare to walk the first hash table */
    2817       26944 :     select_current_set(aggstate, batch->setno, true);
    2818       26944 :     ResetTupleHashIterator(aggstate->perhash[batch->setno].hashtable,
    2819             :                            &aggstate->perhash[batch->setno].hashiter);
    2820             : 
    2821       26944 :     pfree(batch);
    2822             : 
    2823       26944 :     return true;
    2824             : }
    2825             : 
    2826             : /*
    2827             :  * ExecAgg for hashed case: retrieving groups from hash table
    2828             :  *
    2829             :  * After exhausting in-memory tuples, also try refilling the hash table using
    2830             :  * previously-spilled tuples. Only returns NULL after all in-memory and
    2831             :  * spilled tuples are exhausted.
    2832             :  */
    2833             : static TupleTableSlot *
    2834      511048 : agg_retrieve_hash_table(AggState *aggstate)
    2835             : {
    2836      511048 :     TupleTableSlot *result = NULL;
    2837             : 
    2838     1031212 :     while (result == NULL)
    2839             :     {
    2840      537992 :         result = agg_retrieve_hash_table_in_memory(aggstate);
    2841      537992 :         if (result == NULL)
    2842             :         {
    2843       44772 :             if (!agg_refill_hash_table(aggstate))
    2844             :             {
    2845       17828 :                 aggstate->agg_done = true;
    2846       17828 :                 break;
    2847             :             }
    2848             :         }
    2849             :     }
    2850             : 
    2851      511048 :     return result;
    2852             : }
    2853             : 
    2854             : /*
    2855             :  * Retrieve the groups from the in-memory hash tables without considering any
    2856             :  * spilled tuples.
    2857             :  */
    2858             : static TupleTableSlot *
    2859      537992 : agg_retrieve_hash_table_in_memory(AggState *aggstate)
    2860             : {
    2861             :     ExprContext *econtext;
    2862             :     AggStatePerAgg peragg;
    2863             :     AggStatePerGroup pergroup;
    2864             :     TupleHashEntry entry;
    2865             :     TupleTableSlot *firstSlot;
    2866             :     TupleTableSlot *result;
    2867             :     AggStatePerHash perhash;
    2868             : 
    2869             :     /*
    2870             :      * get state info from node.
    2871             :      *
    2872             :      * econtext is the per-output-tuple expression context.
    2873             :      */
    2874      537992 :     econtext = aggstate->ss.ps.ps_ExprContext;
    2875      537992 :     peragg = aggstate->peragg;
    2876      537992 :     firstSlot = aggstate->ss.ss_ScanTupleSlot;
    2877             : 
    2878             :     /*
    2879             :      * Note that perhash (and therefore anything accessed through it) can
    2880             :      * change inside the loop, as we change between grouping sets.
    2881             :      */
    2882      537992 :     perhash = &aggstate->perhash[aggstate->current_set];
    2883             : 
    2884             :     /*
    2885             :      * We loop retrieving groups until we find one satisfying
    2886             :      * aggstate->ss.ps.qual
    2887             :      */
    2888             :     for (;;)
    2889      135954 :     {
    2890      673946 :         TupleTableSlot *hashslot = perhash->hashslot;
    2891      673946 :         TupleHashTable hashtable = perhash->hashtable;
    2892             :         int         i;
    2893             : 
    2894      673946 :         CHECK_FOR_INTERRUPTS();
    2895             : 
    2896             :         /*
    2897             :          * Find the next entry in the hash table
    2898             :          */
    2899      673946 :         entry = ScanTupleHashTable(hashtable, &perhash->hashiter);
    2900      673946 :         if (entry == NULL)
    2901             :         {
    2902      145064 :             int         nextset = aggstate->current_set + 1;
    2903             : 
    2904      145064 :             if (nextset < aggstate->num_hashes)
    2905             :             {
    2906             :                 /*
    2907             :                  * Switch to next grouping set, reinitialize, and restart the
    2908             :                  * loop.
    2909             :                  */
    2910      100292 :                 select_current_set(aggstate, nextset, true);
    2911             : 
    2912      100292 :                 perhash = &aggstate->perhash[aggstate->current_set];
    2913             : 
    2914      100292 :                 ResetTupleHashIterator(hashtable, &perhash->hashiter);
    2915             : 
    2916      100292 :                 continue;
    2917             :             }
    2918             :             else
    2919             :             {
    2920       44772 :                 return NULL;
    2921             :             }
    2922             :         }
    2923             : 
    2924             :         /*
    2925             :          * Clear the per-output-tuple context for each group
    2926             :          *
    2927             :          * We intentionally don't use ReScanExprContext here; if any aggs have
    2928             :          * registered shutdown callbacks, they mustn't be called yet, since we
    2929             :          * might not be done with that agg.
    2930             :          */
    2931      528882 :         ResetExprContext(econtext);
    2932             : 
    2933             :         /*
    2934             :          * Transform representative tuple back into one with the right
    2935             :          * columns.
    2936             :          */
    2937      528882 :         ExecStoreMinimalTuple(TupleHashEntryGetTuple(entry), hashslot, false);
    2938      528882 :         slot_getallattrs(hashslot);
    2939             : 
    2940      528882 :         ExecClearTuple(firstSlot);
    2941      528882 :         memset(firstSlot->tts_isnull, true,
    2942      528882 :                firstSlot->tts_tupleDescriptor->natts * sizeof(bool));
    2943             : 
    2944     1389260 :         for (i = 0; i < perhash->numhashGrpCols; i++)
    2945             :         {
    2946      860378 :             int         varNumber = perhash->hashGrpColIdxInput[i] - 1;
    2947             : 
    2948      860378 :             firstSlot->tts_values[varNumber] = hashslot->tts_values[i];
    2949      860378 :             firstSlot->tts_isnull[varNumber] = hashslot->tts_isnull[i];
    2950             :         }
    2951      528882 :         ExecStoreVirtualTuple(firstSlot);
    2952             : 
    2953      528882 :         pergroup = (AggStatePerGroup) TupleHashEntryGetAdditional(hashtable, entry);
    2954             : 
    2955             :         /*
    2956             :          * Use the representative input tuple for any references to
    2957             :          * non-aggregated input columns in the qual and tlist.
    2958             :          */
    2959      528882 :         econtext->ecxt_outertuple = firstSlot;
    2960             : 
    2961      528882 :         prepare_projection_slot(aggstate,
    2962             :                                 econtext->ecxt_outertuple,
    2963             :                                 aggstate->current_set);
    2964             : 
    2965      528882 :         finalize_aggregates(aggstate, peragg, pergroup);
    2966             : 
    2967      528882 :         result = project_aggregates(aggstate);
    2968      528882 :         if (result)
    2969      493220 :             return result;
    2970             :     }
    2971             : 
    2972             :     /* No more groups */
    2973             :     return NULL;
    2974             : }
    2975             : 
    2976             : /*
    2977             :  * hashagg_spill_init
    2978             :  *
    2979             :  * Called after we determined that spilling is necessary. Chooses the number
    2980             :  * of partitions to create, and initializes them.
    2981             :  */
    2982             : static void
    2983       12622 : hashagg_spill_init(HashAggSpill *spill, LogicalTapeSet *tapeset, int used_bits,
    2984             :                    double input_groups, double hashentrysize)
    2985             : {
    2986             :     int         npartitions;
    2987             :     int         partition_bits;
    2988             : 
    2989       12622 :     npartitions = hash_choose_num_partitions(input_groups, hashentrysize,
    2990             :                                              used_bits, &partition_bits);
    2991             : 
    2992             : #ifdef USE_INJECTION_POINTS
    2993       12622 :     if (IS_INJECTION_POINT_ATTACHED("hash-aggregate-single-partition"))
    2994             :     {
    2995          10 :         npartitions = 1;
    2996          10 :         partition_bits = 0;
    2997          10 :         INJECTION_POINT_CACHED("hash-aggregate-single-partition", NULL);
    2998             :     }
    2999             : #endif
    3000             : 
    3001       12622 :     spill->partitions = palloc0(sizeof(LogicalTape *) * npartitions);
    3002       12622 :     spill->ntuples = palloc0(sizeof(int64) * npartitions);
    3003       12622 :     spill->hll_card = palloc0(sizeof(hyperLogLogState) * npartitions);
    3004             : 
    3005       63080 :     for (int i = 0; i < npartitions; i++)
    3006       50458 :         spill->partitions[i] = LogicalTapeCreate(tapeset);
    3007             : 
    3008       12622 :     spill->shift = 32 - used_bits - partition_bits;
    3009       12622 :     if (spill->shift < 32)
    3010       12612 :         spill->mask = (npartitions - 1) << spill->shift;
    3011             :     else
    3012          10 :         spill->mask = 0;
    3013       12622 :     spill->npartitions = npartitions;
    3014             : 
    3015       63080 :     for (int i = 0; i < npartitions; i++)
    3016       50458 :         initHyperLogLog(&spill->hll_card[i], HASHAGG_HLL_BIT_WIDTH);
    3017       12622 : }
    3018             : 
    3019             : /*
    3020             :  * hashagg_spill_tuple
    3021             :  *
    3022             :  * No room for new groups in the hash table. Save for later in the appropriate
    3023             :  * partition.
    3024             :  */
    3025             : static Size
    3026     1216776 : hashagg_spill_tuple(AggState *aggstate, HashAggSpill *spill,
    3027             :                     TupleTableSlot *inputslot, uint32 hash)
    3028             : {
    3029             :     TupleTableSlot *spillslot;
    3030             :     int         partition;
    3031             :     MinimalTuple tuple;
    3032             :     LogicalTape *tape;
    3033     1216776 :     int         total_written = 0;
    3034             :     bool        shouldFree;
    3035             : 
    3036             :     Assert(spill->partitions != NULL);
    3037             : 
    3038             :     /* spill only attributes that we actually need */
    3039     1216776 :     if (!aggstate->all_cols_needed)
    3040             :     {
    3041        1572 :         spillslot = aggstate->hash_spill_wslot;
    3042        1572 :         slot_getsomeattrs(inputslot, aggstate->max_colno_needed);
    3043        1572 :         ExecClearTuple(spillslot);
    3044        4716 :         for (int i = 0; i < spillslot->tts_tupleDescriptor->natts; i++)
    3045             :         {
    3046        3144 :             if (bms_is_member(i + 1, aggstate->colnos_needed))
    3047             :             {
    3048        1572 :                 spillslot->tts_values[i] = inputslot->tts_values[i];
    3049        1572 :                 spillslot->tts_isnull[i] = inputslot->tts_isnull[i];
    3050             :             }
    3051             :             else
    3052        1572 :                 spillslot->tts_isnull[i] = true;
    3053             :         }
    3054        1572 :         ExecStoreVirtualTuple(spillslot);
    3055             :     }
    3056             :     else
    3057     1215204 :         spillslot = inputslot;
    3058             : 
    3059     1216776 :     tuple = ExecFetchSlotMinimalTuple(spillslot, &shouldFree);
    3060             : 
    3061     1216776 :     if (spill->shift < 32)
    3062     1195776 :         partition = (hash & spill->mask) >> spill->shift;
    3063             :     else
    3064       21000 :         partition = 0;
    3065             : 
    3066     1216776 :     spill->ntuples[partition]++;
    3067             : 
    3068             :     /*
    3069             :      * All hash values destined for a given partition have some bits in
    3070             :      * common, which causes bad HLL cardinality estimates. Hash the hash to
    3071             :      * get a more uniform distribution.
    3072             :      */
    3073     1216776 :     addHyperLogLog(&spill->hll_card[partition], hash_bytes_uint32(hash));
    3074             : 
    3075     1216776 :     tape = spill->partitions[partition];
    3076             : 
    3077     1216776 :     LogicalTapeWrite(tape, &hash, sizeof(uint32));
    3078     1216776 :     total_written += sizeof(uint32);
    3079             : 
    3080     1216776 :     LogicalTapeWrite(tape, tuple, tuple->t_len);
    3081     1216776 :     total_written += tuple->t_len;
    3082             : 
    3083     1216776 :     if (shouldFree)
    3084      767236 :         pfree(tuple);
    3085             : 
    3086     1216776 :     return total_written;
    3087             : }
    3088             : 
    3089             : /*
    3090             :  * hashagg_batch_new
    3091             :  *
    3092             :  * Construct a HashAggBatch item, which represents one iteration of HashAgg to
    3093             :  * be done.
    3094             :  */
    3095             : static HashAggBatch *
    3096       26944 : hashagg_batch_new(LogicalTape *input_tape, int setno,
    3097             :                   int64 input_tuples, double input_card, int used_bits)
    3098             : {
    3099       26944 :     HashAggBatch *batch = palloc0(sizeof(HashAggBatch));
    3100             : 
    3101       26944 :     batch->setno = setno;
    3102       26944 :     batch->used_bits = used_bits;
    3103       26944 :     batch->input_tape = input_tape;
    3104       26944 :     batch->input_tuples = input_tuples;
    3105       26944 :     batch->input_card = input_card;
    3106             : 
    3107       26944 :     return batch;
    3108             : }
    3109             : 
    3110             : /*
    3111             :  * hashagg_batch_read
    3112             :  *      read the next tuple from a batch's tape.  Return NULL if no more.
    3113             :  */
    3114             : static MinimalTuple
    3115     1243720 : hashagg_batch_read(HashAggBatch *batch, uint32 *hashp)
    3116             : {
    3117     1243720 :     LogicalTape *tape = batch->input_tape;
    3118             :     MinimalTuple tuple;
    3119             :     uint32      t_len;
    3120             :     size_t      nread;
    3121             :     uint32      hash;
    3122             : 
    3123     1243720 :     nread = LogicalTapeRead(tape, &hash, sizeof(uint32));
    3124     1243720 :     if (nread == 0)
    3125       26944 :         return NULL;
    3126     1216776 :     if (nread != sizeof(uint32))
    3127           0 :         ereport(ERROR,
    3128             :                 (errcode_for_file_access(),
    3129             :                  errmsg_internal("unexpected EOF for tape %p: requested %zu bytes, read %zu bytes",
    3130             :                                  tape, sizeof(uint32), nread)));
    3131     1216776 :     if (hashp != NULL)
    3132     1216776 :         *hashp = hash;
    3133             : 
    3134     1216776 :     nread = LogicalTapeRead(tape, &t_len, sizeof(t_len));
    3135     1216776 :     if (nread != sizeof(uint32))
    3136           0 :         ereport(ERROR,
    3137             :                 (errcode_for_file_access(),
    3138             :                  errmsg_internal("unexpected EOF for tape %p: requested %zu bytes, read %zu bytes",
    3139             :                                  tape, sizeof(uint32), nread)));
    3140             : 
    3141     1216776 :     tuple = (MinimalTuple) palloc(t_len);
    3142     1216776 :     tuple->t_len = t_len;
    3143             : 
    3144     1216776 :     nread = LogicalTapeRead(tape,
    3145             :                             (char *) tuple + sizeof(uint32),
    3146             :                             t_len - sizeof(uint32));
    3147     1216776 :     if (nread != t_len - sizeof(uint32))
    3148           0 :         ereport(ERROR,
    3149             :                 (errcode_for_file_access(),
    3150             :                  errmsg_internal("unexpected EOF for tape %p: requested %zu bytes, read %zu bytes",
    3151             :                                  tape, t_len - sizeof(uint32), nread)));
    3152             : 
    3153     1216776 :     return tuple;
    3154             : }
    3155             : 
    3156             : /*
    3157             :  * hashagg_finish_initial_spills
    3158             :  *
    3159             :  * After a HashAggBatch has been processed, it may have spilled tuples to
    3160             :  * disk. If so, turn the spilled partitions into new batches that must later
    3161             :  * be executed.
    3162             :  */
    3163             : static void
    3164       16998 : hashagg_finish_initial_spills(AggState *aggstate)
    3165             : {
    3166             :     int         setno;
    3167       16998 :     int         total_npartitions = 0;
    3168             : 
    3169       16998 :     if (aggstate->hash_spills != NULL)
    3170             :     {
    3171         184 :         for (setno = 0; setno < aggstate->num_hashes; setno++)
    3172             :         {
    3173         122 :             HashAggSpill *spill = &aggstate->hash_spills[setno];
    3174             : 
    3175         122 :             total_npartitions += spill->npartitions;
    3176         122 :             hashagg_spill_finish(aggstate, spill, setno);
    3177             :         }
    3178             : 
    3179             :         /*
    3180             :          * We're not processing tuples from outer plan any more; only
    3181             :          * processing batches of spilled tuples. The initial spill structures
    3182             :          * are no longer needed.
    3183             :          */
    3184          62 :         pfree(aggstate->hash_spills);
    3185          62 :         aggstate->hash_spills = NULL;
    3186             :     }
    3187             : 
    3188       16998 :     hash_agg_update_metrics(aggstate, false, total_npartitions);
    3189       16998 :     aggstate->hash_spill_mode = false;
    3190       16998 : }
    3191             : 
    3192             : /*
    3193             :  * hashagg_spill_finish
    3194             :  *
    3195             :  * Transform spill partitions into new batches.
    3196             :  */
    3197             : static void
    3198       12622 : hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno)
    3199             : {
    3200             :     int         i;
    3201       12622 :     int         used_bits = 32 - spill->shift;
    3202             : 
    3203       12622 :     if (spill->npartitions == 0)
    3204           0 :         return;                 /* didn't spill */
    3205             : 
    3206       63080 :     for (i = 0; i < spill->npartitions; i++)
    3207             :     {
    3208       50458 :         LogicalTape *tape = spill->partitions[i];
    3209             :         HashAggBatch *new_batch;
    3210             :         double      cardinality;
    3211             : 
    3212             :         /* if the partition is empty, don't create a new batch of work */
    3213       50458 :         if (spill->ntuples[i] == 0)
    3214       23514 :             continue;
    3215             : 
    3216       26944 :         cardinality = estimateHyperLogLog(&spill->hll_card[i]);
    3217       26944 :         freeHyperLogLog(&spill->hll_card[i]);
    3218             : 
    3219             :         /* rewinding frees the buffer while not in use */
    3220       26944 :         LogicalTapeRewindForRead(tape, HASHAGG_READ_BUFFER_SIZE);
    3221             : 
    3222       26944 :         new_batch = hashagg_batch_new(tape, setno,
    3223       26944 :                                       spill->ntuples[i], cardinality,
    3224             :                                       used_bits);
    3225       26944 :         aggstate->hash_batches = lappend(aggstate->hash_batches, new_batch);
    3226       26944 :         aggstate->hash_batches_used++;
    3227             :     }
    3228             : 
    3229       12622 :     pfree(spill->ntuples);
    3230       12622 :     pfree(spill->hll_card);
    3231       12622 :     pfree(spill->partitions);
    3232             : }
    3233             : 
    3234             : /*
    3235             :  * Free resources related to a spilled HashAgg.
    3236             :  */
    3237             : static void
    3238       59918 : hashagg_reset_spill_state(AggState *aggstate)
    3239             : {
    3240             :     /* free spills from initial pass */
    3241       59918 :     if (aggstate->hash_spills != NULL)
    3242             :     {
    3243             :         int         setno;
    3244             : 
    3245           0 :         for (setno = 0; setno < aggstate->num_hashes; setno++)
    3246             :         {
    3247           0 :             HashAggSpill *spill = &aggstate->hash_spills[setno];
    3248             : 
    3249           0 :             pfree(spill->ntuples);
    3250           0 :             pfree(spill->partitions);
    3251             :         }
    3252           0 :         pfree(aggstate->hash_spills);
    3253           0 :         aggstate->hash_spills = NULL;
    3254             :     }
    3255             : 
    3256             :     /* free batches */
    3257       59918 :     list_free_deep(aggstate->hash_batches);
    3258       59918 :     aggstate->hash_batches = NIL;
    3259             : 
    3260             :     /* close tape set */
    3261       59918 :     if (aggstate->hash_tapeset != NULL)
    3262             :     {
    3263          62 :         LogicalTapeSetClose(aggstate->hash_tapeset);
    3264          62 :         aggstate->hash_tapeset = NULL;
    3265             :     }
    3266       59918 : }
    3267             : 
    3268             : 
    3269             : /* -----------------
    3270             :  * ExecInitAgg
    3271             :  *
    3272             :  *  Creates the run-time information for the agg node produced by the
    3273             :  *  planner and initializes its outer subtree.
    3274             :  *
    3275             :  * -----------------
    3276             :  */
    3277             : AggState *
    3278       47726 : ExecInitAgg(Agg *node, EState *estate, int eflags)
    3279             : {
    3280             :     AggState   *aggstate;
    3281             :     AggStatePerAgg peraggs;
    3282             :     AggStatePerTrans pertransstates;
    3283             :     AggStatePerGroup *pergroups;
    3284             :     Plan       *outerPlan;
    3285             :     ExprContext *econtext;
    3286             :     TupleDesc   scanDesc;
    3287             :     int         max_aggno;
    3288             :     int         max_transno;
    3289             :     int         numaggrefs;
    3290             :     int         numaggs;
    3291             :     int         numtrans;
    3292             :     int         phase;
    3293             :     int         phaseidx;
    3294             :     ListCell   *l;
    3295       47726 :     Bitmapset  *all_grouped_cols = NULL;
    3296       47726 :     int         numGroupingSets = 1;
    3297             :     int         numPhases;
    3298             :     int         numHashes;
    3299       47726 :     int         i = 0;
    3300       47726 :     int         j = 0;
    3301       88974 :     bool        use_hashing = (node->aggstrategy == AGG_HASHED ||
    3302       41248 :                                node->aggstrategy == AGG_MIXED);
    3303             : 
    3304             :     /* check for unsupported flags */
    3305             :     Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
    3306             : 
    3307             :     /*
    3308             :      * create state structure
    3309             :      */
    3310       47726 :     aggstate = makeNode(AggState);
    3311       47726 :     aggstate->ss.ps.plan = (Plan *) node;
    3312       47726 :     aggstate->ss.ps.state = estate;
    3313       47726 :     aggstate->ss.ps.ExecProcNode = ExecAgg;
    3314             : 
    3315       47726 :     aggstate->aggs = NIL;
    3316       47726 :     aggstate->numaggs = 0;
    3317       47726 :     aggstate->numtrans = 0;
    3318       47726 :     aggstate->aggstrategy = node->aggstrategy;
    3319       47726 :     aggstate->aggsplit = node->aggsplit;
    3320       47726 :     aggstate->maxsets = 0;
    3321       47726 :     aggstate->projected_set = -1;
    3322       47726 :     aggstate->current_set = 0;
    3323       47726 :     aggstate->peragg = NULL;
    3324       47726 :     aggstate->pertrans = NULL;
    3325       47726 :     aggstate->curperagg = NULL;
    3326       47726 :     aggstate->curpertrans = NULL;
    3327       47726 :     aggstate->input_done = false;
    3328       47726 :     aggstate->agg_done = false;
    3329       47726 :     aggstate->pergroups = NULL;
    3330       47726 :     aggstate->grp_firstTuple = NULL;
    3331       47726 :     aggstate->sort_in = NULL;
    3332       47726 :     aggstate->sort_out = NULL;
    3333             : 
    3334             :     /*
    3335             :      * phases[0] always exists, but is dummy in sorted/plain mode
    3336             :      */
    3337       47726 :     numPhases = (use_hashing ? 1 : 2);
    3338       47726 :     numHashes = (use_hashing ? 1 : 0);
    3339             : 
    3340             :     /*
    3341             :      * Calculate the maximum number of grouping sets in any phase; this
    3342             :      * determines the size of some allocations.  Also calculate the number of
    3343             :      * phases, since all hashed/mixed nodes contribute to only a single phase.
    3344             :      */
    3345       47726 :     if (node->groupingSets)
    3346             :     {
    3347         866 :         numGroupingSets = list_length(node->groupingSets);
    3348             : 
    3349        1852 :         foreach(l, node->chain)
    3350             :         {
    3351         986 :             Agg        *agg = lfirst(l);
    3352             : 
    3353         986 :             numGroupingSets = Max(numGroupingSets,
    3354             :                                   list_length(agg->groupingSets));
    3355             : 
    3356             :             /*
    3357             :              * additional AGG_HASHED aggs become part of phase 0, but all
    3358             :              * others add an extra phase.
    3359             :              */
    3360         986 :             if (agg->aggstrategy != AGG_HASHED)
    3361         472 :                 ++numPhases;
    3362             :             else
    3363         514 :                 ++numHashes;
    3364             :         }
    3365             :     }
    3366             : 
    3367       47726 :     aggstate->maxsets = numGroupingSets;
    3368       47726 :     aggstate->numphases = numPhases;
    3369             : 
    3370       47726 :     aggstate->aggcontexts = (ExprContext **)
    3371       47726 :         palloc0(sizeof(ExprContext *) * numGroupingSets);
    3372             : 
    3373             :     /*
    3374             :      * Create expression contexts.  We need three or more, one for
    3375             :      * per-input-tuple processing, one for per-output-tuple processing, one
    3376             :      * for all the hashtables, and one for each grouping set.  The per-tuple
    3377             :      * memory context of the per-grouping-set ExprContexts (aggcontexts)
    3378             :      * replaces the standalone memory context formerly used to hold transition
    3379             :      * values.  We cheat a little by using ExecAssignExprContext() to build
    3380             :      * all of them.
    3381             :      *
    3382             :      * NOTE: the details of what is stored in aggcontexts and what is stored
    3383             :      * in the regular per-query memory context are driven by a simple
    3384             :      * decision: we want to reset the aggcontext at group boundaries (if not
    3385             :      * hashing) and in ExecReScanAgg to recover no-longer-wanted space.
    3386             :      */
    3387       47726 :     ExecAssignExprContext(estate, &aggstate->ss.ps);
    3388       47726 :     aggstate->tmpcontext = aggstate->ss.ps.ps_ExprContext;
    3389             : 
    3390       96292 :     for (i = 0; i < numGroupingSets; ++i)
    3391             :     {
    3392       48566 :         ExecAssignExprContext(estate, &aggstate->ss.ps);
    3393       48566 :         aggstate->aggcontexts[i] = aggstate->ss.ps.ps_ExprContext;
    3394             :     }
    3395             : 
    3396       47726 :     if (use_hashing)
    3397        6710 :         hash_create_memory(aggstate);
    3398             : 
    3399       47726 :     ExecAssignExprContext(estate, &aggstate->ss.ps);
    3400             : 
    3401             :     /*
    3402             :      * Initialize child nodes.
    3403             :      *
    3404             :      * If we are doing a hashed aggregation then the child plan does not need
    3405             :      * to handle REWIND efficiently; see ExecReScanAgg.
    3406             :      */
    3407       47726 :     if (node->aggstrategy == AGG_HASHED)
    3408        6478 :         eflags &= ~EXEC_FLAG_REWIND;
    3409       47726 :     outerPlan = outerPlan(node);
    3410       47726 :     outerPlanState(aggstate) = ExecInitNode(outerPlan, estate, eflags);
    3411             : 
    3412             :     /*
    3413             :      * initialize source tuple type.
    3414             :      */
    3415       47726 :     aggstate->ss.ps.outerops =
    3416       47726 :         ExecGetResultSlotOps(outerPlanState(&aggstate->ss),
    3417             :                              &aggstate->ss.ps.outeropsfixed);
    3418       47726 :     aggstate->ss.ps.outeropsset = true;
    3419             : 
    3420       47726 :     ExecCreateScanSlotFromOuterPlan(estate, &aggstate->ss,
    3421             :                                     aggstate->ss.ps.outerops);
    3422       47726 :     scanDesc = aggstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
    3423             : 
    3424             :     /*
    3425             :      * If there are more than two phases (including a potential dummy phase
    3426             :      * 0), input will be resorted using tuplesort. Need a slot for that.
    3427             :      */
    3428       47726 :     if (numPhases > 2)
    3429             :     {
    3430         198 :         aggstate->sort_slot = ExecInitExtraTupleSlot(estate, scanDesc,
    3431             :                                                      &TTSOpsMinimalTuple);
    3432             : 
    3433             :         /*
    3434             :          * The output of the tuplesort, and the output from the outer child
    3435             :          * might not use the same type of slot. In most cases the child will
    3436             :          * be a Sort, and thus return a TTSOpsMinimalTuple type slot - but the
    3437             :          * input can also be presorted due an index, in which case it could be
    3438             :          * a different type of slot.
    3439             :          *
    3440             :          * XXX: For efficiency it would be good to instead/additionally
    3441             :          * generate expressions with corresponding settings of outerops* for
    3442             :          * the individual phases - deforming is often a bottleneck for
    3443             :          * aggregations with lots of rows per group. If there's multiple
    3444             :          * sorts, we know that all but the first use TTSOpsMinimalTuple (via
    3445             :          * the nodeAgg.c internal tuplesort).
    3446             :          */
    3447         198 :         if (aggstate->ss.ps.outeropsfixed &&
    3448         198 :             aggstate->ss.ps.outerops != &TTSOpsMinimalTuple)
    3449          12 :             aggstate->ss.ps.outeropsfixed = false;
    3450             :     }
    3451             : 
    3452             :     /*
    3453             :      * Initialize result type, slot and projection.
    3454             :      */
    3455       47726 :     ExecInitResultTupleSlotTL(&aggstate->ss.ps, &TTSOpsVirtual);
    3456       47726 :     ExecAssignProjectionInfo(&aggstate->ss.ps, NULL);
    3457             : 
    3458             :     /*
    3459             :      * initialize child expressions
    3460             :      *
    3461             :      * We expect the parser to have checked that no aggs contain other agg
    3462             :      * calls in their arguments (and just to be sure, we verify it again while
    3463             :      * initializing the plan node).  This would make no sense under SQL
    3464             :      * semantics, and it's forbidden by the spec.  Because it is true, we
    3465             :      * don't need to worry about evaluating the aggs in any particular order.
    3466             :      *
    3467             :      * Note: execExpr.c finds Aggrefs for us, and adds them to aggstate->aggs.
    3468             :      * Aggrefs in the qual are found here; Aggrefs in the targetlist are found
    3469             :      * during ExecAssignProjectionInfo, above.
    3470             :      */
    3471       47726 :     aggstate->ss.ps.qual =
    3472       47726 :         ExecInitQual(node->plan.qual, (PlanState *) aggstate);
    3473             : 
    3474             :     /*
    3475             :      * We should now have found all Aggrefs in the targetlist and quals.
    3476             :      */
    3477       47726 :     numaggrefs = list_length(aggstate->aggs);
    3478       47726 :     max_aggno = -1;
    3479       47726 :     max_transno = -1;
    3480      101860 :     foreach(l, aggstate->aggs)
    3481             :     {
    3482       54134 :         Aggref     *aggref = (Aggref *) lfirst(l);
    3483             : 
    3484       54134 :         max_aggno = Max(max_aggno, aggref->aggno);
    3485       54134 :         max_transno = Max(max_transno, aggref->aggtransno);
    3486             :     }
    3487       47726 :     aggstate->numaggs = numaggs = max_aggno + 1;
    3488       47726 :     aggstate->numtrans = numtrans = max_transno + 1;
    3489             : 
    3490             :     /*
    3491             :      * For each phase, prepare grouping set data and fmgr lookup data for
    3492             :      * compare functions.  Accumulate all_grouped_cols in passing.
    3493             :      */
    3494       47726 :     aggstate->phases = palloc0(numPhases * sizeof(AggStatePerPhaseData));
    3495             : 
    3496       47726 :     aggstate->num_hashes = numHashes;
    3497       47726 :     if (numHashes)
    3498             :     {
    3499        6710 :         aggstate->perhash = palloc0(sizeof(AggStatePerHashData) * numHashes);
    3500        6710 :         aggstate->phases[0].numsets = 0;
    3501        6710 :         aggstate->phases[0].gset_lengths = palloc(numHashes * sizeof(int));
    3502        6710 :         aggstate->phases[0].grouped_cols = palloc(numHashes * sizeof(Bitmapset *));
    3503             :     }
    3504             : 
    3505       47726 :     phase = 0;
    3506       96438 :     for (phaseidx = 0; phaseidx <= list_length(node->chain); ++phaseidx)
    3507             :     {
    3508             :         Agg        *aggnode;
    3509             :         Sort       *sortnode;
    3510             : 
    3511       48712 :         if (phaseidx > 0)
    3512             :         {
    3513         986 :             aggnode = list_nth_node(Agg, node->chain, phaseidx - 1);
    3514         986 :             sortnode = castNode(Sort, outerPlan(aggnode));
    3515             :         }
    3516             :         else
    3517             :         {
    3518       47726 :             aggnode = node;
    3519       47726 :             sortnode = NULL;
    3520             :         }
    3521             : 
    3522             :         Assert(phase <= 1 || sortnode);
    3523             : 
    3524       48712 :         if (aggnode->aggstrategy == AGG_HASHED
    3525       41720 :             || aggnode->aggstrategy == AGG_MIXED)
    3526        7224 :         {
    3527        7224 :             AggStatePerPhase phasedata = &aggstate->phases[0];
    3528             :             AggStatePerHash perhash;
    3529        7224 :             Bitmapset  *cols = NULL;
    3530             : 
    3531             :             Assert(phase == 0);
    3532        7224 :             i = phasedata->numsets++;
    3533        7224 :             perhash = &aggstate->perhash[i];
    3534             : 
    3535             :             /* phase 0 always points to the "real" Agg in the hash case */
    3536        7224 :             phasedata->aggnode = node;
    3537        7224 :             phasedata->aggstrategy = node->aggstrategy;
    3538             : 
    3539             :             /* but the actual Agg node representing this hash is saved here */
    3540        7224 :             perhash->aggnode = aggnode;
    3541             : 
    3542        7224 :             phasedata->gset_lengths[i] = perhash->numCols = aggnode->numCols;
    3543             : 
    3544       17752 :             for (j = 0; j < aggnode->numCols; ++j)
    3545       10528 :                 cols = bms_add_member(cols, aggnode->grpColIdx[j]);
    3546             : 
    3547        7224 :             phasedata->grouped_cols[i] = cols;
    3548             : 
    3549        7224 :             all_grouped_cols = bms_add_members(all_grouped_cols, cols);
    3550        7224 :             continue;
    3551             :         }
    3552             :         else
    3553             :         {
    3554       41488 :             AggStatePerPhase phasedata = &aggstate->phases[++phase];
    3555             :             int         num_sets;
    3556             : 
    3557       41488 :             phasedata->numsets = num_sets = list_length(aggnode->groupingSets);
    3558             : 
    3559       41488 :             if (num_sets)
    3560             :             {
    3561         940 :                 phasedata->gset_lengths = palloc(num_sets * sizeof(int));
    3562         940 :                 phasedata->grouped_cols = palloc(num_sets * sizeof(Bitmapset *));
    3563             : 
    3564         940 :                 i = 0;
    3565        2792 :                 foreach(l, aggnode->groupingSets)
    3566             :                 {
    3567        1852 :                     int         current_length = list_length(lfirst(l));
    3568        1852 :                     Bitmapset  *cols = NULL;
    3569             : 
    3570             :                     /* planner forces this to be correct */
    3571        3648 :                     for (j = 0; j < current_length; ++j)
    3572        1796 :                         cols = bms_add_member(cols, aggnode->grpColIdx[j]);
    3573             : 
    3574        1852 :                     phasedata->grouped_cols[i] = cols;
    3575        1852 :                     phasedata->gset_lengths[i] = current_length;
    3576             : 
    3577        1852 :                     ++i;
    3578             :                 }
    3579             : 
    3580         940 :                 all_grouped_cols = bms_add_members(all_grouped_cols,
    3581         940 :                                                    phasedata->grouped_cols[0]);
    3582             :             }
    3583             :             else
    3584             :             {
    3585             :                 Assert(phaseidx == 0);
    3586             : 
    3587       40548 :                 phasedata->gset_lengths = NULL;
    3588       40548 :                 phasedata->grouped_cols = NULL;
    3589             :             }
    3590             : 
    3591             :             /*
    3592             :              * If we are grouping, precompute fmgr lookup data for inner loop.
    3593             :              */
    3594       41488 :             if (aggnode->aggstrategy == AGG_SORTED)
    3595             :             {
    3596             :                 /*
    3597             :                  * Build a separate function for each subset of columns that
    3598             :                  * need to be compared.
    3599             :                  */
    3600        2468 :                 phasedata->eqfunctions =
    3601        2468 :                     (ExprState **) palloc0(aggnode->numCols * sizeof(ExprState *));
    3602             : 
    3603             :                 /* for each grouping set */
    3604        4036 :                 for (int k = 0; k < phasedata->numsets; k++)
    3605             :                 {
    3606        1568 :                     int         length = phasedata->gset_lengths[k];
    3607             : 
    3608             :                     /* nothing to do for empty grouping set */
    3609        1568 :                     if (length == 0)
    3610         326 :                         continue;
    3611             : 
    3612             :                     /* if we already had one of this length, it'll do */
    3613        1242 :                     if (phasedata->eqfunctions[length - 1] != NULL)
    3614         138 :                         continue;
    3615             : 
    3616        1104 :                     phasedata->eqfunctions[length - 1] =
    3617        1104 :                         execTuplesMatchPrepare(scanDesc,
    3618             :                                                length,
    3619        1104 :                                                aggnode->grpColIdx,
    3620        1104 :                                                aggnode->grpOperators,
    3621        1104 :                                                aggnode->grpCollations,
    3622             :                                                (PlanState *) aggstate);
    3623             :                 }
    3624             : 
    3625             :                 /* and for all grouped columns, unless already computed */
    3626        2468 :                 if (aggnode->numCols > 0 &&
    3627        2374 :                     phasedata->eqfunctions[aggnode->numCols - 1] == NULL)
    3628             :                 {
    3629        1646 :                     phasedata->eqfunctions[aggnode->numCols - 1] =
    3630        1646 :                         execTuplesMatchPrepare(scanDesc,
    3631             :                                                aggnode->numCols,
    3632        1646 :                                                aggnode->grpColIdx,
    3633        1646 :                                                aggnode->grpOperators,
    3634        1646 :                                                aggnode->grpCollations,
    3635             :                                                (PlanState *) aggstate);
    3636             :                 }
    3637             :             }
    3638             : 
    3639       41488 :             phasedata->aggnode = aggnode;
    3640       41488 :             phasedata->aggstrategy = aggnode->aggstrategy;
    3641       41488 :             phasedata->sortnode = sortnode;
    3642             :         }
    3643             :     }
    3644             : 
    3645             :     /*
    3646             :      * Convert all_grouped_cols to a descending-order list.
    3647             :      */
    3648       47726 :     i = -1;
    3649       58814 :     while ((i = bms_next_member(all_grouped_cols, i)) >= 0)
    3650       11088 :         aggstate->all_grouped_cols = lcons_int(i, aggstate->all_grouped_cols);
    3651             : 
    3652             :     /*
    3653             :      * Set up aggregate-result storage in the output expr context, and also
    3654             :      * allocate my private per-agg working storage
    3655             :      */
    3656       47726 :     econtext = aggstate->ss.ps.ps_ExprContext;
    3657       47726 :     econtext->ecxt_aggvalues = (Datum *) palloc0(sizeof(Datum) * numaggs);
    3658       47726 :     econtext->ecxt_aggnulls = (bool *) palloc0(sizeof(bool) * numaggs);
    3659             : 
    3660       47726 :     peraggs = (AggStatePerAgg) palloc0(sizeof(AggStatePerAggData) * numaggs);
    3661       47726 :     pertransstates = (AggStatePerTrans) palloc0(sizeof(AggStatePerTransData) * numtrans);
    3662             : 
    3663       47726 :     aggstate->peragg = peraggs;
    3664       47726 :     aggstate->pertrans = pertransstates;
    3665             : 
    3666             : 
    3667       47726 :     aggstate->all_pergroups =
    3668       47726 :         (AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup)
    3669       47726 :                                      * (numGroupingSets + numHashes));
    3670       47726 :     pergroups = aggstate->all_pergroups;
    3671             : 
    3672       47726 :     if (node->aggstrategy != AGG_HASHED)
    3673             :     {
    3674       83336 :         for (i = 0; i < numGroupingSets; i++)
    3675             :         {
    3676       42088 :             pergroups[i] = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData)
    3677             :                                                       * numaggs);
    3678             :         }
    3679             : 
    3680       41248 :         aggstate->pergroups = pergroups;
    3681       41248 :         pergroups += numGroupingSets;
    3682             :     }
    3683             : 
    3684             :     /*
    3685             :      * Hashing can only appear in the initial phase.
    3686             :      */
    3687       47726 :     if (use_hashing)
    3688             :     {
    3689        6710 :         Plan       *outerplan = outerPlan(node);
    3690        6710 :         uint64      totalGroups = 0;
    3691             : 
    3692        6710 :         aggstate->hash_spill_rslot = ExecInitExtraTupleSlot(estate, scanDesc,
    3693             :                                                             &TTSOpsMinimalTuple);
    3694        6710 :         aggstate->hash_spill_wslot = ExecInitExtraTupleSlot(estate, scanDesc,
    3695             :                                                             &TTSOpsVirtual);
    3696             : 
    3697             :         /* this is an array of pointers, not structures */
    3698        6710 :         aggstate->hash_pergroup = pergroups;
    3699             : 
    3700       13420 :         aggstate->hashentrysize = hash_agg_entry_size(aggstate->numtrans,
    3701        6710 :                                                       outerplan->plan_width,
    3702             :                                                       node->transitionSpace);
    3703             : 
    3704             :         /*
    3705             :          * Consider all of the grouping sets together when setting the limits
    3706             :          * and estimating the number of partitions. This can be inaccurate
    3707             :          * when there is more than one grouping set, but should still be
    3708             :          * reasonable.
    3709             :          */
    3710       13934 :         for (int k = 0; k < aggstate->num_hashes; k++)
    3711        7224 :             totalGroups += aggstate->perhash[k].aggnode->numGroups;
    3712             : 
    3713        6710 :         hash_agg_set_limits(aggstate->hashentrysize, totalGroups, 0,
    3714             :                             &aggstate->hash_mem_limit,
    3715             :                             &aggstate->hash_ngroups_limit,
    3716             :                             &aggstate->hash_planned_partitions);
    3717        6710 :         find_hash_columns(aggstate);
    3718             : 
    3719             :         /* Skip massive memory allocation if we are just doing EXPLAIN */
    3720        6710 :         if (!(eflags & EXEC_FLAG_EXPLAIN_ONLY))
    3721        4864 :             build_hash_tables(aggstate);
    3722             : 
    3723        6710 :         aggstate->table_filled = false;
    3724             : 
    3725             :         /* Initialize this to 1, meaning nothing spilled, yet */
    3726        6710 :         aggstate->hash_batches_used = 1;
    3727             :     }
    3728             : 
    3729             :     /*
    3730             :      * Initialize current phase-dependent values to initial phase. The initial
    3731             :      * phase is 1 (first sort pass) for all strategies that use sorting (if
    3732             :      * hashing is being done too, then phase 0 is processed last); but if only
    3733             :      * hashing is being done, then phase 0 is all there is.
    3734             :      */
    3735       47726 :     if (node->aggstrategy == AGG_HASHED)
    3736             :     {
    3737        6478 :         aggstate->current_phase = 0;
    3738        6478 :         initialize_phase(aggstate, 0);
    3739        6478 :         select_current_set(aggstate, 0, true);
    3740             :     }
    3741             :     else
    3742             :     {
    3743       41248 :         aggstate->current_phase = 1;
    3744       41248 :         initialize_phase(aggstate, 1);
    3745       41248 :         select_current_set(aggstate, 0, false);
    3746             :     }
    3747             : 
    3748             :     /*
    3749             :      * Perform lookups of aggregate function info, and initialize the
    3750             :      * unchanging fields of the per-agg and per-trans data.
    3751             :      */
    3752      101854 :     foreach(l, aggstate->aggs)
    3753             :     {
    3754       54134 :         Aggref     *aggref = lfirst(l);
    3755             :         AggStatePerAgg peragg;
    3756             :         AggStatePerTrans pertrans;
    3757             :         Oid         aggTransFnInputTypes[FUNC_MAX_ARGS];
    3758             :         int         numAggTransFnArgs;
    3759             :         int         numDirectArgs;
    3760             :         HeapTuple   aggTuple;
    3761             :         Form_pg_aggregate aggform;
    3762             :         AclResult   aclresult;
    3763             :         Oid         finalfn_oid;
    3764             :         Oid         serialfn_oid,
    3765             :                     deserialfn_oid;
    3766             :         Oid         aggOwner;
    3767             :         Expr       *finalfnexpr;
    3768             :         Oid         aggtranstype;
    3769             : 
    3770             :         /* Planner should have assigned aggregate to correct level */
    3771             :         Assert(aggref->agglevelsup == 0);
    3772             :         /* ... and the split mode should match */
    3773             :         Assert(aggref->aggsplit == aggstate->aggsplit);
    3774             : 
    3775       54134 :         peragg = &peraggs[aggref->aggno];
    3776             : 
    3777             :         /* Check if we initialized the state for this aggregate already. */
    3778       54134 :         if (peragg->aggref != NULL)
    3779         484 :             continue;
    3780             : 
    3781       53650 :         peragg->aggref = aggref;
    3782       53650 :         peragg->transno = aggref->aggtransno;
    3783             : 
    3784             :         /* Fetch the pg_aggregate row */
    3785       53650 :         aggTuple = SearchSysCache1(AGGFNOID,
    3786             :                                    ObjectIdGetDatum(aggref->aggfnoid));
    3787       53650 :         if (!HeapTupleIsValid(aggTuple))
    3788           0 :             elog(ERROR, "cache lookup failed for aggregate %u",
    3789             :                  aggref->aggfnoid);
    3790       53650 :         aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple);
    3791             : 
    3792             :         /* Check permission to call aggregate function */
    3793       53650 :         aclresult = object_aclcheck(ProcedureRelationId, aggref->aggfnoid, GetUserId(),
    3794             :                                     ACL_EXECUTE);
    3795       53650 :         if (aclresult != ACLCHECK_OK)
    3796           6 :             aclcheck_error(aclresult, OBJECT_AGGREGATE,
    3797           6 :                            get_func_name(aggref->aggfnoid));
    3798       53644 :         InvokeFunctionExecuteHook(aggref->aggfnoid);
    3799             : 
    3800             :         /* planner recorded transition state type in the Aggref itself */
    3801       53644 :         aggtranstype = aggref->aggtranstype;
    3802             :         Assert(OidIsValid(aggtranstype));
    3803             : 
    3804             :         /* Final function only required if we're finalizing the aggregates */
    3805       53644 :         if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
    3806        5388 :             peragg->finalfn_oid = finalfn_oid = InvalidOid;
    3807             :         else
    3808       48256 :             peragg->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
    3809             : 
    3810       53644 :         serialfn_oid = InvalidOid;
    3811       53644 :         deserialfn_oid = InvalidOid;
    3812             : 
    3813             :         /*
    3814             :          * Check if serialization/deserialization is required.  We only do it
    3815             :          * for aggregates that have transtype INTERNAL.
    3816             :          */
    3817       53644 :         if (aggtranstype == INTERNALOID)
    3818             :         {
    3819             :             /*
    3820             :              * The planner should only have generated a serialize agg node if
    3821             :              * every aggregate with an INTERNAL state has a serialization
    3822             :              * function.  Verify that.
    3823             :              */
    3824       22636 :             if (DO_AGGSPLIT_SERIALIZE(aggstate->aggsplit))
    3825             :             {
    3826             :                 /* serialization only valid when not running finalfn */
    3827             :                 Assert(DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit));
    3828             : 
    3829         336 :                 if (!OidIsValid(aggform->aggserialfn))
    3830           0 :                     elog(ERROR, "serialfunc not provided for serialization aggregation");
    3831         336 :                 serialfn_oid = aggform->aggserialfn;
    3832             :             }
    3833             : 
    3834             :             /* Likewise for deserialization functions */
    3835       22636 :             if (DO_AGGSPLIT_DESERIALIZE(aggstate->aggsplit))
    3836             :             {
    3837             :                 /* deserialization only valid when combining states */
    3838             :                 Assert(DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
    3839             : 
    3840         120 :                 if (!OidIsValid(aggform->aggdeserialfn))
    3841           0 :                     elog(ERROR, "deserialfunc not provided for deserialization aggregation");
    3842         120 :                 deserialfn_oid = aggform->aggdeserialfn;
    3843             :             }
    3844             :         }
    3845             : 
    3846             :         /* Check that aggregate owner has permission to call component fns */
    3847             :         {
    3848             :             HeapTuple   procTuple;
    3849             : 
    3850       53644 :             procTuple = SearchSysCache1(PROCOID,
    3851             :                                         ObjectIdGetDatum(aggref->aggfnoid));
    3852       53644 :             if (!HeapTupleIsValid(procTuple))
    3853           0 :                 elog(ERROR, "cache lookup failed for function %u",
    3854             :                      aggref->aggfnoid);
    3855       53644 :             aggOwner = ((Form_pg_proc) GETSTRUCT(procTuple))->proowner;
    3856       53644 :             ReleaseSysCache(procTuple);
    3857             : 
    3858       53644 :             if (OidIsValid(finalfn_oid))
    3859             :             {
    3860       24186 :                 aclresult = object_aclcheck(ProcedureRelationId, finalfn_oid, aggOwner,
    3861             :                                             ACL_EXECUTE);
    3862       24186 :                 if (aclresult != ACLCHECK_OK)
    3863           0 :                     aclcheck_error(aclresult, OBJECT_FUNCTION,
    3864           0 :                                    get_func_name(finalfn_oid));
    3865       24186 :                 InvokeFunctionExecuteHook(finalfn_oid);
    3866             :             }
    3867       53644 :             if (OidIsValid(serialfn_oid))
    3868             :             {
    3869         336 :                 aclresult = object_aclcheck(ProcedureRelationId, serialfn_oid, aggOwner,
    3870             :                                             ACL_EXECUTE);
    3871         336 :                 if (aclresult != ACLCHECK_OK)
    3872           0 :                     aclcheck_error(aclresult, OBJECT_FUNCTION,
    3873           0 :                                    get_func_name(serialfn_oid));
    3874         336 :                 InvokeFunctionExecuteHook(serialfn_oid);
    3875             :             }
    3876       53644 :             if (OidIsValid(deserialfn_oid))
    3877             :             {
    3878         120 :                 aclresult = object_aclcheck(ProcedureRelationId, deserialfn_oid, aggOwner,
    3879             :                                             ACL_EXECUTE);
    3880         120 :                 if (aclresult != ACLCHECK_OK)
    3881           0 :                     aclcheck_error(aclresult, OBJECT_FUNCTION,
    3882           0 :                                    get_func_name(deserialfn_oid));
    3883         120 :                 InvokeFunctionExecuteHook(deserialfn_oid);
    3884             :             }
    3885             :         }
    3886             : 
    3887             :         /*
    3888             :          * Get actual datatypes of the (nominal) aggregate inputs.  These
    3889             :          * could be different from the agg's declared input types, when the
    3890             :          * agg accepts ANY or a polymorphic type.
    3891             :          */
    3892       53644 :         numAggTransFnArgs = get_aggregate_argtypes(aggref,
    3893             :                                                    aggTransFnInputTypes);
    3894             : 
    3895             :         /* Count the "direct" arguments, if any */
    3896       53644 :         numDirectArgs = list_length(aggref->aggdirectargs);
    3897             : 
    3898             :         /* Detect how many arguments to pass to the finalfn */
    3899       53644 :         if (aggform->aggfinalextra)
    3900       16228 :             peragg->numFinalArgs = numAggTransFnArgs + 1;
    3901             :         else
    3902       37416 :             peragg->numFinalArgs = numDirectArgs + 1;
    3903             : 
    3904             :         /* Initialize any direct-argument expressions */
    3905       53644 :         peragg->aggdirectargs = ExecInitExprList(aggref->aggdirectargs,
    3906             :                                                  (PlanState *) aggstate);
    3907             : 
    3908             :         /*
    3909             :          * build expression trees using actual argument & result types for the
    3910             :          * finalfn, if it exists and is required.
    3911             :          */
    3912       53644 :         if (OidIsValid(finalfn_oid))
    3913             :         {
    3914       24186 :             build_aggregate_finalfn_expr(aggTransFnInputTypes,
    3915             :                                          peragg->numFinalArgs,
    3916             :                                          aggtranstype,
    3917             :                                          aggref->aggtype,
    3918             :                                          aggref->inputcollid,
    3919             :                                          finalfn_oid,
    3920             :                                          &finalfnexpr);
    3921       24186 :             fmgr_info(finalfn_oid, &peragg->finalfn);
    3922       24186 :             fmgr_info_set_expr((Node *) finalfnexpr, &peragg->finalfn);
    3923             :         }
    3924             : 
    3925             :         /* get info about the output value's datatype */
    3926       53644 :         get_typlenbyval(aggref->aggtype,
    3927             :                         &peragg->resulttypeLen,
    3928             :                         &peragg->resulttypeByVal);
    3929             : 
    3930             :         /*
    3931             :          * Build working state for invoking the transition function, if we
    3932             :          * haven't done it already.
    3933             :          */
    3934       53644 :         pertrans = &pertransstates[aggref->aggtransno];
    3935       53644 :         if (pertrans->aggref == NULL)
    3936             :         {
    3937             :             Datum       textInitVal;
    3938             :             Datum       initValue;
    3939             :             bool        initValueIsNull;
    3940             :             Oid         transfn_oid;
    3941             : 
    3942             :             /*
    3943             :              * If this aggregation is performing state combines, then instead
    3944             :              * of using the transition function, we'll use the combine
    3945             :              * function.
    3946             :              */
    3947       53386 :             if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
    3948             :             {
    3949        2192 :                 transfn_oid = aggform->aggcombinefn;
    3950             : 
    3951             :                 /* If not set then the planner messed up */
    3952        2192 :                 if (!OidIsValid(transfn_oid))
    3953           0 :                     elog(ERROR, "combinefn not set for aggregate function");
    3954             :             }
    3955             :             else
    3956       51194 :                 transfn_oid = aggform->aggtransfn;
    3957             : 
    3958       53386 :             aclresult = object_aclcheck(ProcedureRelationId, transfn_oid, aggOwner, ACL_EXECUTE);
    3959       53386 :             if (aclresult != ACLCHECK_OK)
    3960           0 :                 aclcheck_error(aclresult, OBJECT_FUNCTION,
    3961           0 :                                get_func_name(transfn_oid));
    3962       53386 :             InvokeFunctionExecuteHook(transfn_oid);
    3963             : 
    3964             :             /*
    3965             :              * initval is potentially null, so don't try to access it as a
    3966             :              * struct field. Must do it the hard way with SysCacheGetAttr.
    3967             :              */
    3968       53386 :             textInitVal = SysCacheGetAttr(AGGFNOID, aggTuple,
    3969             :                                           Anum_pg_aggregate_agginitval,
    3970             :                                           &initValueIsNull);
    3971       53386 :             if (initValueIsNull)
    3972       31834 :                 initValue = (Datum) 0;
    3973             :             else
    3974       21552 :                 initValue = GetAggInitVal(textInitVal, aggtranstype);
    3975             : 
    3976       53386 :             if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
    3977             :             {
    3978        2192 :                 Oid         combineFnInputTypes[] = {aggtranstype,
    3979             :                 aggtranstype};
    3980             : 
    3981             :                 /*
    3982             :                  * When combining there's only one input, the to-be-combined
    3983             :                  * transition value.  The transition value is not counted
    3984             :                  * here.
    3985             :                  */
    3986        2192 :                 pertrans->numTransInputs = 1;
    3987             : 
    3988             :                 /* aggcombinefn always has two arguments of aggtranstype */
    3989        2192 :                 build_pertrans_for_aggref(pertrans, aggstate, estate,
    3990             :                                           aggref, transfn_oid, aggtranstype,
    3991             :                                           serialfn_oid, deserialfn_oid,
    3992             :                                           initValue, initValueIsNull,
    3993             :                                           combineFnInputTypes, 2);
    3994             : 
    3995             :                 /*
    3996             :                  * Ensure that a combine function to combine INTERNAL states
    3997             :                  * is not strict. This should have been checked during CREATE
    3998             :                  * AGGREGATE, but the strict property could have been changed
    3999             :                  * since then.
    4000             :                  */
    4001        2192 :                 if (pertrans->transfn.fn_strict && aggtranstype == INTERNALOID)
    4002           0 :                     ereport(ERROR,
    4003             :                             (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
    4004             :                              errmsg("combine function with transition type %s must not be declared STRICT",
    4005             :                                     format_type_be(aggtranstype))));
    4006             :             }
    4007             :             else
    4008             :             {
    4009             :                 /* Detect how many arguments to pass to the transfn */
    4010       51194 :                 if (AGGKIND_IS_ORDERED_SET(aggref->aggkind))
    4011         252 :                     pertrans->numTransInputs = list_length(aggref->args);
    4012             :                 else
    4013       50942 :                     pertrans->numTransInputs = numAggTransFnArgs;
    4014             : 
    4015       51194 :                 build_pertrans_for_aggref(pertrans, aggstate, estate,
    4016             :                                           aggref, transfn_oid, aggtranstype,
    4017             :                                           serialfn_oid, deserialfn_oid,
    4018             :                                           initValue, initValueIsNull,
    4019             :                                           aggTransFnInputTypes,
    4020             :                                           numAggTransFnArgs);
    4021             : 
    4022             :                 /*
    4023             :                  * If the transfn is strict and the initval is NULL, make sure
    4024             :                  * input type and transtype are the same (or at least
    4025             :                  * binary-compatible), so that it's OK to use the first
    4026             :                  * aggregated input value as the initial transValue.  This
    4027             :                  * should have been checked at agg definition time, but we
    4028             :                  * must check again in case the transfn's strictness property
    4029             :                  * has been changed.
    4030             :                  */
    4031       51194 :                 if (pertrans->transfn.fn_strict && pertrans->initValueIsNull)
    4032             :                 {
    4033        5064 :                     if (numAggTransFnArgs <= numDirectArgs ||
    4034        5064 :                         !IsBinaryCoercible(aggTransFnInputTypes[numDirectArgs],
    4035             :                                            aggtranstype))
    4036           0 :                         ereport(ERROR,
    4037             :                                 (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
    4038             :                                  errmsg("aggregate %u needs to have compatible input type and transition type",
    4039             :                                         aggref->aggfnoid)));
    4040             :                 }
    4041             :             }
    4042             :         }
    4043             :         else
    4044         258 :             pertrans->aggshared = true;
    4045       53644 :         ReleaseSysCache(aggTuple);
    4046             :     }
    4047             : 
    4048             :     /*
    4049             :      * Last, check whether any more aggregates got added onto the node while
    4050             :      * we processed the expressions for the aggregate arguments (including not
    4051             :      * only the regular arguments and FILTER expressions handled immediately
    4052             :      * above, but any direct arguments we might've handled earlier).  If so,
    4053             :      * we have nested aggregate functions, which is semantically nonsensical,
    4054             :      * so complain.  (This should have been caught by the parser, so we don't
    4055             :      * need to work hard on a helpful error message; but we defend against it
    4056             :      * here anyway, just to be sure.)
    4057             :      */
    4058       47720 :     if (numaggrefs != list_length(aggstate->aggs))
    4059           0 :         ereport(ERROR,
    4060             :                 (errcode(ERRCODE_GROUPING_ERROR),
    4061             :                  errmsg("aggregate function calls cannot be nested")));
    4062             : 
    4063             :     /*
    4064             :      * Build expressions doing all the transition work at once. We build a
    4065             :      * different one for each phase, as the number of transition function
    4066             :      * invocation can differ between phases. Note this'll work both for
    4067             :      * transition and combination functions (although there'll only be one
    4068             :      * phase in the latter case).
    4069             :      */
    4070      136922 :     for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
    4071             :     {
    4072       89202 :         AggStatePerPhase phase = &aggstate->phases[phaseidx];
    4073       89202 :         bool        dohash = false;
    4074       89202 :         bool        dosort = false;
    4075             : 
    4076             :         /* phase 0 doesn't necessarily exist */
    4077       89202 :         if (!phase->aggnode)
    4078       41010 :             continue;
    4079             : 
    4080       48192 :         if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 1)
    4081             :         {
    4082             :             /*
    4083             :              * Phase one, and only phase one, in a mixed agg performs both
    4084             :              * sorting and aggregation.
    4085             :              */
    4086         232 :             dohash = true;
    4087         232 :             dosort = true;
    4088             :         }
    4089       47960 :         else if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 0)
    4090             :         {
    4091             :             /*
    4092             :              * No need to compute a transition function for an AGG_MIXED phase
    4093             :              * 0 - the contents of the hashtables will have been computed
    4094             :              * during phase 1.
    4095             :              */
    4096         232 :             continue;
    4097             :         }
    4098       47728 :         else if (phase->aggstrategy == AGG_PLAIN ||
    4099        8884 :                  phase->aggstrategy == AGG_SORTED)
    4100             :         {
    4101       41250 :             dohash = false;
    4102       41250 :             dosort = true;
    4103             :         }
    4104        6478 :         else if (phase->aggstrategy == AGG_HASHED)
    4105             :         {
    4106        6478 :             dohash = true;
    4107        6478 :             dosort = false;
    4108             :         }
    4109             :         else
    4110             :             Assert(false);
    4111             : 
    4112       47960 :         phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash,
    4113             :                                              false);
    4114             : 
    4115             :         /* cache compiled expression for outer slot without NULL check */
    4116       47960 :         phase->evaltrans_cache[0][0] = phase->evaltrans;
    4117             :     }
    4118             : 
    4119       47720 :     return aggstate;
    4120             : }
    4121             : 
    4122             : /*
    4123             :  * Build the state needed to calculate a state value for an aggregate.
    4124             :  *
    4125             :  * This initializes all the fields in 'pertrans'. 'aggref' is the aggregate
    4126             :  * to initialize the state for. 'transfn_oid', 'aggtranstype', and the rest
    4127             :  * of the arguments could be calculated from 'aggref', but the caller has
    4128             :  * calculated them already, so might as well pass them.
    4129             :  *
    4130             :  * 'transfn_oid' may be either the Oid of the aggtransfn or the aggcombinefn.
    4131             :  */
    4132             : static void
    4133       53386 : build_pertrans_for_aggref(AggStatePerTrans pertrans,
    4134             :                           AggState *aggstate, EState *estate,
    4135             :                           Aggref *aggref,
    4136             :                           Oid transfn_oid, Oid aggtranstype,
    4137             :                           Oid aggserialfn, Oid aggdeserialfn,
    4138             :                           Datum initValue, bool initValueIsNull,
    4139             :                           Oid *inputTypes, int numArguments)
    4140             : {
    4141       53386 :     int         numGroupingSets = Max(aggstate->maxsets, 1);
    4142             :     Expr       *transfnexpr;
    4143             :     int         numTransArgs;
    4144       53386 :     Expr       *serialfnexpr = NULL;
    4145       53386 :     Expr       *deserialfnexpr = NULL;
    4146             :     ListCell   *lc;
    4147             :     int         numInputs;
    4148             :     int         numDirectArgs;
    4149             :     List       *sortlist;
    4150             :     int         numSortCols;
    4151             :     int         numDistinctCols;
    4152             :     int         i;
    4153             : 
    4154             :     /* Begin filling in the pertrans data */
    4155       53386 :     pertrans->aggref = aggref;
    4156       53386 :     pertrans->aggshared = false;
    4157       53386 :     pertrans->aggCollation = aggref->inputcollid;
    4158       53386 :     pertrans->transfn_oid = transfn_oid;
    4159       53386 :     pertrans->serialfn_oid = aggserialfn;
    4160       53386 :     pertrans->deserialfn_oid = aggdeserialfn;
    4161       53386 :     pertrans->initValue = initValue;
    4162       53386 :     pertrans->initValueIsNull = initValueIsNull;
    4163             : 
    4164             :     /* Count the "direct" arguments, if any */
    4165       53386 :     numDirectArgs = list_length(aggref->aggdirectargs);
    4166             : 
    4167             :     /* Count the number of aggregated input columns */
    4168       53386 :     pertrans->numInputs = numInputs = list_length(aggref->args);
    4169             : 
    4170       53386 :     pertrans->aggtranstype = aggtranstype;
    4171             : 
    4172             :     /* account for the current transition state */
    4173       53386 :     numTransArgs = pertrans->numTransInputs + 1;
    4174             : 
    4175             :     /*
    4176             :      * Set up infrastructure for calling the transfn.  Note that invtransfn is
    4177             :      * not needed here.
    4178             :      */
    4179       53386 :     build_aggregate_transfn_expr(inputTypes,
    4180             :                                  numArguments,
    4181             :                                  numDirectArgs,
    4182       53386 :                                  aggref->aggvariadic,
    4183             :                                  aggtranstype,
    4184             :                                  aggref->inputcollid,
    4185             :                                  transfn_oid,
    4186             :                                  InvalidOid,
    4187             :                                  &transfnexpr,
    4188             :                                  NULL);
    4189             : 
    4190       53386 :     fmgr_info(transfn_oid, &pertrans->transfn);
    4191       53386 :     fmgr_info_set_expr((Node *) transfnexpr, &pertrans->transfn);
    4192             : 
    4193       53386 :     pertrans->transfn_fcinfo =
    4194       53386 :         (FunctionCallInfo) palloc(SizeForFunctionCallInfo(numTransArgs));
    4195       53386 :     InitFunctionCallInfoData(*pertrans->transfn_fcinfo,
    4196             :                              &pertrans->transfn,
    4197             :                              numTransArgs,
    4198             :                              pertrans->aggCollation,
    4199             :                              (Node *) aggstate, NULL);
    4200             : 
    4201             :     /* get info about the state value's datatype */
    4202       53386 :     get_typlenbyval(aggtranstype,
    4203             :                     &pertrans->transtypeLen,
    4204             :                     &pertrans->transtypeByVal);
    4205             : 
    4206       53386 :     if (OidIsValid(aggserialfn))
    4207             :     {
    4208         336 :         build_aggregate_serialfn_expr(aggserialfn,
    4209             :                                       &serialfnexpr);
    4210         336 :         fmgr_info(aggserialfn, &pertrans->serialfn);
    4211         336 :         fmgr_info_set_expr((Node *) serialfnexpr, &pertrans->serialfn);
    4212             : 
    4213         336 :         pertrans->serialfn_fcinfo =
    4214         336 :             (FunctionCallInfo) palloc(SizeForFunctionCallInfo(1));
    4215         336 :         InitFunctionCallInfoData(*pertrans->serialfn_fcinfo,
    4216             :                                  &pertrans->serialfn,
    4217             :                                  1,
    4218             :                                  InvalidOid,
    4219             :                                  (Node *) aggstate, NULL);
    4220             :     }
    4221             : 
    4222       53386 :     if (OidIsValid(aggdeserialfn))
    4223             :     {
    4224         120 :         build_aggregate_deserialfn_expr(aggdeserialfn,
    4225             :                                         &deserialfnexpr);
    4226         120 :         fmgr_info(aggdeserialfn, &pertrans->deserialfn);
    4227         120 :         fmgr_info_set_expr((Node *) deserialfnexpr, &pertrans->deserialfn);
    4228             : 
    4229         120 :         pertrans->deserialfn_fcinfo =
    4230         120 :             (FunctionCallInfo) palloc(SizeForFunctionCallInfo(2));
    4231         120 :         InitFunctionCallInfoData(*pertrans->deserialfn_fcinfo,
    4232             :                                  &pertrans->deserialfn,
    4233             :                                  2,
    4234             :                                  InvalidOid,
    4235             :                                  (Node *) aggstate, NULL);
    4236             :     }
    4237             : 
    4238             :     /*
    4239             :      * If we're doing either DISTINCT or ORDER BY for a plain agg, then we
    4240             :      * have a list of SortGroupClause nodes; fish out the data in them and
    4241             :      * stick them into arrays.  We ignore ORDER BY for an ordered-set agg,
    4242             :      * however; the agg's transfn and finalfn are responsible for that.
    4243             :      *
    4244             :      * When the planner has set the aggpresorted flag, the input to the
    4245             :      * aggregate is already correctly sorted.  For ORDER BY aggregates we can
    4246             :      * simply treat these as normal aggregates.  For presorted DISTINCT
    4247             :      * aggregates an extra step must be added to remove duplicate consecutive
    4248             :      * inputs.
    4249             :      *
    4250             :      * Note that by construction, if there is a DISTINCT clause then the ORDER
    4251             :      * BY clause is a prefix of it (see transformDistinctClause).
    4252             :      */
    4253       53386 :     if (AGGKIND_IS_ORDERED_SET(aggref->aggkind))
    4254             :     {
    4255         252 :         sortlist = NIL;
    4256         252 :         numSortCols = numDistinctCols = 0;
    4257         252 :         pertrans->aggsortrequired = false;
    4258             :     }
    4259       53134 :     else if (aggref->aggpresorted && aggref->aggdistinct == NIL)
    4260             :     {
    4261        2092 :         sortlist = NIL;
    4262        2092 :         numSortCols = numDistinctCols = 0;
    4263        2092 :         pertrans->aggsortrequired = false;
    4264             :     }
    4265       51042 :     else if (aggref->aggdistinct)
    4266             :     {
    4267         582 :         sortlist = aggref->aggdistinct;
    4268         582 :         numSortCols = numDistinctCols = list_length(sortlist);
    4269             :         Assert(numSortCols >= list_length(aggref->aggorder));
    4270         582 :         pertrans->aggsortrequired = !aggref->aggpresorted;
    4271             :     }
    4272             :     else
    4273             :     {
    4274       50460 :         sortlist = aggref->aggorder;
    4275       50460 :         numSortCols = list_length(sortlist);
    4276       50460 :         numDistinctCols = 0;
    4277       50460 :         pertrans->aggsortrequired = (numSortCols > 0);
    4278             :     }
    4279             : 
    4280       53386 :     pertrans->numSortCols = numSortCols;
    4281       53386 :     pertrans->numDistinctCols = numDistinctCols;
    4282             : 
    4283             :     /*
    4284             :      * If we have either sorting or filtering to do, create a tupledesc and
    4285             :      * slot corresponding to the aggregated inputs (including sort
    4286             :      * expressions) of the agg.
    4287             :      */
    4288       53386 :     if (numSortCols > 0 || aggref->aggfilter)
    4289             :     {
    4290        1430 :         pertrans->sortdesc = ExecTypeFromTL(aggref->args);
    4291        1430 :         pertrans->sortslot =
    4292        1430 :             ExecInitExtraTupleSlot(estate, pertrans->sortdesc,
    4293             :                                    &TTSOpsMinimalTuple);
    4294             :     }
    4295             : 
    4296       53386 :     if (numSortCols > 0)
    4297             :     {
    4298             :         /*
    4299             :          * We don't implement DISTINCT or ORDER BY aggs in the HASHED case
    4300             :          * (yet)
    4301             :          */
    4302             :         Assert(aggstate->aggstrategy != AGG_HASHED && aggstate->aggstrategy != AGG_MIXED);
    4303             : 
    4304             :         /* ORDER BY aggregates are not supported with partial aggregation */
    4305             :         Assert(!DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
    4306             : 
    4307             :         /* If we have only one input, we need its len/byval info. */
    4308         720 :         if (numInputs == 1)
    4309             :         {
    4310         570 :             get_typlenbyval(inputTypes[numDirectArgs],
    4311             :                             &pertrans->inputtypeLen,
    4312             :                             &pertrans->inputtypeByVal);
    4313             :         }
    4314         150 :         else if (numDistinctCols > 0)
    4315             :         {
    4316             :             /* we will need an extra slot to store prior values */
    4317         108 :             pertrans->uniqslot =
    4318         108 :                 ExecInitExtraTupleSlot(estate, pertrans->sortdesc,
    4319             :                                        &TTSOpsMinimalTuple);
    4320             :         }
    4321             : 
    4322             :         /* Extract the sort information for use later */
    4323         720 :         pertrans->sortColIdx =
    4324         720 :             (AttrNumber *) palloc(numSortCols * sizeof(AttrNumber));
    4325         720 :         pertrans->sortOperators =
    4326         720 :             (Oid *) palloc(numSortCols * sizeof(Oid));
    4327         720 :         pertrans->sortCollations =
    4328         720 :             (Oid *) palloc(numSortCols * sizeof(Oid));
    4329         720 :         pertrans->sortNullsFirst =
    4330         720 :             (bool *) palloc(numSortCols * sizeof(bool));
    4331             : 
    4332         720 :         i = 0;
    4333        1638 :         foreach(lc, sortlist)
    4334             :         {
    4335         918 :             SortGroupClause *sortcl = (SortGroupClause *) lfirst(lc);
    4336         918 :             TargetEntry *tle = get_sortgroupclause_tle(sortcl, aggref->args);
    4337             : 
    4338             :             /* the parser should have made sure of this */
    4339             :             Assert(OidIsValid(sortcl->sortop));
    4340             : 
    4341         918 :             pertrans->sortColIdx[i] = tle->resno;
    4342         918 :             pertrans->sortOperators[i] = sortcl->sortop;
    4343         918 :             pertrans->sortCollations[i] = exprCollation((Node *) tle->expr);
    4344         918 :             pertrans->sortNullsFirst[i] = sortcl->nulls_first;
    4345         918 :             i++;
    4346             :         }
    4347             :         Assert(i == numSortCols);
    4348             :     }
    4349             : 
    4350       53386 :     if (aggref->aggdistinct)
    4351             :     {
    4352             :         Oid        *ops;
    4353             : 
    4354             :         Assert(numArguments > 0);
    4355             :         Assert(list_length(aggref->aggdistinct) == numDistinctCols);
    4356             : 
    4357         582 :         ops = palloc(numDistinctCols * sizeof(Oid));
    4358             : 
    4359         582 :         i = 0;
    4360        1344 :         foreach(lc, aggref->aggdistinct)
    4361         762 :             ops[i++] = ((SortGroupClause *) lfirst(lc))->eqop;
    4362             : 
    4363             :         /* lookup / build the necessary comparators */
    4364         582 :         if (numDistinctCols == 1)
    4365         474 :             fmgr_info(get_opcode(ops[0]), &pertrans->equalfnOne);
    4366             :         else
    4367         108 :             pertrans->equalfnMulti =
    4368         108 :                 execTuplesMatchPrepare(pertrans->sortdesc,
    4369             :                                        numDistinctCols,
    4370         108 :                                        pertrans->sortColIdx,
    4371             :                                        ops,
    4372         108 :                                        pertrans->sortCollations,
    4373             :                                        &aggstate->ss.ps);
    4374         582 :         pfree(ops);
    4375             :     }
    4376             : 
    4377       53386 :     pertrans->sortstates = (Tuplesortstate **)
    4378       53386 :         palloc0(sizeof(Tuplesortstate *) * numGroupingSets);
    4379       53386 : }
    4380             : 
    4381             : 
    4382             : static Datum
    4383       21552 : GetAggInitVal(Datum textInitVal, Oid transtype)
    4384             : {
    4385             :     Oid         typinput,
    4386             :                 typioparam;
    4387             :     char       *strInitVal;
    4388             :     Datum       initVal;
    4389             : 
    4390       21552 :     getTypeInputInfo(transtype, &typinput, &typioparam);
    4391       21552 :     strInitVal = TextDatumGetCString(textInitVal);
    4392       21552 :     initVal = OidInputFunctionCall(typinput, strInitVal,
    4393             :                                    typioparam, -1);
    4394       21552 :     pfree(strInitVal);
    4395       21552 :     return initVal;
    4396             : }
    4397             : 
    4398             : void
    4399       47516 : ExecEndAgg(AggState *node)
    4400             : {
    4401             :     PlanState  *outerPlan;
    4402             :     int         transno;
    4403       47516 :     int         numGroupingSets = Max(node->maxsets, 1);
    4404             :     int         setno;
    4405             : 
    4406             :     /*
    4407             :      * When ending a parallel worker, copy the statistics gathered by the
    4408             :      * worker back into shared memory so that it can be picked up by the main
    4409             :      * process to report in EXPLAIN ANALYZE.
    4410             :      */
    4411       47516 :     if (node->shared_info && IsParallelWorker())
    4412             :     {
    4413             :         AggregateInstrumentation *si;
    4414             : 
    4415             :         Assert(ParallelWorkerNumber <= node->shared_info->num_workers);
    4416         168 :         si = &node->shared_info->sinstrument[ParallelWorkerNumber];
    4417         168 :         si->hash_batches_used = node->hash_batches_used;
    4418         168 :         si->hash_disk_used = node->hash_disk_used;
    4419         168 :         si->hash_mem_peak = node->hash_mem_peak;
    4420             :     }
    4421             : 
    4422             :     /* Make sure we have closed any open tuplesorts */
    4423             : 
    4424       47516 :     if (node->sort_in)
    4425         156 :         tuplesort_end(node->sort_in);
    4426       47516 :     if (node->sort_out)
    4427          42 :         tuplesort_end(node->sort_out);
    4428             : 
    4429       47516 :     hashagg_reset_spill_state(node);
    4430             : 
    4431       47516 :     if (node->hash_metacxt != NULL)
    4432             :     {
    4433        6702 :         MemoryContextDelete(node->hash_metacxt);
    4434        6702 :         node->hash_metacxt = NULL;
    4435             :     }
    4436       47516 :     if (node->hash_tablecxt != NULL)
    4437             :     {
    4438        6702 :         MemoryContextDelete(node->hash_tablecxt);
    4439        6702 :         node->hash_tablecxt = NULL;
    4440             :     }
    4441             : 
    4442             : 
    4443      100694 :     for (transno = 0; transno < node->numtrans; transno++)
    4444             :     {
    4445       53178 :         AggStatePerTrans pertrans = &node->pertrans[transno];
    4446             : 
    4447      107370 :         for (setno = 0; setno < numGroupingSets; setno++)
    4448             :         {
    4449       54192 :             if (pertrans->sortstates[setno])
    4450           0 :                 tuplesort_end(pertrans->sortstates[setno]);
    4451             :         }
    4452             :     }
    4453             : 
    4454             :     /* And ensure any agg shutdown callbacks have been called */
    4455       95872 :     for (setno = 0; setno < numGroupingSets; setno++)
    4456       48356 :         ReScanExprContext(node->aggcontexts[setno]);
    4457       47516 :     if (node->hashcontext)
    4458        6702 :         ReScanExprContext(node->hashcontext);
    4459             : 
    4460       47516 :     outerPlan = outerPlanState(node);
    4461       47516 :     ExecEndNode(outerPlan);
    4462       47516 : }
    4463             : 
    4464             : void
    4465       54668 : ExecReScanAgg(AggState *node)
    4466             : {
    4467       54668 :     ExprContext *econtext = node->ss.ps.ps_ExprContext;
    4468       54668 :     PlanState  *outerPlan = outerPlanState(node);
    4469       54668 :     Agg        *aggnode = (Agg *) node->ss.ps.plan;
    4470             :     int         transno;
    4471       54668 :     int         numGroupingSets = Max(node->maxsets, 1);
    4472             :     int         setno;
    4473             : 
    4474       54668 :     node->agg_done = false;
    4475             : 
    4476       54668 :     if (node->aggstrategy == AGG_HASHED)
    4477             :     {
    4478             :         /*
    4479             :          * In the hashed case, if we haven't yet built the hash table then we
    4480             :          * can just return; nothing done yet, so nothing to undo. If subnode's
    4481             :          * chgParam is not NULL then it will be re-scanned by ExecProcNode,
    4482             :          * else no reason to re-scan it at all.
    4483             :          */
    4484       13424 :         if (!node->table_filled)
    4485         150 :             return;
    4486             : 
    4487             :         /*
    4488             :          * If we do have the hash table, and it never spilled, and the subplan
    4489             :          * does not have any parameter changes, and none of our own parameter
    4490             :          * changes affect input expressions of the aggregated functions, then
    4491             :          * we can just rescan the existing hash table; no need to build it
    4492             :          * again.
    4493             :          */
    4494       13274 :         if (outerPlan->chgParam == NULL && !node->hash_ever_spilled &&
    4495         926 :             !bms_overlap(node->ss.ps.chgParam, aggnode->aggParams))
    4496             :         {
    4497         902 :             ResetTupleHashIterator(node->perhash[0].hashtable,
    4498             :                                    &node->perhash[0].hashiter);
    4499         902 :             select_current_set(node, 0, true);
    4500         902 :             return;
    4501             :         }
    4502             :     }
    4503             : 
    4504             :     /* Make sure we have closed any open tuplesorts */
    4505      121968 :     for (transno = 0; transno < node->numtrans; transno++)
    4506             :     {
    4507      136740 :         for (setno = 0; setno < numGroupingSets; setno++)
    4508             :         {
    4509       68388 :             AggStatePerTrans pertrans = &node->pertrans[transno];
    4510             : 
    4511       68388 :             if (pertrans->sortstates[setno])
    4512             :             {
    4513           0 :                 tuplesort_end(pertrans->sortstates[setno]);
    4514           0 :                 pertrans->sortstates[setno] = NULL;
    4515             :             }
    4516             :         }
    4517             :     }
    4518             : 
    4519             :     /*
    4520             :      * We don't need to ReScanExprContext the output tuple context here;
    4521             :      * ExecReScan already did it. But we do need to reset our per-grouping-set
    4522             :      * contexts, which may have transvalues stored in them. (We use rescan
    4523             :      * rather than just reset because transfns may have registered callbacks
    4524             :      * that need to be run now.) For the AGG_HASHED case, see below.
    4525             :      */
    4526             : 
    4527      107268 :     for (setno = 0; setno < numGroupingSets; setno++)
    4528             :     {
    4529       53652 :         ReScanExprContext(node->aggcontexts[setno]);
    4530             :     }
    4531             : 
    4532             :     /* Release first tuple of group, if we have made a copy */
    4533       53616 :     if (node->grp_firstTuple != NULL)
    4534             :     {
    4535           0 :         heap_freetuple(node->grp_firstTuple);
    4536           0 :         node->grp_firstTuple = NULL;
    4537             :     }
    4538       53616 :     ExecClearTuple(node->ss.ss_ScanTupleSlot);
    4539             : 
    4540             :     /* Forget current agg values */
    4541      121968 :     MemSet(econtext->ecxt_aggvalues, 0, sizeof(Datum) * node->numaggs);
    4542       53616 :     MemSet(econtext->ecxt_aggnulls, 0, sizeof(bool) * node->numaggs);
    4543             : 
    4544             :     /*
    4545             :      * With AGG_HASHED/MIXED, the hash table is allocated in a sub-context of
    4546             :      * the hashcontext. This used to be an issue, but now, resetting a context
    4547             :      * automatically deletes sub-contexts too.
    4548             :      */
    4549       53616 :     if (node->aggstrategy == AGG_HASHED || node->aggstrategy == AGG_MIXED)
    4550             :     {
    4551       12402 :         hashagg_reset_spill_state(node);
    4552             : 
    4553       12402 :         node->hash_ever_spilled = false;
    4554       12402 :         node->hash_spill_mode = false;
    4555       12402 :         node->hash_ngroups_current = 0;
    4556             : 
    4557       12402 :         ReScanExprContext(node->hashcontext);
    4558       12402 :         MemoryContextReset(node->hash_tablecxt);
    4559             :         /* Rebuild an empty hash table */
    4560       12402 :         build_hash_tables(node);
    4561       12402 :         node->table_filled = false;
    4562             :         /* iterator will be reset when the table is filled */
    4563             : 
    4564       12402 :         hashagg_recompile_expressions(node, false, false);
    4565             :     }
    4566             : 
    4567       53616 :     if (node->aggstrategy != AGG_HASHED)
    4568             :     {
    4569             :         /*
    4570             :          * Reset the per-group state (in particular, mark transvalues null)
    4571             :          */
    4572       82524 :         for (setno = 0; setno < numGroupingSets; setno++)
    4573             :         {
    4574      177960 :             MemSet(node->pergroups[setno], 0,
    4575             :                    sizeof(AggStatePerGroupData) * node->numaggs);
    4576             :         }
    4577             : 
    4578             :         /* reset to phase 1 */
    4579       41244 :         initialize_phase(node, 1);
    4580             : 
    4581       41244 :         node->input_done = false;
    4582       41244 :         node->projected_set = -1;
    4583             :     }
    4584             : 
    4585       53616 :     if (outerPlan->chgParam == NULL)
    4586         218 :         ExecReScan(outerPlan);
    4587             : }
    4588             : 
    4589             : 
    4590             : /***********************************************************************
    4591             :  * API exposed to aggregate functions
    4592             :  ***********************************************************************/
    4593             : 
    4594             : 
    4595             : /*
    4596             :  * AggCheckCallContext - test if a SQL function is being called as an aggregate
    4597             :  *
    4598             :  * The transition and/or final functions of an aggregate may want to verify
    4599             :  * that they are being called as aggregates, rather than as plain SQL
    4600             :  * functions.  They should use this function to do so.  The return value
    4601             :  * is nonzero if being called as an aggregate, or zero if not.  (Specific
    4602             :  * nonzero values are AGG_CONTEXT_AGGREGATE or AGG_CONTEXT_WINDOW, but more
    4603             :  * values could conceivably appear in future.)
    4604             :  *
    4605             :  * If aggcontext isn't NULL, the function also stores at *aggcontext the
    4606             :  * identity of the memory context that aggregate transition values are being
    4607             :  * stored in.  Note that the same aggregate call site (flinfo) may be called
    4608             :  * interleaved on different transition values in different contexts, so it's
    4609             :  * not kosher to cache aggcontext under fn_extra.  It is, however, kosher to
    4610             :  * cache it in the transvalue itself (for internal-type transvalues).
    4611             :  */
    4612             : int
    4613     5440040 : AggCheckCallContext(FunctionCallInfo fcinfo, MemoryContext *aggcontext)
    4614             : {
    4615     5440040 :     if (fcinfo->context && IsA(fcinfo->context, AggState))
    4616             :     {
    4617     5428654 :         if (aggcontext)
    4618             :         {
    4619     2751824 :             AggState   *aggstate = ((AggState *) fcinfo->context);
    4620     2751824 :             ExprContext *cxt = aggstate->curaggcontext;
    4621             : 
    4622     2751824 :             *aggcontext = cxt->ecxt_per_tuple_memory;
    4623             :         }
    4624     5428654 :         return AGG_CONTEXT_AGGREGATE;
    4625             :     }
    4626       11386 :     if (fcinfo->context && IsA(fcinfo->context, WindowAggState))
    4627             :     {
    4628        9512 :         if (aggcontext)
    4629         710 :             *aggcontext = ((WindowAggState *) fcinfo->context)->curaggcontext;
    4630        9512 :         return AGG_CONTEXT_WINDOW;
    4631             :     }
    4632             : 
    4633             :     /* this is just to prevent "uninitialized variable" warnings */
    4634        1874 :     if (aggcontext)
    4635        1826 :         *aggcontext = NULL;
    4636        1874 :     return 0;
    4637             : }
    4638             : 
    4639             : /*
    4640             :  * AggGetAggref - allow an aggregate support function to get its Aggref
    4641             :  *
    4642             :  * If the function is being called as an aggregate support function,
    4643             :  * return the Aggref node for the aggregate call.  Otherwise, return NULL.
    4644             :  *
    4645             :  * Aggregates sharing the same inputs and transition functions can get
    4646             :  * merged into a single transition calculation.  If the transition function
    4647             :  * calls AggGetAggref, it will get some one of the Aggrefs for which it is
    4648             :  * executing.  It must therefore not pay attention to the Aggref fields that
    4649             :  * relate to the final function, as those are indeterminate.  But if a final
    4650             :  * function calls AggGetAggref, it will get a precise result.
    4651             :  *
    4652             :  * Note that if an aggregate is being used as a window function, this will
    4653             :  * return NULL.  We could provide a similar function to return the relevant
    4654             :  * WindowFunc node in such cases, but it's not needed yet.
    4655             :  */
    4656             : Aggref *
    4657         246 : AggGetAggref(FunctionCallInfo fcinfo)
    4658             : {
    4659         246 :     if (fcinfo->context && IsA(fcinfo->context, AggState))
    4660             :     {
    4661         246 :         AggState   *aggstate = (AggState *) fcinfo->context;
    4662             :         AggStatePerAgg curperagg;
    4663             :         AggStatePerTrans curpertrans;
    4664             : 
    4665             :         /* check curperagg (valid when in a final function) */
    4666         246 :         curperagg = aggstate->curperagg;
    4667             : 
    4668         246 :         if (curperagg)
    4669           0 :             return curperagg->aggref;
    4670             : 
    4671             :         /* check curpertrans (valid when in a transition function) */
    4672         246 :         curpertrans = aggstate->curpertrans;
    4673             : 
    4674         246 :         if (curpertrans)
    4675         246 :             return curpertrans->aggref;
    4676             :     }
    4677           0 :     return NULL;
    4678             : }
    4679             : 
    4680             : /*
    4681             :  * AggGetTempMemoryContext - fetch short-term memory context for aggregates
    4682             :  *
    4683             :  * This is useful in agg final functions; the context returned is one that
    4684             :  * the final function can safely reset as desired.  This isn't useful for
    4685             :  * transition functions, since the context returned MAY (we don't promise)
    4686             :  * be the same as the context those are called in.
    4687             :  *
    4688             :  * As above, this is currently not useful for aggs called as window functions.
    4689             :  */
    4690             : MemoryContext
    4691           0 : AggGetTempMemoryContext(FunctionCallInfo fcinfo)
    4692             : {
    4693           0 :     if (fcinfo->context && IsA(fcinfo->context, AggState))
    4694             :     {
    4695           0 :         AggState   *aggstate = (AggState *) fcinfo->context;
    4696             : 
    4697           0 :         return aggstate->tmpcontext->ecxt_per_tuple_memory;
    4698             :     }
    4699           0 :     return NULL;
    4700             : }
    4701             : 
    4702             : /*
    4703             :  * AggStateIsShared - find out whether transition state is shared
    4704             :  *
    4705             :  * If the function is being called as an aggregate support function,
    4706             :  * return true if the aggregate's transition state is shared across
    4707             :  * multiple aggregates, false if it is not.
    4708             :  *
    4709             :  * Returns true if not called as an aggregate support function.
    4710             :  * This is intended as a conservative answer, ie "no you'd better not
    4711             :  * scribble on your input".  In particular, will return true if the
    4712             :  * aggregate is being used as a window function, which is a scenario
    4713             :  * in which changing the transition state is a bad idea.  We might
    4714             :  * want to refine the behavior for the window case in future.
    4715             :  */
    4716             : bool
    4717         246 : AggStateIsShared(FunctionCallInfo fcinfo)
    4718             : {
    4719         246 :     if (fcinfo->context && IsA(fcinfo->context, AggState))
    4720             :     {
    4721         246 :         AggState   *aggstate = (AggState *) fcinfo->context;
    4722             :         AggStatePerAgg curperagg;
    4723             :         AggStatePerTrans curpertrans;
    4724             : 
    4725             :         /* check curperagg (valid when in a final function) */
    4726         246 :         curperagg = aggstate->curperagg;
    4727             : 
    4728         246 :         if (curperagg)
    4729           0 :             return aggstate->pertrans[curperagg->transno].aggshared;
    4730             : 
    4731             :         /* check curpertrans (valid when in a transition function) */
    4732         246 :         curpertrans = aggstate->curpertrans;
    4733             : 
    4734         246 :         if (curpertrans)
    4735         246 :             return curpertrans->aggshared;
    4736             :     }
    4737           0 :     return true;
    4738             : }
    4739             : 
    4740             : /*
    4741             :  * AggRegisterCallback - register a cleanup callback for an aggregate
    4742             :  *
    4743             :  * This is useful for aggs to register shutdown callbacks, which will ensure
    4744             :  * that non-memory resources are freed.  The callback will occur just before
    4745             :  * the associated aggcontext (as returned by AggCheckCallContext) is reset,
    4746             :  * either between groups or as a result of rescanning the query.  The callback
    4747             :  * will NOT be called on error paths.  The typical use-case is for freeing of
    4748             :  * tuplestores or tuplesorts maintained in aggcontext, or pins held by slots
    4749             :  * created by the agg functions.  (The callback will not be called until after
    4750             :  * the result of the finalfn is no longer needed, so it's safe for the finalfn
    4751             :  * to return data that will be freed by the callback.)
    4752             :  *
    4753             :  * As above, this is currently not useful for aggs called as window functions.
    4754             :  */
    4755             : void
    4756         660 : AggRegisterCallback(FunctionCallInfo fcinfo,
    4757             :                     ExprContextCallbackFunction func,
    4758             :                     Datum arg)
    4759             : {
    4760         660 :     if (fcinfo->context && IsA(fcinfo->context, AggState))
    4761             :     {
    4762         660 :         AggState   *aggstate = (AggState *) fcinfo->context;
    4763         660 :         ExprContext *cxt = aggstate->curaggcontext;
    4764             : 
    4765         660 :         RegisterExprContextCallback(cxt, func, arg);
    4766             : 
    4767         660 :         return;
    4768             :     }
    4769           0 :     elog(ERROR, "aggregate function cannot register a callback in this context");
    4770             : }
    4771             : 
    4772             : 
    4773             : /* ----------------------------------------------------------------
    4774             :  *                      Parallel Query Support
    4775             :  * ----------------------------------------------------------------
    4776             :  */
    4777             : 
    4778             :  /* ----------------------------------------------------------------
    4779             :   *     ExecAggEstimate
    4780             :   *
    4781             :   *     Estimate space required to propagate aggregate statistics.
    4782             :   * ----------------------------------------------------------------
    4783             :   */
    4784             : void
    4785         560 : ExecAggEstimate(AggState *node, ParallelContext *pcxt)
    4786             : {
    4787             :     Size        size;
    4788             : 
    4789             :     /* don't need this if not instrumenting or no workers */
    4790         560 :     if (!node->ss.ps.instrument || pcxt->nworkers == 0)
    4791         458 :         return;
    4792             : 
    4793         102 :     size = mul_size(pcxt->nworkers, sizeof(AggregateInstrumentation));
    4794         102 :     size = add_size(size, offsetof(SharedAggInfo, sinstrument));
    4795         102 :     shm_toc_estimate_chunk(&pcxt->estimator, size);
    4796         102 :     shm_toc_estimate_keys(&pcxt->estimator, 1);
    4797             : }
    4798             : 
    4799             : /* ----------------------------------------------------------------
    4800             :  *      ExecAggInitializeDSM
    4801             :  *
    4802             :  *      Initialize DSM space for aggregate statistics.
    4803             :  * ----------------------------------------------------------------
    4804             :  */
    4805             : void
    4806         560 : ExecAggInitializeDSM(AggState *node, ParallelContext *pcxt)
    4807             : {
    4808             :     Size        size;
    4809             : 
    4810             :     /* don't need this if not instrumenting or no workers */
    4811         560 :     if (!node->ss.ps.instrument || pcxt->nworkers == 0)
    4812         458 :         return;
    4813             : 
    4814         102 :     size = offsetof(SharedAggInfo, sinstrument)
    4815         102 :         + pcxt->nworkers * sizeof(AggregateInstrumentation);
    4816         102 :     node->shared_info = shm_toc_allocate(pcxt->toc, size);
    4817             :     /* ensure any unfilled slots will contain zeroes */
    4818         102 :     memset(node->shared_info, 0, size);
    4819         102 :     node->shared_info->num_workers = pcxt->nworkers;
    4820         102 :     shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id,
    4821         102 :                    node->shared_info);
    4822             : }
    4823             : 
    4824             : /* ----------------------------------------------------------------
    4825             :  *      ExecAggInitializeWorker
    4826             :  *
    4827             :  *      Attach worker to DSM space for aggregate statistics.
    4828             :  * ----------------------------------------------------------------
    4829             :  */
    4830             : void
    4831        1560 : ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt)
    4832             : {
    4833        1560 :     node->shared_info =
    4834        1560 :         shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true);
    4835        1560 : }
    4836             : 
    4837             : /* ----------------------------------------------------------------
    4838             :  *      ExecAggRetrieveInstrumentation
    4839             :  *
    4840             :  *      Transfer aggregate statistics from DSM to private memory.
    4841             :  * ----------------------------------------------------------------
    4842             :  */
    4843             : void
    4844         102 : ExecAggRetrieveInstrumentation(AggState *node)
    4845             : {
    4846             :     Size        size;
    4847             :     SharedAggInfo *si;
    4848             : 
    4849         102 :     if (node->shared_info == NULL)
    4850           0 :         return;
    4851             : 
    4852         102 :     size = offsetof(SharedAggInfo, sinstrument)
    4853         102 :         + node->shared_info->num_workers * sizeof(AggregateInstrumentation);
    4854         102 :     si = palloc(size);
    4855         102 :     memcpy(si, node->shared_info, size);
    4856         102 :     node->shared_info = si;
    4857             : }

Generated by: LCOV version 1.16