#include "access/parallel.h"
#include "nodes/execnodes.h"

Include dependency graph for nodeAgg.h:

This graph shows which files directly or indirectly include this file:

Data Structures
struct	AggStatePerTransData

struct	AggStatePerAggData

struct	AggStatePerGroupData

struct	AggStatePerPhaseData

struct	AggStatePerHashData

Macros
#define	FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUE 0

#define	FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUEISNULL 1

#define	FIELDNO_AGGSTATEPERGROUPDATA_NOTRANSVALUE 2

Typedefs
typedef struct AggStatePerTransData	AggStatePerTransData

typedef struct AggStatePerAggData	AggStatePerAggData

typedef struct AggStatePerGroupData	AggStatePerGroupData

typedef struct AggStatePerPhaseData	AggStatePerPhaseData

typedef struct AggStatePerHashData	AggStatePerHashData

Functions
AggState *	ExecInitAgg (Agg node, EState estate, int eflags)

void	ExecEndAgg (AggState *node)

void	ExecReScanAgg (AggState *node)

Size	hash_agg_entry_size (int numTrans, Size tupleWidth, Size transitionSpace)

void	hash_agg_set_limits (double hashentrysize, double input_groups, int used_bits, Size mem_limit, uint64 ngroups_limit, int *num_partitions)

void	ExecAggEstimate (AggState node, ParallelContext pcxt)

void	ExecAggInitializeDSM (AggState node, ParallelContext pcxt)

void	ExecAggInitializeWorker (AggState node, ParallelWorkerContext pwcxt)

void	ExecAggRetrieveInstrumentation (AggState *node)

Macro Definition Documentation

◆ FIELDNO_AGGSTATEPERGROUPDATA_NOTRANSVALUE

#define FIELDNO_AGGSTATEPERGROUPDATA_NOTRANSVALUE 2

Definition at line 249 of file nodeAgg.h.

Referenced by llvm_compile_expr().

◆ FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUE

#define FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUE 0

Definition at line 244 of file nodeAgg.h.

Referenced by llvm_compile_expr().

◆ FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUEISNULL

#define FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUEISNULL 1

Definition at line 246 of file nodeAgg.h.

Referenced by llvm_compile_expr().

Typedef Documentation

◆ AggStatePerAggData

typedef struct AggStatePerAggData AggStatePerAggData

◆ AggStatePerGroupData

typedef struct AggStatePerGroupData AggStatePerGroupData

◆ AggStatePerHashData

typedef struct AggStatePerHashData AggStatePerHashData

◆ AggStatePerPhaseData

typedef struct AggStatePerPhaseData AggStatePerPhaseData

◆ AggStatePerTransData

typedef struct AggStatePerTransData AggStatePerTransData

Function Documentation

◆ ExecAggEstimate()

void ExecAggEstimate	(	AggState *	node,
		ParallelContext *	pcxt
	)

Definition at line 4964 of file nodeAgg.c.

References add_size(), ParallelContext::estimator, PlanState::instrument, mul_size(), ParallelContext::nworkers, offsetof, ScanState::ps, shm_toc_estimate_chunk, shm_toc_estimate_keys, and AggState::ss.

Referenced by ExecParallelEstimate().

 {
     Size        size;
 
     /* don't need this if not instrumenting or no workers */
     if (!node->ss.ps.instrument || pcxt->nworkers == 0)
         return;
 
     size = mul_size(pcxt->nworkers, sizeof(AggregateInstrumentation));
     size = add_size(size, offsetof(SharedAggInfo, sinstrument));
     shm_toc_estimate_chunk(&pcxt->estimator, size);
     shm_toc_estimate_keys(&pcxt->estimator, 1);
 }

◆ ExecAggInitializeDSM()

void ExecAggInitializeDSM	(	AggState *	node,
		ParallelContext *	pcxt
	)

Definition at line 4985 of file nodeAgg.c.

References PlanState::instrument, SharedAggInfo::num_workers, ParallelContext::nworkers, offsetof, PlanState::plan, Plan::plan_node_id, ScanState::ps, AggState::shared_info, shm_toc_allocate(), shm_toc_insert(), AggState::ss, and ParallelContext::toc.

Referenced by ExecParallelInitializeDSM().

 {
     Size        size;
 
     /* don't need this if not instrumenting or no workers */
     if (!node->ss.ps.instrument || pcxt->nworkers == 0)
         return;
 
     size = offsetof(SharedAggInfo, sinstrument)
         + pcxt->nworkers * sizeof(AggregateInstrumentation);
     node->shared_info = shm_toc_allocate(pcxt->toc, size);
     /* ensure any unfilled slots will contain zeroes */
     memset(node->shared_info, 0, size);
     node->shared_info->num_workers = pcxt->nworkers;
     shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id,
                    node->shared_info);
 }

◆ ExecAggInitializeWorker()

void ExecAggInitializeWorker	(	AggState *	node,
		ParallelWorkerContext *	pwcxt
	)

Definition at line 5010 of file nodeAgg.c.

References PlanState::plan, Plan::plan_node_id, ScanState::ps, AggState::shared_info, shm_toc_lookup(), AggState::ss, and ParallelWorkerContext::toc.

Referenced by ExecParallelInitializeWorker().

 {
     node->shared_info =
         shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true);
 }

◆ ExecAggRetrieveInstrumentation()

void ExecAggRetrieveInstrumentation ( AggState * node )

Definition at line 5023 of file nodeAgg.c.

References SharedAggInfo::num_workers, offsetof, palloc(), and AggState::shared_info.

Referenced by ExecParallelRetrieveInstrumentation().

 {
     Size        size;
     SharedAggInfo *si;
 
     if (node->shared_info == NULL)
         return;
 
     size = offsetof(SharedAggInfo, sinstrument)
         + node->shared_info->num_workers * sizeof(AggregateInstrumentation);
     si = palloc(size);
     memcpy(si, node->shared_info, size);
     node->shared_info = si;
 }

◆ ExecEndAgg()

void ExecEndAgg ( AggState * node )

Definition at line 4557 of file nodeAgg.c.

