PostgreSQL Source Code  git master
nodeAgg.h File Reference
#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:

Go to the source code of 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

AggStateExecInitAgg (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 257 of file nodeAgg.h.

◆ FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUE

#define FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUE   0

Definition at line 252 of file nodeAgg.h.

◆ FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUEISNULL

#define FIELDNO_AGGSTATEPERGROUPDATA_TRANSVALUEISNULL   1

Definition at line 254 of file nodeAgg.h.

Typedef Documentation

◆ AggStatePerAggData

◆ AggStatePerGroupData

◆ AggStatePerHashData

◆ AggStatePerPhaseData

◆ AggStatePerTransData

Function Documentation

◆ ExecAggEstimate()

void ExecAggEstimate ( AggState node,
ParallelContext pcxt 
)

Definition at line 4683 of file nodeAgg.c.

4684 {
4685  Size size;
4686 
4687  /* don't need this if not instrumenting or no workers */
4688  if (!node->ss.ps.instrument || pcxt->nworkers == 0)
4689  return;
4690 
4691  size = mul_size(pcxt->nworkers, sizeof(AggregateInstrumentation));
4692  size = add_size(size, offsetof(SharedAggInfo, sinstrument));
4694  shm_toc_estimate_keys(&pcxt->estimator, 1);
4695 }
size_t Size
Definition: c.h:605
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
Size add_size(Size s1, Size s2)
Definition: shmem.c:493
Size mul_size(Size s1, Size s2)
Definition: shmem.c:510
static pg_noinline void Size size
Definition: slab.c:607
ScanState ss
Definition: execnodes.h:2463
shm_toc_estimator estimator
Definition: parallel.h:41
Instrumentation * instrument
Definition: execnodes.h:1127
PlanState ps
Definition: execnodes.h:1564

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

Referenced by ExecParallelEstimate().

◆ ExecAggInitializeDSM()

void ExecAggInitializeDSM ( AggState node,
ParallelContext pcxt 
)

Definition at line 4704 of file nodeAgg.c.

4705 {
4706  Size size;
4707 
4708  /* don't need this if not instrumenting or no workers */
4709  if (!node->ss.ps.instrument || pcxt->nworkers == 0)
4710  return;
4711 
4712  size = offsetof(SharedAggInfo, sinstrument)
4713  + pcxt->nworkers * sizeof(AggregateInstrumentation);
4714  node->shared_info = shm_toc_allocate(pcxt->toc, size);
4715  /* ensure any unfilled slots will contain zeroes */
4716  memset(node->shared_info, 0, size);
4717  node->shared_info->num_workers = pcxt->nworkers;
4718  shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id,
4719  node->shared_info);
4720 }
struct AggregateInstrumentation AggregateInstrumentation
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
SharedAggInfo * shared_info
Definition: execnodes.h:2534
shm_toc * toc
Definition: parallel.h:44
Plan * plan
Definition: execnodes.h:1117
int plan_node_id
Definition: plannodes.h:151

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

Referenced by ExecParallelInitializeDSM().

◆ ExecAggInitializeWorker()

void ExecAggInitializeWorker ( AggState node,
ParallelWorkerContext pwcxt 
)

Definition at line 4729 of file nodeAgg.c.

4730 {
4731  node->shared_info =
4732  shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true);
4733 }
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232

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

Referenced by ExecParallelInitializeWorker().

◆ ExecAggRetrieveInstrumentation()

void ExecAggRetrieveInstrumentation ( AggState node)

Definition at line 4742 of file nodeAgg.c.

4743 {
4744  Size size;
4745  SharedAggInfo *si;
4746 
4747  if (node->shared_info == NULL)
4748  return;
4749 
4750  size = offsetof(SharedAggInfo, sinstrument)
4752  si = palloc(size);
4753  memcpy(si, node->shared_info, size);
4754  node->shared_info = si;
4755 }
void * palloc(Size size)
Definition: mcxt.c:1316

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

Referenced by ExecParallelRetrieveInstrumentation().

◆ ExecEndAgg()

void ExecEndAgg ( AggState node)

Definition at line 4304 of file nodeAgg.c.

4305 {
4307  int transno;
4308  int numGroupingSets = Max(node->maxsets, 1);
4309  int setno;
4310 
4311  /*
4312  * When ending a parallel worker, copy the statistics gathered by the
4313  * worker back into shared memory so that it can be picked up by the main
4314  * process to report in EXPLAIN ANALYZE.
4315  */
4316  if (node->shared_info && IsParallelWorker())
4317  {
4319 
4320  Assert(ParallelWorkerNumber <= node->shared_info->num_workers);
4323  si->hash_disk_used = node->hash_disk_used;
4324  si->hash_mem_peak = node->hash_mem_peak;
4325  }
4326 
4327  /* Make sure we have closed any open tuplesorts */
4328 
4329  if (node->sort_in)
4330  tuplesort_end(node->sort_in);
4331  if (node->sort_out)
4332  tuplesort_end(node->sort_out);
4333 
4335 
4336  if (node->hash_metacxt != NULL)
4337  {
4339  node->hash_metacxt = NULL;
4340  }
4341 
4342  for (transno = 0; transno < node->numtrans; transno++)
4343  {
4344  AggStatePerTrans pertrans = &node->pertrans[transno];
4345 
4346  for (setno = 0; setno < numGroupingSets; setno++)
4347  {
4348  if (pertrans->sortstates[setno])
4349  tuplesort_end(pertrans->sortstates[setno]);
4350  }
4351  }
4352 
4353  /* And ensure any agg shutdown callbacks have been called */
4354  for (setno = 0; setno < numGroupingSets; setno++)
4355  ReScanExprContext(node->aggcontexts[setno]);
4356  if (node->hashcontext)
4358 
4359  outerPlan = outerPlanState(node);
4361 }
int ParallelWorkerNumber
Definition: parallel.c:112
#define Max(x, y)
Definition: c.h:998
#define Assert(condition)
Definition: c.h:858
void ExecEndNode(PlanState *node)
Definition: execProcnode.c:557
void ReScanExprContext(ExprContext *econtext)
Definition: execUtils.c:441
#define outerPlanState(node)
Definition: execnodes.h:1213
#define IsParallelWorker()
Definition: parallel.h:60
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:454
static void hashagg_reset_spill_state(AggState *aggstate)
Definition: nodeAgg.c:3133
#define outerPlan(node)
Definition: plannodes.h:182
Tuplesortstate ** sortstates
Definition: nodeAgg.h:162
MemoryContext hash_metacxt
Definition: execnodes.h:2505
Tuplesortstate * sort_out
Definition: execnodes.h:2496
uint64 hash_disk_used
Definition: execnodes.h:2523
AggStatePerTrans pertrans
Definition: execnodes.h:2473
int numtrans
Definition: execnodes.h:2466
ExprContext * hashcontext
Definition: execnodes.h:2474
Size hash_mem_peak
Definition: execnodes.h:2520
int hash_batches_used
Definition: execnodes.h:2524
Tuplesortstate * sort_in
Definition: execnodes.h:2495
int maxsets
Definition: execnodes.h:2493
ExprContext ** aggcontexts
Definition: execnodes.h:2475
AggregateInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]
Definition: execnodes.h:2439
void tuplesort_end(Tuplesortstate *state)
Definition: tuplesort.c:971

References AggState::aggcontexts, Assert, ExecEndNode(), 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, ReScanExprContext(), AggState::shared_info, SharedAggInfo::sinstrument, AggState::sort_in, AggState::sort_out, AggStatePerTransData::sortstates, and tuplesort_end().

Referenced by ExecEndNode().

◆ ExecInitAgg()

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

Definition at line 3173 of file nodeAgg.c.

