From 69c3936a1499b772a749ae629fc59b2d72722332 Mon Sep 17 00:00:00 2001 From: Andres Freund Date: Tue, 9 Jan 2018 13:25:38 -0800 Subject: [PATCH] Expression evaluation based aggregate transition invocation. Previously aggregate transition and combination functions were invoked by special case code in nodeAgg.c, evaluating input and filters separately using the expression evaluation machinery. That turns out to not be great for performance for several reasons: - repeated expression evaluations have some cost - the transition functions invocations are poorly predicted, as commonly there are multiple aggregates in a query, resulting in the same call-stack invoking different functions. - filter and input computation had to be done separately - the special case code made it hard to implement JITing of the whole transition function invocation Address this by building one large expression that computes input, evaluates filters, and invokes transition functions. This leads to moderate speedups in queries bottlenecked by aggregate computations, and enables large speedups for similar cases once JITing is done. There's potential for further improvement: - It'd be nice if we could simplify the somewhat expensive aggstate->all_pergroups lookups. - right now there's still an advance_transition_function invocation in nodeAgg.c, leading to some code duplication. Author: Andres Freund Discussion: https://postgr.es/m/20170901064131.tazjxwus3k2w3ybh@alap3.anarazel.de --- src/backend/executor/execExpr.c | 429 ++++++++++++- src/backend/executor/execExprInterp.c | 356 ++++++++++- src/backend/executor/nodeAgg.c | 864 +++----------------------- src/include/executor/execExpr.h | 76 ++- src/include/executor/executor.h | 4 +- src/include/executor/nodeAgg.h | 284 +++++++++ src/include/nodes/execnodes.h | 5 +- src/tools/pgindent/typedefs.list | 1 + 8 files changed, 1230 insertions(+), 789 deletions(-) diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c index 16f908037c..794573803d 100644 --- a/src/backend/executor/execExpr.c +++ b/src/backend/executor/execExpr.c @@ -43,6 +43,7 @@ #include "optimizer/planner.h" #include "pgstat.h" #include "utils/builtins.h" +#include "utils/datum.h" #include "utils/lsyscache.h" #include "utils/typcache.h" @@ -61,6 +62,7 @@ static void ExecInitFunc(ExprEvalStep *scratch, Expr *node, List *args, Oid funcid, Oid inputcollid, ExprState *state); static void ExecInitExprSlots(ExprState *state, Node *node); +static void ExecPushExprSlots(ExprState *state, LastAttnumInfo *info); static bool get_last_attnums_walker(Node *node, LastAttnumInfo *info); static void ExecInitWholeRowVar(ExprEvalStep *scratch, Var *variable, ExprState *state); @@ -71,6 +73,10 @@ static bool isAssignmentIndirectionExpr(Expr *expr); static void ExecInitCoerceToDomain(ExprEvalStep *scratch, CoerceToDomain *ctest, ExprState *state, Datum *resv, bool *resnull); +static void ExecBuildAggTransCall(ExprState *state, AggState *aggstate, + ExprEvalStep *scratch, + FunctionCallInfo fcinfo, AggStatePerTrans pertrans, + int transno, int setno, int setoff, bool ishash); /* @@ -2250,30 +2256,42 @@ static void ExecInitExprSlots(ExprState *state, Node *node) { LastAttnumInfo info = {0, 0, 0}; - ExprEvalStep scratch; /* * Figure out which attributes we're going to need. */ get_last_attnums_walker(node, &info); + ExecPushExprSlots(state, &info); +} + +/* + * Add steps deforming the ExprState's inner/out/scan slots as much as + * indicated by info. This is useful when building an ExprState covering more + * than one expression. + */ +static void +ExecPushExprSlots(ExprState *state, LastAttnumInfo *info) +{ + ExprEvalStep scratch; + /* Emit steps as needed */ - if (info.last_inner > 0) + if (info->last_inner > 0) { scratch.opcode = EEOP_INNER_FETCHSOME; - scratch.d.fetch.last_var = info.last_inner; + scratch.d.fetch.last_var = info->last_inner; ExprEvalPushStep(state, &scratch); } - if (info.last_outer > 0) + if (info->last_outer > 0) { scratch.opcode = EEOP_OUTER_FETCHSOME; - scratch.d.fetch.last_var = info.last_outer; + scratch.d.fetch.last_var = info->last_outer; ExprEvalPushStep(state, &scratch); } - if (info.last_scan > 0) + if (info->last_scan > 0) { scratch.opcode = EEOP_SCAN_FETCHSOME; - scratch.d.fetch.last_var = info.last_scan; + scratch.d.fetch.last_var = info->last_scan; ExprEvalPushStep(state, &scratch); } } @@ -2775,3 +2793,400 @@ ExecInitCoerceToDomain(ExprEvalStep *scratch, CoerceToDomain *ctest, } } } + +/* + * Build transition/combine function invocations for all aggregate transition + * / combination function invocations in a grouping sets phase. This has to + * invoke all sort based transitions in a phase (if doSort is true), all hash + * based transitions (if doHash is true), or both (both true). + * + * The resulting expression will, for each set of transition values, first + * check for filters, evaluate aggregate input, check that that input is not + * NULL for a strict transition function, and then finally invoke the + * transition for each of the concurrently computed grouping sets. + */ +ExprState * +ExecBuildAggTrans(AggState *aggstate, AggStatePerPhase phase, + bool doSort, bool doHash) +{ + ExprState *state = makeNode(ExprState); + PlanState *parent = &aggstate->ss.ps; + ExprEvalStep scratch; + int transno = 0; + int setoff = 0; + bool isCombine = DO_AGGSPLIT_COMBINE(aggstate->aggsplit); + LastAttnumInfo deform = {0, 0, 0}; + + state->expr = (Expr *) aggstate; + state->parent = parent; + + scratch.resvalue = &state->resvalue; + scratch.resnull = &state->resnull; + + /* + * First figure out which slots, and how many columns from each, we're + * going to need. + */ + for (transno = 0; transno < aggstate->numtrans; transno++) + { + AggStatePerTrans pertrans = &aggstate->pertrans[transno]; + + get_last_attnums_walker((Node *) pertrans->aggref->aggdirectargs, + &deform); + get_last_attnums_walker((Node *) pertrans->aggref->args, + &deform); + get_last_attnums_walker((Node *) pertrans->aggref->aggorder, + &deform); + get_last_attnums_walker((Node *) pertrans->aggref->aggdistinct, + &deform); + get_last_attnums_walker((Node *) pertrans->aggref->aggfilter, + &deform); + } + ExecPushExprSlots(state, &deform); + + /* + * Emit instructions for each transition value / grouping set combination. + */ + for (transno = 0; transno < aggstate->numtrans; transno++) + { + AggStatePerTrans pertrans = &aggstate->pertrans[transno]; + int numInputs = pertrans->numInputs; + int argno; + int setno; + FunctionCallInfo trans_fcinfo = &pertrans->transfn_fcinfo; + ListCell *arg; + ListCell *bail; + List *adjust_bailout = NIL; + bool *strictnulls = NULL; + + /* + * If filter present, emit. Do so before evaluating the input, to + * avoid potentially unneeded computations, or even worse, unintended + * side-effects. When combining, all the necessary filtering has + * already been done. + */ + if (pertrans->aggref->aggfilter && !isCombine) + { + /* evaluate filter expression */ + ExecInitExprRec(pertrans->aggref->aggfilter, state, + &state->resvalue, &state->resnull); + /* and jump out if false */ + scratch.opcode = EEOP_JUMP_IF_NOT_TRUE; + scratch.d.jump.jumpdone = -1; /* adjust later */ + ExprEvalPushStep(state, &scratch); + adjust_bailout = lappend_int(adjust_bailout, + state->steps_len - 1); + } + + /* + * Evaluate arguments to aggregate/combine function. + */ + argno = 0; + if (isCombine) + { + /* + * Combining two aggregate transition values. Instead of directly + * coming from a tuple the input is a, potentially deserialized, + * transition value. + */ + TargetEntry *source_tle; + + Assert(pertrans->numSortCols == 0); + Assert(list_length(pertrans->aggref->args) == 1); + + strictnulls = trans_fcinfo->argnull + 1; + source_tle = (TargetEntry *) linitial(pertrans->aggref->args); + + /* + * deserialfn_oid will be set if we must deserialize the input + * state before calling the combine function. + */ + if (!OidIsValid(pertrans->deserialfn_oid)) + { + /* + * Start from 1, since the 0th arg will be the transition + * value + */ + ExecInitExprRec(source_tle->expr, state, + &trans_fcinfo->arg[argno + 1], + &trans_fcinfo->argnull[argno + 1]); + } + else + { + FunctionCallInfo ds_fcinfo = &pertrans->deserialfn_fcinfo; + + /* evaluate argument */ + ExecInitExprRec(source_tle->expr, state, + &ds_fcinfo->arg[0], + &ds_fcinfo->argnull[0]); + + /* Dummy second argument for type-safety reasons */ + ds_fcinfo->arg[1] = PointerGetDatum(NULL); + ds_fcinfo->argnull[1] = false; + + /* + * Don't call a strict deserialization function with NULL + * input + */ + if (pertrans->deserialfn.fn_strict) + scratch.opcode = EEOP_AGG_STRICT_DESERIALIZE; + else + scratch.opcode = EEOP_AGG_DESERIALIZE; + + scratch.d.agg_deserialize.aggstate = aggstate; + scratch.d.agg_deserialize.fcinfo_data = ds_fcinfo; + scratch.d.agg_deserialize.jumpnull = -1; /* adjust later */ + scratch.resvalue = &trans_fcinfo->arg[argno + 1]; + scratch.resnull = &trans_fcinfo->argnull[argno + 1]; + + ExprEvalPushStep(state, &scratch); + adjust_bailout = lappend_int(adjust_bailout, + state->steps_len - 1); + + /* restore normal settings of scratch fields */ + scratch.resvalue = &state->resvalue; + scratch.resnull = &state->resnull; + } + argno++; + } + else if (pertrans->numSortCols == 0) + { + /* + * Normal transition function without ORDER BY / DISTINCT. + */ + strictnulls = trans_fcinfo->argnull + 1; + + foreach(arg, pertrans->aggref->args) + { + TargetEntry *source_tle = (TargetEntry *) lfirst(arg); + + /* + * Start from 1, since the 0th arg will be the transition + * value + */ + ExecInitExprRec(source_tle->expr, state, + &trans_fcinfo->arg[argno + 1], + &trans_fcinfo->argnull[argno + 1]); + argno++; + } + } + else if (pertrans->numInputs == 1) + { + /* + * DISTINCT and/or ORDER BY case, with a single column sorted on. + */ + TargetEntry *source_tle = + (TargetEntry *) linitial(pertrans->aggref->args); + + Assert(list_length(pertrans->aggref->args) == 1); + + ExecInitExprRec(source_tle->expr, state, + &state->resvalue, + &state->resnull); + strictnulls = &state->resnull; + argno++; + } + else + { + /* + * DISTINCT and/or ORDER BY case, with multiple columns sorted on. + */ + Datum *values = pertrans->sortslot->tts_values; + bool *nulls = pertrans->sortslot->tts_isnull; + + strictnulls = nulls; + + foreach(arg, pertrans->aggref->args) + { + TargetEntry *source_tle = (TargetEntry *) lfirst(arg); + + ExecInitExprRec(source_tle->expr, state, + &values[argno], &nulls[argno]); + argno++; + } + } + Assert(numInputs == argno); + + /* + * For a strict transfn, nothing happens when there's a NULL input; we + * just keep the prior transValue. This is true for both plain and + * sorted/distinct aggregates. + */ + if (trans_fcinfo->flinfo->fn_strict && numInputs > 0) + { + scratch.opcode = EEOP_AGG_STRICT_INPUT_CHECK; + scratch.d.agg_strict_input_check.nulls = strictnulls; + scratch.d.agg_strict_input_check.jumpnull = -1; /* adjust later */ + scratch.d.agg_strict_input_check.nargs = numInputs; + ExprEvalPushStep(state, &scratch); + adjust_bailout = lappend_int(adjust_bailout, + state->steps_len - 1); + } + + /* + * Call transition function (once for each concurrently evaluated + * grouping set). Do so for both sort and hash based computations, as + * applicable. + */ + setoff = 0; + if (doSort) + { + int processGroupingSets = Max(phase->numsets, 1); + + for (setno = 0; setno < processGroupingSets; setno++) + { + ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo, + pertrans, transno, setno, setoff, false); + setoff++; + } + } + + if (doHash) + { + int numHashes = aggstate->num_hashes; + + /* in MIXED mode, there'll be preceding transition values */ + if (aggstate->aggstrategy != AGG_HASHED) + setoff = aggstate->maxsets; + else + setoff = 0; + + for (setno = 0; setno < numHashes; setno++) + { + ExecBuildAggTransCall(state, aggstate, &scratch, trans_fcinfo, + pertrans, transno, setno, setoff, true); + setoff++; + } + } + + /* adjust early bail out jump target(s) */ + foreach(bail, adjust_bailout) + { + ExprEvalStep *as = &state->steps[lfirst_int(bail)]; + + if (as->opcode == EEOP_JUMP_IF_NOT_TRUE) + { + Assert(as->d.jump.jumpdone == -1); + as->d.jump.jumpdone = state->steps_len; + } + else if (as->opcode == EEOP_AGG_STRICT_INPUT_CHECK) + { + Assert(as->d.agg_strict_input_check.jumpnull == -1); + as->d.agg_strict_input_check.jumpnull = state->steps_len; + } + else if (as->opcode == EEOP_AGG_STRICT_DESERIALIZE) + { + Assert(as->d.agg_deserialize.jumpnull == -1); + as->d.agg_deserialize.jumpnull = state->steps_len; + } + } + } + + scratch.resvalue = NULL; + scratch.resnull = NULL; + scratch.opcode = EEOP_DONE; + ExprEvalPushStep(state, &scratch); + + ExecReadyExpr(state); + + return state; +} + +/* + * Build transition/combine function invocation for a single transition + * value. This is separated from ExecBuildAggTrans() because there are + * multiple callsites (hash and sort in some grouping set cases). + */ +static void +ExecBuildAggTransCall(ExprState *state, AggState *aggstate, + ExprEvalStep *scratch, + FunctionCallInfo fcinfo, AggStatePerTrans pertrans, + int transno, int setno, int setoff, bool ishash) +{ + int adjust_init_jumpnull = -1; + int adjust_strict_jumpnull = -1; + ExprContext *aggcontext; + + if (ishash) + aggcontext = aggstate->hashcontext; + else + aggcontext = aggstate->aggcontexts[setno]; + + /* + * If the initial value for the transition state doesn't exist in the + * pg_aggregate table then we will let the first non-NULL value returned + * from the outer procNode become the initial value. (This is useful for + * aggregates like max() and min().) The noTransValue flag signals that we + * still need to do this. + */ + if (pertrans->numSortCols == 0 && + fcinfo->flinfo->fn_strict && + pertrans->initValueIsNull) + { + scratch->opcode = EEOP_AGG_INIT_TRANS; + scratch->d.agg_init_trans.aggstate = aggstate; + scratch->d.agg_init_trans.pertrans = pertrans; + scratch->d.agg_init_trans.setno = setno; + scratch->d.agg_init_trans.setoff = setoff; + scratch->d.agg_init_trans.transno = transno; + scratch->d.agg_init_trans.aggcontext = aggcontext; + scratch->d.agg_init_trans.jumpnull = -1; /* adjust later */ + ExprEvalPushStep(state, scratch); + + /* see comment about jumping out below */ + adjust_init_jumpnull = state->steps_len - 1; + } + + if (pertrans->numSortCols == 0 && + fcinfo->flinfo->fn_strict) + { + scratch->opcode = EEOP_AGG_STRICT_TRANS_CHECK; + scratch->d.agg_strict_trans_check.aggstate = aggstate; + scratch->d.agg_strict_trans_check.setno = setno; + scratch->d.agg_strict_trans_check.setoff = setoff; + scratch->d.agg_strict_trans_check.transno = transno; + scratch->d.agg_strict_trans_check.jumpnull = -1; /* adjust later */ + ExprEvalPushStep(state, scratch); + + /* + * Note, we don't push into adjust_bailout here - those jump to the + * end of all transition value computations. Here a single transition + * value is NULL, so just skip processing the individual value. + */ + adjust_strict_jumpnull = state->steps_len - 1; + } + + /* invoke appropriate transition implementation */ + if (pertrans->numSortCols == 0 && pertrans->transtypeByVal) + scratch->opcode = EEOP_AGG_PLAIN_TRANS_BYVAL; + else if (pertrans->numSortCols == 0) + scratch->opcode = EEOP_AGG_PLAIN_TRANS; + else if (pertrans->numInputs == 1) + scratch->opcode = EEOP_AGG_ORDERED_TRANS_DATUM; + else + scratch->opcode = EEOP_AGG_ORDERED_TRANS_TUPLE; + + scratch->d.agg_trans.aggstate = aggstate; + scratch->d.agg_trans.pertrans = pertrans; + scratch->d.agg_trans.setno = setno; + scratch->d.agg_trans.setoff = setoff; + scratch->d.agg_trans.transno = transno; + scratch->d.agg_trans.aggcontext = aggcontext; + ExprEvalPushStep(state, scratch); + + /* adjust jumps so they jump till after transition invocation */ + if (adjust_init_jumpnull != -1) + { + ExprEvalStep *as = &state->steps[adjust_init_jumpnull]; + + Assert(as->d.agg_init_trans.jumpnull == -1); + as->d.agg_init_trans.jumpnull = state->steps_len; + } + if (adjust_strict_jumpnull != -1) + { + ExprEvalStep *as = &state->steps[adjust_strict_jumpnull]; + + Assert(as->d.agg_strict_trans_check.jumpnull == -1); + as->d.agg_strict_trans_check.jumpnull = state->steps_len; + } +} diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c index 2e88417265..f646fd9c51 100644 --- a/src/backend/executor/execExprInterp.c +++ b/src/backend/executor/execExprInterp.c @@ -62,12 +62,14 @@ #include "executor/execExpr.h" #include "executor/nodeSubplan.h" #include "funcapi.h" +#include "utils/memutils.h" #include "miscadmin.h" #include "nodes/nodeFuncs.h" #include "parser/parsetree.h" #include "pgstat.h" #include "utils/builtins.h" #include "utils/date.h" +#include "utils/datum.h" #include "utils/lsyscache.h" #include "utils/timestamp.h" #include "utils/typcache.h" @@ -99,11 +101,12 @@ typedef struct ExprEvalOpLookup { const void *opcode; - ExprEvalOp op; + ExprEvalOp op; } ExprEvalOpLookup; /* to make dispatch_table accessible outside ExecInterpExpr() */ static const void **dispatch_table = NULL; + /* jump target -> opcode lookup table */ static ExprEvalOpLookup reverse_dispatch_table[EEOP_LAST]; @@ -379,6 +382,15 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull) &&CASE_EEOP_WINDOW_FUNC, &&CASE_EEOP_SUBPLAN, &&CASE_EEOP_ALTERNATIVE_SUBPLAN, + &&CASE_EEOP_AGG_STRICT_DESERIALIZE, + &&CASE_EEOP_AGG_DESERIALIZE, + &&CASE_EEOP_AGG_STRICT_INPUT_CHECK, + &&CASE_EEOP_AGG_INIT_TRANS, + &&CASE_EEOP_AGG_STRICT_TRANS_CHECK, + &&CASE_EEOP_AGG_PLAIN_TRANS_BYVAL, + &&CASE_EEOP_AGG_PLAIN_TRANS, + &&CASE_EEOP_AGG_ORDERED_TRANS_DATUM, + &&CASE_EEOP_AGG_ORDERED_TRANS_TUPLE, &&CASE_EEOP_LAST }; @@ -1514,6 +1526,235 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull) EEO_NEXT(); } + /* evaluate a strict aggregate deserialization function */ + EEO_CASE(EEOP_AGG_STRICT_DESERIALIZE) + { + bool *argnull = op->d.agg_deserialize.fcinfo_data->argnull; + + /* Don't call a strict deserialization function with NULL input */ + if (argnull[0]) + EEO_JUMP(op->d.agg_deserialize.jumpnull); + + /* fallthrough */ + } + + /* evaluate aggregate deserialization function (non-strict portion) */ + EEO_CASE(EEOP_AGG_DESERIALIZE) + { + FunctionCallInfo fcinfo = op->d.agg_deserialize.fcinfo_data; + AggState *aggstate = op->d.agg_deserialize.aggstate; + MemoryContext oldContext; + + /* + * We run the deserialization functions in per-input-tuple memory + * context. + */ + oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory); + fcinfo->isnull = false; + *op->resvalue = FunctionCallInvoke(fcinfo); + *op->resnull = fcinfo->isnull; + MemoryContextSwitchTo(oldContext); + + EEO_NEXT(); + } + + /* + * Check that a strict aggregate transition / combination function's + * input is not NULL. + */ + EEO_CASE(EEOP_AGG_STRICT_INPUT_CHECK) + { + int argno; + bool *nulls = op->d.agg_strict_input_check.nulls; + int nargs = op->d.agg_strict_input_check.nargs; + + for (argno = 0; argno < nargs; argno++) + { + if (nulls[argno]) + EEO_JUMP(op->d.agg_strict_input_check.jumpnull); + } + EEO_NEXT(); + } + + /* + * Initialize an aggregate's first value if necessary. + */ + EEO_CASE(EEOP_AGG_INIT_TRANS) + { + AggState *aggstate; + AggStatePerGroup pergroup; + + aggstate = op->d.agg_init_trans.aggstate; + pergroup = &aggstate->all_pergroups + [op->d.agg_init_trans.setoff] + [op->d.agg_init_trans.transno]; + + /* If transValue has not yet been initialized, do so now. */ + if (pergroup->noTransValue) + { + AggStatePerTrans pertrans = op->d.agg_init_trans.pertrans; + + aggstate->curaggcontext = op->d.agg_init_trans.aggcontext; + aggstate->current_set = op->d.agg_init_trans.setno; + + ExecAggInitGroup(aggstate, pertrans, pergroup); + + /* copied trans value from input, done this round */ + EEO_JUMP(op->d.agg_init_trans.jumpnull); + } + + EEO_NEXT(); + } + + /* check that a strict aggregate's input isn't NULL */ + EEO_CASE(EEOP_AGG_STRICT_TRANS_CHECK) + { + AggState *aggstate; + AggStatePerGroup pergroup; + + aggstate = op->d.agg_strict_trans_check.aggstate; + pergroup = &aggstate->all_pergroups + [op->d.agg_strict_trans_check.setoff] + [op->d.agg_strict_trans_check.transno]; + + if (unlikely(pergroup->transValueIsNull)) + EEO_JUMP(op->d.agg_strict_trans_check.jumpnull); + + EEO_NEXT(); + } + + /* + * Evaluate aggregate transition / combine function that has a + * by-value transition type. That's a seperate case from the + * by-reference implementation because it's a bit simpler. + */ + EEO_CASE(EEOP_AGG_PLAIN_TRANS_BYVAL) + { + AggState *aggstate; + AggStatePerTrans pertrans; + AggStatePerGroup pergroup; + FunctionCallInfo fcinfo; + MemoryContext oldContext; + Datum newVal; + + aggstate = op->d.agg_trans.aggstate; + pertrans = op->d.agg_trans.pertrans; + + pergroup = &aggstate->all_pergroups + [op->d.agg_trans.setoff] + [op->d.agg_trans.transno]; + + Assert(pertrans->transtypeByVal); + + fcinfo = &pertrans->transfn_fcinfo; + + /* cf. select_current_set() */ + aggstate->curaggcontext = op->d.agg_trans.aggcontext; + aggstate->current_set = op->d.agg_trans.setno; + + /* set up aggstate->curpertrans for AggGetAggref() */ + aggstate->curpertrans = pertrans; + + /* invoke transition function in per-tuple context */ + oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory); + + fcinfo->arg[0] = pergroup->transValue; + fcinfo->argnull[0] = pergroup->transValueIsNull; + fcinfo->isnull = false; /* just in case transfn doesn't set it */ + + newVal = FunctionCallInvoke(fcinfo); + + pergroup->transValue = newVal; + pergroup->transValueIsNull = fcinfo->isnull; + + MemoryContextSwitchTo(oldContext); + + EEO_NEXT(); + } + + /* + * Evaluate aggregate transition / combine function that has a + * by-reference transition type. + * + * Could optimize a bit further by splitting off by-reference + * fixed-length types, but currently that doesn't seem worth it. + */ + EEO_CASE(EEOP_AGG_PLAIN_TRANS) + { + AggState *aggstate; + AggStatePerTrans pertrans; + AggStatePerGroup pergroup; + FunctionCallInfo fcinfo; + MemoryContext oldContext; + Datum newVal; + + aggstate = op->d.agg_trans.aggstate; + pertrans = op->d.agg_trans.pertrans; + + pergroup = &aggstate->all_pergroups + [op->d.agg_trans.setoff] + [op->d.agg_trans.transno]; + + Assert(!pertrans->transtypeByVal); + + fcinfo = &pertrans->transfn_fcinfo; + + /* cf. select_current_set() */ + aggstate->curaggcontext = op->d.agg_trans.aggcontext; + aggstate->current_set = op->d.agg_trans.setno; + + /* set up aggstate->curpertrans for AggGetAggref() */ + aggstate->curpertrans = pertrans; + + /* invoke transition function in per-tuple context */ + oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory); + + fcinfo->arg[0] = pergroup->transValue; + fcinfo->argnull[0] = pergroup->transValueIsNull; + fcinfo->isnull = false; /* just in case transfn doesn't set it */ + + newVal = FunctionCallInvoke(fcinfo); + + /* + * For pass-by-ref datatype, must copy the new value into + * aggcontext and free the prior transValue. But if transfn + * returned a pointer to its first input, we don't need to do + * anything. Also, if transfn returned a pointer to a R/W + * expanded object that is already a child of the aggcontext, + * assume we can adopt that value without copying it. + */ + if (DatumGetPointer(newVal) != DatumGetPointer(pergroup->transValue)) + newVal = ExecAggTransReparent(aggstate, pertrans, + newVal, fcinfo->isnull, + pergroup->transValue, + pergroup->transValueIsNull); + + pergroup->transValue = newVal; + pergroup->transValueIsNull = fcinfo->isnull; + + MemoryContextSwitchTo(oldContext); + + EEO_NEXT(); + } + + /* process single-column ordered aggregate datum */ + EEO_CASE(EEOP_AGG_ORDERED_TRANS_DATUM) + { + /* too complex for an inline implementation */ + ExecEvalAggOrderedTransDatum(state, op, econtext); + + EEO_NEXT(); + } + + /* process multi-column ordered aggregate tuple */ + EEO_CASE(EEOP_AGG_ORDERED_TRANS_TUPLE) + { + /* too complex for an inline implementation */ + ExecEvalAggOrderedTransTuple(state, op, econtext); + + EEO_NEXT(); + } + EEO_CASE(EEOP_LAST) { /* unreachable */ @@ -1536,8 +1777,8 @@ Datum ExecInterpExprStillValid(ExprState *state, ExprContext *econtext, bool *isNull) { /* - * First time through, check whether attribute matches Var. Might - * not be ok anymore, due to schema changes. + * First time through, check whether attribute matches Var. Might not be + * ok anymore, due to schema changes. */ CheckExprStillValid(state, econtext); @@ -1555,7 +1796,7 @@ ExecInterpExprStillValid(ExprState *state, ExprContext *econtext, bool *isNull) void CheckExprStillValid(ExprState *state, ExprContext *econtext) { - int i = 0; + int i = 0; TupleTableSlot *innerslot; TupleTableSlot *outerslot; TupleTableSlot *scanslot; @@ -1564,9 +1805,9 @@ CheckExprStillValid(ExprState *state, ExprContext *econtext) outerslot = econtext->ecxt_outertuple; scanslot = econtext->ecxt_scantuple; - for (i = 0; i < state->steps_len;i++) + for (i = 0; i < state->steps_len; i++) { - ExprEvalStep *op = &state->steps[i]; + ExprEvalStep *op = &state->steps[i]; switch (ExecEvalStepOp(state, op)) { @@ -1859,7 +2100,7 @@ ExecJustApplyFuncToCase(ExprState *state, ExprContext *econtext, bool *isnull) * ExecEvalStepOp() in the threaded dispatch case. */ static int -dispatch_compare_ptr(const void* a, const void *b) +dispatch_compare_ptr(const void *a, const void *b) { const ExprEvalOpLookup *la = (const ExprEvalOpLookup *) a; const ExprEvalOpLookup *lb = (const ExprEvalOpLookup *) b; @@ -1896,7 +2137,7 @@ ExecInitInterpreter(void) /* make it bsearch()able */ qsort(reverse_dispatch_table, - EEOP_LAST /* nmembers */, + EEOP_LAST /* nmembers */ , sizeof(ExprEvalOpLookup), dispatch_compare_ptr); } @@ -1918,13 +2159,13 @@ ExecEvalStepOp(ExprState *state, ExprEvalStep *op) ExprEvalOpLookup key; ExprEvalOpLookup *res; - key.opcode = (void *) op->opcode; + key.opcode = (void *) op->opcode; res = bsearch(&key, reverse_dispatch_table, - EEOP_LAST /* nmembers */, + EEOP_LAST /* nmembers */ , sizeof(ExprEvalOpLookup), dispatch_compare_ptr); - Assert(res); /* unknown ops shouldn't get looked up */ + Assert(res); /* unknown ops shouldn't get looked up */ return res->op; } #endif @@ -3691,3 +3932,96 @@ ExecEvalWholeRowVar(ExprState *state, ExprEvalStep *op, ExprContext *econtext) *op->resvalue = PointerGetDatum(dtuple); *op->resnull = false; } + +/* + * Transition value has not been initialized. This is the first non-NULL input + * value for a group. We use it as the initial value for transValue. + */ +void +ExecAggInitGroup(AggState *aggstate, AggStatePerTrans pertrans, AggStatePerGroup pergroup) +{ + FunctionCallInfo fcinfo = &pertrans->transfn_fcinfo; + MemoryContext oldContext; + + /* + * We must copy the datum into aggcontext if it is pass-by-ref. We do not + * need to pfree the old transValue, since it's NULL. (We already checked + * that the agg's input type is binary-compatible with its transtype, so + * straight copy here is OK.) + */ + oldContext = MemoryContextSwitchTo( + aggstate->curaggcontext->ecxt_per_tuple_memory); + pergroup->transValue = datumCopy(fcinfo->arg[1], + pertrans->transtypeByVal, + pertrans->transtypeLen); + pergroup->transValueIsNull = false; + pergroup->noTransValue = false; + MemoryContextSwitchTo(oldContext); +} + +/* + * Ensure that the current transition value is a child of the aggcontext, + * rather than the per-tuple context. + * + * NB: This can change the current memory context. + */ +Datum +ExecAggTransReparent(AggState *aggstate, AggStatePerTrans pertrans, + Datum newValue, bool newValueIsNull, + Datum oldValue, bool oldValueIsNull) +{ + if (!newValueIsNull) + { + MemoryContextSwitchTo(aggstate->curaggcontext->ecxt_per_tuple_memory); + if (DatumIsReadWriteExpandedObject(newValue, + false, + pertrans->transtypeLen) && + MemoryContextGetParent(DatumGetEOHP(newValue)->eoh_context) == CurrentMemoryContext) + /* do nothing */ ; + else + newValue = datumCopy(newValue, + pertrans->transtypeByVal, + pertrans->transtypeLen); + } + if (!oldValueIsNull) + { + if (DatumIsReadWriteExpandedObject(oldValue, + false, + pertrans->transtypeLen)) + DeleteExpandedObject(oldValue); + else + pfree(DatumGetPointer(oldValue)); + } + + return newValue; +} + +/* + * Invoke ordered transition function, with a datum argument. + */ +void +ExecEvalAggOrderedTransDatum(ExprState *state, ExprEvalStep *op, + ExprContext *econtext) +{ + AggStatePerTrans pertrans = op->d.agg_trans.pertrans; + int setno = op->d.agg_trans.setno; + + tuplesort_putdatum(pertrans->sortstates[setno], + *op->resvalue, *op->resnull); +} + +/* + * Invoke ordered transition function, with a tuple argument. + */ +void +ExecEvalAggOrderedTransTuple(ExprState *state, ExprEvalStep *op, + ExprContext *econtext) +{ + AggStatePerTrans pertrans = op->d.agg_trans.pertrans; + int setno = op->d.agg_trans.setno; + + ExecClearTuple(pertrans->sortslot); + pertrans->sortslot->tts_nvalid = pertrans->numInputs; + ExecStoreVirtualTuple(pertrans->sortslot); + tuplesort_puttupleslot(pertrans->sortstates[setno], pertrans->sortslot); +} diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c index 46ee880415..061acad80f 100644 --- a/src/backend/executor/nodeAgg.c +++ b/src/backend/executor/nodeAgg.c @@ -90,7 +90,7 @@ * but in the aggregate case we know the left input is either the initial * transition value or a previous function result, and in either case its * value need not be preserved. See int8inc() for an example. Notice that - * advance_transition_function() is coded to avoid a data copy step when + * the EEOP_AGG_PLAIN_TRANS step is coded to avoid a data copy step when * the previous transition value pointer is returned. It is also possible * to avoid repeated data copying when the transition value is an expanded * object: to do that, the transition function must take care to return @@ -194,6 +194,16 @@ * transition values. hashcontext is the single context created to support * all hash tables. * + * Transition / Combine function invocation: + * + * For performance reasons transition functions, including combine + * functions, aren't invoked one-by-one from nodeAgg.c after computing + * arguments using the expression evaluation engine. Instead + * ExecBuildAggTrans() builds one large expression that does both argument + * evaluation and transition function invocation. That avoids performance + * issues due to repeated uses of expression evaluation, complications due + * to filter expressions having to be evaluated early, and allows to JIT + * the entire expression into one native function. * * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California @@ -229,305 +239,6 @@ #include "utils/datum.h" -/* - * AggStatePerTransData - per aggregate state value information - * - * Working state for updating the aggregate's state value, by calling the - * transition function with an input row. This struct does not store the - * information needed to produce the final aggregate result from the transition - * state, that's stored in AggStatePerAggData instead. This separation allows - * multiple aggregate results to be produced from a single state value. - */ -typedef struct AggStatePerTransData -{ - /* - * These values are set up during ExecInitAgg() and do not change - * thereafter: - */ - - /* - * Link to an Aggref expr this state value is for. - * - * There can be multiple Aggref's sharing the same state value, so long as - * the inputs and transition functions are identical and the final - * functions are not read-write. This points to the first one of them. - */ - Aggref *aggref; - - /* - * Is this state value actually being shared by more than one Aggref? - */ - bool aggshared; - - /* - * Number of aggregated input columns. This includes ORDER BY expressions - * in both the plain-agg and ordered-set cases. Ordered-set direct args - * are not counted, though. - */ - int numInputs; - - /* - * Number of aggregated input columns to pass to the transfn. This - * includes the ORDER BY columns for ordered-set aggs, but not for plain - * aggs. (This doesn't count the transition state value!) - */ - int numTransInputs; - - /* - * At each input row, we perform a single ExecProject call to evaluate all - * argument expressions that will certainly be needed at this row; that - * includes this aggregate's filter expression if it has one, or its - * regular argument expressions (including any ORDER BY columns) if it - * doesn't. inputoff is the starting index of this aggregate's required - * expressions in the resulting tuple. - */ - int inputoff; - - /* Oid of the state transition or combine function */ - Oid transfn_oid; - - /* Oid of the serialization function or InvalidOid */ - Oid serialfn_oid; - - /* Oid of the deserialization function or InvalidOid */ - Oid deserialfn_oid; - - /* Oid of state value's datatype */ - Oid aggtranstype; - - /* - * fmgr lookup data for transition function or combine function. Note in - * particular that the fn_strict flag is kept here. - */ - FmgrInfo transfn; - - /* fmgr lookup data for serialization function */ - FmgrInfo serialfn; - - /* fmgr lookup data for deserialization function */ - FmgrInfo deserialfn; - - /* Input collation derived for aggregate */ - Oid aggCollation; - - /* number of sorting columns */ - int numSortCols; - - /* number of sorting columns to consider in DISTINCT comparisons */ - /* (this is either zero or the same as numSortCols) */ - int numDistinctCols; - - /* deconstructed sorting information (arrays of length numSortCols) */ - AttrNumber *sortColIdx; - Oid *sortOperators; - Oid *sortCollations; - bool *sortNullsFirst; - - /* - * fmgr lookup data for input columns' equality operators --- only - * set/used when aggregate has DISTINCT flag. Note that these are in - * order of sort column index, not parameter index. - */ - FmgrInfo *equalfns; /* array of length numDistinctCols */ - - /* - * initial value from pg_aggregate entry - */ - Datum initValue; - bool initValueIsNull; - - /* - * We need the len and byval info for the agg's input and transition data - * types in order to know how to copy/delete values. - * - * Note that the info for the input type is used only when handling - * DISTINCT aggs with just one argument, so there is only one input type. - */ - int16 inputtypeLen, - transtypeLen; - bool inputtypeByVal, - transtypeByVal; - - /* - * Stuff for evaluation of aggregate inputs, when they must be evaluated - * separately because there's a FILTER expression. In such cases we will - * create a sortslot and the result will be stored there, whether or not - * we're actually sorting. - */ - ProjectionInfo *evalproj; /* projection machinery */ - - /* - * Slots for holding the evaluated input arguments. These are set up - * during ExecInitAgg() and then used for each input row requiring either - * FILTER or ORDER BY/DISTINCT processing. - */ - TupleTableSlot *sortslot; /* current input tuple */ - TupleTableSlot *uniqslot; /* used for multi-column DISTINCT */ - TupleDesc sortdesc; /* descriptor of input tuples */ - - /* - * These values are working state that is initialized at the start of an - * input tuple group and updated for each input tuple. - * - * For a simple (non DISTINCT/ORDER BY) aggregate, we just feed the input - * values straight to the transition function. If it's DISTINCT or - * requires ORDER BY, we pass the input values into a Tuplesort object; - * then at completion of the input tuple group, we scan the sorted values, - * eliminate duplicates if needed, and run the transition function on the - * rest. - * - * We need a separate tuplesort for each grouping set. - */ - - Tuplesortstate **sortstates; /* sort objects, if DISTINCT or ORDER BY */ - - /* - * This field is a pre-initialized FunctionCallInfo struct used for - * calling this aggregate's transfn. We save a few cycles per row by not - * re-initializing the unchanging fields; which isn't much, but it seems - * worth the extra space consumption. - */ - FunctionCallInfoData transfn_fcinfo; - - /* Likewise for serialization and deserialization functions */ - FunctionCallInfoData serialfn_fcinfo; - - FunctionCallInfoData deserialfn_fcinfo; -} AggStatePerTransData; - -/* - * AggStatePerAggData - per-aggregate information - * - * This contains the information needed to call the final function, to produce - * a final aggregate result from the state value. If there are multiple - * identical Aggrefs in the query, they can all share the same per-agg data. - * - * These values are set up during ExecInitAgg() and do not change thereafter. - */ -typedef struct AggStatePerAggData -{ - /* - * Link to an Aggref expr this state value is for. - * - * There can be multiple identical Aggref's sharing the same per-agg. This - * points to the first one of them. - */ - Aggref *aggref; - - /* index to the state value which this agg should use */ - int transno; - - /* Optional Oid of final function (may be InvalidOid) */ - Oid finalfn_oid; - - /* - * fmgr lookup data for final function --- only valid when finalfn_oid is - * not InvalidOid. - */ - FmgrInfo finalfn; - - /* - * Number of arguments to pass to the finalfn. This is always at least 1 - * (the transition state value) plus any ordered-set direct args. If the - * finalfn wants extra args then we pass nulls corresponding to the - * aggregated input columns. - */ - int numFinalArgs; - - /* ExprStates for any direct-argument expressions */ - List *aggdirectargs; - - /* - * We need the len and byval info for the agg's result data type in order - * to know how to copy/delete values. - */ - int16 resulttypeLen; - bool resulttypeByVal; - - /* - * "sharable" is false if this agg cannot share state values with other - * aggregates because the final function is read-write. - */ - bool sharable; -} AggStatePerAggData; - -/* - * AggStatePerGroupData - per-aggregate-per-group working state - * - * These values are working state that is initialized at the start of - * an input tuple group and updated for each input tuple. - * - * In AGG_PLAIN and AGG_SORTED modes, we have a single array of these - * structs (pointed to by aggstate->pergroup); we re-use the array for - * each input group, if it's AGG_SORTED mode. In AGG_HASHED mode, the - * hash table contains an array of these structs for each tuple group. - * - * Logically, the sortstate field belongs in this struct, but we do not - * keep it here for space reasons: we don't support DISTINCT aggregates - * in AGG_HASHED mode, so there's no reason to use up a pointer field - * in every entry of the hashtable. - */ -typedef struct AggStatePerGroupData -{ - Datum transValue; /* current transition value */ - bool transValueIsNull; - - bool noTransValue; /* true if transValue not set yet */ - - /* - * Note: noTransValue initially has the same value as transValueIsNull, - * and if true both are cleared to false at the same time. They are not - * the same though: if transfn later returns a NULL, we want to keep that - * NULL and not auto-replace it with a later input value. Only the first - * non-NULL input will be auto-substituted. - */ -} AggStatePerGroupData; - -/* - * AggStatePerPhaseData - per-grouping-set-phase state - * - * Grouping sets are divided into "phases", where a single phase can be - * processed in one pass over the input. If there is more than one phase, then - * at the end of input from the current phase, state is reset and another pass - * taken over the data which has been re-sorted in the mean time. - * - * Accordingly, each phase specifies a list of grouping sets and group clause - * information, plus each phase after the first also has a sort order. - */ -typedef struct AggStatePerPhaseData -{ - AggStrategy aggstrategy; /* strategy for this phase */ - int numsets; /* number of grouping sets (or 0) */ - int *gset_lengths; /* lengths of grouping sets */ - Bitmapset **grouped_cols; /* column groupings for rollup */ - FmgrInfo *eqfunctions; /* per-grouping-field equality fns */ - Agg *aggnode; /* Agg node for phase data */ - Sort *sortnode; /* Sort node for input ordering for phase */ -} AggStatePerPhaseData; - -/* - * AggStatePerHashData - per-hashtable state - * - * When doing grouping sets with hashing, we have one of these for each - * grouping set. (When doing hashing without grouping sets, we have just one of - * them.) - */ -typedef struct AggStatePerHashData -{ - TupleHashTable hashtable; /* hash table with one entry per group */ - TupleHashIterator hashiter; /* for iterating through hash table */ - TupleTableSlot *hashslot; /* slot for loading hash table */ - FmgrInfo *hashfunctions; /* per-grouping-field hash fns */ - FmgrInfo *eqfunctions; /* per-grouping-field equality fns */ - int numCols; /* number of hash key columns */ - int numhashGrpCols; /* number of columns in hash table */ - int largestGrpColIdx; /* largest col required for hashing */ - AttrNumber *hashGrpColIdxInput; /* hash col indices in input slot */ - AttrNumber *hashGrpColIdxHash; /* indices in hashtbl tuples */ - Agg *aggnode; /* original Agg node, for numGroups etc. */ -} AggStatePerHashData; - - static void select_current_set(AggState *aggstate, int setno, bool is_hash); static void initialize_phase(AggState *aggstate, int newphase); static TupleTableSlot *fetch_input_tuple(AggState *aggstate); @@ -537,13 +248,7 @@ static void initialize_aggregates(AggState *aggstate, static void advance_transition_function(AggState *aggstate, AggStatePerTrans pertrans, AggStatePerGroup pergroupstate); -static void advance_aggregates(AggState *aggstate, - AggStatePerGroup *sort_pergroups, - AggStatePerGroup *hash_pergroups); -static void advance_combine_function(AggState *aggstate, - AggStatePerTrans pertrans, - AggStatePerGroup pergroupstate); -static void combine_aggregates(AggState *aggstate, AggStatePerGroup pergroup); +static void advance_aggregates(AggState *aggstate); static void process_ordered_aggregate_single(AggState *aggstate, AggStatePerTrans pertrans, AggStatePerGroup pergroupstate); @@ -569,7 +274,7 @@ static Bitmapset *find_unaggregated_cols(AggState *aggstate); static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos); static void build_hash_table(AggState *aggstate); static TupleHashEntryData *lookup_hash_entry(AggState *aggstate); -static AggStatePerGroup *lookup_hash_entries(AggState *aggstate); +static void lookup_hash_entries(AggState *aggstate); static TupleTableSlot *agg_retrieve_direct(AggState *aggstate); static void agg_fill_hash_table(AggState *aggstate); static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate); @@ -597,6 +302,7 @@ static int find_compatible_pertrans(AggState *aggstate, Aggref *newagg, static void select_current_set(AggState *aggstate, int setno, bool is_hash) { + /* when changing this, also adapt ExecInterpExpr() and friends */ if (is_hash) aggstate->curaggcontext = aggstate->hashcontext; else @@ -967,350 +673,15 @@ advance_transition_function(AggState *aggstate, * When called, CurrentMemoryContext should be the per-query context. */ static void -advance_aggregates(AggState *aggstate, - AggStatePerGroup *sort_pergroups, - AggStatePerGroup *hash_pergroups) +advance_aggregates(AggState *aggstate) { - int transno; - int setno = 0; - int numGroupingSets = Max(aggstate->phase->numsets, 1); - int numHashes = aggstate->num_hashes; - int numTrans = aggstate->numtrans; - TupleTableSlot *combinedslot; - - /* compute required inputs for all aggregates */ - combinedslot = ExecProject(aggstate->combinedproj); - - for (transno = 0; transno < numTrans; transno++) - { - AggStatePerTrans pertrans = &aggstate->pertrans[transno]; - int numTransInputs = pertrans->numTransInputs; - int inputoff = pertrans->inputoff; - TupleTableSlot *slot; - int i; - - /* Skip anything FILTERed out */ - if (pertrans->aggref->aggfilter) - { - /* Check the result of the filter expression */ - if (combinedslot->tts_isnull[inputoff] || - !DatumGetBool(combinedslot->tts_values[inputoff])) - continue; - - /* Now it's safe to evaluate this agg's arguments */ - slot = ExecProject(pertrans->evalproj); - /* There's no offset needed in this slot, of course */ - inputoff = 0; - } - else - { - /* arguments are already evaluated into combinedslot @ inputoff */ - slot = combinedslot; - } - - if (pertrans->numSortCols > 0) - { - /* DISTINCT and/or ORDER BY case */ - Assert(slot->tts_nvalid >= (pertrans->numInputs + inputoff)); - Assert(!hash_pergroups); - - /* - * If the transfn is strict, we want to check for nullity before - * storing the row in the sorter, to save space if there are a lot - * of nulls. Note that we must only check numTransInputs columns, - * not numInputs, since nullity in columns used only for sorting - * is not relevant here. - */ - if (pertrans->transfn.fn_strict) - { - for (i = 0; i < numTransInputs; i++) - { - if (slot->tts_isnull[i + inputoff]) - break; - } - if (i < numTransInputs) - continue; - } - - for (setno = 0; setno < numGroupingSets; setno++) - { - /* OK, put the tuple into the tuplesort object */ - if (pertrans->numInputs == 1) - tuplesort_putdatum(pertrans->sortstates[setno], - slot->tts_values[inputoff], - slot->tts_isnull[inputoff]); - else if (pertrans->aggref->aggfilter) - { - /* - * When filtering and ordering, we already have a slot - * containing just the argument columns. - */ - Assert(slot == pertrans->sortslot); - tuplesort_puttupleslot(pertrans->sortstates[setno], slot); - } - else - { - /* - * Copy argument columns from combined slot, starting at - * inputoff, into sortslot, so that we can store just the - * columns we want. - */ - ExecClearTuple(pertrans->sortslot); - memcpy(pertrans->sortslot->tts_values, - &slot->tts_values[inputoff], - pertrans->numInputs * sizeof(Datum)); - memcpy(pertrans->sortslot->tts_isnull, - &slot->tts_isnull[inputoff], - pertrans->numInputs * sizeof(bool)); - ExecStoreVirtualTuple(pertrans->sortslot); - tuplesort_puttupleslot(pertrans->sortstates[setno], - pertrans->sortslot); - } - } - } - else - { - /* We can apply the transition function immediately */ - FunctionCallInfo fcinfo = &pertrans->transfn_fcinfo; + bool dummynull; - /* Load values into fcinfo */ - /* Start from 1, since the 0th arg will be the transition value */ - Assert(slot->tts_nvalid >= (numTransInputs + inputoff)); - - for (i = 0; i < numTransInputs; i++) - { - fcinfo->arg[i + 1] = slot->tts_values[i + inputoff]; - fcinfo->argnull[i + 1] = slot->tts_isnull[i + inputoff]; - } - - if (sort_pergroups) - { - /* advance transition states for ordered grouping */ - - for (setno = 0; setno < numGroupingSets; setno++) - { - AggStatePerGroup pergroupstate; - - select_current_set(aggstate, setno, false); - - pergroupstate = &sort_pergroups[setno][transno]; - - advance_transition_function(aggstate, pertrans, pergroupstate); - } - } - - if (hash_pergroups) - { - /* advance transition states for hashed grouping */ - - for (setno = 0; setno < numHashes; setno++) - { - AggStatePerGroup pergroupstate; - - select_current_set(aggstate, setno, true); - - pergroupstate = &hash_pergroups[setno][transno]; - - advance_transition_function(aggstate, pertrans, pergroupstate); - } - } - } - } + ExecEvalExprSwitchContext(aggstate->phase->evaltrans, + aggstate->tmpcontext, + &dummynull); } -/* - * combine_aggregates replaces advance_aggregates in DO_AGGSPLIT_COMBINE - * mode. The principal difference is that here we may need to apply the - * deserialization function before running the transfn (which, in this mode, - * is actually the aggregate's combinefn). Also, we know we don't need to - * handle FILTER, DISTINCT, ORDER BY, or grouping sets. - */ -static void -combine_aggregates(AggState *aggstate, AggStatePerGroup pergroup) -{ - int transno; - int numTrans = aggstate->numtrans; - TupleTableSlot *slot; - - /* combine not supported with grouping sets */ - Assert(aggstate->phase->numsets <= 1); - - /* compute input for all aggregates */ - slot = ExecProject(aggstate->combinedproj); - - for (transno = 0; transno < numTrans; transno++) - { - AggStatePerTrans pertrans = &aggstate->pertrans[transno]; - AggStatePerGroup pergroupstate = &pergroup[transno]; - FunctionCallInfo fcinfo = &pertrans->transfn_fcinfo; - int inputoff = pertrans->inputoff; - - Assert(slot->tts_nvalid > inputoff); - - /* - * deserialfn_oid will be set if we must deserialize the input state - * before calling the combine function - */ - if (OidIsValid(pertrans->deserialfn_oid)) - { - /* Don't call a strict deserialization function with NULL input */ - if (pertrans->deserialfn.fn_strict && slot->tts_isnull[inputoff]) - { - fcinfo->arg[1] = slot->tts_values[inputoff]; - fcinfo->argnull[1] = slot->tts_isnull[inputoff]; - } - else - { - FunctionCallInfo dsinfo = &pertrans->deserialfn_fcinfo; - MemoryContext oldContext; - - dsinfo->arg[0] = slot->tts_values[inputoff]; - dsinfo->argnull[0] = slot->tts_isnull[inputoff]; - /* Dummy second argument for type-safety reasons */ - dsinfo->arg[1] = PointerGetDatum(NULL); - dsinfo->argnull[1] = false; - - /* - * We run the deserialization functions in per-input-tuple - * memory context. - */ - oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory); - - fcinfo->arg[1] = FunctionCallInvoke(dsinfo); - fcinfo->argnull[1] = dsinfo->isnull; - - MemoryContextSwitchTo(oldContext); - } - } - else - { - fcinfo->arg[1] = slot->tts_values[inputoff]; - fcinfo->argnull[1] = slot->tts_isnull[inputoff]; - } - - advance_combine_function(aggstate, pertrans, pergroupstate); - } -} - -/* - * Perform combination of states between 2 aggregate states. Effectively this - * 'adds' two states together by whichever logic is defined in the aggregate - * function's combine function. - * - * Note that in this case transfn is set to the combination function. This - * perhaps should be changed to avoid confusion, but one field is ok for now - * as they'll never be needed at the same time. - */ -static void -advance_combine_function(AggState *aggstate, - AggStatePerTrans pertrans, - AggStatePerGroup pergroupstate) -{ - FunctionCallInfo fcinfo = &pertrans->transfn_fcinfo; - MemoryContext oldContext; - Datum newVal; - - if (pertrans->transfn.fn_strict) - { - /* if we're asked to merge to a NULL state, then do nothing */ - if (fcinfo->argnull[1]) - return; - - if (pergroupstate->noTransValue) - { - /* - * transValue has not yet been initialized. If pass-by-ref - * datatype we must copy the combining state value into - * aggcontext. - */ - if (!pertrans->transtypeByVal) - { - oldContext = MemoryContextSwitchTo( - aggstate->curaggcontext->ecxt_per_tuple_memory); - pergroupstate->transValue = datumCopy(fcinfo->arg[1], - pertrans->transtypeByVal, - pertrans->transtypeLen); - MemoryContextSwitchTo(oldContext); - } - else - pergroupstate->transValue = fcinfo->arg[1]; - - pergroupstate->transValueIsNull = false; - pergroupstate->noTransValue = false; - return; - } - - if (pergroupstate->transValueIsNull) - { - /* - * Don't call a strict function with NULL inputs. Note it is - * possible to get here despite the above tests, if the combinefn - * is strict *and* returned a NULL on a prior cycle. If that - * happens we will propagate the NULL all the way to the end. - */ - return; - } - } - - /* We run the combine functions in per-input-tuple memory context */ - oldContext = MemoryContextSwitchTo(aggstate->tmpcontext->ecxt_per_tuple_memory); - - /* set up aggstate->curpertrans for AggGetAggref() */ - aggstate->curpertrans = pertrans; - - /* - * OK to call the combine function - */ - fcinfo->arg[0] = pergroupstate->transValue; - fcinfo->argnull[0] = pergroupstate->transValueIsNull; - fcinfo->isnull = false; /* just in case combine func doesn't set it */ - - newVal = FunctionCallInvoke(fcinfo); - - aggstate->curpertrans = NULL; - - /* - * If pass-by-ref datatype, must copy the new value into aggcontext and - * free the prior transValue. But if the combine function returned a - * pointer to its first input, we don't need to do anything. Also, if the - * combine function returned a pointer to a R/W expanded object that is - * already a child of the aggcontext, assume we can adopt that value - * without copying it. - */ - if (!pertrans->transtypeByVal && - DatumGetPointer(newVal) != DatumGetPointer(pergroupstate->transValue)) - { - if (!fcinfo->isnull) - { - MemoryContextSwitchTo(aggstate->curaggcontext->ecxt_per_tuple_memory); - if (DatumIsReadWriteExpandedObject(newVal, - false, - pertrans->transtypeLen) && - MemoryContextGetParent(DatumGetEOHP(newVal)->eoh_context) == CurrentMemoryContext) - /* do nothing */ ; - else - newVal = datumCopy(newVal, - pertrans->transtypeByVal, - pertrans->transtypeLen); - } - if (!pergroupstate->transValueIsNull) - { - if (DatumIsReadWriteExpandedObject(pergroupstate->transValue, - false, - pertrans->transtypeLen)) - DeleteExpandedObject(pergroupstate->transValue); - else - pfree(DatumGetPointer(pergroupstate->transValue)); - } - } - - pergroupstate->transValue = newVal; - pergroupstate->transValueIsNull = fcinfo->isnull; - - MemoryContextSwitchTo(oldContext); -} - - /* * Run the transition function for a DISTINCT or ORDER BY aggregate * with only one input. This is called after we have completed @@ -2118,7 +1489,7 @@ lookup_hash_entry(AggState *aggstate) * * Be aware that lookup_hash_entry can reset the tmpcontext. */ -static AggStatePerGroup * +static void lookup_hash_entries(AggState *aggstate) { int numHashes = aggstate->num_hashes; @@ -2130,8 +1501,6 @@ lookup_hash_entries(AggState *aggstate) select_current_set(aggstate, setno, true); pergroup[setno] = lookup_hash_entry(aggstate)->additional; } - - return pergroup; } /* @@ -2191,7 +1560,6 @@ agg_retrieve_direct(AggState *aggstate) ExprContext *tmpcontext; AggStatePerAgg peragg; AggStatePerGroup *pergroups; - AggStatePerGroup *hash_pergroups = NULL; TupleTableSlot *outerslot; TupleTableSlot *firstSlot; TupleTableSlot *result; @@ -2446,15 +1814,11 @@ agg_retrieve_direct(AggState *aggstate) if (aggstate->aggstrategy == AGG_MIXED && aggstate->current_phase == 1) { - hash_pergroups = lookup_hash_entries(aggstate); + lookup_hash_entries(aggstate); } - else - hash_pergroups = NULL; - if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit)) - combine_aggregates(aggstate, pergroups[0]); - else - advance_aggregates(aggstate, pergroups, hash_pergroups); + /* Advance the aggregates (or combine functions) */ + advance_aggregates(aggstate); /* Reset per-input-tuple context after each tuple */ ResetExprContext(tmpcontext); @@ -2548,8 +1912,6 @@ agg_fill_hash_table(AggState *aggstate) */ for (;;) { - AggStatePerGroup *pergroups; - outerslot = fetch_input_tuple(aggstate); if (TupIsNull(outerslot)) break; @@ -2558,13 +1920,10 @@ agg_fill_hash_table(AggState *aggstate) tmpcontext->ecxt_outertuple = outerslot; /* Find or build hashtable entries */ - pergroups = lookup_hash_entries(aggstate); + lookup_hash_entries(aggstate); - /* Advance the aggregates */ - if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit)) - combine_aggregates(aggstate, pergroups[0]); - else - advance_aggregates(aggstate, NULL, pergroups); + /* Advance the aggregates (or combine functions) */ + advance_aggregates(aggstate); /* * Reset per-input-tuple context after each tuple, but note that the @@ -2716,6 +2075,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) AggState *aggstate; AggStatePerAgg peraggs; AggStatePerTrans pertransstates; + AggStatePerGroup *pergroups; Plan *outerPlan; ExprContext *econtext; int numaggs, @@ -2723,15 +2083,11 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) aggno; int phase; int phaseidx; - List *combined_inputeval; - TupleDesc combineddesc; - TupleTableSlot *combinedslot; ListCell *l; Bitmapset *all_grouped_cols = NULL; int numGroupingSets = 1; int numPhases; int numHashes; - int column_offset; int i = 0; int j = 0; bool use_hashing = (node->aggstrategy == AGG_HASHED || @@ -3033,6 +2389,24 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) 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. */ @@ -3049,27 +2423,13 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) } /* this is an array of pointers, not structures */ - aggstate->hash_pergroup = palloc0(sizeof(AggStatePerGroup) * numHashes); + aggstate->hash_pergroup = pergroups; find_hash_columns(aggstate); build_hash_table(aggstate); aggstate->table_filled = false; } - if (node->aggstrategy != AGG_HASHED) - { - AggStatePerGroup *pergroups; - - pergroups = (AggStatePerGroup *) palloc0(sizeof(AggStatePerGroup) * - numGroupingSets); - - for (i = 0; i < numGroupingSets; i++) - pergroups[i] = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData) - * numaggs); - - aggstate->pergroups = pergroups; - } - /* * Initialize current phase-dependent values to initial phase. The initial * phase is 1 (first sort pass) for all strategies that use sorting (if @@ -3409,98 +2769,72 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) aggstate->numtrans = transno + 1; /* - * Build a single projection computing the required arguments for all - * aggregates at once; if there's more than one, that's considerably - * faster than doing it separately for each. - * - * First create a targetlist representing the values to compute. + * 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.) */ - combined_inputeval = NIL; - column_offset = 0; - for (transno = 0; transno < aggstate->numtrans; transno++) + 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++) { - AggStatePerTrans pertrans = &pertransstates[transno]; + AggStatePerPhase phase = &aggstate->phases[phaseidx]; + bool dohash = false; + bool dosort = false; - /* - * Mark this per-trans state with its starting column in the combined - * slot. - */ - pertrans->inputoff = column_offset; + /* phase 0 doesn't necessarily exist */ + if (!phase->aggnode) + continue; - /* - * If the aggregate has a FILTER, we can only evaluate the filter - * expression, not the actual input expressions, during the combined - * eval step --- unless we're ignoring the filter because this node is - * running combinefns not transfns. - */ - if (pertrans->aggref->aggfilter && - !DO_AGGSPLIT_COMBINE(aggstate->aggsplit)) + if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 1) { - TargetEntry *tle; - - tle = makeTargetEntry(pertrans->aggref->aggfilter, - column_offset + 1, NULL, false); - combined_inputeval = lappend(combined_inputeval, tle); - column_offset++; - /* - * We'll need separate projection machinery for the real args. - * Arrange to evaluate them into the sortslot previously created. + * Phase one, and only phase one, in a mixed agg performs both + * sorting and aggregation. */ - Assert(pertrans->sortslot); - pertrans->evalproj = ExecBuildProjectionInfo(pertrans->aggref->args, - aggstate->tmpcontext, - pertrans->sortslot, - &aggstate->ss.ps, - NULL); + dohash = true; + dosort = true; } - else + else if (aggstate->aggstrategy == AGG_MIXED && phaseidx == 0) { /* - * Add agg's input expressions to combined_inputeval, adjusting - * resnos in the copied target entries to match the combined slot. + * 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. */ - ListCell *arg; - - foreach(arg, pertrans->aggref->args) - { - TargetEntry *source_tle = lfirst_node(TargetEntry, arg); - TargetEntry *tle; - - tle = flatCopyTargetEntry(source_tle); - tle->resno += column_offset; - - combined_inputeval = lappend(combined_inputeval, tle); - } - - column_offset += list_length(pertrans->aggref->args); + 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); - /* Now create a projection for the combined targetlist */ - combineddesc = ExecTypeFromTL(combined_inputeval, false); - combinedslot = ExecInitExtraTupleSlot(estate); - ExecSetSlotDescriptor(combinedslot, combineddesc); - aggstate->combinedproj = ExecBuildProjectionInfo(combined_inputeval, - aggstate->tmpcontext, - combinedslot, - &aggstate->ss.ps, - NULL); + phase->evaltrans = ExecBuildAggTrans(aggstate, phase, dosort, dohash); - /* - * 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"))); + } return aggstate; } @@ -3557,8 +2891,6 @@ build_pertrans_for_aggref(AggStatePerTrans pertrans, else pertrans->numTransInputs = numArguments; - /* inputoff and evalproj will be set up later, in ExecInitAgg */ - /* * When combining states, we have no use at all for the aggregate * function's transfn. Instead we use the combinefn. In this case, the diff --git a/src/include/executor/execExpr.h b/src/include/executor/execExpr.h index b0c7bda76f..117fc892f4 100644 --- a/src/include/executor/execExpr.h +++ b/src/include/executor/execExpr.h @@ -14,6 +14,7 @@ #ifndef EXEC_EXPR_H #define EXEC_EXPR_H +#include "executor/nodeAgg.h" #include "nodes/execnodes.h" /* forward references to avoid circularity */ @@ -64,9 +65,9 @@ typedef enum ExprEvalOp EEOP_WHOLEROW, /* - * Compute non-system Var value, assign it into ExprState's - * resultslot. These are not used if a CheckVarSlotCompatibility() check - * would be needed. + * Compute non-system Var value, assign it into ExprState's resultslot. + * These are not used if a CheckVarSlotCompatibility() check would be + * needed. */ EEOP_ASSIGN_INNER_VAR, EEOP_ASSIGN_OUTER_VAR, @@ -218,6 +219,17 @@ typedef enum ExprEvalOp EEOP_SUBPLAN, EEOP_ALTERNATIVE_SUBPLAN, + /* aggregation related nodes */ + EEOP_AGG_STRICT_DESERIALIZE, + EEOP_AGG_DESERIALIZE, + EEOP_AGG_STRICT_INPUT_CHECK, + EEOP_AGG_INIT_TRANS, + EEOP_AGG_STRICT_TRANS_CHECK, + EEOP_AGG_PLAIN_TRANS_BYVAL, + EEOP_AGG_PLAIN_TRANS, + EEOP_AGG_ORDERED_TRANS_DATUM, + EEOP_AGG_ORDERED_TRANS_TUPLE, + /* non-existent operation, used e.g. to check array lengths */ EEOP_LAST } ExprEvalOp; @@ -573,6 +585,55 @@ typedef struct ExprEvalStep /* out-of-line state, created by nodeSubplan.c */ AlternativeSubPlanState *asstate; } alternative_subplan; + + /* for EEOP_AGG_*DESERIALIZE */ + struct + { + AggState *aggstate; + FunctionCallInfo fcinfo_data; + int jumpnull; + } agg_deserialize; + + /* for EEOP_AGG_STRICT_INPUT_CHECK */ + struct + { + bool *nulls; + int nargs; + int jumpnull; + } agg_strict_input_check; + + /* for EEOP_AGG_INIT_TRANS */ + struct + { + AggState *aggstate; + AggStatePerTrans pertrans; + ExprContext *aggcontext; + int setno; + int transno; + int setoff; + int jumpnull; + } agg_init_trans; + + /* for EEOP_AGG_STRICT_TRANS_CHECK */ + struct + { + AggState *aggstate; + int setno; + int transno; + int setoff; + int jumpnull; + } agg_strict_trans_check; + + /* for EEOP_AGG_{PLAIN,ORDERED}_TRANS* */ + struct + { + AggState *aggstate; + AggStatePerTrans pertrans; + ExprContext *aggcontext; + int setno; + int transno; + int setoff; + } agg_trans; } d; } ExprEvalStep; @@ -669,4 +730,13 @@ extern void ExecEvalAlternativeSubPlan(ExprState *state, ExprEvalStep *op, extern void ExecEvalWholeRowVar(ExprState *state, ExprEvalStep *op, ExprContext *econtext); +extern void ExecAggInitGroup(AggState *aggstate, AggStatePerTrans pertrans, AggStatePerGroup pergroup); +extern Datum ExecAggTransReparent(AggState *aggstate, AggStatePerTrans pertrans, + Datum newValue, bool newValueIsNull, + Datum oldValue, bool oldValueIsNull); +extern void ExecEvalAggOrderedTransDatum(ExprState *state, ExprEvalStep *op, + ExprContext *econtext); +extern void ExecEvalAggOrderedTransTuple(ExprState *state, ExprEvalStep *op, + ExprContext *econtext); + #endif /* EXEC_EXPR_H */ diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index a782fae0f8..6545a80222 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -192,7 +192,7 @@ extern void ExecConstraints(ResultRelInfo *resultRelInfo, extern bool ExecPartitionCheck(ResultRelInfo *resultRelInfo, TupleTableSlot *slot, EState *estate); extern void ExecPartitionCheckEmitError(ResultRelInfo *resultRelInfo, - TupleTableSlot *slot, EState *estate); + TupleTableSlot *slot, EState *estate); extern void ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo, TupleTableSlot *slot, EState *estate); extern LockTupleMode ExecUpdateLockMode(EState *estate, ResultRelInfo *relinfo); @@ -254,6 +254,8 @@ extern ExprState *ExecInitExprWithParams(Expr *node, ParamListInfo ext_params); extern ExprState *ExecInitQual(List *qual, PlanState *parent); extern ExprState *ExecInitCheck(List *qual, PlanState *parent); extern List *ExecInitExprList(List *nodes, PlanState *parent); +extern ExprState *ExecBuildAggTrans(AggState *aggstate, struct AggStatePerPhaseData *phase, + bool doSort, bool doHash); extern ProjectionInfo *ExecBuildProjectionInfo(List *targetList, ExprContext *econtext, TupleTableSlot *slot, diff --git a/src/include/executor/nodeAgg.h b/src/include/executor/nodeAgg.h index 90c68795f1..3b06db86fd 100644 --- a/src/include/executor/nodeAgg.h +++ b/src/include/executor/nodeAgg.h @@ -16,6 +16,290 @@ #include "nodes/execnodes.h" + +/* + * AggStatePerTransData - per aggregate state value information + * + * Working state for updating the aggregate's state value, by calling the + * transition function with an input row. This struct does not store the + * information needed to produce the final aggregate result from the transition + * state, that's stored in AggStatePerAggData instead. This separation allows + * multiple aggregate results to be produced from a single state value. + */ +typedef struct AggStatePerTransData +{ + /* + * These values are set up during ExecInitAgg() and do not change + * thereafter: + */ + + /* + * Link to an Aggref expr this state value is for. + * + * There can be multiple Aggref's sharing the same state value, so long as + * the inputs and transition functions are identical and the final + * functions are not read-write. This points to the first one of them. + */ + Aggref *aggref; + + /* + * Is this state value actually being shared by more than one Aggref? + */ + bool aggshared; + + /* + * Number of aggregated input columns. This includes ORDER BY expressions + * in both the plain-agg and ordered-set cases. Ordered-set direct args + * are not counted, though. + */ + int numInputs; + + /* + * Number of aggregated input columns to pass to the transfn. This + * includes the ORDER BY columns for ordered-set aggs, but not for plain + * aggs. (This doesn't count the transition state value!) + */ + int numTransInputs; + + /* Oid of the state transition or combine function */ + Oid transfn_oid; + + /* Oid of the serialization function or InvalidOid */ + Oid serialfn_oid; + + /* Oid of the deserialization function or InvalidOid */ + Oid deserialfn_oid; + + /* Oid of state value's datatype */ + Oid aggtranstype; + + /* + * fmgr lookup data for transition function or combine function. Note in + * particular that the fn_strict flag is kept here. + */ + FmgrInfo transfn; + + /* fmgr lookup data for serialization function */ + FmgrInfo serialfn; + + /* fmgr lookup data for deserialization function */ + FmgrInfo deserialfn; + + /* Input collation derived for aggregate */ + Oid aggCollation; + + /* number of sorting columns */ + int numSortCols; + + /* number of sorting columns to consider in DISTINCT comparisons */ + /* (this is either zero or the same as numSortCols) */ + int numDistinctCols; + + /* deconstructed sorting information (arrays of length numSortCols) */ + AttrNumber *sortColIdx; + Oid *sortOperators; + Oid *sortCollations; + bool *sortNullsFirst; + + /* + * fmgr lookup data for input columns' equality operators --- only + * set/used when aggregate has DISTINCT flag. Note that these are in + * order of sort column index, not parameter index. + */ + FmgrInfo *equalfns; /* array of length numDistinctCols */ + + /* + * initial value from pg_aggregate entry + */ + Datum initValue; + bool initValueIsNull; + + /* + * We need the len and byval info for the agg's input and transition data + * types in order to know how to copy/delete values. + * + * Note that the info for the input type is used only when handling + * DISTINCT aggs with just one argument, so there is only one input type. + */ + int16 inputtypeLen, + transtypeLen; + bool inputtypeByVal, + transtypeByVal; + + /* + * Slots for holding the evaluated input arguments. These are set up + * during ExecInitAgg() and then used for each input row requiring either + * FILTER or ORDER BY/DISTINCT processing. + */ + TupleTableSlot *sortslot; /* current input tuple */ + TupleTableSlot *uniqslot; /* used for multi-column DISTINCT */ + TupleDesc sortdesc; /* descriptor of input tuples */ + + /* + * These values are working state that is initialized at the start of an + * input tuple group and updated for each input tuple. + * + * For a simple (non DISTINCT/ORDER BY) aggregate, we just feed the input + * values straight to the transition function. If it's DISTINCT or + * requires ORDER BY, we pass the input values into a Tuplesort object; + * then at completion of the input tuple group, we scan the sorted values, + * eliminate duplicates if needed, and run the transition function on the + * rest. + * + * We need a separate tuplesort for each grouping set. + */ + + Tuplesortstate **sortstates; /* sort objects, if DISTINCT or ORDER BY */ + + /* + * This field is a pre-initialized FunctionCallInfo struct used for + * calling this aggregate's transfn. We save a few cycles per row by not + * re-initializing the unchanging fields; which isn't much, but it seems + * worth the extra space consumption. + */ + FunctionCallInfoData transfn_fcinfo; + + /* Likewise for serialization and deserialization functions */ + FunctionCallInfoData serialfn_fcinfo; + + FunctionCallInfoData deserialfn_fcinfo; +} AggStatePerTransData; + +/* + * AggStatePerAggData - per-aggregate information + * + * This contains the information needed to call the final function, to produce + * a final aggregate result from the state value. If there are multiple + * identical Aggrefs in the query, they can all share the same per-agg data. + * + * These values are set up during ExecInitAgg() and do not change thereafter. + */ +typedef struct AggStatePerAggData +{ + /* + * Link to an Aggref expr this state value is for. + * + * There can be multiple identical Aggref's sharing the same per-agg. This + * points to the first one of them. + */ + Aggref *aggref; + + /* index to the state value which this agg should use */ + int transno; + + /* Optional Oid of final function (may be InvalidOid) */ + Oid finalfn_oid; + + /* + * fmgr lookup data for final function --- only valid when finalfn_oid is + * not InvalidOid. + */ + FmgrInfo finalfn; + + /* + * Number of arguments to pass to the finalfn. This is always at least 1 + * (the transition state value) plus any ordered-set direct args. If the + * finalfn wants extra args then we pass nulls corresponding to the + * aggregated input columns. + */ + int numFinalArgs; + + /* ExprStates for any direct-argument expressions */ + List *aggdirectargs; + + /* + * We need the len and byval info for the agg's result data type in order + * to know how to copy/delete values. + */ + int16 resulttypeLen; + bool resulttypeByVal; + + /* + * "sharable" is false if this agg cannot share state values with other + * aggregates because the final function is read-write. + */ + bool sharable; +} AggStatePerAggData; + +/* + * AggStatePerGroupData - per-aggregate-per-group working state + * + * These values are working state that is initialized at the start of + * an input tuple group and updated for each input tuple. + * + * In AGG_PLAIN and AGG_SORTED modes, we have a single array of these + * structs (pointed to by aggstate->pergroup); we re-use the array for + * each input group, if it's AGG_SORTED mode. In AGG_HASHED mode, the + * hash table contains an array of these structs for each tuple group. + * + * Logically, the sortstate field belongs in this struct, but we do not + * keep it here for space reasons: we don't support DISTINCT aggregates + * in AGG_HASHED mode, so there's no reason to use up a pointer field + * in every entry of the hashtable. + */ +typedef struct AggStatePerGroupData +{ + Datum transValue; /* current transition value */ + bool transValueIsNull; + + bool noTransValue; /* true if transValue not set yet */ + + /* + * Note: noTransValue initially has the same value as transValueIsNull, + * and if true both are cleared to false at the same time. They are not + * the same though: if transfn later returns a NULL, we want to keep that + * NULL and not auto-replace it with a later input value. Only the first + * non-NULL input will be auto-substituted. + */ +} AggStatePerGroupData; + +/* + * AggStatePerPhaseData - per-grouping-set-phase state + * + * Grouping sets are divided into "phases", where a single phase can be + * processed in one pass over the input. If there is more than one phase, then + * at the end of input from the current phase, state is reset and another pass + * taken over the data which has been re-sorted in the mean time. + * + * Accordingly, each phase specifies a list of grouping sets and group clause + * information, plus each phase after the first also has a sort order. + */ +typedef struct AggStatePerPhaseData +{ + AggStrategy aggstrategy; /* strategy for this phase */ + int numsets; /* number of grouping sets (or 0) */ + int *gset_lengths; /* lengths of grouping sets */ + Bitmapset **grouped_cols; /* column groupings for rollup */ + FmgrInfo *eqfunctions; /* per-grouping-field equality fns */ + Agg *aggnode; /* Agg node for phase data */ + Sort *sortnode; /* Sort node for input ordering for phase */ + + ExprState *evaltrans; /* evaluation of transition functions */ +} AggStatePerPhaseData; + +/* + * AggStatePerHashData - per-hashtable state + * + * When doing grouping sets with hashing, we have one of these for each + * grouping set. (When doing hashing without grouping sets, we have just one of + * them.) + */ +typedef struct AggStatePerHashData +{ + TupleHashTable hashtable; /* hash table with one entry per group */ + TupleHashIterator hashiter; /* for iterating through hash table */ + TupleTableSlot *hashslot; /* slot for loading hash table */ + FmgrInfo *hashfunctions; /* per-grouping-field hash fns */ + FmgrInfo *eqfunctions; /* per-grouping-field equality fns */ + int numCols; /* number of hash key columns */ + int numhashGrpCols; /* number of columns in hash table */ + int largestGrpColIdx; /* largest col required for hashing */ + AttrNumber *hashGrpColIdxInput; /* hash col indices in input slot */ + AttrNumber *hashGrpColIdxHash; /* indices in hashtbl tuples */ + Agg *aggnode; /* original Agg node, for numGroups etc. */ +} AggStatePerHashData; + + extern AggState *ExecInitAgg(Agg *node, EState *estate, int eflags); extern void ExecEndAgg(AggState *node); extern void ExecReScanAgg(AggState *node); diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 2a4f7407a1..4bb5cb163d 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1850,10 +1850,13 @@ typedef struct AggState /* these fields are used in AGG_HASHED and AGG_MIXED modes: */ bool table_filled; /* hash table filled yet? */ int num_hashes; - AggStatePerHash perhash; + AggStatePerHash perhash; /* array of per-hashtable data */ AggStatePerGroup *hash_pergroup; /* grouping set indexed array of * per-group pointers */ + /* support for evaluation of agg input expressions: */ + AggStatePerGroup *all_pergroups; /* array of first ->pergroups, than + * ->hash_pergroup */ ProjectionInfo *combinedproj; /* projection machinery */ } AggState; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index a92c62adde..cc84217dd9 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -604,6 +604,7 @@ ExprContextCallbackFunction ExprContext_CB ExprDoneCond ExprEvalOp +ExprEvalOpLookup ExprEvalStep ExprState ExprStateEvalFunc -- 2.40.0