References AggState::aggcontexts, Assert, ExecClearTuple(), ExecEndNode(), ExecFreeExprContext(), AggregateInstrumentation::hash_batches_used, AggState::hash_batches_used, AggregateInstrumentation::hash_disk_used, AggState::hash_disk_used, AggregateInstrumentation::hash_mem_peak, AggState::hash_mem_peak, AggState::hash_metacxt, hashagg_reset_spill_state(), AggState::hashcontext, IsParallelWorker, Max, AggState::maxsets, MemoryContextDelete(), AggState::numtrans, outerPlan, outerPlanState, ParallelWorkerNumber, AggState::pertrans, ScanState::ps, ReScanExprContext(), AggState::shared_info, SharedAggInfo::sinstrument, AggState::sort_in, AggState::sort_out, AggStatePerTransData::sortstates, AggState::ss, ScanState::ss_ScanTupleSlot, and tuplesort_end().

Referenced by ExecEndNode().

 {
     PlanState  *outerPlan;
     int         transno;
     int         numGroupingSets = Max(node->maxsets, 1);
     int         setno;
 
     /*
      * When ending a parallel worker, copy the statistics gathered by the
      * worker back into shared memory so that it can be picked up by the main
      * process to report in EXPLAIN ANALYZE.
      */
     if (node->shared_info && IsParallelWorker())
     {
         AggregateInstrumentation *si;
 
         Assert(ParallelWorkerNumber <= node->shared_info->num_workers);
         si = &node->shared_info->sinstrument[ParallelWorkerNumber];
         si->hash_batches_used = node->hash_batches_used;
         si->hash_disk_used = node->hash_disk_used;
         si->hash_mem_peak = node->hash_mem_peak;
     }
 
     /* Make sure we have closed any open tuplesorts */
 
     if (node->sort_in)
         tuplesort_end(node->sort_in);
     if (node->sort_out)
         tuplesort_end(node->sort_out);
 
     hashagg_reset_spill_state(node);
 
     if (node->hash_metacxt != NULL)
     {
         MemoryContextDelete(node->hash_metacxt);
         node->hash_metacxt = NULL;
     }
 
     for (transno = 0; transno < node->numtrans; transno++)
     {
         AggStatePerTrans pertrans = &node->pertrans[transno];
 
         for (setno = 0; setno < numGroupingSets; setno++)
         {
             if (pertrans->sortstates[setno])
                 tuplesort_end(pertrans->sortstates[setno]);
         }
     }
 
     /* And ensure any agg shutdown callbacks have been called */
     for (setno = 0; setno < numGroupingSets; setno++)
         ReScanExprContext(node->aggcontexts[setno]);
     if (node->hashcontext)
         ReScanExprContext(node->hashcontext);
 
     /*
      * We don't actually free any ExprContexts here (see comment in
      * ExecFreeExprContext), just unlinking the output one from the plan node
      * suffices.
      */
     ExecFreeExprContext(&node->ss.ps);
 
     /* clean up tuple table */
     ExecClearTuple(node->ss.ss_ScanTupleSlot);
 
     outerPlan = outerPlanState(node);
     ExecEndNode(outerPlan);
 }

◆ ExecInitAgg()

AggState* ExecInitAgg	(	Agg *	node,
		EState *	estate,
		int	eflags
	)

Definition at line 3234 of file nodeAgg.c.