3174 {
3175  AggState *aggstate;
3176  AggStatePerAgg peraggs;
3177  AggStatePerTrans pertransstates;
3178  AggStatePerGroup *pergroups;
3179  Plan *outerPlan;
3180  ExprContext *econtext;
3181  TupleDesc scanDesc;
3182  int max_aggno;
3183  int max_transno;
3184  int numaggrefs;
3185  int numaggs;
3186  int numtrans;
3187  int phase;
3188  int phaseidx;
3189  ListCell *l;
3190  Bitmapset *all_grouped_cols = NULL;
3191  int numGroupingSets = 1;
3192  int numPhases;
3193  int numHashes;
3194  int i = 0;
3195  int j = 0;
3196  bool use_hashing = (node->aggstrategy == AGG_HASHED ||
3197  node->aggstrategy == AGG_MIXED);
3198 
3199  /* check for unsupported flags */
3200  Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
3201 
3202  /*
3203  * create state structure
3204  */
3205  aggstate = makeNode(AggState);
3206  aggstate->ss.ps.plan = (Plan *) node;
3207  aggstate->ss.ps.state = estate;
3208  aggstate->ss.ps.ExecProcNode = ExecAgg;
3209 
3210  aggstate->aggs = NIL;
3211  aggstate->numaggs = 0;
3212  aggstate->numtrans = 0;
3213  aggstate->aggstrategy = node->aggstrategy;
3214  aggstate->aggsplit = node->aggsplit;
3215  aggstate->maxsets = 0;
3216  aggstate->projected_set = -1;
3217  aggstate->current_set = 0;
3218  aggstate->peragg = NULL;
3219  aggstate->pertrans = NULL;
3220  aggstate->curperagg = NULL;
3221  aggstate->curpertrans = NULL;
3222  aggstate->input_done = false;
3223  aggstate->agg_done = false;
3224  aggstate->pergroups = NULL;
3225  aggstate->grp_firstTuple = NULL;
3226  aggstate->sort_in = NULL;
3227  aggstate->sort_out = NULL;
3228 
3229  /*
3230  * phases[0] always exists, but is dummy in sorted/plain mode
3231  */
3232  numPhases = (use_hashing ? 1 : 2);
3233  numHashes = (use_hashing ? 1 : 0);
3234 
3235  /*
3236  * Calculate the maximum number of grouping sets in any phase; this
3237  * determines the size of some allocations. Also calculate the number of
3238  * phases, since all hashed/mixed nodes contribute to only a single phase.
3239  */
3240  if (node->groupingSets)
3241  {
3242  numGroupingSets = list_length(node->groupingSets);
3243 
3244  foreach(l, node->chain)
3245  {
3246  Agg *agg = lfirst(l);
3247 
3248  numGroupingSets = Max(numGroupingSets,
3249  list_length(agg->groupingSets));
3250 
3251  /*
3252  * additional AGG_HASHED aggs become part of phase 0, but all
3253  * others add an extra phase.
3254  */
3255  if (agg->aggstrategy != AGG_HASHED)
3256  ++numPhases;
3257  else
3258  ++numHashes;
3259  }
3260  }
3261 
3262  aggstate->maxsets = numGroupingSets;
3263  aggstate->numphases = numPhases;
3264 
3265  aggstate->aggcontexts = (ExprContext **)
3266  palloc0(sizeof(ExprContext *) * numGroupingSets);
3267 
3268  /*
3269  * Create expression contexts. We need three or more, one for
3270  * per-input-tuple processing, one for per-output-tuple processing, one
3271  * for all the hashtables, and one for each grouping set. The per-tuple
3272  * memory context of the per-grouping-set ExprContexts (aggcontexts)
3273  * replaces the standalone memory context formerly used to hold transition
3274  * values. We cheat a little by using ExecAssignExprContext() to build
3275  * all of them.
3276  *
3277  * NOTE: the details of what is stored in aggcontexts and what is stored
3278  * in the regular per-query memory context are driven by a simple
3279  * decision: we want to reset the aggcontext at group boundaries (if not
3280  * hashing) and in ExecReScanAgg to recover no-longer-wanted space.
3281  */
3282  ExecAssignExprContext(estate, &aggstate->ss.ps);
3283  aggstate->tmpcontext = aggstate->ss.ps.ps_ExprContext;
3284 
3285  for (i = 0; i < numGroupingSets; ++i)
3286  {
3287  ExecAssignExprContext(estate, &aggstate->ss.ps);
3288  aggstate->aggcontexts[i] = aggstate->ss.ps.ps_ExprContext;
3289  }
3290 
3291  if (use_hashing)
3292  aggstate->hashcontext = CreateWorkExprContext(estate);
3293 
3294  ExecAssignExprContext(estate, &aggstate->ss.ps);
3295 
3296  /*
3297  * Initialize child nodes.
3298  *
3299  * If we are doing a hashed aggregation then the child plan does not need
3300  * to handle REWIND efficiently; see ExecReScanAgg.
3301  */
3302  if (node->aggstrategy == AGG_HASHED)
3303  eflags &= ~EXEC_FLAG_REWIND;
3304  outerPlan = outerPlan(node);
3305  outerPlanState(aggstate) = ExecInitNode(outerPlan, estate, eflags);
3306 
3307  /*
3308  * initialize source tuple type.
3309  */
3310  aggstate->ss.ps.outerops =
3312  &aggstate->ss.ps.outeropsfixed);
3313  aggstate->ss.ps.outeropsset = true;
3314 
3315  ExecCreateScanSlotFromOuterPlan(estate, &aggstate->ss,
3316  aggstate->ss.ps.outerops);
3317  scanDesc = aggstate->ss.ss_ScanTupleSlot->tts_tupleDescriptor;
3318 
3319  /*
3320  * If there are more than two phases (including a potential dummy phase
3321  * 0), input will be resorted using tuplesort. Need a slot for that.
3322  */
3323  if (numPhases > 2)
3324  {
3325  aggstate->sort_slot = ExecInitExtraTupleSlot(estate, scanDesc,
3327 
3328  /*
3329  * The output of the tuplesort, and the output from the outer child
3330  * might not use the same type of slot. In most cases the child will
3331  * be a Sort, and thus return a TTSOpsMinimalTuple type slot - but the
3332  * input can also be presorted due an index, in which case it could be
3333  * a different type of slot.
3334  *
3335  * XXX: For efficiency it would be good to instead/additionally
3336  * generate expressions with corresponding settings of outerops* for
3337  * the individual phases - deforming is often a bottleneck for
3338  * aggregations with lots of rows per group. If there's multiple
3339  * sorts, we know that all but the first use TTSOpsMinimalTuple (via
3340  * the nodeAgg.c internal tuplesort).
3341  */
3342  if (aggstate->ss.ps.outeropsfixed &&
3343  aggstate->ss.ps.outerops != &TTSOpsMinimalTuple)
3344  aggstate->ss.ps.outeropsfixed = false;
3345  }
3346 
3347  /*
3348  * Initialize result type, slot and projection.
3349  */
3351  ExecAssignProjectionInfo(&aggstate->ss.ps, NULL);
3352 
3353  /*
3354  * initialize child expressions
3355  *
3356  * We expect the parser to have checked that no aggs contain other agg
3357  * calls in their arguments (and just to be sure, we verify it again while
3358  * initializing the plan node). This would make no sense under SQL
3359  * semantics, and it's forbidden by the spec. Because it is true, we
3360  * don't need to worry about evaluating the aggs in any particular order.
3361  *
3362  * Note: execExpr.c finds Aggrefs for us, and adds them to aggstate->aggs.
3363  * Aggrefs in the qual are found here; Aggrefs in the targetlist are found
3364  * during ExecAssignProjectionInfo, above.
3365  */
3366  aggstate->ss.ps.qual =
3367  ExecInitQual(node->plan.qual, (PlanState *) aggstate);
3368 
3369  /*
3370  * We should now have found all Aggrefs in the targetlist and quals.
3371  */
3372  numaggrefs = list_length(aggstate->aggs);
3373  max_aggno = -1;
3374  max_transno = -1;
3375  foreach(l, aggstate->aggs)
3376  {
3377  Aggref *aggref = (Aggref *) lfirst(l);
3378 
3379  max_aggno = Max(max_aggno, aggref->aggno);
3380  max_transno = Max(max_transno, aggref->aggtransno);
3381  }
3382  numaggs = max_aggno + 1;
3383  numtrans = max_transno + 1;
3384 
3385  /*
3386  * For each phase, prepare grouping set data and fmgr lookup data for
3387  * compare functions. Accumulate all_grouped_cols in passing.
3388  */
3389  aggstate->phases = palloc0(numPhases * sizeof(AggStatePerPhaseData));
3390 
3391  aggstate->num_hashes = numHashes;
3392  if (numHashes)
3393  {
3394  aggstate->perhash = palloc0(sizeof(AggStatePerHashData) * numHashes);
3395  aggstate->phases[0].numsets = 0;
3396  aggstate->phases[0].gset_lengths = palloc(numHashes * sizeof(int));
3397  aggstate->phases[0].grouped_cols = palloc(numHashes * sizeof(Bitmapset *));
3398  }
3399 
3400  phase = 0;
3401  for (phaseidx = 0; phaseidx <= list_length(node->chain); ++phaseidx)
3402  {
3403  Agg *aggnode;
3404  Sort *sortnode;
3405 
3406  if (phaseidx > 0)
3407  {
3408  aggnode = list_nth_node(Agg, node->chain, phaseidx - 1);
3409  sortnode = castNode(Sort, outerPlan(aggnode));
3410  }
3411  else
3412  {
3413  aggnode = node;
3414  sortnode = NULL;
3415  }
3416 
3417  Assert(phase <= 1 || sortnode);
3418 
3419  if (aggnode->aggstrategy == AGG_HASHED
3420  || aggnode->aggstrategy == AGG_MIXED)
3421  {
3422  AggStatePerPhase phasedata = &aggstate->phases[0];
3423  AggStatePerHash perhash;
3424  Bitmapset *cols = NULL;
3425 
3426  Assert(phase == 0);
3427  i = phasedata->numsets++;
3428  perhash = &aggstate->perhash[i];
3429 
3430  /* phase 0 always points to the "real" Agg in the hash case */
3431  phasedata->aggnode = node;
3432  phasedata->aggstrategy = node->aggstrategy;
3433 
3434  /* but the actual Agg node representing this hash is saved here */
3435  perhash->aggnode = aggnode;
3436 
3437  phasedata->gset_lengths[i] = perhash->numCols = aggnode->numCols;
3438 
3439  for (j = 0; j < aggnode->numCols; ++j)
3440  cols = bms_add_member(cols, aggnode->grpColIdx[j]);
3441 
3442  phasedata->grouped_cols[i] = cols;
3443 
3444  all_grouped_cols = bms_add_members(all_grouped_cols, cols);
3445  continue;
3446  }
3447  else
3448  {
3449  AggStatePerPhase phasedata = &aggstate->phases[++phase];
3450  int num_sets;
3451 
3452  phasedata->numsets = num_sets = list_length(aggnode->groupingSets);
3453 
3454  if (num_sets)
3455  {
3456  phasedata->gset_lengths = palloc(num_sets * sizeof(int));
3457  phasedata->grouped_cols = palloc(num_sets * sizeof(Bitmapset *));
3458 
3459  i = 0;
3460  foreach(l, aggnode->groupingSets)
3461  {
3462  int current_length = list_length(lfirst(l));
3463  Bitmapset *cols = NULL;
3464 
3465  /* planner forces this to be correct */
3466  for (j = 0; j < current_length; ++j)
3467  cols = bms_add_member(cols, aggnode->grpColIdx[j]);
3468 
3469  phasedata->grouped_cols[i] = cols;
3470  phasedata->gset_lengths[i] = current_length;
3471 
3472  ++i;
3473  }
3474 
3475  all_grouped_cols = bms_add_members(all_grouped_cols,
3476  phasedata->grouped_cols[0]);
3477  }
3478  else
3479  {
3480  Assert(phaseidx == 0);
3481 
3482  phasedata->gset_lengths = NULL;
3483  phasedata->grouped_cols = NULL;
3484  }
3485 
3486  /*
3487  * If we are grouping, precompute fmgr lookup data for inner loop.
3488  */
3489  if (aggnode->aggstrategy == AGG_SORTED)
3490  {
3491  /*
3492  * Build a separate function for each subset of columns that
3493  * need to be compared.
3494  */
3495  phasedata->eqfunctions =
3496  (ExprState **) palloc0(aggnode->numCols * sizeof(ExprState *));
3497 
3498  /* for each grouping set */
3499  for (int k = 0; k < phasedata->numsets; k++)
3500  {
3501  int length = phasedata->gset_lengths[k];
3502 
3503  /* nothing to do for empty grouping set */
3504  if (length == 0)
3505  continue;
3506 
3507  /* if we already had one of this length, it'll do */
3508  if (phasedata->eqfunctions[length - 1] != NULL)
3509  continue;
3510 
3511  phasedata->eqfunctions[length - 1] =
3512  execTuplesMatchPrepare(scanDesc,
3513  length,
3514  aggnode->grpColIdx,
3515  aggnode->grpOperators,
3516  aggnode->grpCollations,
3517  (PlanState *) aggstate);
3518  }
3519 
3520  /* and for all grouped columns, unless already computed */
3521  if (aggnode->numCols > 0 &&
3522  phasedata->eqfunctions[aggnode->numCols - 1] == NULL)
3523  {
3524  phasedata->eqfunctions[aggnode->numCols - 1] =
3525  execTuplesMatchPrepare(scanDesc,
3526  aggnode->numCols,
3527  aggnode->grpColIdx,
3528  aggnode->grpOperators,
3529  aggnode->grpCollations,
3530  (PlanState *) aggstate);
3531  }
3532  }
3533 
3534  phasedata->aggnode = aggnode;
3535  phasedata->aggstrategy = aggnode->aggstrategy;
3536  phasedata->sortnode = sortnode;
3537  }
3538  }
3539 
3540  /*
3541  * Convert all_grouped_cols to a descending-order list.
3542  */
3543  i = -1;
3544  while ((i = bms_next_member(all_grouped_cols, i)) >= 0)
3545  aggstate->all_grouped_cols = lcons_int(i, aggstate->all_grouped_cols);
3546 
3547  /*
3548  * Set up aggregate-result storage in the output expr context, and also
3549  * allocate my private per-agg working storage
3550  */
3551  econtext = aggstate->ss.ps.ps_ExprContext;
3552  econtext->ecxt_aggvalues = (Datum *) palloc0(sizeof(Datum) * numaggs);
3553  econtext->ecxt_aggnulls = (bool *) palloc0(sizeof(bool) * numaggs);
3554 
3555  peraggs = (AggStatePerAgg) palloc0(sizeof(AggStatePerAggData) * numaggs);
3556  pertransstates = (AggStatePerTrans) palloc0(sizeof(AggStatePerTransData) * numtrans);
3557 
3558  aggstate->peragg = peraggs;
3559  aggstate->pertrans = pertransstates;
3560 
3561 
3562  aggstate->all_pergroups =
3564  * (numGroupingSets + numHashes));
3565  pergroups = aggstate->all_pergroups;
3566 
3567  if (node->aggstrategy != AGG_HASHED)
3568  {
3569  for (i = 0; i < numGroupingSets; i++)
3570  {
3571  pergroups[i] = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData)
3572  * numaggs);
3573  }
3574 
3575  aggstate->pergroups = pergroups;
3576  pergroups += numGroupingSets;
3577  }
3578 
3579  /*
3580  * Hashing can only appear in the initial phase.
3581  */
3582  if (use_hashing)
3583  {
3584  Plan *outerplan = outerPlan(node);
3585  uint64 totalGroups = 0;
3586 
3587  aggstate->hash_metacxt = AllocSetContextCreate(aggstate->ss.ps.state->es_query_cxt,
3588  "HashAgg meta context",
3590  aggstate->hash_spill_rslot = ExecInitExtraTupleSlot(estate, scanDesc,
3592  aggstate->hash_spill_wslot = ExecInitExtraTupleSlot(estate, scanDesc,
3593  &TTSOpsVirtual);
3594 
3595  /* this is an array of pointers, not structures */
3596  aggstate->hash_pergroup = pergroups;
3597 
3598  aggstate->hashentrysize = hash_agg_entry_size(aggstate->numtrans,
3599  outerplan->plan_width,
3600  node->transitionSpace);
3601 
3602  /*
3603  * Consider all of the grouping sets together when setting the limits
3604  * and estimating the number of partitions. This can be inaccurate
3605  * when there is more than one grouping set, but should still be
3606  * reasonable.
3607  */
3608  for (int k = 0; k < aggstate->num_hashes; k++)
3609  totalGroups += aggstate->perhash[k].aggnode->numGroups;
3610 
3611  hash_agg_set_limits(aggstate->hashentrysize, totalGroups, 0,
3612  &aggstate->hash_mem_limit,
3613  &aggstate->hash_ngroups_limit,
3614  &aggstate->hash_planned_partitions);
3615  find_hash_columns(aggstate);
3616 
3617  /* Skip massive memory allocation if we are just doing EXPLAIN */
3618  if (!(eflags & EXEC_FLAG_EXPLAIN_ONLY))
3619  build_hash_tables(aggstate);
3620 
3621  aggstate->table_filled = false;
3622 
3623  /* Initialize this to 1, meaning nothing spilled, yet */
3624  aggstate->hash_batches_used = 1;
3625  }
3626 
3627  /*
3628  * Initialize current phase-dependent values to initial phase. The initial
3629  * phase is 1 (first sort pass) for all strategies that use sorting (if
3630  * hashing is being done too, then phase 0 is processed last); but if only
3631  * hashing is being done, then phase 0 is all there is.
3632  */
3633  if (node->aggstrategy == AGG_HASHED)
3634  {
3635  aggstate->current_phase = 0;
3636  initialize_phase(aggstate, 0);
3637  select_current_set(aggstate, 0, true);
3638  }
3639  else
3640  {
3641  aggstate->current_phase = 1;
3642  initialize_phase(aggstate, 1);
3643  select_current_set(aggstate, 0, false);
3644  }
3645 
3646  /*
3647  * Perform lookups of aggregate function info, and initialize the
3648  * unchanging fields of the per-agg and per-trans data.
3649  */
3650  foreach(l, aggstate->aggs)
3651  {
3652  Aggref *aggref = lfirst(l);
3653  AggStatePerAgg peragg;
3654  AggStatePerTrans pertrans;
3655  Oid aggTransFnInputTypes[FUNC_MAX_ARGS];
3656  int numAggTransFnArgs;
3657  int numDirectArgs;
3658  HeapTuple aggTuple;
3659  Form_pg_aggregate aggform;
3660  AclResult aclresult;
3661  Oid finalfn_oid;
3662  Oid serialfn_oid,
3663  deserialfn_oid;
3664  Oid aggOwner;
3665  Expr *finalfnexpr;
3666  Oid aggtranstype;
3667 
3668  /* Planner should have assigned aggregate to correct level */
3669  Assert(aggref->agglevelsup == 0);
3670  /* ... and the split mode should match */
3671  Assert(aggref->aggsplit == aggstate->aggsplit);
3672 
3673  peragg = &peraggs[aggref->aggno];
3674 
3675  /* Check if we initialized the state for this aggregate already. */
3676  if (peragg->aggref != NULL)
3677  continue;
3678 
3679  peragg->aggref = aggref;
3680  peragg->transno = aggref->aggtransno;
3681 
3682  /* Fetch the pg_aggregate row */
3683  aggTuple = SearchSysCache1(AGGFNOID,
3684  ObjectIdGetDatum(aggref->aggfnoid));
3685  if (!HeapTupleIsValid(aggTuple))
3686  elog(ERROR, "cache lookup failed for aggregate %u",
3687  aggref->aggfnoid);
3688  aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple);
3689 
3690  /* Check permission to call aggregate function */
3691  aclresult = object_aclcheck(ProcedureRelationId, aggref->aggfnoid, GetUserId(),
3692  ACL_EXECUTE);
3693  if (aclresult != ACLCHECK_OK)
3694  aclcheck_error(aclresult, OBJECT_AGGREGATE,
3695  get_func_name(aggref->aggfnoid));
3697 
3698  /* planner recorded transition state type in the Aggref itself */
3699  aggtranstype = aggref->aggtranstype;
3700  Assert(OidIsValid(aggtranstype));
3701 
3702  /* Final function only required if we're finalizing the aggregates */
3703  if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
3704  peragg->finalfn_oid = finalfn_oid = InvalidOid;
3705  else
3706  peragg->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
3707 
3708  serialfn_oid = InvalidOid;
3709  deserialfn_oid = InvalidOid;
3710 
3711  /*
3712  * Check if serialization/deserialization is required. We only do it
3713  * for aggregates that have transtype INTERNAL.
3714  */
3715  if (aggtranstype == INTERNALOID)
3716  {
3717  /*
3718  * The planner should only have generated a serialize agg node if
3719  * every aggregate with an INTERNAL state has a serialization
3720  * function. Verify that.
3721  */
3722  if (DO_AGGSPLIT_SERIALIZE(aggstate->aggsplit))
3723  {
3724  /* serialization only valid when not running finalfn */
3726 
3727  if (!OidIsValid(aggform->aggserialfn))
3728  elog(ERROR, "serialfunc not provided for serialization aggregation");
3729  serialfn_oid = aggform->aggserialfn;
3730  }
3731 
3732  /* Likewise for deserialization functions */
3733  if (DO_AGGSPLIT_DESERIALIZE(aggstate->aggsplit))
3734  {
3735  /* deserialization only valid when combining states */
3736  Assert(DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
3737 
3738  if (!OidIsValid(aggform->aggdeserialfn))
3739  elog(ERROR, "deserialfunc not provided for deserialization aggregation");
3740  deserialfn_oid = aggform->aggdeserialfn;
3741  }
3742  }
3743 
3744  /* Check that aggregate owner has permission to call component fns */
3745  {
3746  HeapTuple procTuple;
3747 
3748  procTuple = SearchSysCache1(PROCOID,
3749  ObjectIdGetDatum(aggref->aggfnoid));
3750  if (!HeapTupleIsValid(procTuple))
3751  elog(ERROR, "cache lookup failed for function %u",
3752  aggref->aggfnoid);
3753  aggOwner = ((Form_pg_proc) GETSTRUCT(procTuple))->proowner;
3754  ReleaseSysCache(procTuple);
3755 
3756  if (OidIsValid(finalfn_oid))
3757  {
3758  aclresult = object_aclcheck(ProcedureRelationId, finalfn_oid, aggOwner,
3759  ACL_EXECUTE);
3760  if (aclresult != ACLCHECK_OK)
3761  aclcheck_error(aclresult, OBJECT_FUNCTION,
3762  get_func_name(finalfn_oid));
3763  InvokeFunctionExecuteHook(finalfn_oid);
3764  }
3765  if (OidIsValid(serialfn_oid))
3766  {
3767  aclresult = object_aclcheck(ProcedureRelationId, serialfn_oid, aggOwner,
3768  ACL_EXECUTE);
3769  if (aclresult != ACLCHECK_OK)
3770  aclcheck_error(aclresult, OBJECT_FUNCTION,
3771  get_func_name(serialfn_oid));
3772  InvokeFunctionExecuteHook(serialfn_oid);
3773  }
3774  if (OidIsValid(deserialfn_oid))
3775  {
3776  aclresult = object_aclcheck(ProcedureRelationId, deserialfn_oid, aggOwner,
3777  ACL_EXECUTE);
3778  if (aclresult != ACLCHECK_OK)
3779  aclcheck_error(aclresult, OBJECT_FUNCTION,
3780  get_func_name(deserialfn_oid));
3781  InvokeFunctionExecuteHook(deserialfn_oid);
3782  }
3783  }
3784 
3785  /*
3786  * Get actual datatypes of the (nominal) aggregate inputs. These
3787  * could be different from the agg's declared input types, when the
3788  * agg accepts ANY or a polymorphic type.
3789  */
3790  numAggTransFnArgs = get_aggregate_argtypes(aggref,
3791  aggTransFnInputTypes);
3792 
3793  /* Count the "direct" arguments, if any */
3794  numDirectArgs = list_length(aggref->aggdirectargs);
3795 
3796  /* Detect how many arguments to pass to the finalfn */
3797  if (aggform->aggfinalextra)
3798  peragg->numFinalArgs = numAggTransFnArgs + 1;
3799  else
3800  peragg->numFinalArgs = numDirectArgs + 1;
3801 
3802  /* Initialize any direct-argument expressions */
3803  peragg->aggdirectargs = ExecInitExprList(aggref->aggdirectargs,
3804  (PlanState *) aggstate);
3805 
3806  /*
3807  * build expression trees using actual argument & result types for the
3808  * finalfn, if it exists and is required.
3809  */
3810  if (OidIsValid(finalfn_oid))
3811  {
3812  build_aggregate_finalfn_expr(aggTransFnInputTypes,
3813  peragg->numFinalArgs,
3814  aggtranstype,
3815  aggref->aggtype,
3816  aggref->inputcollid,
3817  finalfn_oid,
3818  &finalfnexpr);
3819  fmgr_info(finalfn_oid, &peragg->finalfn);
3820  fmgr_info_set_expr((Node *) finalfnexpr, &peragg->finalfn);
3821  }
3822 
3823  /* get info about the output value's datatype */
3824  get_typlenbyval(aggref->aggtype,
3825  &peragg->resulttypeLen,
3826  &peragg->resulttypeByVal);
3827 
3828  /*
3829  * Build working state for invoking the transition function, if we
3830  * haven't done it already.
3831  */
3832  pertrans = &pertransstates[aggref->aggtransno];
3833  if (pertrans->aggref == NULL)
3834  {
3835  Datum textInitVal;
3836  Datum initValue;
3837  bool initValueIsNull;
3838  Oid transfn_oid;
3839 
3840  /*
3841  * If this aggregation is performing state combines, then instead
3842  * of using the transition function, we'll use the combine
3843  * function.
3844  */
3845  if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
3846  {
3847  transfn_oid = aggform->aggcombinefn;
3848 
3849  /* If not set then the planner messed up */
3850  if (!OidIsValid(transfn_oid))
3851  elog(ERROR, "combinefn not set for aggregate function");
3852  }
3853  else
3854  transfn_oid = aggform->aggtransfn;
3855 
3856  aclresult = object_aclcheck(ProcedureRelationId, transfn_oid, aggOwner, ACL_EXECUTE);
3857  if (aclresult != ACLCHECK_OK)
3858  aclcheck_error(aclresult, OBJECT_FUNCTION,
3859  get_func_name(transfn_oid));
3860  InvokeFunctionExecuteHook(transfn_oid);
3861 
3862  /*
3863  * initval is potentially null, so don't try to access it as a
3864  * struct field. Must do it the hard way with SysCacheGetAttr.
3865  */
3866  textInitVal = SysCacheGetAttr(AGGFNOID, aggTuple,
3867  Anum_pg_aggregate_agginitval,
3868  &initValueIsNull);
3869  if (initValueIsNull)
3870  initValue = (Datum) 0;
3871  else
3872  initValue = GetAggInitVal(textInitVal, aggtranstype);
3873 
3874  if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
3875  {
3876  Oid combineFnInputTypes[] = {aggtranstype,
3877  aggtranstype};
3878 
3879  /*
3880  * When combining there's only one input, the to-be-combined
3881  * transition value. The transition value is not counted
3882  * here.
3883  */
3884  pertrans->numTransInputs = 1;
3885 
3886  /* aggcombinefn always has two arguments of aggtranstype */
3887  build_pertrans_for_aggref(pertrans, aggstate, estate,
3888  aggref, transfn_oid, aggtranstype,
3889  serialfn_oid, deserialfn_oid,
3890  initValue, initValueIsNull,
3891  combineFnInputTypes, 2);
3892 
3893  /*
3894  * Ensure that a combine function to combine INTERNAL states
3895  * is not strict. This should have been checked during CREATE
3896  * AGGREGATE, but the strict property could have been changed
3897  * since then.
3898  */
3899  if (pertrans->transfn.fn_strict && aggtranstype == INTERNALOID)
3900  ereport(ERROR,
3901  (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
3902  errmsg("combine function with transition type %s must not be declared STRICT",
3903  format_type_be(aggtranstype))));
3904  }
3905  else
3906  {
3907  /* Detect how many arguments to pass to the transfn */
3908  if (AGGKIND_IS_ORDERED_SET(aggref->aggkind))
3909  pertrans->numTransInputs = list_length(aggref->args);
3910  else
3911  pertrans->numTransInputs = numAggTransFnArgs;
3912 
3913  build_pertrans_for_aggref(pertrans, aggstate, estate,
3914  aggref, transfn_oid, aggtranstype,
3915  serialfn_oid, deserialfn_oid,
3916  initValue, initValueIsNull,
3917  aggTransFnInputTypes,
3918  numAggTransFnArgs);
3919 
3920  /*
3921  * If the transfn is strict and the initval is NULL, make sure
3922  * input type and transtype are the same (or at least
3923  * binary-compatible), so that it's OK to use the first
3924  * aggregated input value as the initial transValue. This
3925  * should have been checked at agg definition time, but we
3926  * must check again in case the transfn's strictness property
3927  * has been changed.
3928  */
3929  if (pertrans->transfn.fn_strict && pertrans->initValueIsNull)
3930  {
3931  if (numAggTransFnArgs <= numDirectArgs ||
3932  !IsBinaryCoercible(aggTransFnInputTypes[numDirectArgs],
3933  aggtranstype))
3934  ereport(ERROR,
3935  (errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
3936  errmsg("aggregate %u needs to have compatible input type and transition type",
3937  aggref->aggfnoid)));
3938  }
3939  }
3940  }
3941  else
3942  pertrans->aggshared = true;
3943  ReleaseSysCache(aggTuple);
3944  }
3945 
3946  /*
3947  * Update aggstate->numaggs to be the number of unique aggregates found.
3948  * Also set numstates to the number of unique transition states found.
3949  */
3950  aggstate->numaggs = numaggs;
3951  aggstate->numtrans = numtrans;
3952 
3953  /*
3954  * Last, check whether any more aggregates got added onto the node while
3955  * we processed the expressions for the aggregate arguments (including not
3956  * only the regular arguments and FILTER expressions handled immediately
3957  * above, but any direct arguments we might've handled earlier). If so,
3958  * we have nested aggregate functions, which is semantically nonsensical,
3959  * so complain. (This should have been caught by the parser, so we don't
3960  * need to work hard on a helpful error message; but we defend against it
3961  * here anyway, just to be sure.)
3962  */
3963  if (numaggrefs != list_length(aggstate->aggs))
3964  ereport(ERROR,
3965  (errcode(ERRCODE_GROUPING_ERROR),
3966  errmsg("aggregate function calls cannot be nested")));
3967 
3968  /*
3969  * Build expressions doing all the transition work at once. We build a
3970  * different one for each phase, as the number of transition function
3971  * invocation can differ between phases. Note this'll work both for
3972  * transition and combination functions (although there'll only be one
3973  * phase in the latter case).
3974  */
3975  for (phaseidx = 0; phaseidx < aggstate->numphases; phaseidx++)
3976  {
3977  AggStatePerPhase phase = &aggstate->phases[phaseidx];
3978  bool dohash = false;
3979  bool dosort = false;
3980 
3981  /* phase 0 doesn't necessarily exist */
3982  if (!phase->aggnode)
3983  continue;
3984 
3985  if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 1)
3986  {
3987  /*
3988  * Phase one, and only phase one, in a mixed agg performs both
3989  * sorting and aggregation.
3990  */
3991  dohash = true;
3992  dosort = true;
3993  }
3994  else if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 0)
3995  {
3996  /*
3997  * No need to compute a transition function for an AGG_MIXED phase
3998  * 0 - the contents of the hashtables will have been computed
3999  * during phase 1.
4000  */
4001  continue;
4002  }
4003  else if (phase->aggstrategy == AGG_PLAIN ||
4004  phase->aggstrategy == AGG_SORTED)
4005  {
4006  dohash = false;
4007  dosort = true;
4008  }
4009  else if (phase->aggstrategy == AGG_HASHED)
4010  {
4011  dohash = true;
4012  dosort = false;
4013  }
4014  else
4015  Assert(false);
4016 
4017  phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash,
4018  false);
4019 
4020  /* cache compiled expression for outer slot without NULL check */
4021  phase->evaltrans_cache[0][0] = phase->evaltrans;
4022  }
4023 
4024  return aggstate;
4025 }
AclResult
Definition: acl.h:182
@ ACLCHECK_OK
Definition: acl.h:183
void aclcheck_error(AclResult aclerr, ObjectType objtype, const char *objectname)
Definition: aclchk.c:2702
AclResult object_aclcheck(Oid classid, Oid objectid, Oid roleid, AclMode mode)
Definition: aclchk.c:3890
int bms_next_member(const Bitmapset *a, int prevbit)
Definition: bitmapset.c:1306
Bitmapset * bms_add_member(Bitmapset *a, int x)
Definition: bitmapset.c:815
Bitmapset * bms_add_members(Bitmapset *a, const Bitmapset *b)
Definition: bitmapset.c:917
#define OidIsValid(objectId)
Definition: c.h:775
int errcode(int sqlerrcode)
Definition: elog.c:857
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:224
#define ereport(elevel,...)
Definition: elog.h:149
List * ExecInitExprList(List *nodes, PlanState *parent)
Definition: execExpr.c:326
ExprState * ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, bool doSort, bool doHash, bool nullcheck)
Definition: execExpr.c:3490
ExprState * ExecInitQual(List *qual, PlanState *parent)
Definition: execExpr.c:220
ExprState * execTuplesMatchPrepare(TupleDesc desc, int numCols, const AttrNumber *keyColIdx, const Oid *eqOperators, const Oid *collations, PlanState *parent)
Definition: execGrouping.c:58
PlanState * ExecInitNode(Plan *node, EState *estate, int eflags)
Definition: execProcnode.c:142
const TupleTableSlotOps TTSOpsVirtual
Definition: execTuples.c:84
TupleTableSlot * ExecInitExtraTupleSlot(EState *estate, TupleDesc tupledesc, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1918
void ExecInitResultTupleSlotTL(PlanState *planstate, const TupleTableSlotOps *tts_ops)
Definition: execTuples.c:1886
const TupleTableSlotOps TTSOpsMinimalTuple
Definition: execTuples.c:86
ExprContext * CreateWorkExprContext(EState *estate)
Definition: execUtils.c:319
const TupleTableSlotOps * ExecGetResultSlotOps(PlanState *planstate, bool *isfixed)
Definition: execUtils.c:502
void ExecCreateScanSlotFromOuterPlan(EState *estate, ScanState *scanstate, const TupleTableSlotOps *tts_ops)
Definition: execUtils.c:659
void ExecAssignExprContext(EState *estate, PlanState *planstate)
Definition: execUtils.c:483
void ExecAssignProjectionInfo(PlanState *planstate, TupleDesc inputDesc)
Definition: execUtils.c:538
struct AggStatePerGroupData * AggStatePerGroup
Definition: execnodes.h:2457
struct AggStatePerTransData * AggStatePerTrans
Definition: execnodes.h:2456
struct AggStatePerAggData * AggStatePerAgg
Definition: execnodes.h:2455
#define EXEC_FLAG_BACKWARD
Definition: executor.h:68
#define EXEC_FLAG_REWIND
Definition: executor.h:67
#define EXEC_FLAG_EXPLAIN_ONLY
Definition: executor.h:65
#define EXEC_FLAG_MARK
Definition: executor.h:69
void fmgr_info(Oid functionId, FmgrInfo *finfo)
Definition: fmgr.c:127
#define fmgr_info_set_expr(expr, finfo)
Definition: fmgr.h:135
char * format_type_be(Oid type_oid)
Definition: format_type.c:343
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
#define GETSTRUCT(TUP)
Definition: htup_details.h:653
static int initValue(long lng_val)
Definition: informix.c:683
int j
Definition: isn.c:74
int i
Definition: isn.c:73
List * lcons_int(int datum, List *list)
Definition: list.c:513
void get_typlenbyval(Oid typid, int16 *typlen, bool *typbyval)
Definition: lsyscache.c:2251
char * get_func_name(Oid funcid)
Definition: lsyscache.c:1608
void * palloc0(Size size)
Definition: mcxt.c:1346
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:160
Oid GetUserId(void)
Definition: miscinit.c:514
static void find_hash_columns(AggState *aggstate)
Definition: nodeAgg.c:1563
static Datum GetAggInitVal(Datum textInitVal, Oid transtype)
Definition: nodeAgg.c:4288
static TupleTableSlot * ExecAgg(PlanState *pstate)
Definition: nodeAgg.c:2158
static void initialize_phase(AggState *aggstate, int newphase)
Definition: nodeAgg.c:477
Size hash_agg_entry_size(int numTrans, Size tupleWidth, Size transitionSpace)
Definition: nodeAgg.c:1694
static void build_pertrans_for_aggref(AggStatePerTrans pertrans, AggState *aggstate, EState *estate, Aggref *aggref, Oid transfn_oid, Oid aggtranstype, Oid aggserialfn, Oid aggdeserialfn, Datum initValue, bool initValueIsNull, Oid *inputTypes, int numArguments)
Definition: nodeAgg.c:4038
static void build_hash_tables(AggState *aggstate)
Definition: nodeAgg.c:1468
static void select_current_set(AggState *aggstate, int setno, bool is_hash)
Definition: nodeAgg.c:455
void hash_agg_set_limits(double hashentrysize, double input_groups, int used_bits, Size *mem_limit, uint64 *ngroups_limit, int *num_partitions)
Definition: nodeAgg.c:1798
#define DO_AGGSPLIT_SKIPFINAL(as)
Definition: nodes.h:385
#define DO_AGGSPLIT_DESERIALIZE(as)
Definition: nodes.h:387
#define DO_AGGSPLIT_COMBINE(as)
Definition: nodes.h:384
@ AGG_SORTED
Definition: nodes.h:354
@ AGG_HASHED
Definition: nodes.h:355
@ AGG_MIXED
Definition: nodes.h:356
@ AGG_PLAIN
Definition: nodes.h:353
#define DO_AGGSPLIT_SERIALIZE(as)
Definition: nodes.h:386
#define makeNode(_type_)
Definition: nodes.h:155
#define castNode(_type_, nodeptr)
Definition: nodes.h:176
#define InvokeFunctionExecuteHook(objectId)
Definition: objectaccess.h:213
void build_aggregate_finalfn_expr(Oid *agg_input_types, int num_finalfn_inputs, Oid agg_state_type, Oid agg_result_type, Oid agg_input_collation, Oid finalfn_oid, Expr **finalfnexpr)
Definition: parse_agg.c:2136
int get_aggregate_argtypes(Aggref *aggref, Oid *inputTypes)
Definition: parse_agg.c:1908
bool IsBinaryCoercible(Oid srctype, Oid targettype)
@ OBJECT_AGGREGATE
Definition: parsenodes.h:2262
@ OBJECT_FUNCTION
Definition: parsenodes.h:2280
#define ACL_EXECUTE
Definition: parsenodes.h:83
FormData_pg_aggregate * Form_pg_aggregate
Definition: pg_aggregate.h:109
#define FUNC_MAX_ARGS
#define lfirst(lc)
Definition: pg_list.h:172
static int list_length(const List *l)
Definition: pg_list.h:152
#define NIL
Definition: pg_list.h:68
#define list_nth_node(type, list, n)
Definition: pg_list.h:327
FormData_pg_proc * Form_pg_proc
Definition: pg_proc.h:136
uintptr_t Datum
Definition: postgres.h:64
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
#define InvalidOid
Definition: postgres_ext.h:36
unsigned int Oid
Definition: postgres_ext.h:31
FmgrInfo finalfn
Definition: nodeAgg.h:207
bool resulttypeByVal
Definition: nodeAgg.h:225
List * aggdirectargs
Definition: nodeAgg.h:218
Aggref * aggref
Definition: nodeAgg.h:195
int16 resulttypeLen
Definition: nodeAgg.h:224
Bitmapset ** grouped_cols
Definition: nodeAgg.h:285
ExprState * evaltrans
Definition: nodeAgg.h:291
ExprState * evaltrans_cache[2][2]
Definition: nodeAgg.h:299
ExprState ** eqfunctions
Definition: nodeAgg.h:286
AggStrategy aggstrategy
Definition: nodeAgg.h:282
FmgrInfo transfn
Definition: nodeAgg.h:86
Aggref * aggref
Definition: nodeAgg.h:44
AggStatePerGroup * all_pergroups
Definition: execnodes.h:2532
AggStatePerGroup * hash_pergroup
Definition: execnodes.h:2527
List * aggs
Definition: execnodes.h:2464
ExprContext * tmpcontext
Definition: execnodes.h:2476
int hash_planned_partitions
Definition: execnodes.h:2517
HeapTuple grp_firstTuple
Definition: execnodes.h:2501
Size hash_mem_limit
Definition: execnodes.h:2515
AggStatePerTrans curpertrans
Definition: execnodes.h:2481
bool table_filled
Definition: execnodes.h:2503
int current_set
Definition: execnodes.h:2486
AggStrategy aggstrategy
Definition: execnodes.h:2467
AggSplit aggsplit
Definition: execnodes.h:2468
int projected_set
Definition: execnodes.h:2484
uint64 hash_ngroups_limit
Definition: execnodes.h:2516
bool input_done
Definition: execnodes.h:2482
AggStatePerPhase phases
Definition: execnodes.h:2494
List * all_grouped_cols
Definition: execnodes.h:2488
AggStatePerGroup * pergroups
Definition: execnodes.h:2499
AggStatePerHash perhash
Definition: execnodes.h:2526
double hashentrysize
Definition: execnodes.h:2519
int numphases
Definition: execnodes.h:2470
TupleTableSlot * hash_spill_wslot
Definition: execnodes.h:2510
AggStatePerAgg curperagg
Definition: execnodes.h:2479
TupleTableSlot * sort_slot
Definition: execnodes.h:2497
int numaggs
Definition: execnodes.h:2465
int num_hashes
Definition: execnodes.h:2504
AggStatePerAgg peragg
Definition: execnodes.h:2472
TupleTableSlot * hash_spill_rslot
Definition: execnodes.h:2509
int current_phase
Definition: execnodes.h:2471
bool agg_done
Definition: execnodes.h:2483
Definition: plannodes.h:997
AggSplit aggsplit
Definition: plannodes.h:1004
List * chain
Definition: plannodes.h:1031
long numGroups
Definition: plannodes.h:1017
List * groupingSets
Definition: plannodes.h:1028
Plan plan
Definition: plannodes.h:998
int numCols
Definition: plannodes.h:1007
uint64 transitionSpace
Definition: plannodes.h:1020
AggStrategy aggstrategy
Definition: plannodes.h:1001
Oid aggfnoid
Definition: primnodes.h:444
List * aggdirectargs
Definition: primnodes.h:465
List * args
Definition: primnodes.h:468
MemoryContext es_query_cxt
Definition: execnodes.h:667
Datum * ecxt_aggvalues
Definition: execnodes.h:274
bool * ecxt_aggnulls
Definition: execnodes.h:276
bool fn_strict
Definition: fmgr.h:61
Definition: nodes.h:129
bool outeropsset
Definition: execnodes.h:1200
const TupleTableSlotOps * outerops
Definition: execnodes.h:1192
ExprState * qual
Definition: execnodes.h:1138
bool outeropsfixed
Definition: execnodes.h:1196
EState * state
Definition: execnodes.h:1119
ExprContext * ps_ExprContext
Definition: execnodes.h:1156
ExecProcNodeMtd ExecProcNode
Definition: execnodes.h:1123
List * qual
Definition: plannodes.h:153
int plan_width
Definition: plannodes.h:135
TupleTableSlot * ss_ScanTupleSlot
Definition: execnodes.h:1567
TupleDesc tts_tupleDescriptor
Definition: tuptable.h:123
void ReleaseSysCache(HeapTuple tuple)
Definition: syscache.c:266
HeapTuple SearchSysCache1(int cacheId, Datum key1)
Definition: syscache.c:218
Datum SysCacheGetAttr(int cacheId, HeapTuple tup, AttrNumber attributeNumber, bool *isNull)
Definition: syscache.c:479

References ACL_EXECUTE, aclcheck_error(), ACLCHECK_OK, AggState::agg_done, AGG_HASHED, AGG_MIXED, AGG_PLAIN, AGG_SORTED, AggState::aggcontexts, AggStatePerAggData::aggdirectargs, Aggref::aggdirectargs, Aggref::aggfnoid, AggStatePerPhaseData::aggnode, AggStatePerHashData::aggnode, AggStatePerTransData::aggref, AggStatePerAggData::aggref, AggState::aggs, AggStatePerTransData::aggshared, AggState::aggsplit, Agg::aggsplit, AggStatePerPhaseData::aggstrategy, AggState::aggstrategy, Agg::aggstrategy, AggState::all_grouped_cols, AggState::all_pergroups, ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Aggref::args, Assert, bms_add_member(), bms_add_members(), bms_next_member(), build_aggregate_finalfn_expr(), build_hash_tables(), build_pertrans_for_aggref(), castNode, Agg::chain, CreateWorkExprContext(), AggState::curperagg, AggState::curpertrans, AggState::current_phase, AggState::current_set, DO_AGGSPLIT_COMBINE, DO_AGGSPLIT_DESERIALIZE, DO_AGGSPLIT_SERIALIZE, DO_AGGSPLIT_SKIPFINAL, ExprContext::ecxt_aggnulls, ExprContext::ecxt_aggvalues, elog, AggStatePerPhaseData::eqfunctions, ereport, errcode(), errmsg(), ERROR, EState::es_query_cxt, AggStatePerPhaseData::evaltrans, AggStatePerPhaseData::evaltrans_cache, EXEC_FLAG_BACKWARD, EXEC_FLAG_EXPLAIN_ONLY, EXEC_FLAG_MARK, EXEC_FLAG_REWIND, ExecAgg(), ExecAssignExprContext(), ExecAssignProjectionInfo(), ExecBuildAggTrans(), ExecCreateScanSlotFromOuterPlan(), ExecGetResultSlotOps(), ExecInitExprList(), ExecInitExtraTupleSlot(), ExecInitNode(), ExecInitQual(), ExecInitResultTupleSlotTL(), PlanState::ExecProcNode, execTuplesMatchPrepare(), AggStatePerAggData::finalfn, AggStatePerAggData::finalfn_oid, find_hash_columns(), fmgr_info(), fmgr_info_set_expr, FmgrInfo::fn_strict, format_type_be(), FUNC_MAX_ARGS, get_aggregate_argtypes(), get_func_name(), get_typlenbyval(), GetAggInitVal(), GETSTRUCT, GetUserId(), AggStatePerPhaseData::grouped_cols, Agg::groupingSets, AggState::grp_firstTuple, AggStatePerPhaseData::gset_lengths, hash_agg_entry_size(), hash_agg_set_limits(), AggState::hash_batches_used, AggState::hash_mem_limit, AggState::hash_metacxt, AggState::hash_ngroups_limit, AggState::hash_pergroup, AggState::hash_planned_partitions, AggState::hash_spill_rslot, AggState::hash_spill_wslot, AggState::hashcontext, AggState::hashentrysize, HeapTupleIsValid, i, initialize_phase(), initValue(), AggStatePerTransData::initValueIsNull, AggState::input_done, InvalidOid, InvokeFunctionExecuteHook, IsBinaryCoercible(), j, lcons_int(), lfirst, list_length(), list_nth_node, makeNode, Max, AggState::maxsets, NIL, AggState::num_hashes, AggState::numaggs, AggStatePerHashData::numCols, Agg::numCols, AggStatePerAggData::numFinalArgs, Agg::numGroups, AggState::numphases, AggStatePerPhaseData::numsets, AggState::numtrans, AggStatePerTransData::numTransInputs, object_aclcheck(), OBJECT_AGGREGATE, OBJECT_FUNCTION, ObjectIdGetDatum(), OidIsValid, PlanState::outerops, PlanState::outeropsfixed, PlanState::outeropsset, outerPlan, outerPlanState, palloc(), palloc0(), AggState::peragg, AggState::pergroups, AggState::perhash, AggState::pertrans, AggState::phases, PlanState::plan, Agg::plan, Plan::plan_width, AggState::projected_set, ScanState::ps, PlanState::ps_ExprContext, PlanState::qual, Plan::qual, ReleaseSysCache(), AggStatePerAggData::resulttypeByVal, AggStatePerAggData::resulttypeLen, SearchSysCache1(), select_current_set(), AggState::sort_in, AggState::sort_out, AggState::sort_slot, AggStatePerPhaseData::sortnode, AggState::ss, ScanState::ss_ScanTupleSlot, PlanState::state, SysCacheGetAttr(), AggState::table_filled, AggState::tmpcontext, AggStatePerTransData::transfn, Agg::transitionSpace, AggStatePerAggData::transno, TupleTableSlot::tts_tupleDescriptor, TTSOpsMinimalTuple, and TTSOpsVirtual.

Referenced by ExecInitNode().

◆ ExecReScanAgg()

void ExecReScanAgg ( AggState node)

Definition at line 4364 of file nodeAgg.c.

4365 {
4366  ExprContext *econtext = node->ss.ps.ps_ExprContext;
4368  Agg *aggnode = (Agg *) node->ss.ps.plan;
4369  int transno;
4370  int numGroupingSets = Max(node->maxsets, 1);
4371  int setno;
4372 
4373  node->agg_done = false;
4374 
4375  if (node->aggstrategy == AGG_HASHED)
4376  {
4377  /*
4378  * In the hashed case, if we haven't yet built the hash table then we
4379  * can just return; nothing done yet, so nothing to undo. If subnode's
4380  * chgParam is not NULL then it will be re-scanned by ExecProcNode,
4381  * else no reason to re-scan it at all.
4382  */
4383  if (!node->table_filled)
4384  return;
4385 
4386  /*
4387  * If we do have the hash table, and it never spilled, and the subplan
4388  * does not have any parameter changes, and none of our own parameter
4389  * changes affect input expressions of the aggregated functions, then
4390  * we can just rescan the existing hash table; no need to build it
4391  * again.
4392  */
4393  if (outerPlan->chgParam == NULL && !node->hash_ever_spilled &&
4394  !bms_overlap(node->ss.ps.chgParam, aggnode->aggParams))
4395  {
4397  &node->perhash[0].hashiter);
4398  select_current_set(node, 0, true);
4399  return;
4400  }
4401  }
4402 
4403  /* Make sure we have closed any open tuplesorts */
4404  for (transno = 0; transno < node->numtrans; transno++)
4405  {
4406  for (setno = 0; setno < numGroupingSets; setno++)
4407  {
4408  AggStatePerTrans pertrans = &node->pertrans[transno];
4409 
4410  if (pertrans->sortstates[setno])
4411  {
4412  tuplesort_end(pertrans->sortstates[setno]);
4413  pertrans->sortstates[setno] = NULL;
4414  }
4415  }
4416  }
4417 
4418  /*
4419  * We don't need to ReScanExprContext the output tuple context here;
4420  * ExecReScan already did it. But we do need to reset our per-grouping-set
4421  * contexts, which may have transvalues stored in them. (We use rescan
4422  * rather than just reset because transfns may have registered callbacks
4423  * that need to be run now.) For the AGG_HASHED case, see below.
4424  */
4425 
4426  for (setno = 0; setno < numGroupingSets; setno++)
4427  {
4428  ReScanExprContext(node->aggcontexts[setno]);
4429  }
4430 
4431  /* Release first tuple of group, if we have made a copy */
4432  if (node->grp_firstTuple != NULL)
4433  {
4435  node->grp_firstTuple = NULL;
4436  }
4438 
4439  /* Forget current agg values */
4440  MemSet(econtext->ecxt_aggvalues, 0, sizeof(Datum) * node->numaggs);
4441  MemSet(econtext->ecxt_aggnulls, 0, sizeof(bool) * node->numaggs);
4442 
4443  /*
4444  * With AGG_HASHED/MIXED, the hash table is allocated in a sub-context of
4445  * the hashcontext. This used to be an issue, but now, resetting a context
4446  * automatically deletes sub-contexts too.
4447  */
4448  if (node->aggstrategy == AGG_HASHED || node->aggstrategy == AGG_MIXED)
4449  {
4451 
4452  node->hash_ever_spilled = false;
4453  node->hash_spill_mode = false;
4454  node->hash_ngroups_current = 0;
4455 
4457  /* Rebuild an empty hash table */
4458  build_hash_tables(node);
4459  node->table_filled = false;
4460  /* iterator will be reset when the table is filled */
4461 
4462  hashagg_recompile_expressions(node, false, false);
4463  }
4464 
4465  if (node->aggstrategy != AGG_HASHED)
4466  {
4467  /*
4468  * Reset the per-group state (in particular, mark transvalues null)
4469  */
4470  for (setno = 0; setno < numGroupingSets; setno++)
4471  {
4472  MemSet(node->pergroups[setno], 0,
4473  sizeof(AggStatePerGroupData) * node->numaggs);
4474  }
4475 
4476  /* reset to phase 1 */
4477  initialize_phase(node, 1);
4478 
4479  node->input_done = false;
4480  node->projected_set = -1;
4481  }
4482 
4483  if (outerPlan->chgParam == NULL)
4485 }
bool bms_overlap(const Bitmapset *a, const Bitmapset *b)
Definition: bitmapset.c:582
#define MemSet(start, val, len)
Definition: c.h:1020
void ExecReScan(PlanState *node)
Definition: execAmi.c:76
#define ResetTupleHashIterator(htable, iter)
Definition: execnodes.h:846
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1434
static void hashagg_recompile_expressions(AggState *aggstate, bool minslot, bool nullcheck)
Definition: nodeAgg.c:1741
TupleHashTable hashtable
Definition: nodeAgg.h:311
TupleHashIterator hashiter
Definition: nodeAgg.h:312
bool hash_spill_mode
Definition: execnodes.h:2513
uint64 hash_ngroups_current
Definition: execnodes.h:2521
bool hash_ever_spilled
Definition: execnodes.h:2512
Bitmapset * aggParams
Definition: plannodes.h:1023
Bitmapset * chgParam
Definition: execnodes.h:1149
static TupleTableSlot * ExecClearTuple(TupleTableSlot *slot)
Definition: tuptable.h:454

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().

