{
Agg *agg = (Agg *) plan;
- if (agg->finalizeAggs == false)
+ if (DO_AGGSPLIT_SKIPFINAL(agg->aggsplit))
operation = "Partial";
- else if (agg->combineStates == true)
+ else if (DO_AGGSPLIT_COMBINE(agg->aggsplit))
operation = "Finalize";
switch (agg->aggstrategy)
case T_Aggref:
{
AggrefExprState *astate = makeNode(AggrefExprState);
- AggState *aggstate = (AggState *) parent;
- Aggref *aggref = (Aggref *) node;
astate->xprstate.evalfunc = (ExprStateEvalFunc) ExecEvalAggref;
- if (!aggstate || !IsA(aggstate, AggState))
+ if (parent && IsA(parent, AggState))
{
- /* planner messed up */
- elog(ERROR, "Aggref found in non-Agg plan node");
- }
- if (aggref->aggpartial == aggstate->finalizeAggs)
- {
- /* planner messed up */
- if (aggref->aggpartial)
- elog(ERROR, "partial Aggref found in finalize agg plan node");
- else
- elog(ERROR, "non-partial Aggref found in non-finalize agg plan node");
- }
+ AggState *aggstate = (AggState *) parent;
- if (aggref->aggcombine != aggstate->combineStates)
+ aggstate->aggs = lcons(astate, aggstate->aggs);
+ aggstate->numaggs++;
+ }
+ else
{
/* planner messed up */
- if (aggref->aggcombine)
- elog(ERROR, "combine Aggref found in non-combine agg plan node");
- else
- elog(ERROR, "non-combine Aggref found in combine agg plan node");
+ elog(ERROR, "Aggref found in non-Agg plan node");
}
-
- aggstate->aggs = lcons(astate, aggstate->aggs);
- aggstate->numaggs++;
state = (ExprState *) astate;
}
break;
* transvalue = transfunc(transvalue, input_value(s))
* result = finalfunc(transvalue, direct_argument(s))
*
- * If a finalfunc is not supplied or finalizeAggs is false, then the result
- * is just the ending value of transvalue.
- *
- * Other behavior is also supported and is controlled by the 'combineStates'
- * and 'finalizeAggs'. 'combineStates' controls whether the trans func or
- * the combine func is used during aggregation. When 'combineStates' is
- * true we expect other (previously) aggregated states as input rather than
- * input tuples. This mode facilitates multiple aggregate stages which
- * allows us to support pushing aggregation down deeper into the plan rather
- * than leaving it for the final stage. For example with a query such as:
- *
- * SELECT count(*) FROM (SELECT * FROM a UNION ALL SELECT * FROM b);
- *
- * with this functionality the planner has the flexibility to generate a
- * plan which performs count(*) on table a and table b separately and then
- * add a combine phase to combine both results. In this case the combine
- * function would simply add both counts together.
- *
- * When multiple aggregate stages exist the planner should have set the
- * 'finalizeAggs' to true only for the final aggregtion state, and each
- * stage, apart from the very first one should have 'combineStates' set to
- * true. This permits plans such as:
- *
- * Finalize Aggregate
- * -> Partial Aggregate
- * -> Partial Aggregate
- *
- * Combine functions which use pass-by-ref states should be careful to
- * always update the 1st state parameter by adding the 2nd parameter to it,
- * rather than the other way around. If the 1st state is NULL, then it's not
- * sufficient to simply return the 2nd state, as the memory context is
- * incorrect. Instead a new state should be created in the correct aggregate
- * memory context and the 2nd state should be copied over.
- *
- * The 'serialStates' option can be used to allow multi-stage aggregation
- * for aggregates with an INTERNAL state type. When this mode is disabled
- * only a pointer to the INTERNAL aggregate states are passed around the
- * executor. When enabled, INTERNAL states are serialized and deserialized
- * as required; this is useful when data must be passed between processes.
+ * If a finalfunc is not supplied then the result is just the ending
+ * value of transvalue.
+ *
+ * Other behaviors can be selected by the "aggsplit" mode, which exists
+ * to support partial aggregation. It is possible to:
+ * * Skip running the finalfunc, so that the output is always the
+ * final transvalue state.
+ * * Substitute the combinefunc for the transfunc, so that transvalue
+ * states (propagated up from a child partial-aggregation step) are merged
+ * rather than processing raw input rows. (The statements below about
+ * the transfunc apply equally to the combinefunc, when it's selected.)
+ * * Apply the serializefunc to the output values (this only makes sense
+ * when skipping the finalfunc, since the serializefunc works on the
+ * transvalue data type).
+ * * Apply the deserializefunc to the input values (this only makes sense
+ * when using the combinefunc, for similar reasons).
+ * It is the planner's responsibility to connect up Agg nodes using these
+ * alternate behaviors in a way that makes sense, with partial aggregation
+ * results being fed to nodes that expect them.
*
* If a normal aggregate call specifies DISTINCT or ORDER BY, we sort the
* input tuples and eliminate duplicates (if required) before performing
* the above-depicted process. (However, we don't do that for ordered-set
* aggregates; their "ORDER BY" inputs are ordinary aggregate arguments
- * so far as this module is concerned.)
+ * so far as this module is concerned.) Note that partial aggregation
+ * is not supported in these cases, since we couldn't ensure global
+ * ordering or distinctness of the inputs.
*
* If transfunc is marked "strict" in pg_proc and initcond is NULL,
* then the first non-NULL input_value is assigned directly to transvalue,
int numGroupingSets = Max(aggstate->phase->numsets, 1);
int numTrans = aggstate->numtrans;
- Assert(!aggstate->combineStates);
-
for (transno = 0; transno < numTrans; transno++)
{
AggStatePerTrans pertrans = &aggstate->pertrans[transno];
}
/*
- * combine_aggregates is used when running in 'combineState' mode. This
- * advances each aggregate transition state by adding another transition state
- * to it.
+ * 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)
/* combine not supported with grouping sets */
Assert(aggstate->phase->numsets == 0);
- Assert(aggstate->combineStates);
for (transno = 0; transno < numTrans; transno++)
{
AggStatePerTrans pertrans = &aggstate->pertrans[transno];
+ AggStatePerGroup pergroupstate = &pergroup[transno];
TupleTableSlot *slot;
FunctionCallInfo fcinfo = &pertrans->transfn_fcinfo;
- AggStatePerGroup pergroupstate = &pergroup[transno];
/* Evaluate the current input expressions for this aggregate */
slot = ExecProject(pertrans->evalproj, NULL);
*/
if (OidIsValid(pertrans->deserialfn_oid))
{
- /*
- * Don't call a strict deserialization function with NULL input. A
- * strict deserialization function and a null value means we skip
- * calling the combine function for this state. We assume that
- * this would be a waste of time and effort anyway so just skip
- * it.
- */
+ /* Don't call a strict deserialization function with NULL input */
if (pertrans->deserialfn.fn_strict && slot->tts_isnull[0])
- continue;
+ {
+ fcinfo->arg[1] = slot->tts_values[0];
+ fcinfo->argnull[1] = slot->tts_isnull[0];
+ }
else
{
FunctionCallInfo dsinfo = &pertrans->deserialfn_fcinfo;
pergroupstate->transValueIsNull = fcinfo->isnull;
MemoryContextSwitchTo(oldContext);
-
}
}
/*
- * Compute the final value of one partial aggregate.
+ * Compute the output value of one partial aggregate.
*
* The serialization function will be run, and the result delivered, in the
* output-tuple context; caller's CurrentMemoryContext does not matter.
oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
/*
- * serialfn_oid will be set if we must serialize the input state before
- * calling the combine function on the state.
+ * serialfn_oid will be set if we must serialize the transvalue before
+ * returning it
*/
if (OidIsValid(pertrans->serialfn_oid))
{
pergroupstate);
}
- if (aggstate->finalizeAggs)
- finalize_aggregate(aggstate, peragg, pergroupstate,
- &aggvalues[aggno], &aggnulls[aggno]);
- else
+ if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
finalize_partialaggregate(aggstate, peragg, pergroupstate,
&aggvalues[aggno], &aggnulls[aggno]);
+ else
+ finalize_aggregate(aggstate, peragg, pergroupstate,
+ &aggvalues[aggno], &aggnulls[aggno]);
}
}
*/
for (;;)
{
- if (!aggstate->combineStates)
- advance_aggregates(aggstate, pergroup);
- else
+ if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
combine_aggregates(aggstate, pergroup);
+ else
+ advance_aggregates(aggstate, pergroup);
/* Reset per-input-tuple context after each tuple */
ResetExprContext(tmpcontext);
entry = lookup_hash_entry(aggstate, outerslot);
/* Advance the aggregates */
- if (!aggstate->combineStates)
- advance_aggregates(aggstate, entry->pergroup);
- else
+ if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
combine_aggregates(aggstate, entry->pergroup);
+ else
+ advance_aggregates(aggstate, entry->pergroup);
/* Reset per-input-tuple context after each tuple */
ResetExprContext(tmpcontext);
aggstate->aggs = NIL;
aggstate->numaggs = 0;
aggstate->numtrans = 0;
+ aggstate->aggsplit = node->aggsplit;
aggstate->maxsets = 0;
aggstate->hashfunctions = NULL;
aggstate->projected_set = -1;
aggstate->peragg = NULL;
aggstate->pertrans = NULL;
aggstate->curpertrans = NULL;
- aggstate->agg_done = false;
- aggstate->combineStates = node->combineStates;
- aggstate->finalizeAggs = node->finalizeAggs;
- aggstate->serialStates = node->serialStates;
aggstate->input_done = false;
+ aggstate->agg_done = false;
aggstate->pergroup = NULL;
aggstate->grp_firstTuple = NULL;
aggstate->hashtable = NULL;
/* Planner should have assigned aggregate to correct level */
Assert(aggref->agglevelsup == 0);
+ /* ... and the split mode should match */
+ Assert(aggref->aggsplit == aggstate->aggsplit);
/* 1. Check for already processed aggs which can be re-used */
existing_aggno = find_compatible_peragg(aggref, aggstate, aggno,
* If this aggregation is performing state combines, then instead of
* using the transition function, we'll use the combine function
*/
- if (aggstate->combineStates)
+ if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
{
transfn_oid = aggform->aggcombinefn;
transfn_oid = aggform->aggtransfn;
/* Final function only required if we're finalizing the aggregates */
- if (aggstate->finalizeAggs)
- peragg->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
- else
+ if (DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit))
peragg->finalfn_oid = finalfn_oid = InvalidOid;
+ else
+ peragg->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
serialfn_oid = InvalidOid;
deserialfn_oid = InvalidOid;
/*
- * Determine if we require serialization or deserialization of the
- * aggregate states. This is only required if the aggregate state is
- * internal.
+ * Check if serialization/deserialization is required. We only do it
+ * for aggregates that have transtype INTERNAL.
*/
- if (aggstate->serialStates && aggtranstype == INTERNALOID)
+ if (aggtranstype == INTERNALOID)
{
/*
- * The planner should only have generated an agg node with
- * serialStates if every aggregate with an INTERNAL state has
- * serialization/deserialization functions. Verify that.
+ * The planner should only have generated a serialize agg node if
+ * every aggregate with an INTERNAL state has a serialization
+ * function. Verify that.
*/
- if (!OidIsValid(aggform->aggserialfn))
- elog(ERROR, "serialfunc not set during serialStates aggregation step");
-
- if (!OidIsValid(aggform->aggdeserialfn))
- elog(ERROR, "deserialfunc not set during serialStates aggregation step");
+ if (DO_AGGSPLIT_SERIALIZE(aggstate->aggsplit))
+ {
+ /* serialization only valid when not running finalfn */
+ Assert(DO_AGGSPLIT_SKIPFINAL(aggstate->aggsplit));
- /* serialization func only required when not finalizing aggs */
- if (!aggstate->finalizeAggs)
+ if (!OidIsValid(aggform->aggserialfn))
+ elog(ERROR, "serialfunc not provided for serialization aggregation");
serialfn_oid = aggform->aggserialfn;
+ }
+
+ /* Likewise for deserialization functions */
+ if (DO_AGGSPLIT_DESERIALIZE(aggstate->aggsplit))
+ {
+ /* deserialization only valid when combining states */
+ Assert(DO_AGGSPLIT_COMBINE(aggstate->aggsplit));
- /* deserialization func only required when combining states */
- if (aggstate->combineStates)
+ if (!OidIsValid(aggform->aggdeserialfn))
+ elog(ERROR, "deserialfunc not provided for deserialization aggregation");
deserialfn_oid = aggform->aggdeserialfn;
+ }
}
/* Check that aggregate owner has permission to call component fns */
}
/* get info about the output value's datatype */
- get_typlenbyval(aggref->aggoutputtype,
+ get_typlenbyval(aggref->aggtype,
&peragg->resulttypeLen,
&peragg->resulttypeByVal);
* transfn and transfn_oid fields of pertrans refer to the combine
* function rather than the transition function.
*/
- if (aggstate->combineStates)
+ if (DO_AGGSPLIT_COMBINE(aggstate->aggsplit))
{
Expr *combinefnexpr;
CopyPlanFields((const Plan *) from, (Plan *) newnode);
COPY_SCALAR_FIELD(aggstrategy);
- COPY_SCALAR_FIELD(combineStates);
- COPY_SCALAR_FIELD(finalizeAggs);
- COPY_SCALAR_FIELD(serialStates);
+ COPY_SCALAR_FIELD(aggsplit);
COPY_SCALAR_FIELD(numCols);
if (from->numCols > 0)
{
COPY_SCALAR_FIELD(aggfnoid);
COPY_SCALAR_FIELD(aggtype);
- COPY_SCALAR_FIELD(aggoutputtype);
COPY_SCALAR_FIELD(aggcollid);
COPY_SCALAR_FIELD(inputcollid);
COPY_SCALAR_FIELD(aggtranstype);
COPY_NODE_FIELD(aggfilter);
COPY_SCALAR_FIELD(aggstar);
COPY_SCALAR_FIELD(aggvariadic);
- COPY_SCALAR_FIELD(aggcombine);
- COPY_SCALAR_FIELD(aggpartial);
COPY_SCALAR_FIELD(aggkind);
COPY_SCALAR_FIELD(agglevelsup);
+ COPY_SCALAR_FIELD(aggsplit);
COPY_LOCATION_FIELD(location);
return newnode;
{
COMPARE_SCALAR_FIELD(aggfnoid);
COMPARE_SCALAR_FIELD(aggtype);
- COMPARE_SCALAR_FIELD(aggoutputtype);
COMPARE_SCALAR_FIELD(aggcollid);
COMPARE_SCALAR_FIELD(inputcollid);
/* ignore aggtranstype since it might not be set yet */
COMPARE_NODE_FIELD(aggfilter);
COMPARE_SCALAR_FIELD(aggstar);
COMPARE_SCALAR_FIELD(aggvariadic);
- COMPARE_SCALAR_FIELD(aggcombine);
- COMPARE_SCALAR_FIELD(aggpartial);
COMPARE_SCALAR_FIELD(aggkind);
COMPARE_SCALAR_FIELD(agglevelsup);
+ COMPARE_SCALAR_FIELD(aggsplit);
COMPARE_LOCATION_FIELD(location);
return true;
type = ((const Param *) expr)->paramtype;
break;
case T_Aggref:
- type = ((const Aggref *) expr)->aggoutputtype;
+ type = ((const Aggref *) expr)->aggtype;
break;
case T_GroupingFunc:
type = INT4OID;
_outPlanInfo(str, (const Plan *) node);
WRITE_ENUM_FIELD(aggstrategy, AggStrategy);
- WRITE_BOOL_FIELD(combineStates);
- WRITE_BOOL_FIELD(finalizeAggs);
- WRITE_BOOL_FIELD(serialStates);
+ WRITE_ENUM_FIELD(aggsplit, AggSplit);
WRITE_INT_FIELD(numCols);
appendStringInfoString(str, " :grpColIdx");
WRITE_OID_FIELD(aggfnoid);
WRITE_OID_FIELD(aggtype);
- WRITE_OID_FIELD(aggoutputtype);
WRITE_OID_FIELD(aggcollid);
WRITE_OID_FIELD(inputcollid);
WRITE_OID_FIELD(aggtranstype);
WRITE_NODE_FIELD(aggfilter);
WRITE_BOOL_FIELD(aggstar);
WRITE_BOOL_FIELD(aggvariadic);
- WRITE_BOOL_FIELD(aggcombine);
- WRITE_BOOL_FIELD(aggpartial);
WRITE_CHAR_FIELD(aggkind);
WRITE_UINT_FIELD(agglevelsup);
+ WRITE_ENUM_FIELD(aggsplit, AggSplit);
WRITE_LOCATION_FIELD(location);
}
WRITE_NODE_FIELD(subpath);
WRITE_ENUM_FIELD(aggstrategy, AggStrategy);
+ WRITE_ENUM_FIELD(aggsplit, AggSplit);
WRITE_FLOAT_FIELD(numGroups, "%.0f");
WRITE_NODE_FIELD(groupClause);
WRITE_NODE_FIELD(qual);
READ_OID_FIELD(aggfnoid);
READ_OID_FIELD(aggtype);
- READ_OID_FIELD(aggoutputtype);
READ_OID_FIELD(aggcollid);
READ_OID_FIELD(inputcollid);
READ_OID_FIELD(aggtranstype);
READ_NODE_FIELD(aggfilter);
READ_BOOL_FIELD(aggstar);
READ_BOOL_FIELD(aggvariadic);
- READ_BOOL_FIELD(aggcombine);
- READ_BOOL_FIELD(aggpartial);
READ_CHAR_FIELD(aggkind);
READ_UINT_FIELD(agglevelsup);
+ READ_ENUM_FIELD(aggsplit, AggSplit);
READ_LOCATION_FIELD(location);
READ_DONE();
ReadCommonPlan(&local_node->plan);
READ_ENUM_FIELD(aggstrategy, AggStrategy);
- READ_BOOL_FIELD(combineStates);
- READ_BOOL_FIELD(finalizeAggs);
- READ_BOOL_FIELD(serialStates);
+ READ_ENUM_FIELD(aggsplit, AggSplit);
READ_INT_FIELD(numCols);
READ_ATTRNUMBER_ARRAY(grpColIdx, local_node->numCols);
READ_OID_ARRAY(grpOperators, local_node->numCols);
plan = (Plan *) make_agg(build_path_tlist(root, &best_path->path),
NIL,
AGG_HASHED,
- false,
- true,
- false,
+ AGGSPLIT_SIMPLE,
numGroupCols,
groupColIdx,
groupOperators,
plan = make_agg(tlist, quals,
best_path->aggstrategy,
- best_path->combineStates,
- best_path->finalizeAggs,
- best_path->serialStates,
+ best_path->aggsplit,
list_length(best_path->groupClause),
extract_grouping_cols(best_path->groupClause,
subplan->targetlist),
agg_plan = (Plan *) make_agg(NIL,
NIL,
AGG_SORTED,
- false,
- true,
- false,
+ AGGSPLIT_SIMPLE,
list_length((List *) linitial(gsets)),
new_grpColIdx,
extract_grouping_ops(groupClause),
plan = make_agg(build_path_tlist(root, &best_path->path),
best_path->qual,
(numGroupCols > 0) ? AGG_SORTED : AGG_PLAIN,
- false,
- true,
- false,
+ AGGSPLIT_SIMPLE,
numGroupCols,
top_grpColIdx,
extract_grouping_ops(groupClause),
Agg *
make_agg(List *tlist, List *qual,
- AggStrategy aggstrategy,
- bool combineStates, bool finalizeAggs, bool serialStates,
+ AggStrategy aggstrategy, AggSplit aggsplit,
int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators,
List *groupingSets, List *chain,
double dNumGroups, Plan *lefttree)
numGroups = (long) Min(dNumGroups, (double) LONG_MAX);
node->aggstrategy = aggstrategy;
- node->combineStates = combineStates;
- node->finalizeAggs = finalizeAggs;
- node->serialStates = serialStates;
+ node->aggsplit = aggsplit;
node->numCols = numGroupCols;
node->grpColIdx = grpColIdx;
node->grpOperators = grpOperators;
MemSet(&agg_costs, 0, sizeof(AggClauseCosts));
if (parse->hasAggs)
{
- count_agg_clauses(root, (Node *) target->exprs, &agg_costs, true,
- false, false);
- count_agg_clauses(root, parse->havingQual, &agg_costs, true, false,
- false);
+ get_agg_clause_costs(root, (Node *) target->exprs, AGGSPLIT_SIMPLE,
+ &agg_costs);
+ get_agg_clause_costs(root, parse->havingQual, AGGSPLIT_SIMPLE,
+ &agg_costs);
}
/*
if (parse->hasAggs)
{
/* partial phase */
- count_agg_clauses(root, (Node *) partial_grouping_target->exprs,
- &agg_partial_costs, false, false, true);
+ get_agg_clause_costs(root, (Node *) partial_grouping_target->exprs,
+ AGGSPLIT_INITIAL_SERIAL,
+ &agg_partial_costs);
/* final phase */
- count_agg_clauses(root, (Node *) target->exprs, &agg_final_costs,
- true, true, true);
- count_agg_clauses(root, parse->havingQual, &agg_final_costs, true,
- true, true);
+ get_agg_clause_costs(root, (Node *) target->exprs,
+ AGGSPLIT_FINAL_DESERIAL,
+ &agg_final_costs);
+ get_agg_clause_costs(root, parse->havingQual,
+ AGGSPLIT_FINAL_DESERIAL,
+ &agg_final_costs);
}
if (can_sort)
path,
partial_grouping_target,
parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_INITIAL_SERIAL,
parse->groupClause,
NIL,
&agg_partial_costs,
- dNumPartialGroups,
- false,
- false,
- true));
+ dNumPartialGroups));
else
add_partial_path(grouped_rel, (Path *)
create_group_path(root,
cheapest_partial_path,
partial_grouping_target,
AGG_HASHED,
+ AGGSPLIT_INITIAL_SERIAL,
parse->groupClause,
NIL,
&agg_partial_costs,
- dNumPartialGroups,
- false,
- false,
- true));
+ dNumPartialGroups));
}
}
}
path,
target,
parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_SIMPLE,
parse->groupClause,
(List *) parse->havingQual,
&agg_costs,
- dNumGroups,
- false,
- true,
- false));
+ dNumGroups));
}
else if (parse->groupClause)
{
path,
target,
parse->groupClause ? AGG_SORTED : AGG_PLAIN,
+ AGGSPLIT_FINAL_DESERIAL,
parse->groupClause,
(List *) parse->havingQual,
&agg_final_costs,
- dNumGroups,
- true,
- true,
- true));
+ dNumGroups));
else
add_path(grouped_rel, (Path *)
create_group_path(root,
cheapest_path,
target,
AGG_HASHED,
+ AGGSPLIT_SIMPLE,
parse->groupClause,
(List *) parse->havingQual,
&agg_costs,
- dNumGroups,
- false,
- true,
- false));
+ dNumGroups));
}
/*
path,
target,
AGG_HASHED,
+ AGGSPLIT_FINAL_DESERIAL,
parse->groupClause,
(List *) parse->havingQual,
&agg_final_costs,
- dNumGroups,
- true,
- true,
- true));
+ dNumGroups));
}
}
}
cheapest_input_path,
cheapest_input_path->pathtarget,
AGG_HASHED,
+ AGGSPLIT_SIMPLE,
parse->distinctClause,
NIL,
NULL,
- numDistinctRows,
- false,
- true,
- false));
+ numDistinctRows));
}
/* Give a helpful error if we failed to find any implementation */
newaggref = makeNode(Aggref);
memcpy(newaggref, aggref, sizeof(Aggref));
- /* XXX assume serialization required */
- mark_partial_aggref(newaggref, true);
+ /* For now, assume serialization is required */
+ mark_partial_aggref(newaggref, AGGSPLIT_INITIAL_SERIAL);
lfirst(lc) = newaggref;
}
/*
* mark_partial_aggref
- * Adjust an Aggref to make it represent the output of partial aggregation.
+ * Adjust an Aggref to make it represent a partial-aggregation step.
*
* The Aggref node is modified in-place; caller must do any copying required.
*/
void
-mark_partial_aggref(Aggref *agg, bool serialize)
+mark_partial_aggref(Aggref *agg, AggSplit aggsplit)
{
/* aggtranstype should be computed by this point */
Assert(OidIsValid(agg->aggtranstype));
+ /* ... but aggsplit should still be as the parser left it */
+ Assert(agg->aggsplit == AGGSPLIT_SIMPLE);
+
+ /* Mark the Aggref with the intended partial-aggregation mode */
+ agg->aggsplit = aggsplit;
/*
- * Normally, a partial aggregate returns the aggregate's transition type;
- * but if that's INTERNAL and we're serializing, it returns BYTEA instead.
+ * Adjust result type if needed. Normally, a partial aggregate returns
+ * the aggregate's transition type; but if that's INTERNAL and we're
+ * serializing, it returns BYTEA instead.
*/
- if (agg->aggtranstype == INTERNALOID && serialize)
- agg->aggoutputtype = BYTEAOID;
- else
- agg->aggoutputtype = agg->aggtranstype;
-
- /* flag it as partial */
- agg->aggpartial = true;
+ if (DO_AGGSPLIT_SKIPFINAL(aggsplit))
+ {
+ if (agg->aggtranstype == INTERNALOID && DO_AGGSPLIT_SERIALIZE(aggsplit))
+ agg->aggtype = BYTEAOID;
+ else
+ agg->aggtype = agg->aggtranstype;
+ }
}
/*
* partial-aggregate subexpressions that will be available
* from the child plan node.
*/
- if (agg->combineStates)
+ if (DO_AGGSPLIT_COMBINE(agg->aggsplit))
{
plan->targetlist = (List *)
convert_combining_aggrefs((Node *) plan->targetlist,
/*
* Now, set up child_agg to represent the first phase of partial
- * aggregation. XXX assume serialization required.
+ * aggregation. For now, assume serialization is required.
*/
- mark_partial_aggref(child_agg, true);
+ mark_partial_aggref(child_agg, AGGSPLIT_INITIAL_SERIAL);
/*
* And set up parent_agg to represent the second phase.
*/
parent_agg->args = list_make1(makeTargetEntry((Expr *) child_agg,
1, NULL, false));
- parent_agg->aggcombine = true;
+ mark_partial_aggref(parent_agg, AGGSPLIT_FINAL_DESERIAL);
return (Node *) parent_agg;
}
path,
create_pathtarget(root, tlist),
AGG_HASHED,
+ AGGSPLIT_SIMPLE,
groupList,
NIL,
NULL,
- dNumGroups,
- false,
- true,
- false);
+ dNumGroups);
}
else
{
typedef struct
{
PlannerInfo *root;
+ AggSplit aggsplit;
AggClauseCosts *costs;
- bool finalizeAggs;
- bool combineStates;
- bool serialStates;
-} count_agg_clauses_context;
+} get_agg_clause_costs_context;
typedef struct
{
static bool aggregates_allow_partial_walker(Node *node,
partial_agg_context *context);
static bool contain_agg_clause_walker(Node *node, void *context);
-static bool count_agg_clauses_walker(Node *node,
- count_agg_clauses_context *context);
+static bool get_agg_clause_costs_walker(Node *node,
+ get_agg_clause_costs_context *context);
static bool find_window_functions_walker(Node *node, WindowFuncLists *lists);
static bool expression_returns_set_rows_walker(Node *node, double *count);
static bool contain_subplans_walker(Node *node, void *context);
}
/*
- * count_agg_clauses
- * Recursively count the Aggref nodes in an expression tree, and
- * accumulate other information about them too.
+ * get_agg_clause_costs
+ * Recursively find the Aggref nodes in an expression tree, and
+ * accumulate cost information about them.
*
- * Note: this also checks for nested aggregates, which are an error.
+ * 'aggsplit' tells us the expected partial-aggregation mode, which affects
+ * the cost estimates.
*
- * We not only count the nodes, but estimate their execution costs, and
- * attempt to estimate the total space needed for their transition state
- * values if all are evaluated in parallel (as would be done in a HashAgg
- * plan). See AggClauseCosts for the exact set of statistics collected.
+ * NOTE that the counts/costs are ADDED to those already in *costs ... so
+ * the caller is responsible for zeroing the struct initially.
+ *
+ * We count the nodes, estimate their execution costs, and estimate the total
+ * space needed for their transition state values if all are evaluated in
+ * parallel (as would be done in a HashAgg plan). See AggClauseCosts for
+ * the exact set of statistics collected.
*
* In addition, we mark Aggref nodes with the correct aggtranstype, so
* that that doesn't need to be done repeatedly. (That makes this function's
* name a bit of a misnomer.)
*
- * NOTE that the counts/costs are ADDED to those already in *costs ... so
- * the caller is responsible for zeroing the struct initially.
- *
* This does not descend into subqueries, and so should be used only after
* reduction of sublinks to subplans, or in contexts where it's known there
* are no subqueries. There mustn't be outer-aggregate references either.
*/
void
-count_agg_clauses(PlannerInfo *root, Node *clause, AggClauseCosts *costs,
- bool finalizeAggs, bool combineStates, bool serialStates)
+get_agg_clause_costs(PlannerInfo *root, Node *clause, AggSplit aggsplit,
+ AggClauseCosts *costs)
{
- count_agg_clauses_context context;
+ get_agg_clause_costs_context context;
context.root = root;
+ context.aggsplit = aggsplit;
context.costs = costs;
- context.finalizeAggs = finalizeAggs;
- context.combineStates = combineStates;
- context.serialStates = serialStates;
- (void) count_agg_clauses_walker(clause, &context);
+ (void) get_agg_clause_costs_walker(clause, &context);
}
static bool
-count_agg_clauses_walker(Node *node, count_agg_clauses_context *context)
+get_agg_clause_costs_walker(Node *node, get_agg_clause_costs_context *context)
{
if (node == NULL)
return false;
* Add the appropriate component function execution costs to
* appropriate totals.
*/
- if (context->combineStates)
+ if (DO_AGGSPLIT_COMBINE(context->aggsplit))
{
/* charge for combining previously aggregated states */
costs->transCost.per_tuple += get_func_cost(aggcombinefn) * cpu_operator_cost;
-
- /* charge for deserialization, when appropriate */
- if (context->serialStates && OidIsValid(aggdeserialfn))
- costs->transCost.per_tuple += get_func_cost(aggdeserialfn) * cpu_operator_cost;
}
else
costs->transCost.per_tuple += get_func_cost(aggtransfn) * cpu_operator_cost;
-
- if (context->finalizeAggs)
- {
- if (OidIsValid(aggfinalfn))
- costs->finalCost += get_func_cost(aggfinalfn) * cpu_operator_cost;
- }
- else if (context->serialStates)
- {
- if (OidIsValid(aggserialfn))
- costs->finalCost += get_func_cost(aggserialfn) * cpu_operator_cost;
- }
+ if (DO_AGGSPLIT_DESERIALIZE(context->aggsplit) &&
+ OidIsValid(aggdeserialfn))
+ costs->transCost.per_tuple += get_func_cost(aggdeserialfn) * cpu_operator_cost;
+ if (DO_AGGSPLIT_SERIALIZE(context->aggsplit) &&
+ OidIsValid(aggserialfn))
+ costs->finalCost += get_func_cost(aggserialfn) * cpu_operator_cost;
+ if (!DO_AGGSPLIT_SKIPFINAL(context->aggsplit) &&
+ OidIsValid(aggfinalfn))
+ costs->finalCost += get_func_cost(aggfinalfn) * cpu_operator_cost;
/*
- * Some costs will already have been incurred by the initial aggregate
- * node, so we mustn't include these again.
+ * These costs are incurred only by the initial aggregate node, so we
+ * mustn't include them again at upper levels.
*/
- if (!context->combineStates)
+ if (!DO_AGGSPLIT_COMBINE(context->aggsplit))
{
/* add the input expressions' cost to per-input-row costs */
cost_qual_eval_node(&argcosts, (Node *) aggref->args, context->root);
/*
* We assume that the parser checked that there are no aggregates (of
* this level anyway) in the aggregated arguments, direct arguments,
- * or filter clause. Hence, we need not recurse into any of them. (If
- * either the parser or the planner screws up on this point, the
- * executor will still catch it; see ExecInitExpr.)
+ * or filter clause. Hence, we need not recurse into any of them.
*/
return false;
}
Assert(!IsA(node, SubLink));
- return expression_tree_walker(node, count_agg_clauses_walker,
+ return expression_tree_walker(node, get_agg_clause_costs_walker,
(void *) context);
}
* 'subpath' is the path representing the source of data
* 'target' is the PathTarget to be computed
* 'aggstrategy' is the Agg node's basic implementation strategy
+ * 'aggsplit' is the Agg node's aggregate-splitting mode
* 'groupClause' is a list of SortGroupClause's representing the grouping
* 'qual' is the HAVING quals if any
* 'aggcosts' contains cost info about the aggregate functions to be computed
* 'numGroups' is the estimated number of groups (1 if not grouping)
- * 'combineStates' is set to true if the Agg node should combine agg states
- * 'finalizeAggs' is set to false if the Agg node should not call the finalfn
*/
AggPath *
create_agg_path(PlannerInfo *root,
Path *subpath,
PathTarget *target,
AggStrategy aggstrategy,
+ AggSplit aggsplit,
List *groupClause,
List *qual,
const AggClauseCosts *aggcosts,
- double numGroups,
- bool combineStates,
- bool finalizeAggs,
- bool serialStates)
+ double numGroups)
{
AggPath *pathnode = makeNode(AggPath);
pathnode->subpath = subpath;
pathnode->aggstrategy = aggstrategy;
+ pathnode->aggsplit = aggsplit;
pathnode->numGroups = numGroups;
pathnode->groupClause = groupClause;
pathnode->qual = qual;
- pathnode->finalizeAggs = finalizeAggs;
- pathnode->combineStates = combineStates;
- pathnode->serialStates = serialStates;
cost_agg(&pathnode->path, root,
aggstrategy, aggcosts,
Aggref *aggref = makeNode(Aggref);
aggref->aggfnoid = funcid;
- /* default the outputtype to be the same as aggtype */
- aggref->aggtype = aggref->aggoutputtype = rettype;
+ aggref->aggtype = rettype;
/* aggcollid and inputcollid will be set by parse_collate.c */
aggref->aggtranstype = InvalidOid; /* will be set by planner */
/* aggargtypes will be set by transformAggregateCall */
aggref->aggfilter = agg_filter;
aggref->aggstar = agg_star;
aggref->aggvariadic = func_variadic;
- /* at this point, the Aggref is never partial or combining */
- aggref->aggcombine = aggref->aggpartial = false;
aggref->aggkind = aggkind;
/* agglevelsup will be set by transformAggregateCall */
+ aggref->aggsplit = AGGSPLIT_SIMPLE; /* planner might change this */
aggref->location = location;
/*
* one element, which will point to a partial Aggref that supplies us with
* transition states to combine.
*/
- if (aggref->aggcombine)
+ if (DO_AGGSPLIT_COMBINE(aggref->aggsplit))
{
TargetEntry *tle = linitial(aggref->args);
return;
}
- /* Mark as PARTIAL, if appropriate. */
- if (original_aggref->aggpartial)
+ /*
+ * Mark as PARTIAL, if appropriate. We look to the original aggref so as
+ * to avoid printing this when recursing from the code just above.
+ */
+ if (DO_AGGSPLIT_SKIPFINAL(original_aggref->aggsplit))
appendStringInfoString(buf, "PARTIAL ");
/* Extract the argument types as seen by the parser */
*/
/* yyyymmddN */
-#define CATALOG_VERSION_NO 201606221
+#define CATALOG_VERSION_NO 201606261
#endif
List *aggs; /* all Aggref nodes in targetlist & quals */
int numaggs; /* length of list (could be zero!) */
int numtrans; /* number of pertrans items */
+ AggSplit aggsplit; /* agg-splitting mode, see nodes.h */
AggStatePerPhase phase; /* pointer to current phase data */
int numphases; /* number of phases */
int current_phase; /* current phase number */
AggStatePerTrans curpertrans; /* currently active trans state */
bool input_done; /* indicates end of input */
bool agg_done; /* indicates completion of Agg scan */
- bool combineStates; /* input tuples contain transition states */
- bool finalizeAggs; /* should we call the finalfn on agg states? */
- bool serialStates; /* should agg states be (de)serialized? */
int projected_set; /* The last projected grouping set */
int current_set; /* The current grouping set being evaluated */
Bitmapset *grouped_cols; /* grouped cols in current projection */
AGG_HASHED /* grouped agg, use internal hashtable */
} AggStrategy;
+/*
+ * AggSplit -
+ * splitting (partial aggregation) modes for Agg plan nodes
+ *
+ * This is needed in both plannodes.h and relation.h, so put it here...
+ */
+
+/* Primitive options supported by nodeAgg.c: */
+#define AGGSPLITOP_COMBINE 0x01 /* substitute combinefn for transfn */
+#define AGGSPLITOP_SKIPFINAL 0x02 /* skip finalfn, return state as-is */
+#define AGGSPLITOP_SERIALIZE 0x04 /* apply serializefn to output */
+#define AGGSPLITOP_DESERIALIZE 0x08 /* apply deserializefn to input */
+
+/* Supported operating modes (i.e., useful combinations of these options): */
+typedef enum AggSplit
+{
+ /* Basic, non-split aggregation: */
+ AGGSPLIT_SIMPLE = 0,
+ /* Initial phase of partial aggregation, with serialization: */
+ AGGSPLIT_INITIAL_SERIAL = AGGSPLITOP_SKIPFINAL | AGGSPLITOP_SERIALIZE,
+ /* Final phase of partial aggregation, with deserialization: */
+ AGGSPLIT_FINAL_DESERIAL = AGGSPLITOP_COMBINE | AGGSPLITOP_DESERIALIZE
+} AggSplit;
+
+/* Test whether an AggSplit value selects each primitive option: */
+#define DO_AGGSPLIT_COMBINE(as) (((as) & AGGSPLITOP_COMBINE) != 0)
+#define DO_AGGSPLIT_SKIPFINAL(as) (((as) & AGGSPLITOP_SKIPFINAL) != 0)
+#define DO_AGGSPLIT_SERIALIZE(as) (((as) & AGGSPLITOP_SERIALIZE) != 0)
+#define DO_AGGSPLIT_DESERIALIZE(as) (((as) & AGGSPLITOP_DESERIALIZE) != 0)
+
/*
* SetOpCmd and SetOpStrategy -
* overall semantics and execution strategies for SetOp plan nodes
{
Plan plan;
AggStrategy aggstrategy; /* basic strategy, see nodes.h */
- bool combineStates; /* input tuples contain transition states */
- bool finalizeAggs; /* should we call the finalfn on agg states? */
- bool serialStates; /* should agg states be (de)serialized? */
+ AggSplit aggsplit; /* agg-splitting mode, see nodes.h */
int numCols; /* number of grouping columns */
AttrNumber *grpColIdx; /* their indexes in the target list */
Oid *grpOperators; /* equality operators to compare with */
* replaced with a single argument representing the partial-aggregate
* transition values.
*
- * XXX need more documentation about partial aggregation here
- *
- * 'aggtype' and 'aggoutputtype' are the same except when we're performing
- * partal aggregation; in that case, we output transition states. Nothing
- * interesting happens in the Aggref itself, but we must set the output data
- * type to whatever type is used for transition values.
- *
- * Note: If you are adding fields here you may also need to add a comparison
- * in search_indexed_tlist_for_partial_aggref()
+ * aggsplit indicates the expected partial-aggregation mode for the Aggref's
+ * parent plan node. It's always set to AGGSPLIT_SIMPLE in the parser, but
+ * the planner might change it to something else. We use this mainly as
+ * a crosscheck that the Aggrefs match the plan; but note that when aggsplit
+ * indicates a non-final mode, aggtype reflects the transition data type
+ * not the SQL-level output type of the aggregate.
*/
typedef struct Aggref
{
Expr xpr;
Oid aggfnoid; /* pg_proc Oid of the aggregate */
- Oid aggtype; /* type Oid of final result of the aggregate */
- Oid aggoutputtype; /* type Oid of result of this aggregate */
+ Oid aggtype; /* type Oid of result of the aggregate */
Oid aggcollid; /* OID of collation of result */
Oid inputcollid; /* OID of collation that function should use */
Oid aggtranstype; /* type Oid of aggregate's transition value */
bool aggstar; /* TRUE if argument list was really '*' */
bool aggvariadic; /* true if variadic arguments have been
* combined into an array last argument */
- bool aggcombine; /* combining agg; input is a transvalue */
- bool aggpartial; /* partial agg; output is a transvalue */
char aggkind; /* aggregate kind (see pg_aggregate.h) */
Index agglevelsup; /* > 0 if agg belongs to outer query */
+ AggSplit aggsplit; /* expected agg-splitting mode of parent Agg */
int location; /* token location, or -1 if unknown */
} Aggref;
Path path;
Path *subpath; /* path representing input source */
AggStrategy aggstrategy; /* basic strategy, see nodes.h */
+ AggSplit aggsplit; /* agg-splitting mode, see nodes.h */
double numGroups; /* estimated number of groups in input */
List *groupClause; /* a list of SortGroupClause's */
List *qual; /* quals (HAVING quals), if any */
- bool combineStates; /* input is partially aggregated agg states */
- bool finalizeAggs; /* should the executor call the finalfn? */
- bool serialStates; /* should agg states be (de)serialized? */
} AggPath;
/*
extern PartialAggType aggregates_allow_partial(Node *clause);
extern bool contain_agg_clause(Node *clause);
-extern void count_agg_clauses(PlannerInfo *root, Node *clause,
- AggClauseCosts *costs, bool finalizeAggs,
- bool combineStates, bool serialStates);
+extern void get_agg_clause_costs(PlannerInfo *root, Node *clause,
+ AggSplit aggsplit, AggClauseCosts *costs);
extern bool contain_window_function(Node *clause);
extern WindowFuncLists *find_window_functions(Node *clause, Index maxWinRef);
Path *subpath,
PathTarget *target,
AggStrategy aggstrategy,
+ AggSplit aggsplit,
List *groupClause,
List *qual,
const AggClauseCosts *aggcosts,
- double numGroups,
- bool combineStates,
- bool finalizeAggs,
- bool serialStates);
+ double numGroups);
extern GroupingSetsPath *create_groupingsets_path(PlannerInfo *root,
RelOptInfo *rel,
Path *subpath,
/* External use of these functions is deprecated: */
extern Sort *make_sort_from_sortclauses(List *sortcls, Plan *lefttree);
-extern Agg *make_agg(List *tlist, List *qual, AggStrategy aggstrategy,
- bool combineStates, bool finalizeAggs, bool serialStates,
+extern Agg *make_agg(List *tlist, List *qual,
+ AggStrategy aggstrategy, AggSplit aggsplit,
int numGroupCols, AttrNumber *grpColIdx, Oid *grpOperators,
List *groupingSets, List *chain,
double dNumGroups, Plan *lefttree);
extern RowMarkType select_rowmark_type(RangeTblEntry *rte,
LockClauseStrength strength);
-extern void mark_partial_aggref(Aggref *agg, bool serialize);
+extern void mark_partial_aggref(Aggref *agg, AggSplit aggsplit);
extern Path *get_cheapest_fractional_path(RelOptInfo *rel,
double tuple_fraction);
projects / postgresql / commitdiff