Referenced by ExecInitNode().

 {
     AggState   *aggstate;
     AggStatePerAgg peraggs;
     AggStatePerTrans pertransstates;
     AggStatePerGroup *pergroups;
     Plan       *outerPlan;
     ExprContext *econtext;
     TupleDesc   scanDesc;
     int         numaggs,
                 transno,
                 aggno;
     int         phase;
     int         phaseidx;
     ListCell   *l;
     Bitmapset  *all_grouped_cols = NULL;
     int         numGroupingSets = 1;
     int         numPhases;
     int         numHashes;
     int         i = 0;
     int         j = 0;
     bool        use_hashing = (node->aggstrategy == AGG_HASHED ||
                                node->aggstrategy == AGG_MIXED);
 
     /* check for unsupported flags */
     Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
 
     /*
      * create state structure
      */
     aggstate = makeNode(AggState);
     aggstate->ss.ps.plan = (Plan *) node;
     aggstate->ss.ps.state = estate;
     aggstate->ss.ps.ExecProcNode = ExecAgg;
 
     aggstate->aggs = NIL;
     aggstate->numaggs = 0;
     aggstate->numtrans = 0;
     aggstate->aggstrategy = node->aggstrategy;
     aggstate->aggsplit = node->aggsplit;
     aggstate->maxsets = 0;
     aggstate->projected_set = -1;
     aggstate->current_set = 0;
     aggstate->peragg = NULL;
     aggstate->pertrans = NULL;
     aggstate->curperagg = NULL;
     aggstate->curpertrans = NULL;
     aggstate->input_done = false;
     aggstate->agg_done = false;
     aggstate->pergroups = NULL;
     aggstate->grp_firstTuple = NULL;
     aggstate->sort_in = NULL;
     aggstate->sort_out = NULL;
 
     /*
      * phases[0] always exists, but is dummy in sorted/plain mode
      */
     numPhases = (use_hashing ? 1 : 2);
     numHashes = (use_hashing ? 1 : 0);
 
     /*
      * Calculate the maximum number of grouping sets in any phase; this
      * determines the size of some allocations.  Also calculate the number of
      * phases, since all hashed/mixed nodes contribute to only a single phase.
      */
     if (node->groupingSets)
     {
         numGroupingSets = list_length(node->groupingSets);
 
         foreach(l, node->chain)
         {
             Agg        *agg = lfirst(l);
 
             numGroupingSets = Max(numGroupingSets,
                                   list_length(agg->groupingSets));
 
             /*
              * additional AGG_HASHED aggs become part of phase 0, but all
              * others add an extra phase.
              */
             if (agg->aggstrategy != AGG_HASHED)
                 ++numPhases;
             else
                 ++numHashes;
         }
     }
 
     aggstate->maxsets = numGroupingSets;
     aggstate->numphases = numPhases;
 
     aggstate->aggcontexts = (ExprContext **)
         palloc0(sizeof(ExprContext *) * numGroupingSets);
 
     /*
      * Create expression contexts.  We need three or more, one for
      * per-input-tuple processing, one for per-output-tuple processing, one
      * for all the hashtables, and one for each grouping set.  The per-tuple
      * memory context of the per-grouping-set ExprContexts (aggcontexts)
      * replaces the standalone memory context formerly used to hold transition
      * values.  We cheat a little by using ExecAssignExprContext() to build
      * all of them.
      *
      * NOTE: the details of what is stored in aggcontexts and what is stored
      * in the regular per-query memory context are driven by a simple
      * decision: we want to reset the aggcontext at group boundaries (if not
      * hashing) and in ExecReScanAgg to recover no-longer-wanted space.
      */
     ExecAssignExprContext(estate, &aggstate->ss.ps);
     aggstate->tmpcontext = aggstate->ss.ps.ps_ExprContext;
 
     for (i = 0; i < numGroupingSets; ++i)
     {
         ExecAssignExprContext(estate, &aggstate->ss.ps);
         aggstate->aggcontexts[i] = aggstate->ss.ps.ps_ExprContext;
     }
 
     if (use_hashing)
         aggstate->hashcontext = CreateWorkExprContext(estate);
 
     ExecAssignExprContext(estate, &aggstate->ss.ps);
 
     /*
      * Initialize child nodes.
      *
      * If we are doing a hashed aggregation then the child plan does not need
      * to handle REWIND efficiently; see ExecReScanAgg.
      */
     if (node->aggstrategy == AGG_HASHED)
         eflags &= ~EXEC_FLAG_REWIND;
     outerPlan = outerPlan(node);
     outerPlanState(aggstate) = ExecInitNode(outerPlan, estate, eflags);
 
     /*
      * initialize source tuple type.
      */
     aggstate->ss.ps.outerops =
         ExecGetResultSlotOps(outerPlanState(&aggstate->ss),
                              &aggstate->ss.ps.outeropsfixed);
     aggstate->ss.ps.outeropsset = true;
 
     ExecCreateScanSlotFromOuterPlan(estate, &aggstate->ss,
                                     aggstate->ss.ps.outerops);
     scanDesc = aggstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
 
     /*
      * If there are more than two phases (including a potential dummy phase
      * 0), input will be resorted using tuplesort. Need a slot for that.
      */
     if (numPhases > 2)
     {
         aggstate->sort_slot = ExecInitExtraTupleSlot(estate, scanDesc,
                                                      &TTSOpsMinimalTuple);
 
         /*
          * The output of the tuplesort, and the output from the outer child
          * might not use the same type of slot. In most cases the child will
          * be a Sort, and thus return a TTSOpsMinimalTuple type slot - but the
          * input can also be presorted due an index, in which case it could be
          * a different type of slot.
          *
          * XXX: For efficiency it would be good to instead/additionally
          * generate expressions with corresponding settings of outerops* for
          * the individual phases - deforming is often a bottleneck for
          * aggregations with lots of rows per group. If there's multiple
          * sorts, we know that all but the first use TTSOpsMinimalTuple (via
          * the nodeAgg.c internal tuplesort).
          */
         if (aggstate->ss.ps.outeropsfixed &&
             aggstate->ss.ps.outerops != &TTSOpsMinimalTuple)
             aggstate->ss.ps.outeropsfixed = false;
     }
 
     /*
      * Initialize result type, slot and projection.
      */
     ExecInitResultTupleSlotTL(&aggstate->ss.ps, &TTSOpsVirtual);
     ExecAssignProjectionInfo(&aggstate->ss.ps, NULL);
 
     /*
      * initialize child expressions
      *
      * We expect the parser to have checked that no aggs contain other agg
      * calls in their arguments (and just to be sure, we verify it again while
      * initializing the plan node).  This would make no sense under SQL
      * semantics, and it's forbidden by the spec.  Because it is true, we
      * don't need to worry about evaluating the aggs in any particular order.
      *
      * Note: execExpr.c finds Aggrefs for us, and adds their AggrefExprState
      * nodes to aggstate->aggs.  Aggrefs in the qual are found here; Aggrefs
      * in the targetlist are found during ExecAssignProjectionInfo, below.
      */
     aggstate->ss.ps.qual =
         ExecInitQual(node->plan.qual, (PlanState *) aggstate);
 
     /*
      * We should now have found all Aggrefs in the targetlist and quals.
      */
     numaggs = aggstate->numaggs;
     Assert(numaggs == list_length(aggstate->aggs));
 
     /*
      * For each phase, prepare grouping set data and fmgr lookup data for
      * compare functions.  Accumulate all_grouped_cols in passing.
      */
     aggstate->phases = palloc0(numPhases * sizeof(AggStatePerPhaseData));
 
     aggstate->num_hashes = numHashes;
     if (numHashes)
     {
         aggstate->perhash = palloc0(sizeof(AggStatePerHashData) * numHashes);
         aggstate->phases[0].numsets = 0;
         aggstate->phases[0].gset_lengths = palloc(numHashes * sizeof(int));
         aggstate->phases[0].grouped_cols = palloc(numHashes * sizeof(Bitmapset *));
     }
 
     phase = 0;
     for (phaseidx = 0; phaseidx <= list_length(node->chain); ++phaseidx)
     {
         Agg        *aggnode;
         Sort       *sortnode;
 
         if (phaseidx > 0)
         {
             aggnode = list_nth_node(Agg, node->chain, phaseidx - 1);
             sortnode = castNode(Sort, aggnode->plan.lefttree);
         }
         else
         {
             aggnode = node;
             sortnode = NULL;
         }
 
         Assert(phase <= 1 || sortnode);
 
         if (aggnode->aggstrategy == AGG_HASHED
             || aggnode->aggstrategy == AGG_MIXED)
         {
             AggStatePerPhase phasedata = &aggstate->phases[0];
             AggStatePerHash perhash;
             Bitmapset  *cols = NULL;
 
             Assert(phase == 0);
             i = phasedata->numsets++;
             perhash = &aggstate->perhash[i];
 
             /* phase 0 always points to the "real" Agg in the hash case */
             phasedata->aggnode = node;
             phasedata->aggstrategy = node->aggstrategy;
 
             /* but the actual Agg node representing this hash is saved here */
             perhash->aggnode = aggnode;
 
             phasedata->gset_lengths[i] = perhash->numCols = aggnode->numCols;
 
             for (j = 0; j < aggnode->numCols; ++j)
                 cols = bms_add_member(cols, aggnode->grpColIdx[j]);
 
             phasedata->grouped_cols[i] = cols;
 
             all_grouped_cols = bms_add_members(all_grouped_cols, cols);
             continue;
         }
         else
         {
             AggStatePerPhase phasedata = &aggstate->phases[++phase];
             int         num_sets;
 
             phasedata->numsets = num_sets = list_length(aggnode->groupingSets);
 
             if (num_sets)
             {
                 phasedata->gset_lengths = palloc(num_sets * sizeof(int));
                 phasedata->grouped_cols = palloc(num_sets * sizeof(Bitmapset *));
 
                 i = 0;
                 foreach(l, aggnode->groupingSets)
                 {
                     int         current_length = list_length(lfirst(l));
                     Bitmapset  *cols = NULL;
 
                     /* planner forces this to be correct */
                     for (j = 0; j < current_length; ++j)
                         cols = bms_add_member(cols, aggnode->grpColIdx[j]);
 
                     phasedata->grouped_cols[i] = cols;
                     phasedata->gset_lengths[i] = current_length;
 
                     ++i;
                 }
 
                 all_grouped_cols = bms_add_members(all_grouped_cols,
                                                    phasedata->grouped_cols[0]);
             }
             else
             {
                 Assert(phaseidx == 0);
 
                 phasedata->gset_lengths = NULL;
                 phasedata->grouped_cols = NULL;
             }
 
             /*
              * If we are grouping, precompute fmgr lookup data for inner loop.
              */
             if (aggnode->aggstrategy == AGG_SORTED)
             {
                 int         i = 0;
 
                 Assert(aggnode->numCols > 0);
 
                 /*
                  * Build a separate function for each subset of columns that
                  * need to be compared.
                  */
                 phasedata->eqfunctions =
                     (ExprState **) palloc0(aggnode->numCols * sizeof(ExprState *));
 
                 /* for each grouping set */
                 for (i = 0; i < phasedata->numsets; i++)
                 {
                     int         length = phasedata->gset_lengths[i];
 
                     if (phasedata->eqfunctions[length - 1] != NULL)
                         continue;
 
                     phasedata->eqfunctions[length - 1] =
                         execTuplesMatchPrepare(scanDesc,
                                                length,
                                                aggnode->grpColIdx,
                                                aggnode->grpOperators,
                                                aggnode->grpCollations,
                                                (PlanState *) aggstate);
                 }
 
                 /* and for all grouped columns, unless already computed */
                 if (phasedata->eqfunctions[aggnode->numCols - 1] == NULL)
                 {
                     phasedata->eqfunctions[aggnode->numCols - 1] =
                         execTuplesMatchPrepare(scanDesc,
                                                aggnode->numCols,
                                                aggnode->grpColIdx,
                                                aggnode->grpOperators,
                                                aggnode->grpCollations,
                                                (PlanState *) aggstate);
                 }
             }
 
             phasedata->aggnode = aggnode;
             phasedata->aggstrategy = aggnode->aggstrategy;
             phasedata->sortnode = sortnode;
         }
     }
 
     /*
      * Convert all_grouped_cols to a descending-order list.
      */
     i = -1;
     while ((i = bms_next_member(all_grouped_cols, i)) >= 0)
         aggstate->all_grouped_cols = lcons_int(i, aggstate->all_grouped_cols);
 
     /*
      * Set up aggregate-result storage in the output expr context, and also
      * allocate my private per-agg working storage
      */
     econtext = aggstate->ss.ps.ps_ExprContext;
     econtext->ecxt_aggvalues = (Datum *) palloc0(sizeof(Datum) * numaggs);
     econtext->ecxt_aggnulls = (bool *) palloc0(sizeof(bool) * numaggs);
 
     peraggs = (AggStatePerAgg) palloc0(sizeof(AggStatePerAggData) * numaggs);
     pertransstates = (AggStatePerTrans) palloc0(sizeof(AggStatePerTransData) * numaggs);
 
     aggstate->peragg = peraggs;
     aggstate->pertrans = pertransstates;
 
 
     aggstate->all_pergroups =
         (AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup)
                                      * (numGroupingSets + numHashes));
     pergroups = aggstate->all_pergroups;
 
     if (node->aggstrategy != AGG_HASHED)
     {
         for (i = 0; i < numGroupingSets; i++)
         {
             pergroups[i] = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData)
                                                       * numaggs);
         }
 
         aggstate->pergroups = pergroups;
         pergroups += numGroupingSets;
     }
 
     /*
      * Hashing can only appear in the initial phase.
      */
     if (use_hashing)
     {
         Plan       *outerplan = outerPlan(node);
         uint64      totalGroups = 0;
         int         i;
 
         aggstate->hash_metacxt = AllocSetContextCreate(aggstate->ss.ps.state->es_query_cxt,
                                                        "HashAgg meta context",
                                                        ALLOCSET_DEFAULT_SIZES);
         aggstate->hash_spill_rslot = ExecInitExtraTupleSlot(estate, scanDesc,
                                                             &TTSOpsMinimalTuple);
         aggstate->hash_spill_wslot = ExecInitExtraTupleSlot(estate, scanDesc,
                                                             &TTSOpsVirtual);
 
         /* this is an array of pointers, not structures */
         aggstate->hash_pergroup = pergroups;
 
         aggstate->hashentrysize = hash_agg_entry_size(aggstate->numtrans,
                                                       outerplan->plan_width,
                                                       node->transitionSpace);
 
         /*
          * Consider all of the grouping sets together when setting the limits
          * and estimating the number of partitions. This can be inaccurate
          * when there is more than one grouping set, but should still be
          * reasonable.
          */
         for (i = 0; i < aggstate->num_hashes; i++)
             totalGroups += aggstate->perhash[i].aggnode->numGroups;
 
         hash_agg_set_limits(aggstate->hashentrysize, totalGroups, 0,
                             &aggstate->hash_mem_limit,
                             &aggstate->hash_ngroups_limit,
                             &aggstate->hash_planned_partitions);
         find_hash_columns(aggstate);
         build_hash_tables(aggstate);
         aggstate->table_filled = false;
 
         /* Initialize this to 1, meaning nothing spilled, yet */
         aggstate->hash_batches_used = 1;
     }
 
     /*
      * Initialize current phase-dependent values to initial phase. The initial
      * phase is 1 (first sort pass) for all strategies that use sorting (if
      * hashing is being done too, then phase 0 is processed last); but if only
      * hashing is being done, then phase 0 is all there is.
      */
     if (node->aggstrategy == AGG_HASHED)
     {
         aggstate->current_phase = 0;
         initialize_phase(aggstate, 0);
         select_current_set(aggstate, 0, true);
     }
     else
     {
         aggstate->current_phase = 1;
         initialize_phase(aggstate, 1);
         select_current_set(aggstate, 0, false);
     }
 
     /* -----------------
      * Perform lookups of aggregate function info, and initialize the
      * unchanging fields of the per-agg and per-trans data.
      *
      * We try to optimize by detecting duplicate aggregate functions so that
      * their state and final values are re-used, rather than needlessly being
      * re-calculated independently. We also detect aggregates that are not
      * the same, but which can share the same transition state.
      *
      * Scenarios:
      *
      * 1. Identical aggregate function calls appear in the query:
      *
      *    SELECT SUM(x) FROM ... HAVING SUM(x) > 0
      *
      *    Since these aggregates are identical, we only need to calculate
      *    the value once. Both aggregates will share the same 'aggno' value.
      *
      * 2. Two different aggregate functions appear in the query, but the
      *    aggregates have the same arguments, transition functions and
      *    initial values (and, presumably, different final functions):
      *
      *    SELECT AVG(x), STDDEV(x) FROM ...
      *
      *    In this case we must create a new peragg for the varying aggregate,
      *    and we need to call the final functions separately, but we need
      *    only run the transition function once.  (This requires that the
      *    final functions be nondestructive of the transition state, but
      *    that's required anyway for other reasons.)
      *
      * For either of these optimizations to be valid, all aggregate properties
      * used in the transition phase must be the same, including any modifiers
      * such as ORDER BY, DISTINCT and FILTER, and the arguments mustn't
      * contain any volatile functions.
      * -----------------
      */
     aggno = -1;
     transno = -1;
     foreach(l, aggstate->aggs)
     {
         AggrefExprState *aggrefstate = (AggrefExprState *) lfirst(l);
         Aggref     *aggref = aggrefstate->aggref;
         AggStatePerAgg peragg;
         AggStatePerTrans pertrans;
         int         existing_aggno;
         int         existing_transno;
         List       *same_input_transnos;
         Oid         inputTypes[FUNC_MAX_ARGS];
         int         numArguments;
         int         numDirectArgs;
         HeapTuple   aggTuple;
         Form_pg_aggregate aggform;
         AclResult   aclresult;
         Oid         transfn_oid,
                     finalfn_oid;
         bool        shareable;
         Oid         serialfn_oid,
                     deserialfn_oid;
         Expr       *finalfnexpr;
         Oid         aggtranstype;
         Datum       textInitVal;
         Datum       initValue;
         bool        initValueIsNull;
 
         /* Planner should have assigned aggregate to correct level */
         Assert(aggref->agglevelsup == 0);
         /* ... and the split mode should match */
         Assert(aggref->aggsplit == aggstate->aggsplit);
 
         /* 1. Check for already processed aggs which can be re-used */
         existing_aggno = find_compatible_peragg(aggref, aggstate, aggno,
                                                 &same_input_transnos);
         if (existing_aggno != -1)
         {
             /*
              * Existing compatible agg found. so just point the Aggref to the
              * same per-agg struct.
              */
             aggrefstate->aggno = existing_aggno;
             continue;
         }
 
         /* Mark Aggref state node with assigned index in the result array */
         peragg = &peraggs[++aggno];
         peragg->aggref = aggref;
         aggrefstate->aggno = aggno;
 
         /* Fetch the pg_aggregate row */
         aggTuple = SearchSysCache1(AGGFNOID,
                                    ObjectIdGetDatum(aggref->aggfnoid));
         if (!HeapTupleIsValid(aggTuple))
             elog(ERROR, "cache lookup failed for aggregate %u",
                  aggref->aggfnoid);
         aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple);
 
         /* Check permission to call aggregate function */
         aclresult = pg_proc_aclcheck(aggref->aggfnoid, GetUserId(),
                                      ACL_EXECUTE);
         if (aclresult != ACLCHECK_OK)
             aclcheck_error(aclresult, OBJECT_AGGREGATE,
                            get_func_name(aggref->aggfnoid));
         InvokeFunctionExecuteHook(aggref->aggfnoid);
 
         /* planner recorded transition state type in the Aggref itself */
         aggtranstype = aggref->aggtranstype;
         Assert(OidIsValid(aggtranstype));
 
         /*
          * If this aggregation is performing state combines, then instead of
          * using the transition function, we'll use the combine function
          */
         if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
         {
             transfn_oid = aggform->aggcombinefn;
 
             /* If not set then the planner messed up */
             if (!OidIsValid(transfn_oid))
                 elog(ERROR, "combinefn not set for aggregate function");
         }
         else
             transfn_oid = aggform->aggtransfn;
 
         /* Final function only required if we're finalizing the aggregates */
         if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
             peragg->finalfn_oid = finalfn_oid = InvalidOid;
         else
             peragg->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
 
         /*
          * If finalfn is marked read-write, we can't share transition states;
          * but it is okay to share states for AGGMODIFY_SHAREABLE aggs.  Also,
          * if we're not executing the finalfn here, we can share regardless.
          */
         shareable = (aggform->aggfinalmodify != AGGMODIFY_READ_WRITE) ||
             (finalfn_oid == InvalidOid);
         peragg->shareable = shareable;
 
         serialfn_oid = InvalidOid;
         deserialfn_oid = InvalidOid;
 
         /*
          * Check if serialization/deserialization is required.  We only do it
          * for aggregates that have transtype INTERNAL.
          */
         if (aggtranstype == INTERNALOID)
         {
             /*
              * The planner should only have generated a serialize agg node if
              * every aggregate with an INTERNAL state has a serialization
              * function.  Verify that.
              */
             if (DO_AGGSPLIT_SERIALIZE(aggstate->aggsplit))
             {
                 /* serialization only valid when not running finalfn */
                 Assert(DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit));
 
                 if (!OidIsValid(aggform->aggserialfn))
                     elog(ERROR, "serialfunc not provided for serialization aggregation");
                 serialfn_oid = aggform->aggserialfn;
             }
 
             /* Likewise for deserialization functions */
             if (DO_AGGSPLIT_DESERIALIZE(aggstate->aggsplit))
             {
                 /* deserialization only valid when combining states */
                 Assert(DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
 
                 if (!OidIsValid(aggform->aggdeserialfn))
                     elog(ERROR, "deserialfunc not provided for deserialization aggregation");
                 deserialfn_oid = aggform->aggdeserialfn;
             }
         }
 
         /* Check that aggregate owner has permission to call component fns */
         {
             HeapTuple   procTuple;
             Oid         aggOwner;
 
             procTuple = SearchSysCache1(PROCOID,
                                         ObjectIdGetDatum(aggref->aggfnoid));
             if (!HeapTupleIsValid(procTuple))
                 elog(ERROR, "cache lookup failed for function %u",
                      aggref->aggfnoid);
             aggOwner = ((Form_pg_proc) GETSTRUCT(procTuple))->proowner;
             ReleaseSysCache(procTuple);
 
             aclresult = pg_proc_aclcheck(transfn_oid, aggOwner,
                                          ACL_EXECUTE);
             if (aclresult != ACLCHECK_OK)
                 aclcheck_error(aclresult, OBJECT_FUNCTION,
                                get_func_name(transfn_oid));
             InvokeFunctionExecuteHook(transfn_oid);
             if (OidIsValid(finalfn_oid))
             {
                 aclresult = pg_proc_aclcheck(finalfn_oid, aggOwner,
                                              ACL_EXECUTE);
                 if (aclresult != ACLCHECK_OK)
                     aclcheck_error(aclresult, OBJECT_FUNCTION,
                                    get_func_name(finalfn_oid));
                 InvokeFunctionExecuteHook(finalfn_oid);
             }
             if (OidIsValid(serialfn_oid))
             {
                 aclresult = pg_proc_aclcheck(serialfn_oid, aggOwner,
                                              ACL_EXECUTE);
                 if (aclresult != ACLCHECK_OK)
                     aclcheck_error(aclresult, OBJECT_FUNCTION,
                                    get_func_name(serialfn_oid));
                 InvokeFunctionExecuteHook(serialfn_oid);
             }
             if (OidIsValid(deserialfn_oid))
             {
                 aclresult = pg_proc_aclcheck(deserialfn_oid, aggOwner,
                                              ACL_EXECUTE);
                 if (aclresult != ACLCHECK_OK)
                     aclcheck_error(aclresult, OBJECT_FUNCTION,
                                    get_func_name(deserialfn_oid));
                 InvokeFunctionExecuteHook(deserialfn_oid);
             }
         }
 
         /*
          * Get actual datatypes of the (nominal) aggregate inputs.  These
          * could be different from the agg's declared input types, when the
          * agg accepts ANY or a polymorphic type.
          */
         numArguments = get_aggregate_argtypes(aggref, inputTypes);
 
         /* Count the "direct" arguments, if any */
         numDirectArgs = list_length(aggref->aggdirectargs);
 
         /* Detect how many arguments to pass to the finalfn */
         if (aggform->aggfinalextra)
             peragg->numFinalArgs = numArguments + 1;
         else
             peragg->numFinalArgs = numDirectArgs + 1;
 
         /* Initialize any direct-argument expressions */
         peragg->aggdirectargs = ExecInitExprList(aggref->aggdirectargs,
                                                  (PlanState *) aggstate);
 
         /*
          * build expression trees using actual argument & result types for the
          * finalfn, if it exists and is required.
          */
         if (OidIsValid(finalfn_oid))
         {
             build_aggregate_finalfn_expr(inputTypes,
                                          peragg->numFinalArgs,
                                          aggtranstype,
                                          aggref->aggtype,
                                          aggref->inputcollid,
                                          finalfn_oid,
                                          &finalfnexpr);
             fmgr_info(finalfn_oid, &peragg->finalfn);
             fmgr_info_set_expr((Node *) finalfnexpr, &peragg->finalfn);
         }
 
         /* get info about the output value's datatype */
         get_typlenbyval(aggref->aggtype,
                         &peragg->resulttypeLen,
                         &peragg->resulttypeByVal);
 
         /*
          * initval is potentially null, so don't try to access it as a struct
          * field. Must do it the hard way with SysCacheGetAttr.
          */
         textInitVal = SysCacheGetAttr(AGGFNOID, aggTuple,
                                       Anum_pg_aggregate_agginitval,
                                       &initValueIsNull);
         if (initValueIsNull)
             initValue = (Datum) 0;
         else
             initValue = GetAggInitVal(textInitVal, aggtranstype);
 
         /*
          * 2. Build working state for invoking the transition function, or
          * look up previously initialized working state, if we can share it.
          *
          * find_compatible_peragg() already collected a list of shareable
          * per-Trans's with the same inputs. Check if any of them have the
          * same transition function and initial value.
          */
         existing_transno = find_compatible_pertrans(aggstate, aggref,
                                                     shareable,
                                                     transfn_oid, aggtranstype,
                                                     serialfn_oid, deserialfn_oid,
                                                     initValue, initValueIsNull,
                                                     same_input_transnos);
         if (existing_transno != -1)
         {
             /*
              * Existing compatible trans found, so just point the 'peragg' to
              * the same per-trans struct, and mark the trans state as shared.
              */
             pertrans = &pertransstates[existing_transno];
             pertrans->aggshared = true;
             peragg->transno = existing_transno;
         }
         else
         {
             pertrans = &pertransstates[++transno];
             build_pertrans_for_aggref(pertrans, aggstate, estate,
                                       aggref, transfn_oid, aggtranstype,
                                       serialfn_oid, deserialfn_oid,
                                       initValue, initValueIsNull,
                                       inputTypes, numArguments);
             peragg->transno = transno;
         }
         ReleaseSysCache(aggTuple);
     }
 
     /*
      * Update aggstate->numaggs to be the number of unique aggregates found.
      * Also set numstates to the number of unique transition states found.
      */
     aggstate->numaggs = aggno + 1;
     aggstate->numtrans = transno + 1;
 
     /*
      * Last, check whether any more aggregates got added onto the node while
      * we processed the expressions for the aggregate arguments (including not
      * only the regular arguments and FILTER expressions handled immediately
      * above, but any direct arguments we might've handled earlier).  If so,
      * we have nested aggregate functions, which is semantically nonsensical,
      * so complain.  (This should have been caught by the parser, so we don't
      * need to work hard on a helpful error message; but we defend against it
      * here anyway, just to be sure.)
      */
     if (numaggs != list_length(aggstate->aggs))
         ereport(ERROR,
                 (errcode(ERRCODE_GROUPING_ERROR),
                  errmsg("aggregate function calls cannot be nested")));
 
     /*
      * Build expressions doing all the transition work at once. We build a
      * different one for each phase, as the number of transition function
      * invocation can differ between phases. Note this'll work both for
      * transition and combination functions (although there'll only be one
      * phase in the latter case).
      */
     for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
     {
         AggStatePerPhase phase = &aggstate->phases[phaseidx];
         bool        dohash = false;
         bool        dosort = false;
 
         /* phase 0 doesn't necessarily exist */
         if (!phase->aggnode)
             continue;
 
         if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 1)
         {
             /*
              * Phase one, and only phase one, in a mixed agg performs both
              * sorting and aggregation.
              */
             dohash = true;
             dosort = true;
         }
         else if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 0)
         {
             /*
              * No need to compute a transition function for an AGG_MIXED phase
              * 0 - the contents of the hashtables will have been computed
              * during phase 1.
              */
             continue;
         }
         else if (phase->aggstrategy == AGG_PLAIN ||
                  phase->aggstrategy == AGG_SORTED)
         {
             dohash = false;
             dosort = true;
         }
         else if (phase->aggstrategy == AGG_HASHED)
         {
             dohash = true;
             dosort = false;
         }
         else
             Assert(false);
 
         phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash,
                                              false);
 
         /* cache compiled expression for outer slot without NULL check */
         phase->evaltrans_cache[0][0] = phase->evaltrans;
     }
 
     return aggstate;
 }