◆ hash_agg_entry_size()

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

Definition at line 1694 of file nodeAgg.c.

1695 {
1696  Size tupleChunkSize;
1697  Size pergroupChunkSize;
1698  Size transitionChunkSize;
1699  Size tupleSize = (MAXALIGN(SizeofMinimalTupleHeader) +
1700  tupleWidth);
1701  Size pergroupSize = numTrans * sizeof(AggStatePerGroupData);
1702 
1703  tupleChunkSize = CHUNKHDRSZ + tupleSize;
1704 
1705  if (pergroupSize > 0)
1706  pergroupChunkSize = CHUNKHDRSZ + pergroupSize;
1707  else
1708  pergroupChunkSize = 0;
1709 
1710  if (transitionSpace > 0)
1711  transitionChunkSize = CHUNKHDRSZ + transitionSpace;
1712  else
1713  transitionChunkSize = 0;
1714 
1715  return
1716  sizeof(TupleHashEntryData) +
1717  tupleChunkSize +
1718  pergroupChunkSize +
1719  transitionChunkSize;
1720 }
#define MAXALIGN(LEN)
Definition: c.h:811
struct TupleHashEntryData TupleHashEntryData
#define SizeofMinimalTupleHeader
Definition: htup_details.h:647
#define CHUNKHDRSZ
Definition: nodeAgg.c:320
struct AggStatePerGroupData AggStatePerGroupData

