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

Generated by: LCOV version 1.14