◆ ExecReScanAgg()

void ExecReScanAgg ( AggState * node )

Definition at line 4627 of file nodeAgg.c.

References AggState::agg_done, AGG_HASHED, AGG_MIXED, AggState::aggcontexts, Agg::aggParams, AggState::aggstrategy, bms_overlap(), build_hash_tables(), PlanState::chgParam, ExprContext::ecxt_aggnulls, ExprContext::ecxt_aggvalues, ExecClearTuple(), ExecReScan(), AggState::grp_firstTuple, AggState::hash_ever_spilled, AggState::hash_ngroups_current, AggState::hash_spill_mode, hashagg_recompile_expressions(), hashagg_reset_spill_state(), AggState::hashcontext, AggStatePerHashData::hashiter, AggStatePerHashData::hashtable, heap_freetuple(), initialize_phase(), AggState::input_done, Max, AggState::maxsets, MemSet, AggState::numaggs, AggState::numtrans, outerPlan, outerPlanState, AggState::pergroups, AggState::perhash, AggState::pertrans, PlanState::plan, AggState::projected_set, ScanState::ps, PlanState::ps_ExprContext, ReScanExprContext(), ResetTupleHashIterator, select_current_set(), AggStatePerTransData::sortstates, AggState::ss, ScanState::ss_ScanTupleSlot, AggState::table_filled, and tuplesort_end().