References CHUNKHDRSZ, MAXALIGN, and SizeofMinimalTupleHeader.

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

◆ 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 1798 of file nodeAgg.c.

1801 {
1802  int npartitions;
1803  Size partition_mem;
1804  Size hash_mem_limit = get_hash_memory_limit();
1805 
1806  /* if not expected to spill, use all of hash_mem */
1807  if (input_groups * hashentrysize <= hash_mem_limit)
1808  {
1809  if (num_partitions != NULL)
1810  *num_partitions = 0;
1811  *mem_limit = hash_mem_limit;
1812  *ngroups_limit = hash_mem_limit / hashentrysize;
1813  return;
1814  }
1815 
1816  /*
1817  * Calculate expected memory requirements for spilling, which is the size
1818  * of the buffers needed for all the tapes that need to be open at once.
1819  * Then, subtract that from the memory available for holding hash tables.
1820  */
1821  npartitions = hash_choose_num_partitions(input_groups,
1822  hashentrysize,
1823  used_bits,
1824  NULL);
1825  if (num_partitions != NULL)
1826  *num_partitions = npartitions;
1827 
1828  partition_mem =
1830  HASHAGG_WRITE_BUFFER_SIZE * npartitions;
1831 
1832  /*
1833  * Don't set the limit below 3/4 of hash_mem. In that case, we are at the
1834  * minimum number of partitions, so we aren't going to dramatically exceed
1835  * work mem anyway.
1836  */
1837  if (hash_mem_limit > 4 * partition_mem)
1838  *mem_limit = hash_mem_limit - partition_mem;
1839  else
1840  *mem_limit = hash_mem_limit * 0.75;
1841 
1842  if (*mem_limit > hashentrysize)
1843  *ngroups_limit = *mem_limit / hashentrysize;
1844  else
1845  *ngroups_limit = 1;
1846 }
#define HASHAGG_READ_BUFFER_SIZE
Definition: nodeAgg.c:305
#define HASHAGG_WRITE_BUFFER_SIZE
Definition: nodeAgg.c:306
static int hash_choose_num_partitions(double input_groups, double hashentrysize, int used_bits, int *log2_npartitions)
Definition: nodeAgg.c:1991
size_t get_hash_memory_limit(void)
Definition: nodeHash.c:3595

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

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