Referenced by ExecReScan().

 {
     ExprContext *econtext = node->ss.ps.ps_ExprContext;
     PlanState  *outerPlan = outerPlanState(node);
     Agg        *aggnode = (Agg *) node->ss.ps.plan;
     int         transno;
     int         numGroupingSets = Max(node->maxsets, 1);
     int         setno;
 
     node->agg_done = false;
 
     if (node->aggstrategy == AGG_HASHED)
     {
         /*
          * In the hashed case, if we haven't yet built the hash table then we
          * can just return; nothing done yet, so nothing to undo. If subnode's
          * chgParam is not NULL then it will be re-scanned by ExecProcNode,
          * else no reason to re-scan it at all.
          */
         if (!node->table_filled)
             return;
 
         /*
          * If we do have the hash table, and it never spilled, and the subplan
          * does not have any parameter changes, and none of our own parameter
          * changes affect input expressions of the aggregated functions, then
          * we can just rescan the existing hash table; no need to build it
          * again.
          */
         if (outerPlan->chgParam == NULL && !node->hash_ever_spilled &&
             !bms_overlap(node->ss.ps.chgParam, aggnode->aggParams))
         {
             ResetTupleHashIterator(node->perhash[0].hashtable,
                                    &node->perhash[0].hashiter);
             select_current_set(node, 0, true);
             return;
         }
     }
 
     /* Make sure we have closed any open tuplesorts */
     for (transno = 0; transno < node->numtrans; transno++)
     {
         for (setno = 0; setno < numGroupingSets; setno++)
         {
             AggStatePerTrans pertrans = &node->pertrans[transno];
 
             if (pertrans->sortstates[setno])
             {
                 tuplesort_end(pertrans->sortstates[setno]);
                 pertrans->sortstates[setno] = NULL;
             }
         }
     }
 
     /*
      * We don't need to ReScanExprContext the output tuple context here;
      * ExecReScan already did it. But we do need to reset our per-grouping-set
      * contexts, which may have transvalues stored in them. (We use rescan
      * rather than just reset because transfns may have registered callbacks
      * that need to be run now.) For the AGG_HASHED case, see below.
      */
 
     for (setno = 0; setno < numGroupingSets; setno++)
     {
         ReScanExprContext(node->aggcontexts[setno]);
     }
 
     /* Release first tuple of group, if we have made a copy */
     if (node->grp_firstTuple != NULL)
     {
         heap_freetuple(node->grp_firstTuple);
         node->grp_firstTuple = NULL;
     }
     ExecClearTuple(node->ss.ss_ScanTupleSlot);
 
     /* Forget current agg values */
     MemSet(econtext->ecxt_aggvalues, 0, sizeof(Datum) * node->numaggs);
     MemSet(econtext->ecxt_aggnulls, 0, sizeof(bool) * node->numaggs);
 
     /*
      * With AGG_HASHED/MIXED, the hash table is allocated in a sub-context of
      * the hashcontext. This used to be an issue, but now, resetting a context
      * automatically deletes sub-contexts too.
      */
     if (node->aggstrategy == AGG_HASHED || node->aggstrategy == AGG_MIXED)
     {
         hashagg_reset_spill_state(node);
 
         node->hash_ever_spilled = false;
         node->hash_spill_mode = false;
         node->hash_ngroups_current = 0;
 
         ReScanExprContext(node->hashcontext);
         /* Rebuild an empty hash table */
         build_hash_tables(node);
         node->table_filled = false;
         /* iterator will be reset when the table is filled */
 
         hashagg_recompile_expressions(node, false, false);
     }
 
     if (node->aggstrategy != AGG_HASHED)
     {
         /*
          * Reset the per-group state (in particular, mark transvalues null)
          */
         for (setno = 0; setno < numGroupingSets; setno++)
         {
             MemSet(node->pergroups[setno], 0,
                    sizeof(AggStatePerGroupData) * node->numaggs);
         }
 
         /* reset to phase 1 */
         initialize_phase(node, 1);
 
         node->input_done = false;
         node->projected_set = -1;
     }
 
     if (outerPlan->chgParam == NULL)
         ExecReScan(outerPlan);
 }

◆ hash_agg_entry_size()

Size hash_agg_entry_size	(	int	numTrans,
		Size	tupleWidth,
		Size	transitionSpace
	)

Definition at line 1695 of file nodeAgg.c.

References CHUNKHDRSZ, MAXALIGN, and SizeofMinimalTupleHeader.

Referenced by cost_agg(), estimate_hashagg_tablesize(), and ExecInitAgg().

 {
     Size        tupleChunkSize;
     Size        pergroupChunkSize;
     Size        transitionChunkSize;
     Size        tupleSize = (MAXALIGN(SizeofMinimalTupleHeader) +
                              tupleWidth);
     Size        pergroupSize = numTrans * sizeof(AggStatePerGroupData);
 
     tupleChunkSize = CHUNKHDRSZ + tupleSize;
 
     if (pergroupSize > 0)
         pergroupChunkSize = CHUNKHDRSZ + pergroupSize;
     else
         pergroupChunkSize = 0;
 
     if (transitionSpace > 0)
         transitionChunkSize = CHUNKHDRSZ + transitionSpace;
     else
         transitionChunkSize = 0;
 
     return
         sizeof(TupleHashEntryData) +
         tupleChunkSize +
         pergroupChunkSize +
         transitionChunkSize;
 }

◆ hash_agg_set_limits()

void hash_agg_set_limits	(	double	hashentrysize,
		double	input_groups,
		int	used_bits,
		Size *	mem_limit,
		uint64 *	ngroups_limit,
		int *	num_partitions
	)

Definition at line 1793 of file nodeAgg.c.

References get_hash_mem(), hash_choose_num_partitions(), HASHAGG_READ_BUFFER_SIZE, and HASHAGG_WRITE_BUFFER_SIZE.

Referenced by agg_refill_hash_table(), cost_agg(), and ExecInitAgg().

 {
     int         npartitions;
     Size        partition_mem;
     int         hash_mem = get_hash_mem();
 
     /* if not expected to spill, use all of hash_mem */
     if (input_groups * hashentrysize < hash_mem * 1024L)
     {
         if (num_partitions != NULL)
             *num_partitions = 0;
         *mem_limit = hash_mem * 1024L;
         *ngroups_limit = *mem_limit / hashentrysize;
         return;
     }
 
     /*
      * Calculate expected memory requirements for spilling, which is the size
      * of the buffers needed for all the tapes that need to be open at once.
      * Then, subtract that from the memory available for holding hash tables.
      */
     npartitions = hash_choose_num_partitions(input_groups,
                                              hashentrysize,
                                              used_bits,
                                              NULL);
     if (num_partitions != NULL)
         *num_partitions = npartitions;
 
     partition_mem =
         HASHAGG_READ_BUFFER_SIZE +
         HASHAGG_WRITE_BUFFER_SIZE * npartitions;
 
     /*
      * Don't set the limit below 3/4 of hash_mem. In that case, we are at the
      * minimum number of partitions, so we aren't going to dramatically exceed
      * work mem anyway.
      */
     if (hash_mem * 1024L > 4 * partition_mem)
         *mem_limit = hash_mem * 1024L - partition_mem;
     else
         *mem_limit = hash_mem * 1024L * 0.75;
 
     if (*mem_limit > hashentrysize)
         *ngroups_limit = *mem_limit / hashentrysize;
     else
         *ngroups_limit = 1;
 }

Data Structures

Macros

Typedefs

Functions

Macro Definition Documentation

◆ FIELDNO_AGGSTATEPERGROUPDATA_NOTRANSVALUE

◆ FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUE

◆ FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUEISNULL

Typedef Documentation

◆ AggStatePerAggData

◆ AggStatePerGroupData

◆ AggStatePerHashData

◆ AggStatePerPhaseData

◆ AggStatePerTransData

Function Documentation

◆ ExecAggEstimate()

◆ ExecAggInitializeDSM()

◆ ExecAggInitializeWorker()

◆ ExecAggRetrieveInstrumentation()

◆ ExecEndAgg()

◆ ExecInitAgg()

◆ ExecReScanAgg()

◆ hash_agg_entry_size()

◆ hash_agg_set_limits()