Google Git
Sign in
apache / hawq / 91c45cab0e5844abfe0f398f2378637f4028fe45 / . / src / backend / executor / nodeAgg.c
blob: e4eb79171b70d530af262ab27962f6241acb3fc1 [file] [log] [blame]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
/*-------------------------------------------------------------------------
*
* nodeAgg.c
* Routines to handle aggregate nodes.
*
* ExecAgg evaluates each aggregate in the following steps:
*
* transvalue = initcond
* foreach input_tuple do
* transvalue = transfunc(transvalue, input_value(s))
* result = finalfunc(transvalue)
*
* If a finalfunc is not supplied then the result is just the ending
* value of transvalue.
*
* If transfunc is marked "strict" in pg_proc and initcond is NULL,
* then the first non-NULL input_value is assigned directly to transvalue,
* and transfunc isn't applied until the second non-NULL input_value.
* The agg's first input type and transtype must be the same in this case!
*
* If transfunc is marked "strict" then NULL input_values are skipped,
* keeping the previous transvalue. If transfunc is not strict then it
* is called for every input tuple and must deal with NULL initcond
* or NULL input_values for itself.
*
* If finalfunc is marked "strict" then it is not called when the
* ending transvalue is NULL, instead a NULL result is created
* automatically (this is just the usual handling of strict functions,
* of course). A non-strict finalfunc can make its own choice of
* what to return for a NULL ending transvalue.
*
* We compute aggregate input expressions and run the transition functions
* in a temporary econtext (aggstate->tmpcontext). This is reset at
* least once per input tuple, so when the transvalue datatype is
* pass-by-reference, we have to be careful to copy it into a longer-lived
* memory context, and free the prior value to avoid memory leakage.
*
* Postgres stores transvalues in the memory context aggstate->aggcontext,
* which is also used for the hashtable structures in AGG_HASHED mode.
* MPP (in order to support hybrid hash aggregation) stores hash table
* entries and associated transition values in aggstate->aggcontext.
*
* The node's regular econtext (aggstate->ss.ps.ps_ExprContext)
* is used to run finalize functions and compute the output tuple;
* this context can be reset once per output tuple.
*
* Beginning in PostgreSQL 8.1, the executor's AggState node is passed as
* the fmgr "context" value in all transfunc and finalfunc calls. It is
* not really intended that the transition functions will look into the
* AggState node, but they can use code like
* if (fcinfo->context && IsA(fcinfo->context, AggState))
* to verify that they are being called by nodeAgg.c and not as ordinary
* SQL functions. The main reason a transition function might want to know
* that is that it can avoid palloc'ing a fixed-size pass-by-ref transition
* value on every call: it can instead just scribble on and return its left
* input. Ordinarily it is completely forbidden for functions to modify
* pass-by-ref inputs, 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 previous transition value pointer is returned.
*
* In Greenplum 4.2.2, we add PercentileExpr support along with Aggref.
* It is used to implement inverse distribution function support, namely
* percentile_cont, percentile_disc and median. The semantics for them
* is almost same as Aggref, where the aggregate process is handled by
* an individual function and the expression node only returns a pre-computed
* result. PercentileExpr is used in Agg node because we cannot change
* the catalog in this release, and it may be removed and integrated to
* standard Aggref itself.
*
* Portions Copyright (c) 2007-2008, Greenplum inc
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/executor/nodeAgg.c,v 1.146.2.2 2007/08/08 18:07:03 neilc Exp $
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/heapam.h"
#include "catalog/catquery.h"
#include "catalog/pg_aggregate.h"
#include "catalog/pg_proc.h"
#include "catalog/pg_type.h"
#include "executor/executor.h"
#include "executor/execHHashagg.h"
#include "executor/nodeAgg.h"
#include "lib/stringinfo.h" /* StringInfo */
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "optimizer/clauses.h"
#include "optimizer/tlist.h"
#include "parser/parse_agg.h"
#include "parser/parse_coerce.h"
#include "parser/parse_expr.h"
#include "parser/parse_oper.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/syscache.h"
#include "utils/tuplesort.h"
#include "utils/datum.h"
#include "cdb/cdbexplain.h"
#include "cdb/cdbvars.h" /* mpp_hybrid_hash_agg */
#define IS_HASHAGG(aggstate) (((Agg *) (aggstate)->ss.ps.plan)->aggstrategy == AGG_HASHED)
/*
* AggStatePerAggData -- per-aggregate working state
* AggStatePerGroupData - per-aggregate-per-group working state
*
* Definition moved to nodeAgg.c to provide visibility to execHHashagg.c
*/
/*
* To implement hashed aggregation, we need a hashtable that stores a
* representative tuple and an array of AggStatePerGroup structs for each
* distinct set of GROUP BY column values. We compute the hash key from
* the GROUP BY columns.
*/
typedef struct AggHashEntryData *AggHashEntry;
typedef struct AggHashEntryData
{
TupleHashEntryData shared; /* common header for hash table entries */
/* per-aggregate transition status array - must be last! */
AggStatePerGroupData pergroup[1]; /* VARIABLE LENGTH ARRAY */
} AggHashEntryData; /* VARIABLE LENGTH STRUCT */
static void advance_transition_function(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate,
FunctionCallInfoData *fcinfo,
MemoryManagerContainer *mem_manager);
static void process_ordered_aggregate_single(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate);
static void process_ordered_aggregate_multi(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate);
static void finalize_aggregate(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate,
Datum *resultVal, bool *resultIsNull);
static void finalize_aggregates(AggState *aggstate, AggStatePerGroup pergroup);
static Bitmapset *find_unaggregated_cols(AggState *aggstate);
static bool find_unaggregated_cols_walker(Node *node, Bitmapset **colnos);
static TupleTableSlot *agg_retrieve_direct(AggState *aggstate);
static TupleTableSlot *agg_retrieve_hash_table(AggState *aggstate);
static void ExecAggExplainEnd(PlanState *planstate, struct StringInfoData *buf);
static int count_extra_agg_slots(Node *node);
static bool count_extra_agg_slots_walker(Node *node, int *count);
static TupleTableSlot *computeTupleWithFinalAggregate(AggState *aggstate,
TupleTableSlot *firstSlot);
Datum
datumCopyWithMemManager(Datum oldvalue, Datum value, bool typByVal, int typLen,
MemoryManagerContainer *mem_manager)
{
Datum res;
if (typByVal)
res = value;
else
{
Size realSize;
Size old_realSize = 0;
char *s;
if (DatumGetPointer(value) == NULL)
return PointerGetDatum(NULL);
if (DatumGetPointer(oldvalue) != NULL)
old_realSize = MAXALIGN(datumGetSize(oldvalue, typByVal, typLen));
realSize = datumGetSize(value, typByVal, typLen);
if (old_realSize == 0 || old_realSize < realSize)
{
int alloc_size = MAXALIGN(mem_manager->realloc_ratio * old_realSize);
if (alloc_size < realSize)
alloc_size = MAXALIGN(realSize);
if (mem_manager->free)
{
(*mem_manager->free)(mem_manager->manager, DatumGetPointer(oldvalue));
}
s = (char *) (*mem_manager->alloc)(mem_manager->manager, alloc_size);
}
else
s = (char *) DatumGetPointer(oldvalue);
memcpy(s, DatumGetPointer(value), realSize);
res = PointerGetDatum(s);
}
return res;
}
/*
* Initialize all aggregates for a new group of input values.
*
* When called, CurrentMemoryContext should be the per-query context.
*
* Note that the memory allocation is done through provided memory manager.
*/
void
initialize_aggregates(AggState *aggstate,
AggStatePerAgg peragg,
AggStatePerGroup pergroup,
MemoryManagerContainer *mem_manager)
{
int aggno;
for (aggno = 0; aggno < aggstate->numaggs; aggno++)
{
AggStatePerAgg peraggstate = &peragg[aggno];
AggStatePerGroup pergroupstate = &pergroup[aggno];
/*
* Start a fresh sort operation for each DISTINCT/ORDER BY aggregate.
*/
if (peraggstate->numSortCols > 0)
{
/*
* In case of rescan, maybe there could be an uncompleted sort
* operation? Clean it up if so.
*/
if(gp_enable_mk_sort)
{
if (peraggstate->sortstate)
tuplesort_end_mk((Tuplesortstate_mk *) peraggstate->sortstate);
/*
* We use a plain Datum sorter when there's a single input column;
* otherwise sort the full tuple. (See comments for
* process_ordered_aggregate_single.)
*/
peraggstate->sortstate =
(peraggstate->numInputs == 1) ?
tuplesort_begin_datum_mk(& aggstate->ss,
peraggstate->evaldesc->attrs[0]->atttypid,
peraggstate->sortOperators[0],
PlanStateOperatorMemKB((PlanState *) aggstate), false) :
tuplesort_begin_heap_mk(& aggstate->ss,
peraggstate->evaldesc,
peraggstate->numSortCols,
peraggstate->sortOperators,
peraggstate->sortColIdx,
PlanStateOperatorMemKB((PlanState *) aggstate), false);
/*
* CDB: If EXPLAIN ANALYZE, let all of our tuplesort operations
* share our Instrumentation object and message buffer.
*/
if (aggstate->ss.ps.instrument)
tuplesort_set_instrument_mk((Tuplesortstate_mk *) peraggstate->sortstate,
aggstate->ss.ps.instrument,
aggstate->ss.ps.cdbexplainbuf);
}
else /* gp_enable_mk_sort is off */
{
if (peraggstate->sortstate)
tuplesort_end((Tuplesortstate *) peraggstate->sortstate);
/*
* We use a plain Datum sorter when there's a single input column;
* otherwise sort the full tuple. (See comments for
* process_ordered_aggregate_single.)
*/
peraggstate->sortstate =
(peraggstate->numInputs == 1) ?
tuplesort_begin_datum(peraggstate->evaldesc->attrs[0]->atttypid,
peraggstate->sortOperators[0],
PlanStateOperatorMemKB((PlanState *) aggstate), false) :
tuplesort_begin_heap(peraggstate->evaldesc,
peraggstate->numSortCols,
peraggstate->sortOperators,
peraggstate->sortColIdx,
PlanStateOperatorMemKB((PlanState *) aggstate), false);
/*
* CDB: If EXPLAIN ANALYZE, let all of our tuplesort operations
* share our Instrumentation object and message buffer.
*/
if (aggstate->ss.ps.instrument)
tuplesort_set_instrument((Tuplesortstate *) peraggstate->sortstate,
aggstate->ss.ps.instrument,
aggstate->ss.ps.cdbexplainbuf);
}
/* CDB: Set enhanced sort options. */
{
int64 limit = 0;
int64 offset = 0;
int unique = peragg->aggref->aggdistinct &&
( gp_enable_sort_distinct ? 1 : 0) ;
int sort_flags = gp_sort_flags; /* get the guc */
int maxdistinct = gp_sort_max_distinct; /* get guc */
if(gp_enable_mk_sort)
cdb_tuplesort_init_mk((Tuplesortstate_mk *) peraggstate->sortstate,
offset, limit, unique,
sort_flags, maxdistinct);
else
cdb_tuplesort_init((Tuplesortstate *) peraggstate->sortstate,
offset, limit, unique,
sort_flags, maxdistinct);
}
}
/*
* (Re)set transValue to the initial value.
*
* Note that when the initial value is pass-by-ref, we must copy it
* (into the aggcontext) since we will pfree the transValue later.
*/
if (peraggstate->initValueIsNull)
{
pergroupstate->transValue = peraggstate->initValue;
}
else
{
pergroupstate->transValue =
datumCopyWithMemManager(0,
peraggstate->initValue,
peraggstate->transtypeByVal,
peraggstate->transtypeLen,
mem_manager);
}
pergroupstate->transValueIsNull = peraggstate->initValueIsNull;
/*
* 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.
*/
pergroupstate->noTransValue = peraggstate->initValueIsNull;
}
}
/*
* Given new input value(s), advance the transition function of an aggregate.
*
* The new values (and null flags) have been preloaded into argument positions
* 1 and up in fcinfo, so that we needn't copy them again to pass to the
* transition function. No other fields of fcinfo are assumed valid.
*
* It doesn't matter which memory context this is called in.
*/
static void
advance_transition_function(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate,
FunctionCallInfoData *fcinfo,
MemoryManagerContainer *mem_manager)
{
pergroupstate->transValue =
invoke_agg_trans_func(&(peraggstate->transfn),
peraggstate->numArguments,
pergroupstate->transValue,
&(pergroupstate->noTransValue),
&(pergroupstate->transValueIsNull),
peraggstate->transtypeByVal,
peraggstate->transtypeLen,
fcinfo, (void *)aggstate,
aggstate->tmpcontext->ecxt_per_tuple_memory,
mem_manager);
}
Datum
invoke_agg_trans_func(FmgrInfo *transfn, int numargs, Datum transValue,
bool *noTransvalue, bool *transValueIsNull,
bool transtypeByVal, int16 transtypeLen,
FunctionCallInfoData *fcinfo, void *funcctx,
MemoryContext tuplecontext,
MemoryManagerContainer *mem_manager)
{
MemoryContext oldContext;
Datum newVal;
int i;
if (transfn->fn_strict)
{
/*
* For a strict transfn, nothing happens when there's a NULL input; we
* just keep the prior transValue.
*/
for (i = 1; i <= numargs; i++)
{
if (fcinfo->argnull[i])
return transValue;
}
if (*noTransvalue)
{
/*
* transValue has not been initialized. This is the first non-NULL
* input value. We use it as the initial value for transValue. (We
* already checked that the agg's input type is binary-compatible
* with its transtype, so straight copy here is OK.)
*
* 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.
*/
newVal = datumCopyWithMemManager(transValue, fcinfo->arg[1], transtypeByVal,
transtypeLen, mem_manager);
*transValueIsNull = false;
*noTransvalue = false;
fcinfo->isnull = false;
return newVal;
}
if (*transValueIsNull)
{
/*
* Don't call a strict function with NULL inputs. Note it is
* possible to get here despite the above tests, if the transfn is
* strict *and* returned a NULL on a prior cycle. If that happens
* we will propagate the NULL all the way to the end.
*/
fcinfo->isnull = true;
return transValue;
}
}
/* We run the transition functions in per-input-tuple memory context */
oldContext = MemoryContextSwitchTo(tuplecontext);
/*
* OK to call the transition function
*/
InitFunctionCallInfoData(*fcinfo, transfn,
numargs + 1,
(void *) funcctx, NULL);
fcinfo->arg[0] = transValue;
fcinfo->argnull[0] = *transValueIsNull;
newVal = FunctionCallInvoke(fcinfo);
/*
* If pass-by-ref datatype, must copy the new value into aggcontext and
* pfree the prior transValue. But if transfn returned a pointer to its
* first input, we don't need to do anything.
*/
if (!transtypeByVal &&
DatumGetPointer(newVal) != DatumGetPointer(transValue))
{
if (!fcinfo->isnull)
{
newVal = datumCopyWithMemManager(transValue, newVal, transtypeByVal,
transtypeLen, mem_manager);
}
}
*transValueIsNull = fcinfo->isnull;
if (!fcinfo->isnull)
*noTransvalue = false;
MemoryContextSwitchTo(oldContext);
return newVal;
}
/*
* Advance all the aggregates for one input tuple. The input tuple
* has been stored in tmpcontext->ecxt_scantuple, so that it is accessible
* to ExecEvalExpr. pergroup is the array of per-group structs to use
* (this might be in a hashtable entry).
*
* When called, CurrentMemoryContext should be the per-query context.
*/
void
advance_aggregates(AggState *aggstate, AggStatePerGroup pergroup,
MemoryManagerContainer *mem_manager)
{
int aggno;
for (aggno = 0; aggno < aggstate->numaggs; aggno++)
{
Datum value;
bool isnull;
AggStatePerAgg peraggstate = &aggstate->peragg[aggno];
AggStatePerGroup pergroupstate = &pergroup[aggno];
Aggref *aggref = peraggstate->aggref;
PercentileExpr *perc = peraggstate->perc;
int i;
TupleTableSlot *slot;
int nargs;
if (aggref)
nargs = list_length(aggref->args);
else
{
Assert (perc);
nargs = list_length(perc->args);
}
/* Evaluate the current input expressions for this aggregate */
slot = ExecProject(peraggstate->evalproj, NULL);
slot_getallattrs(slot);
if (peraggstate->numSortCols > 0)
{
/* DISTINCT and/or ORDER BY case */
Assert(slot->PRIVATE_tts_nvalid == peraggstate->numInputs);
Assert(!perc);
/*
* 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 numArguments columns,
* not numInputs, since nullity in columns used only for sorting
* is not relevant here.
*/
if (peraggstate->transfn.fn_strict)
{
for (i = 0; i < nargs; i++)
{
value = slot_getattr(slot, i+1, &isnull);
if (isnull)
break; /* arg loop */
}
if (i < nargs)
continue; /* aggno loop */
}
/* OK, put the tuple into the tuplesort object */
if (peraggstate->numInputs == 1)
{
value = slot_getattr(slot, 1, &isnull);
if (gp_enable_mk_sort)
tuplesort_putdatum_mk((Tuplesortstate_mk*) peraggstate->sortstate,
value,
isnull);
else
tuplesort_putdatum((Tuplesortstate*) peraggstate->sortstate,
value,
isnull);
}
else
{
if (gp_enable_mk_sort)
tuplesort_puttupleslot_mk((Tuplesortstate_mk*) peraggstate->sortstate,
slot);
else
tuplesort_puttupleslot((Tuplesortstate*) peraggstate->sortstate,
slot);
}
}
else
{
/* We can apply the transition function immediately */
FunctionCallInfoData fcinfo;
/* Load values into fcinfo */
/* Start from 1, since the 0th arg will be the transition value */
Assert(slot->PRIVATE_tts_nvalid >= nargs);
if (aggref)
{
for (i = 0; i < nargs; i++)
{
fcinfo.arg[i + 1] = slot_getattr(slot, i+1, &isnull);
fcinfo.argnull[i + 1] = isnull;
}
}
else
{
/*
* In case of percentile functions, put everything into
* fcinfo's argument since there should be the required
* attributes as arguments in the tuple.
*/
int natts;
Assert(perc);
natts = slot->tts_tupleDescriptor->natts;
for (i = 0; i < natts; i++)
{
fcinfo.arg[i + 1] = slot_getattr(slot, i + 1, &isnull);
fcinfo.argnull[i + 1] = isnull;
}
}
advance_transition_function(aggstate, peraggstate, pergroupstate,
&fcinfo, mem_manager);
}
} /* aggno loop */
}
/*
* Run the transition function for a DISTINCT or ORDER BY aggregate
* with only one input. This is called after we have completed
* entering all the input values into the sort object. We complete the
* sort, read out the values in sorted order, and run the transition
* function on each value (applying DISTINCT if appropriate).
*
* Note that the strictness of the transition function was checked when
* entering the values into the sort, so we don't check it again here;
* we just apply standard SQL DISTINCT logic.
*
* The one-input case is handled separately from the multi-input case
* for performance reasons: for single by-value inputs, such as the
* common case of count(distinct id), the tuplesort_getdatum code path
* is around 300% faster. (The speedup for by-reference types is less
* but still noticeable.)
*
* When called, CurrentMemoryContext should be the per-query context.
*/
static void
process_ordered_aggregate_single(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate)
{
Datum oldVal = (Datum) 0;
bool oldIsNull = true;
bool haveOldVal = false;
MemoryContext workcontext = aggstate->tmpcontext->ecxt_per_tuple_memory;
MemoryContext oldContext;
bool isDistinct = peraggstate->aggref->aggdistinct;
Datum *newVal;
bool *isNull;
FunctionCallInfoData fcinfo;
Assert(peraggstate->numInputs == 1);
if(gp_enable_mk_sort)
tuplesort_performsort_mk((Tuplesortstate_mk *) peraggstate->sortstate);
else
tuplesort_performsort((Tuplesortstate *) peraggstate->sortstate);
/* Load the column into argument 1 (arg 0 will be transition value) */
newVal = fcinfo.arg + 1;
isNull = fcinfo.argnull + 1;
/*
* Note: if input type is pass-by-ref, the datums returned by the sort are
* freshly palloc'd in the per-query context, so we must be careful to
* pfree them when they are no longer needed.
*/
while (
gp_enable_mk_sort ?
tuplesort_getdatum_mk((Tuplesortstate_mk *)peraggstate->sortstate, true, newVal, isNull)
:
tuplesort_getdatum((Tuplesortstate *)peraggstate->sortstate, true, newVal, isNull)
)
{
/*
* Clear and select the working context for evaluation of the equality
* function and transition function.
*/
MemoryContextReset(workcontext);
oldContext = MemoryContextSwitchTo(workcontext);
/*
* If DISTINCT mode, and not distinct from prior, skip it.
*/
if (isDistinct && *isNull )
{
/* per SQL, DISTINCT doesn't use nulls */
}
else if (isDistinct &&
haveOldVal &&
((oldIsNull && *isNull) ||
(!oldIsNull && !*isNull &&
DatumGetBool(FunctionCall2(&peraggstate->equalfn,
oldVal, *newVal)))))
{
/* equal to prior, so forget this one */
if (!peraggstate->inputtypeByVal && !*isNull)
pfree(DatumGetPointer(*newVal));
}
else
{
advance_transition_function(aggstate, peraggstate, pergroupstate,
&fcinfo, &(aggstate->mem_manager));
/* forget the old value, if any */
if (!oldIsNull && !peraggstate->inputtypeByVal)
pfree(DatumGetPointer(oldVal));
/* and remember the new one for subsequent equality checks */
oldVal = *newVal;
oldIsNull = *isNull;
haveOldVal = true;
}
MemoryContextSwitchTo(oldContext);
}
if (!oldIsNull && !peraggstate->inputtypeByVal)
pfree(DatumGetPointer(oldVal));
if(gp_enable_mk_sort)
tuplesort_end_mk((Tuplesortstate_mk *) peraggstate->sortstate);
else
tuplesort_end((Tuplesortstate *) peraggstate->sortstate);
peraggstate->sortstate = NULL;
}
/*
* Run the transition function for an ORDER BY aggregate with more than
* one input. In PG DISTINCT aggregates may also have multiple columns,
* but in GPDB, only ORDER BY aggregates do. This is called after we have
* completed entering all the input values into the sort object. We
* complete the sort, read out the values in sorted order, and run the
* transition function on each value.
*
* When called, CurrentMemoryContext should be the per-query context.
*/
static void
process_ordered_aggregate_multi(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate)
{
MemoryContext workcontext = aggstate->tmpcontext->ecxt_per_tuple_memory;
FunctionCallInfoData fcinfo;
TupleTableSlot *slot = peraggstate->evalslot;
int numArguments = peraggstate->numArguments;
int i;
if(gp_enable_mk_sort)
tuplesort_performsort_mk((Tuplesortstate_mk *) peraggstate->sortstate);
else
tuplesort_performsort((Tuplesortstate *) peraggstate->sortstate);
ExecClearTuple(slot);
PG_TRY();
{
while (
gp_enable_mk_sort ?
tuplesort_gettupleslot_mk((Tuplesortstate_mk *)peraggstate->sortstate, true, slot)
:
tuplesort_gettupleslot((Tuplesortstate *)peraggstate->sortstate, true, slot)
)
{
/*
* Extract the first numArguments as datums to pass to the transfn.
* (This will help execTuplesMatch too, so do it immediately.)
*/
slot_getsomeattrs(slot, numArguments);
/* Load values into fcinfo */
/* Start from 1, since the 0th arg will be the transition value */
for (i = 0; i < numArguments; i++)
{
fcinfo.arg[i + 1] = slot_get_values(slot)[i];
fcinfo.argnull[i + 1] = slot_get_isnull(slot)[i];
}
advance_transition_function(aggstate, peraggstate, pergroupstate,
&fcinfo, &(aggstate->mem_manager));
/* Reset context each time */
MemoryContextReset(workcontext);
ExecClearTuple(slot);
}
}
PG_CATCH();
{
/*
* The tuple is stored in a memory context that will be released during
* the error handling. If we don't clear it here we will attempt to
* clear it later after the memory context has been released which would
* be a memory context error.
*/
ExecClearTuple(slot);
/* Carry on with error handling. */
PG_RE_THROW();
}
PG_END_TRY();
if(gp_enable_mk_sort)
tuplesort_end_mk((Tuplesortstate_mk *) peraggstate->sortstate);
else
tuplesort_end((Tuplesortstate *) peraggstate->sortstate);
peraggstate->sortstate = NULL;
}
/*
* finalize_aggregates
* Compute the final value for all aggregate functions.
*/
static void
finalize_aggregates(AggState *aggstate, AggStatePerGroup pergroup)
{
AggStatePerAgg peragg = aggstate->peragg;
ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
Datum *aggvalues = econtext->ecxt_aggvalues;
bool *aggnulls = econtext->ecxt_aggnulls;
for (int aggno = 0; aggno < aggstate->numaggs; aggno++)
{
AggStatePerAgg peraggstate = &peragg[aggno];
AggStatePerGroup pergroupstate = &pergroup[aggno];
if ( peraggstate->numSortCols > 0 )
{
if ( peraggstate->numInputs == 1 )
{
process_ordered_aggregate_single(aggstate, peraggstate, pergroupstate);
}
else
{
process_ordered_aggregate_multi(aggstate, peraggstate, pergroupstate);
}
}
finalize_aggregate(aggstate, peraggstate, pergroupstate,
&aggvalues[aggno], &aggnulls[aggno]);
}
}
/*
* Compute the final value of one aggregate.
*
* The finalfunction will be run, and the result delivered, in the
* output-tuple context; caller's CurrentMemoryContext does not matter.
*/
static void
finalize_aggregate(AggState *aggstate,
AggStatePerAgg peraggstate,
AggStatePerGroup pergroupstate,
Datum *resultVal, bool *resultIsNull)
{
MemoryContext oldContext;
oldContext = MemoryContextSwitchTo(aggstate->ss.ps.ps_ExprContext->ecxt_per_tuple_memory);
/*
* Apply the agg's finalfn if one is provided, else return transValue.
*/
if (OidIsValid(peraggstate->finalfn_oid))
{
FunctionCallInfoData fcinfo;
InitFunctionCallInfoData(fcinfo, &(peraggstate->finalfn), 1,
(void *) aggstate, NULL);
fcinfo.arg[0] = pergroupstate->transValue;
fcinfo.argnull[0] = pergroupstate->transValueIsNull;
if (fcinfo.flinfo->fn_strict && pergroupstate->transValueIsNull)
{
/* don't call a strict function with NULL inputs */
*resultVal = (Datum) 0;
*resultIsNull = true;
}
else
{
*resultVal = FunctionCallInvoke(&fcinfo);
*resultIsNull = fcinfo.isnull;
}
}
else
{
*resultVal = pergroupstate->transValue;
*resultIsNull = pergroupstate->transValueIsNull;
}
/*
* If result is pass-by-ref, make sure it is in the right context.
*/
if (!peraggstate->resulttypeByVal && !*resultIsNull &&
!MemoryContextContainsGenericAllocation(CurrentMemoryContext,
DatumGetPointer(*resultVal)))
*resultVal = datumCopy(*resultVal,
peraggstate->resulttypeByVal,
peraggstate->resulttypeLen);
MemoryContextSwitchTo(oldContext);
}
/*
* find_unaggregated_cols
* Construct a bitmapset of the column numbers of un-aggregated Vars
* appearing in our targetlist and qual (HAVING clause)
*/
static Bitmapset *
find_unaggregated_cols(AggState *aggstate)
{
Agg *node = (Agg *) aggstate->ss.ps.plan;
Bitmapset *colnos;
colnos = NULL;
(void) find_unaggregated_cols_walker((Node *) node->plan.targetlist,
&colnos);
(void) find_unaggregated_cols_walker((Node *) node->plan.qual,
&colnos);
return colnos;
}
static bool
find_unaggregated_cols_walker(Node *node, Bitmapset **colnos)
{
if (node == NULL)
return false;
if (IsA(node, Var))
{
Var *var = (Var *) node;
/* setrefs.c should have set the varno to 0 */
Assert(var->varno == 0);
Assert(var->varlevelsup == 0);
*colnos = bms_add_member(*colnos, var->varattno);
return false;
}
if (IsA(node, Aggref)) /* do not descend into aggregate exprs */
return false;
return expression_tree_walker(node, find_unaggregated_cols_walker,
(void *) colnos);
}
/*
* Create a list of the tuple columns that actually need to be stored
* in hashtable entries. The incoming tuples from the child plan node
* will contain grouping columns, other columns referenced in our
* targetlist and qual, columns used to compute the aggregate functions,
* and perhaps just junk columns we don't use at all. Only columns of the
* first two types need to be stored in the hashtable, and getting rid of
* the others can make the table entries significantly smaller. To avoid
* messing up Var numbering, we keep the same tuple descriptor for
* hashtable entries as the incoming tuples have, but set unwanted columns
* to NULL in the tuples that go into the table.
*
* To eliminate duplicates, we build a bitmapset of the needed columns,
* then convert it to an integer list (cheaper to scan at runtime).
* The list is in decreasing order so that the first entry is the largest;
* lookup_hash_entry depends on this to use slot_getsomeattrs correctly.
*
* Note: at present, searching the tlist/qual is not really necessary
* since the parser should disallow any unaggregated references to
* ungrouped columns. However, the search will be needed when we add
* support for SQL99 semantics that allow use of "functionally dependent"
* columns that haven't been explicitly grouped by.
*/
List *
get_agg_hash_collist(AggState *aggstate)
{
Agg *node = (Agg *) aggstate->ss.ps.plan;
Bitmapset *colnos;
List *collist;
int i;
/* Find Vars that will be needed in tlist and qual */
colnos = find_unaggregated_cols(aggstate);
/* Add in all the grouping columns */
for (i = 0; i < node->numCols; i++)
colnos = bms_add_member(colnos, node->grpColIdx[i]);
/* Convert to list, using lcons so largest element ends up first */
collist = NIL;
while ((i = bms_first_member(colnos)) >= 0)
collist = lcons_int(i, collist);
return collist;
}
/*
* Estimate per-hash-table-entry overhead for the planner.
*
* Note that the estimate does not include space for pass-by-reference
* transition data values, nor for the representative tuple of each group.
*/
Size
hash_agg_entry_size(int numAggs)
{
Size entrysize;
/* This must match build_hash_table */
entrysize = sizeof(AggHashEntryData) +
(numAggs - 1) *sizeof(AggStatePerGroupData);
entrysize = MAXALIGN(entrysize);
/* Account for hashtable overhead (assuming fill factor = 1) */
entrysize += 3 * sizeof(void *);
return entrysize;
}
/*
* ExecAgg -
*
* ExecAgg receives tuples from its outer subplan and aggregates over
* the appropriate attribute for each aggregate function use (Aggref
* node) appearing in the targetlist or qual of the node. The number
* of tuples to aggregate over depends on whether grouped or plain
* aggregation is selected. In grouped aggregation, we produce a result
* row for each group; in plain aggregation there's a single result row
* for the whole query. In either case, the value of each aggregate is
* stored in the expression context to be used when ExecProject evaluates
* the result tuple.
*
* XXX: Fix BTree code.
*
* Streaming bottom: forces end of passes when no tuple for underlying node.
*
* MPP-2614: Btree scan will return null tuple at end of scan. However,
* if one calls ExecProNode again on a btree scan, it will restart from
* beginning even though we did not call rescan. This is a bug on
* btree scan, but mask it off here for v3.1. Really should fix Btree
* code.
*/
TupleTableSlot *
ExecAgg(AggState *node)
{
if (node->agg_done)
{
ExecEagerFreeAgg(node);
return NULL;
}
if (((Agg *) node->ss.ps.plan)->aggstrategy == AGG_HASHED)
{
TupleTableSlot *tuple = NULL;
bool streaming = ((Agg *) node->ss.ps.plan)->streaming;
/*
* ExecAgg processing for hashed aggregation -- returns the
* next result tuple or NULL. When returning NULL also sets
* aggstate to prevent future calls.
*/
if (node->hhashtable == NULL)
{
bool tupremain;
node->hhashtable = create_agg_hash_table(node);
tupremain = agg_hash_initial_pass(node);
if ( streaming )
{
if ( tupremain )
node->hhashtable->state = HASHAGG_STREAMING;
else
node->hhashtable->state = HASHAGG_END_OF_PASSES;
}
else
node->hhashtable->state = HASHAGG_BETWEEN_PASSES;
}
/* On each call we either return a tuple corresponding to a hash
* entry (consuming the entry) or fall through to a state machine
* that tries to make additional hash entries available and continue
* the loop. (This may result in reaching the "exit" state and
* returning a NULL tuple).
*/
for (;;)
{
if (!node->hhashtable->is_spilling)
{
tuple = agg_retrieve_hash_table(node);
node->agg_done = false; /* Not done 'til batches used up. */
if (tuple != NULL)
return tuple;
}
switch (node->hhashtable->state)
{
case HASHAGG_BETWEEN_PASSES:
Assert(!streaming);
if (agg_hash_next_pass(node))
{
node->hhashtable->state = HASHAGG_BETWEEN_PASSES;
continue;
}
node->hhashtable->state = HASHAGG_END_OF_PASSES;
/*
* pass through. Be sure that the next case statment
* is HASHAGG_END_OF_PASSES.
*/
case HASHAGG_END_OF_PASSES:
node->agg_done = true;
if (gp_workfile_caching && node->workfiles_created)
{
/*
* HashAgg closes each spill file after it is done with
* them. Since we got here on the regular path, all
* files should be closed.
*/
Assert(node->hhashtable->work_set);
Assert(node->hhashtable->spill_set == NULL);
agg_hash_close_state_file(node->hhashtable);
agg_hash_mark_spillset_complete(node);
}
ExecEagerFreeAgg(node);
return NULL;
case HASHAGG_STREAMING:
Assert(streaming);
if ( !agg_hash_stream(node) )
node->hhashtable->state = HASHAGG_END_OF_PASSES;
continue;
case HASHAGG_BEFORE_FIRST_PASS:
default:
elog(ERROR,"hybrid hash aggregation sequencing error");
}
}
}
else
{
return agg_retrieve_direct(node);
}
}
/*
* agg_retrieve_scalar
* Compute the scalar aggregates.
*/
static TupleTableSlot *
agg_retrieve_scalar(AggState *aggstate)
{
AggStatePerAgg peragg = aggstate->peragg;
AggStatePerGroup pergroup = aggstate->pergroup ;
initialize_aggregates(aggstate, peragg, pergroup, &(aggstate->mem_manager));
/*
* We loop through input tuples, and compute the aggregates.
*/
while (!aggstate->agg_done)
{
ExprContext *tmpcontext = aggstate->tmpcontext;
/* Reset the per-input-tuple context */
ResetExprContext(tmpcontext);
PlanState *outerPlan = outerPlanState(aggstate);
TupleTableSlot *outerslot = ExecProcNode(outerPlan);
if (TupIsNull(outerslot))
{
aggstate->agg_done = true;
break;
}
Gpmon_M_Incr(GpmonPktFromAggState(aggstate), GPMON_QEXEC_M_ROWSIN);
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
tmpcontext->ecxt_scantuple = outerslot;
advance_aggregates(aggstate, pergroup, &(aggstate->mem_manager));
}
finalize_aggregates(aggstate, pergroup);
ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
Agg *node = (Agg*)aggstate->ss.ps.plan;
econtext->grouping = node->grouping;
econtext->group_id = node->rollupGSTimes;
/* Check the qual (HAVING clause). */
if (ExecQual(aggstate->ss.ps.qual, econtext, false))
{
Gpmon_M_Incr_Rows_Out(GpmonPktFromAggState(aggstate));
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
/*
* Form and return a projection tuple using the aggregate results
* and the representative input tuple.
*/
return ExecProject(aggstate->ss.ps.ps_ProjInfo, NULL);
}
return NULL;
}
/*
* copyFirstTupleInGroup
* Make a copy of a given tuple as the first tuple in a group.
*/
static void
copyFirstTupleInGroup(AggState *aggstate, TupleTableSlot *outerslot)
{
TupleTableSlot *firstSlot = aggstate->ss.ss_ScanTupleSlot;
slot_getallattrs(outerslot);
aggstate->grp_firstTuple = memtuple_form_to(firstSlot->tts_mt_bind,
slot_get_values(outerslot),
slot_get_isnull(outerslot),
NULL /* mtup */,
NULL /* destlen */,
false /* inline_toast */);
}
/*
* processFirstTupleInGroup
* Process the first input tuple in a group: initialize the working state
* for the new group, and accumulate the tuple into the aggregates.
*/
static void
processFirstTupleInGroup(AggState *aggstate)
{
Assert(aggstate->grp_firstTuple != NULL);
TupleTableSlot *firstSlot = aggstate->ss.ss_ScanTupleSlot;
ExecStoreMemTuple(aggstate->grp_firstTuple,
firstSlot,
true);
aggstate->grp_firstTuple = NULL;
/* Clear the per-output-tuple context for each group */
ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
ResetExprContext(econtext);
MemoryContextResetAndDeleteChildren(aggstate->aggcontext);
/* Initialize working state for a new input tuple group */
AggStatePerAgg peragg = aggstate->peragg;
AggStatePerGroup pergroup = aggstate->pergroup;
initialize_aggregates(aggstate, peragg, pergroup, &(aggstate->mem_manager));
ExprContext *tmpcontext = aggstate->tmpcontext;
tmpcontext->ecxt_scantuple = firstSlot;
advance_aggregates(aggstate, pergroup, &(aggstate->mem_manager));
}
/*
* processNextTuple
* Process an input tuple. If this tuple belongs to the current group,
* its value is accumulated into the aggregates for the current group.
* Otherwise, this tuple indicates a new group starts. This function
* makes a copy for this tuple.
*
* This function returns true if a new group starts.
*/
static bool
processNextTuple(AggState *aggstate, TupleTableSlot *outerslot)
{
Assert(aggstate->grp_firstTuple == NULL);
TupleTableSlot *firstSlot = aggstate->ss.ss_ScanTupleSlot;
Assert(!TupIsNull(firstSlot));
Agg *node = (Agg*)aggstate->ss.ps.plan;
ExprContext *tmpcontext = aggstate->tmpcontext;
bool tuple_match = execTuplesMatch(firstSlot,
outerslot,
node->numCols - node->numNullCols,
node->grpColIdx,
aggstate->eqfunctions,
tmpcontext->ecxt_per_tuple_memory);
bool isNewGroup = false;
if (tuple_match)
{
tmpcontext->ecxt_scantuple = outerslot;
AggStatePerGroup pergroup = aggstate->pergroup;
advance_aggregates(aggstate, pergroup, &(aggstate->mem_manager));
}
else
{
copyFirstTupleInGroup(aggstate, outerslot);
isNewGroup = true;
}
return isNewGroup;
}
/*
* agg_retrieve_group
* Compute an aggreate for each group.
*
* This function assumes that input tuples arrive in groups. A new tuple
* with a different grouping key means the start of a new group.
*
* This function also computes Grouping(), Group_id() if they appear in
* the target list.
*/
static TupleTableSlot *
agg_retrieve_group(AggState *aggstate)
{
Assert(((Agg*)aggstate->ss.ps.plan)->aggstrategy == AGG_SORTED);
TupleTableSlot *firstSlot = aggstate->ss.ss_ScanTupleSlot;
PlanState *outerPlan = outerPlanState(aggstate);
TupleTableSlot *outerslot = NULL;
/* Read the first input tuple in a group if this is not done. */
if (TupIsNull(firstSlot) &&
aggstate->grp_firstTuple == NULL)
{
outerslot = ExecProcNode(outerPlan);
if (TupIsNull(outerslot))
{
aggstate->agg_done = true;
return NULL;
}
Gpmon_M_Incr(GpmonPktFromAggState(aggstate), GPMON_QEXEC_M_ROWSIN);
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
copyFirstTupleInGroup(aggstate, outerslot);
}
/*
* We loop through input tuples. If a tuple has a different grouping key,
* a new group starts. At this point, we have seen all tuples for the
* previous group, and is able to compute and return the final aggregate
* value for this group.
*/
while (!aggstate->agg_done)
{
/*
* If grp_firstTuple is not null, it means that we have read the
* first tuple in a new group. Initialize the working state for
* the new group, and advance the aggregate state for this tuple.
*/
if (aggstate->grp_firstTuple != NULL)
{
processFirstTupleInGroup(aggstate);
}
Assert(!TupIsNull(firstSlot));
/*
* Read and process the next input tuple. If this tuple belongs to
* the same group, simply accumlate its value to the intermediate
* state, and go back to the beginning of the loop to process the
* next input tuple. Otherwise, a new group starts. We finalize
* the aggregate value for the current group, and return the value
* if it satisfies the qual.
*/
/* Reset the per-input-tuple context */
ExprContext *tmpcontext = aggstate->tmpcontext;
ResetExprContext(tmpcontext);
bool isNewGroup = false;
outerslot = ExecProcNode(outerPlan);
if (TupIsNull(outerslot))
{
aggstate->agg_done = true;
isNewGroup = true;
}
else
{
Gpmon_M_Incr(GpmonPktFromAggState(aggstate), GPMON_QEXEC_M_ROWSIN);
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
isNewGroup = processNextTuple(aggstate, outerslot);
}
if (isNewGroup)
{
Assert(!TupIsNull(firstSlot));
TupleTableSlot *result =
computeTupleWithFinalAggregate(aggstate, firstSlot);
if (!TupIsNull(result))
{
return result;
}
}
}
return NULL;
}
/*
* isGroupTuple
* Return true when the given tuple is a tuple in a group that
* contribute to the final aggregates.
*
* This is only used in a ROLLUP Agg to distinguish tuples from
* those that are only pass-through.
*/
static bool
isGroupTuple(AggState *aggstate,
TupleTableSlot *outerslot)
{
Agg *node = (Agg*)aggstate->ss.ps.plan;
Assert(node->numCols - node->numNullCols >= 2);
uint64 inputGrouping = node->inputGrouping;
int groupingAttNo = node->grpColIdx[node->numCols - node->numNullCols - 2];
uint64 tupleGrouping =
tuple_grouping(outerslot, inputGrouping, groupingAttNo);
Assert(tupleGrouping <= inputGrouping);
return (tupleGrouping == inputGrouping);
}
/*
* preparePassThruTuple
* Prepare a input tuple to be outputted by the Agg node.
* If this Agg node is the last rollup Agg node, the input tuple
* needs to be finalized.
*
* This function returns NULL if the input tuple does not
* satisfy the qual, or a result tuple that is ready to be
* returned to the upper node.
*/
static TupleTableSlot *
preparePassThruTuple(AggState *aggstate,
TupleTableSlot *outerslot)
{
Agg *node = (Agg*)aggstate->ss.ps.plan;
Assert(node->numCols - node->numNullCols >= 2);
Assert(!TupIsNull(outerslot));
int groupingAttNo = node->grpColIdx[(node->numCols - node->numNullCols) - 2];
int groupIdAttNo = node->grpColIdx[(node->numCols - node->numNullCols) - 1];
TupleTableSlot *result = outerslot;
/*
* If this node is the final rollup, the pass-through input tuples
* need to be finalized before returning.
*/
if (aggstate->aggType == AggTypeFinalRollup)
{
AggStatePerAgg peragg = aggstate->peragg;
AggStatePerGroup perpassthru = aggstate->perpassthru;
ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
initialize_aggregates(aggstate, peragg, perpassthru, &(aggstate->mem_manager));
ExprContext *tmpcontext = aggstate->tmpcontext;
tmpcontext->ecxt_scantuple = outerslot;
advance_aggregates(aggstate, perpassthru, &(aggstate->mem_manager));
finalize_aggregates(aggstate, perpassthru);
econtext->ecxt_scantuple = outerslot;
Assert(node->inputHasGrouping);
econtext->grouping =
get_grouping_groupid(econtext->ecxt_scantuple, groupingAttNo);
econtext->group_id =
get_grouping_groupid(econtext->ecxt_scantuple, groupIdAttNo);
/*
* Check the qual. Form and return a projection tuple using the aggregate results
* and the representative input tuple. Note we do not support
* aggregates returning sets ...
*/
result = NULL;
if (ExecQual(aggstate->ss.ps.qual, econtext, false))
{
Gpmon_M_Incr_Rows_Out(GpmonPktFromAggState(aggstate));
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
result = ExecProject(aggstate->ss.ps.ps_ProjInfo, NULL);
}
}
return result;
}
/*
* computeTupleWithFinalAggregate
* Finalize the aggregate values and check whether the result
* satisfies the given qual. If so, form and return a projection
* tuple.
*/
static TupleTableSlot *
computeTupleWithFinalAggregate(AggState *aggstate,
TupleTableSlot *firstSlot)
{
AggStatePerGroup pergroup = aggstate->pergroup;
finalize_aggregates(aggstate, pergroup);
ExprContext *econtext = aggstate->ss.ps.ps_ExprContext;
econtext->ecxt_scantuple = firstSlot;
/*
* If this is part of a rollup, we need to mask the grouping attributes that
* are supposedly null in this Agg node.
*/
Agg *node = (Agg*)aggstate->ss.ps.plan;
if (node->numNullCols > 0)
{
slot_getallattrs(firstSlot);
bool *isnull = slot_get_isnull(firstSlot);
for (int attno = node->numCols - node->numNullCols; attno < node->numCols; attno++)
{
isnull[node->grpColIdx[attno] - 1] = true;
}
}
econtext->grouping = node->grouping;
econtext->group_id = node->rollupGSTimes;
/*
* In the second stage of a rollup, grouping and group_id values in the
* target list from the input tuples.
*/
if (aggstate->aggType == AggTypeGroup && node->inputHasGrouping)
{
int groupingAttNo = node->grpColIdx[node->numCols - node->numNullCols - 2];
int groupIdAttNo = node->grpColIdx[node->numCols-node->numNullCols - 1];
econtext->grouping =
get_grouping_groupid(econtext->ecxt_scantuple, groupingAttNo);
econtext->group_id =
get_grouping_groupid(econtext->ecxt_scantuple, groupIdAttNo);
}
/*
* Check the qual (HAVING clause). If it is satisfied, form and return
* a projection tuple using the aggregate results and the representative
* input tuple.
*/
TupleTableSlot *result = NULL;
if (ExecQual(aggstate->ss.ps.qual, econtext, false))
{
Gpmon_M_Incr_Rows_Out(GpmonPktFromAggState(aggstate));
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
result = ExecProject(aggstate->ss.ps.ps_ProjInfo, NULL);
}
return result;
}
/*
* agg_retrieve_rollup
* Compute aggregates for a group in a rollup.
*
* The difference between this Agg and the ordinary group agg is that
* (1) not all input tuples need to be aggregated. Some of input tuples
* are result tuples from 2 or more levels lower in the rollup hierarchy.
* These tuples need to be passed up to the upper node (pass-through).
* (2) all input tuples are part of the result tuples for this Agg node.
*/
static TupleTableSlot *
agg_retrieve_rollup(AggState *aggstate)
{
Assert(((Agg*)aggstate->ss.ps.plan)->aggstrategy == AGG_SORTED);
Assert(((Agg*)aggstate->ss.ps.plan)->inputHasGrouping);
Assert(aggstate->aggType == AggTypeIntermediateRollup ||
aggstate->aggType == AggTypeFinalRollup);
TupleTableSlot *firstSlot = aggstate->ss.ss_ScanTupleSlot;
PlanState *outerPlan = outerPlanState(aggstate);
TupleTableSlot *outerslot = NULL;
/*
* Read the input tuple until we read a tuple that is
* the first tuple in a group to be aggregated. In the meanwhile,
* all input tuples need to be returned.
*/
while (TupIsNull(firstSlot) &&
aggstate->grp_firstTuple == NULL)
{
/* Reset the per-input-tuple context */
ExprContext *tmpcontext = aggstate->tmpcontext;
ResetExprContext(tmpcontext);
outerslot = ExecProcNode(outerPlan);
if (TupIsNull(outerslot))
{
aggstate->agg_done = true;
return NULL;
}
Gpmon_M_Incr(GpmonPktFromAggState(aggstate), GPMON_QEXEC_M_ROWSIN);
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
/*
* If this tuple needs to be aggregated, this is the first tuple
* in a group. We make a copy of this tuple. Otherwise,
* outputted this tuple as a pass-through tuple. The first tuple
* in a new group will be passed-through below.
*/
if (isGroupTuple(aggstate, outerslot))
{
copyFirstTupleInGroup(aggstate, outerslot);
}
else
{
/*
* If this tuple needs to be pass-through, do it now.
*/
TupleTableSlot *passThruTuple = preparePassThruTuple(aggstate, outerslot);
if (!TupIsNull(passThruTuple))
{
return passThruTuple;
}
}
}
/*
* We loop through input tuples. If a tuple has a different grouping key,
* a new group starts. At this point, we have seen all tuples for the
* previous group, and is able to compute and return the final aggregate
* value for this group.
*
* Note that input tuples also need to be pass-through.
*/
while (!aggstate->agg_done)
{
/*
* If grp_firstTuple is not null, it means that we have read the
* first tuple in a new group. Initialize the working state for
* the new group, and advance the aggregate state for this tuple.
*
* This tuple also needs to be outputted here.
*/
if (aggstate->grp_firstTuple != NULL)
{
processFirstTupleInGroup(aggstate);
/* Pass through the first tuple in this new group now. */
TupleTableSlot *passThruTuple = preparePassThruTuple(aggstate, firstSlot);
if (!TupIsNull(passThruTuple))
{
return passThruTuple;
}
}
/*
* Read and process the next input tuple. If this tuple is a pass-through
* tuple, return it. If this tuple needs to be aggregated, and belongs to
* the same group, simply accumulate its value to the intermediate
* state, and go back to the beginning of the loop to process the
* next input tuple. Otherwise, a new group starts. We finalize
* the aggregate value for the current group, and return the value
* if it satisfies the qual.
*/
/* Reset the per-input-tuple context */
ExprContext *tmpcontext = aggstate->tmpcontext;
ResetExprContext(tmpcontext);
bool isNewGroup = false;
outerslot = ExecProcNode(outerPlan);
if (TupIsNull(outerslot))
{
aggstate->agg_done = true;
isNewGroup = true;
}
else
{
Assert(!TupIsNull(outerslot));
Gpmon_M_Incr(GpmonPktFromAggState(aggstate), GPMON_QEXEC_M_ROWSIN);
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
if (isGroupTuple(aggstate, outerslot))
{
isNewGroup = processNextTuple(aggstate, outerslot);
}
}
if (isNewGroup)
{
Assert(!TupIsNull(firstSlot));
TupleTableSlot *result = computeTupleWithFinalAggregate(aggstate, firstSlot);
if (!TupIsNull(result))
{
return result;
}
}
if (!TupIsNull(outerslot))
{
TupleTableSlot *passThruTuple = preparePassThruTuple(aggstate, outerslot);
if (!TupIsNull(passThruTuple))
{
return passThruTuple;
}
}
}
return NULL;
}
/*
* ExecAgg for non-hashed case.
*/
static TupleTableSlot *
agg_retrieve_direct(AggState *aggstate)
{
if (aggstate->agg_done)
{
return NULL;
}
switch(aggstate->aggType)
{
case AggTypeScalar:
return agg_retrieve_scalar(aggstate);
case AggTypeGroup:
return agg_retrieve_group(aggstate);
case AggTypeIntermediateRollup:
case AggTypeFinalRollup:
return agg_retrieve_rollup(aggstate);
default:
insist_log(false, "invalid Agg node: type %d", aggstate->aggType);
}
return NULL;
}
/*
* ExecAgg for hashed case: retrieve groups from hash table
*/
static TupleTableSlot *
agg_retrieve_hash_table(AggState *aggstate)
{
ExprContext *econtext;
ProjectionInfo *projInfo;
Datum *aggvalues;
bool *aggnulls;
AggStatePerAgg peragg;
AggStatePerGroup pergroup;
TupleTableSlot *firstSlot;
Agg *node = (Agg *) aggstate->ss.ps.plan;
bool input_has_grouping = node->inputHasGrouping;
bool is_final_rollup_agg =
(node->lastAgg ||
(input_has_grouping && node->numNullCols == 0));
/*
* get state info from node
*/
/* econtext is the per-output-tuple expression context */
econtext = aggstate->ss.ps.ps_ExprContext;
aggvalues = econtext->ecxt_aggvalues;
aggnulls = econtext->ecxt_aggnulls;
projInfo = aggstate->ss.ps.ps_ProjInfo;
peragg = aggstate->peragg;
firstSlot = aggstate->ss.ss_ScanTupleSlot;
if (aggstate->agg_done)
return NULL;
/*
* We loop retrieving groups until we find one satisfying
* aggstate->ss.ps.qual
*/
while (!aggstate->agg_done)
{
HashAggEntry *entry = agg_hash_iter(aggstate);
if (entry == NULL)
{
aggstate->agg_done = TRUE;
return NULL;
}
ResetExprContext(econtext);
/*
* Store the copied first input tuple in the tuple table slot reserved
* for it, so that it can be used in ExecProject.
*/
ExecStoreMemTuple((MemTuple)entry->tuple_and_aggs, firstSlot, false);
pergroup = (AggStatePerGroup)((char *)entry->tuple_and_aggs +
MAXALIGN(memtuple_get_size((MemTuple)entry->tuple_and_aggs,
aggstate->hashslot->tts_mt_bind)));
/*
* Finalize each aggregate calculation, and stash results in the
* per-output-tuple context.
*/
finalize_aggregates(aggstate, pergroup);
/*
* Use the representative input tuple for any references to
* non-aggregated input columns in the qual and tlist.
*/
econtext->ecxt_scantuple = firstSlot;
if (is_final_rollup_agg && input_has_grouping)
{
econtext->group_id =
get_grouping_groupid(econtext->ecxt_scantuple,
node->grpColIdx[node->numCols-node->numNullCols-1]);
econtext->grouping =
get_grouping_groupid(econtext->ecxt_scantuple,
node->grpColIdx[node->numCols-node->numNullCols-2]);
}
else
{
econtext->group_id = node->rollupGSTimes;
econtext->grouping = node->grouping;
}
/*
* Check the qual (HAVING clause); if the group does not match, ignore
* it and loop back to try to process another group.
*/
if (ExecQual(aggstate->ss.ps.qual, econtext, false))
{
/*
* Form and return a projection tuple using the aggregate results
* and the representative input tuple. Note we do not support
* aggregates returning sets ...
*/
Gpmon_M_Incr_Rows_Out(GpmonPktFromAggState(aggstate));
CheckSendPlanStateGpmonPkt(&aggstate->ss.ps);
return ExecProject(projInfo, NULL);
}
}
/* No more groups */
return NULL;
}
/*
* getAggType
* Get the aggType for the given Agg node.
*
* We should really store the type in the Agg struct, and let the planner set the
* correct type. As an intermediate step, we compute the type here.
*/
static int
getAggType(Agg *node)
{
int aggType = AggTypeScalar;
if (node->numCols > 0 &&
(node->lastAgg &&
(node->inputHasGrouping && node->numNullCols > 0)))
{
aggType = AggTypeFinalRollup;
}
else if (node->numCols > 0 &&
(node->inputHasGrouping && node->numNullCols > 0))
{
aggType = AggTypeIntermediateRollup;
}
else if (node->numCols > 0)
{
aggType = AggTypeGroup;
}
else
{
insist_log(node->aggstrategy == AGG_PLAIN, "wrong Agg strategy: %d", node->aggstrategy);
}
return aggType;
}
/* -----------------
* ExecInitAgg
*
* Creates the run-time information for the agg node produced by the
* planner and initializes its outer subtree
* -----------------
*/
AggState *
ExecInitAgg(Agg *node, EState *estate, int eflags)
{
AggState *aggstate;
AggStatePerAgg peragg;
Plan *outerPlan;
ExprContext *econtext;
int numaggs,
aggno;
ListCell *l;
/* check for unsupported flags */
Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));
/*
* create state structure
*/
aggstate = makeNode(AggState);
aggstate->ss.ps.plan = (Plan *) node;
aggstate->ss.ps.state = estate;
aggstate->aggs = NIL;
aggstate->numaggs = 0;
aggstate->eqfunctions = NULL;
aggstate->hashfunctions = NULL;
aggstate->peragg = NULL;
aggstate->agg_done = false;
aggstate->pergroup = NULL;
aggstate->grp_firstTuple = NULL;
aggstate->hashtable = NULL;
agg_hash_reset_workfile_state(aggstate);
/*
* Create expression contexts. We need two, one for per-input-tuple
* processing and one for per-output-tuple processing. We cheat a little
* by using ExecAssignExprContext() to build both.
*/
ExecAssignExprContext(estate, &aggstate->ss.ps);
aggstate->tmpcontext = aggstate->ss.ps.ps_ExprContext;
ExecAssignExprContext(estate, &aggstate->ss.ps);
/*
* We also need a long-lived memory context for holding hashtable data
* structures and transition values. NOTE: the details of what is stored
* in aggcontext and what is stored in the regular per-query memory
* context are driven by a simple decision: we want to reset the
* aggcontext in ExecReScanAgg to recover no-longer-wanted space.
*/
aggstate->aggcontext =
AllocSetContextCreate(CurrentMemoryContext,
"AggContext",
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
#define AGG_NSLOTS 3
/*
* tuple table initialization
*/
aggstate->ss.ss_ScanTupleSlot = ExecInitExtraTupleSlot(estate);
ExecInitResultTupleSlot(estate, &aggstate->ss.ps);
aggstate->hashslot = ExecInitExtraTupleSlot(estate);
/*
* initialize child expressions
*
* Note: ExecInitExpr finds Aggrefs for us, and also checks that no aggs
* contain other agg calls in their arguments. This would make no sense
* under SQL semantics anyway (and it's forbidden by the spec). Because
* that is true, we don't need to worry about evaluating the aggs in any
* particular order.
*/
aggstate->ss.ps.targetlist = (List *)
ExecInitExpr((Expr *) node->plan.targetlist,
(PlanState *) aggstate);
aggstate->ss.ps.qual = (List *)
ExecInitExpr((Expr *) node->plan.qual,
(PlanState *) aggstate);
/*
* CDB: Offer extra info for EXPLAIN ANALYZE.
*/
if (estate->es_instrument)
{
/* Allocate string buffer. */
aggstate->ss.ps.cdbexplainbuf = makeStringInfo();
/* Request a callback at end of query. */
aggstate->ss.ps.cdbexplainfun = ExecAggExplainEnd;
}
/*
* initialize child nodes
*/
outerPlan = outerPlan(node);
outerPlanState(aggstate) = ExecInitNode(outerPlan, estate, eflags);
/*
* initialize source tuple type.
*/
ExecAssignScanTypeFromOuterPlan(&aggstate->ss);
/*
* Initialize result tuple type and projection info.
*/
ExecAssignResultTypeFromTL(&aggstate->ss.ps);
ExecAssignProjectionInfo(&aggstate->ss.ps, NULL);
/*
* get the count of aggregates in targetlist and quals
*/
numaggs = aggstate->numaggs;
Assert(numaggs == list_length(aggstate->aggs) + list_length(aggstate->percs));
if (numaggs <= 0)
{
/*
* This is not an error condition: we might be using the Agg node just
* to do hash-based grouping. Even in the regular case,
* constant-expression simplification could optimize away all of the
* Aggrefs in the targetlist and qual. So keep going, but force local
* copy of numaggs positive so that palloc()s below don't choke.
*/
numaggs = 1;
}
/*
* If we are grouping, precompute fmgr lookup data for inner loop. We need
* both equality and hashing functions to do it by hashing, but only
* equality if not hashing.
*/
if (node->numCols > 0)
{
if (node->aggstrategy == AGG_HASHED)
execTuplesHashPrepare(ExecGetScanType(&aggstate->ss),
node->numCols,
node->grpColIdx,
&aggstate->eqfunctions,
&aggstate->hashfunctions);
else
aggstate->eqfunctions =
execTuplesMatchPrepare(ExecGetScanType(&aggstate->ss),
node->numCols,
node->grpColIdx);
}
/*
* Set up aggregate-result storage in the output expr context, and also
* allocate my private per-agg working storage
*/
econtext = aggstate->ss.ps.ps_ExprContext;
econtext->ecxt_aggvalues = (Datum *) palloc0(sizeof(Datum) * numaggs);
econtext->ecxt_aggnulls = (bool *) palloc0(sizeof(bool) * numaggs);
peragg = (AggStatePerAgg) palloc0(sizeof(AggStatePerAggData) * numaggs);
aggstate->peragg = peragg;
if (node->aggstrategy == AGG_HASHED)
{
aggstate->hash_needed = get_agg_hash_collist(aggstate);
}
else
{
AggStatePerGroup pergroup;
pergroup = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData) * numaggs);
aggstate->pergroup = pergroup;
}
/*
* Perform lookups of aggregate function info, and initialize the
* unchanging fields of the per-agg data. We also detect duplicate
* aggregates (for example, "SELECT sum(x) ... HAVING sum(x) > 0"). When
* duplicates are detected, we only make an AggStatePerAgg struct for the
* first one. The clones are simply pointed at the same result entry by
* giving them duplicate aggno values.
*/
aggno = -1;
foreach(l, aggstate->aggs)
{
AggrefExprState *aggrefstate = (AggrefExprState *) lfirst(l);
Aggref *aggref = (Aggref *) aggrefstate->xprstate.expr;
AggStatePerAgg peraggstate;
List *inputTargets = NIL;
List *inputSortClauses = NIL;
Oid *inputTypes = NULL;
int numInputs;
int numArguments;
int numSortCols;
List *sortlist;
HeapTuple aggTuple;
Form_pg_aggregate aggform;
Oid aggtranstype;
AclResult aclresult;
Oid transfn_oid = InvalidOid,
finalfn_oid = InvalidOid;
Expr *transfnexpr,
*finalfnexpr,
*prelimfnexpr;
Datum textInitVal;
int i;
ListCell *lc;
cqContext *pcqCtx;
/* Planner should have assigned aggregate to correct level */
Assert(aggref->agglevelsup == 0);
/* Look for a previous duplicate aggregate */
for (i = 0; i <= aggno; i++)
{
if (equal(aggref, peragg[i].aggref) &&
!contain_volatile_functions((Node *) aggref))
break;
}
if (i <= aggno)
{
/* Found a match to an existing entry, so just mark it */
aggrefstate->aggno = i;
continue;
}
/* Nope, so assign a new PerAgg record */
peraggstate = &peragg[++aggno];
/* Mark Aggref state node with assigned index in the result array */
aggrefstate->aggno = aggno;
/* Fill in the peraggstate data */
peraggstate->aggrefstate = aggrefstate;
peraggstate->aggref = aggref;
numArguments = list_length(aggref->args);
peraggstate->numArguments = numArguments;
/*
* Use these information from ExecInitExpr for per agg info.
*/
inputTargets = aggrefstate->inputTargets;
inputSortClauses = aggrefstate->inputSortClauses;
numInputs = list_length(inputTargets);
numSortCols = list_length(inputSortClauses);
peraggstate->numSortCols = numSortCols;
peraggstate->numInputs = numInputs;
/* MPP has some restrictions. */
Assert(!(aggref->aggdistinct && aggref->aggorder));
Assert(numArguments == 1 || !aggref->aggdistinct);
/* Get actual datatypes of the inputs. These could be different from
* the agg's declared input types, when the agg accepts ANY, ANYARRAY
* or ANYELEMENT. The result will have argument types at 0 through
* numArguments-1 and sort key types mixed in or at numArguments through
* numInputs.
*/
inputTypes = (Oid*)palloc0(sizeof(Oid)*(numInputs));
i = 0;
foreach(lc, inputTargets)
{
TargetEntry *tle = (TargetEntry *) lfirst(lc);
inputTypes[i++] = exprType((Node*)tle->expr);
}
pcqCtx = caql_beginscan(
NULL,
cql("SELECT * FROM pg_aggregate "
" WHERE aggfnoid = :1 ",
ObjectIdGetDatum(aggref->aggfnoid)));
aggTuple = caql_getnext(pcqCtx);
if (!HeapTupleIsValid(aggTuple))
elog(ERROR, "cache lookup failed for aggregate %u",
aggref->aggfnoid);
aggform = (Form_pg_aggregate) GETSTRUCT(aggTuple);
/* Check permission to call aggregate function */
aclresult = pg_proc_aclcheck(aggref->aggfnoid, GetUserId(),
ACL_EXECUTE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_PROC,
get_func_name(aggref->aggfnoid));
switch ( aggref->aggstage) /* MPP */
{
case AGGSTAGE_NORMAL: /* Single-stage aggregation */
peraggstate->transfn_oid = transfn_oid = aggform->aggtransfn;
peraggstate->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
break;
case AGGSTAGE_PARTIAL:/* Two-stage aggregation -- preliminary stage */
/* the perliminary stage for two-stage aggregation */
peraggstate->transfn_oid = transfn_oid = aggform->aggtransfn;
peraggstate->finalfn_oid = finalfn_oid = InvalidOid;
break;
case AGGSTAGE_INTERMEDIATE:
peraggstate->transfn_oid = transfn_oid = aggform->aggprelimfn;
peraggstate->finalfn_oid = finalfn_oid = InvalidOid;
break;
case AGGSTAGE_FINAL: /* Two-stage aggregation - final stage */
peraggstate->transfn_oid = transfn_oid = aggform->aggprelimfn;
peraggstate->finalfn_oid = finalfn_oid = aggform->aggfinalfn;
break;
}
peraggstate->prelimfn_oid = aggform->aggprelimfn;
/* Check that aggregate owner has permission to call component fns */
{
Oid aggOwner;
int fetchCount;
aggOwner = caql_getoid_plus(
NULL,
&fetchCount,
NULL,
cql("SELECT proowner FROM pg_proc "
" WHERE oid = :1 ",
ObjectIdGetDatum(aggref->aggfnoid)));
if (!fetchCount)
elog(ERROR, "cache lookup failed for function %u",
aggref->aggfnoid);
aclresult = pg_proc_aclcheck(transfn_oid, aggOwner,
ACL_EXECUTE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_PROC,
get_func_name(transfn_oid));
if (OidIsValid(finalfn_oid))
{
aclresult = pg_proc_aclcheck(finalfn_oid, aggOwner,
ACL_EXECUTE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_PROC,
get_func_name(finalfn_oid));
}
if (OidIsValid(peraggstate->prelimfn_oid))
{
aclresult = pg_proc_aclcheck(peraggstate->prelimfn_oid, aggOwner,
ACL_EXECUTE);
if (aclresult != ACLCHECK_OK)
aclcheck_error(aclresult, ACL_KIND_PROC,
get_func_name(peraggstate->prelimfn_oid));
}
}
/* check if the transition type is polymorphic and if so resolve it */
aggtranstype = resolve_polymorphic_transtype(aggform->aggtranstype,
aggref->aggfnoid,
inputTypes);
/* build expression trees using actual argument & result types */
build_aggregate_fnexprs(inputTypes,
numArguments,
aggtranstype,
aggref->aggtype,
transfn_oid,
finalfn_oid,
InvalidOid /* prelim */,
InvalidOid /* invtrans */,
InvalidOid /* invprelim */,
&transfnexpr,
&finalfnexpr,
&prelimfnexpr,
NULL,
NULL);
fmgr_info(transfn_oid, &peraggstate->transfn);
peraggstate->transfn.fn_expr = (Node *) transfnexpr;
if (OidIsValid(finalfn_oid))
{
fmgr_info(finalfn_oid, &peraggstate->finalfn);
peraggstate->finalfn.fn_expr = (Node *) finalfnexpr;
}
if (OidIsValid(peraggstate->prelimfn_oid))
{
fmgr_info(peraggstate->prelimfn_oid, &peraggstate->prelimfn);
peraggstate->prelimfn.fn_expr = (Node *) prelimfnexpr;
}
get_typlenbyval(aggref->aggtype,
&peraggstate->resulttypeLen,
&peraggstate->resulttypeByVal);
get_typlenbyval(aggtranstype,
&peraggstate->transtypeLen,
&peraggstate->transtypeByVal);
/*
* initval is potentially null, so don't try to access it as a struct
* field. Must do it the hard way with caql_getattr
*/
textInitVal = caql_getattr(pcqCtx,
Anum_pg_aggregate_agginitval,
&peraggstate->initValueIsNull);
if (peraggstate->initValueIsNull)
peraggstate->initValue = (Datum) 0;
else
peraggstate->initValue = GetAggInitVal(textInitVal,
aggtranstype);
/*
* If the transfn is strict and the initval is NULL, make sure input
* type and transtype are the same (or at least binary-compatible), so
* that it's OK to use the first input value as the initial
* transValue. This should have been checked at agg definition time,
* but just in case...
*/
if (peraggstate->transfn.fn_strict && peraggstate->initValueIsNull)
{
if (numArguments < 1 ||
!IsBinaryCoercible(inputTypes[0], aggtranstype))
ereport(ERROR,
(errcode(ERRCODE_INVALID_FUNCTION_DEFINITION),
errmsg("aggregate %u needs to have compatible input type and transition type",
aggref->aggfnoid)));
}
/*
* Get a tupledesc corresponding to the inputs (including sort
* expressions) of the agg.
*/
peraggstate->evaldesc = ExecTypeFromTL(inputTargets, false);
/* Create slot we're going to do argument evaluation in */
peraggstate->evalslot = ExecInitExtraTupleSlot(estate);
ExecSetSlotDescriptor(peraggstate->evalslot, peraggstate->evaldesc);
/* Set up projection info for evaluation */
peraggstate->evalproj = ExecBuildProjectionInfo(aggrefstate->args,
aggstate->tmpcontext,
peraggstate->evalslot,
NULL);
/*
* If we're doing either DISTINCT or ORDER BY, then we have a list of
* SortGroupClause nodes; fish out the data in them and stick them
* into arrays.
*
* Note that by construction, if there is a DISTINCT clause then the
* ORDER BY clause is a prefix of it (see transformDistinctClause).
*/
if (aggref->aggdistinct)
{
TargetEntry *tle;
SortClause *sc;
Oid eq_function;
/*
* GPDB 4 doesh't implement DISTINCT aggs for aggs having more than
* than one argument, nor does it allow an ordered aggregate to
* specify distinct, but PG 9 does. The SQL standard allows the
* one-arg-for-DISTINCT restriction, but we really we ought to
* implement it the way PG 9 does eventually.
*
* For now we use the scalar equalfn field of AggStatePerAggData
* for DQAs instead of treating DQAs more generally.
*/
if (numArguments != 1)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("DISTINCT is supported only for single-argument aggregates")));
eq_function = equality_oper_funcid(inputTypes[0]);
fmgr_info(eq_function, &(peraggstate->equalfn));
tle = (TargetEntry*)linitial(inputTargets);
tle->ressortgroupref = 1;
sc = makeNode(SortClause);
sc->tleSortGroupRef = tle->ressortgroupref;
sc->sortop = ordering_oper_opid(inputTypes[0]);
sortlist = list_make1(sc);
numSortCols = 1;
}
else if ( aggref->aggorder )
{
sortlist = aggref->aggorder->sortClause;
numSortCols = list_length(sortlist);
}
else
{
sortlist = NULL;
numSortCols = 0;
}
peraggstate->numSortCols = numSortCols;
if (numSortCols > 0)
{
/*
* We don't implement DISTINCT or ORDER BY aggs in the HASHED case
* (yet)
*/
Assert(node->aggstrategy != AGG_HASHED);
/* If we have only one input, we need its len/byval info. */
if (numInputs == 1)
{
get_typlenbyval(inputTypes[0],
&peraggstate->inputtypeLen,
&peraggstate->inputtypeByVal);
}
/* Extract the sort information for use later */
peraggstate->sortColIdx =
(AttrNumber *) palloc(numSortCols * sizeof(AttrNumber));
peraggstate->sortOperators =
(Oid *) palloc(numSortCols * sizeof(Oid));
i = 0;
foreach(lc, sortlist)
{
SortClause *sortcl = (SortClause *) lfirst(lc);
TargetEntry *tle = get_sortgroupclause_tle(sortcl,
inputTargets);
/* the parser should have made sure of this */
Assert(OidIsValid(sortcl->sortop));
peraggstate->sortColIdx[i] = tle->resno;
peraggstate->sortOperators[i] = sortcl->sortop;
i++;
}
Assert(i == numSortCols);
}
if (aggref->aggdistinct)
{
Oid eqfunc;
Assert(numArguments == 1);
Assert(numSortCols == 1);
/*
* We need the equal function for the DISTINCT comparison we will
* make.
*/
eqfunc = equality_oper_funcid(inputTypes[0]);
fmgr_info(eqfunc, &peraggstate->equalfn);
}
caql_endscan(pcqCtx);
}
/*
* Process percentile expressions. These are treated separately from
* Aggref expressions at the moment as we cannot change the catalog, but
* this will be incorporated into the existing Agggref architecture
* when we can change the catalog. The operation for percentile functions
* is very similar to the Aggref operation except that there is no
* function oid for transition function. We manually manupilate
* FmgrInfo without the oid.
* In case the Agg handles PercentileExpr, there shouldn't be Aggref
* in conjunction with PercentileExpr in the target list (and havingQual),
* or vice versa, from the current design of percentile functions.
* However, we don't assert anything to keep that assumption, for the
* later extensibility.
*/
foreach (l, aggstate->percs)
{
PercentileExprState *percstate = (PercentileExprState *) lfirst(l);
PercentileExpr *perc = (PercentileExpr *) percstate->xprstate.expr;
AggStatePerAgg peraggstate;
FmgrInfo *transfn;
int numArguments;
int i;
Oid trans_argtypes[FUNC_MAX_ARGS];
ListCell *lc;
Expr *dummy_expr;
/* Look for a previous duplicate aggregate */
for (i = 0; i <= aggno; i++)
{
/*
* In practice, percentile expression doesn't contain
* volatile functions since everything is evaluated and
* becomes Var during the preprocess such as ordering operations.
* However, adding a check for volatile may be robust and
* consistent with Aggref initialization.
*/
if (equal(perc, peragg[i].perc) &&
!contain_volatile_functions((Node *) perc))
break;
}
if (i <= aggno)
{
/* Found a match to an existing entry, so just mark it */
percstate->aggno = i;
continue;
}
/* Nope, so assign a new PerAgg record */
peraggstate = &peragg[++aggno];
/* Mark Aggref state node with assigned index in the result array */
percstate->aggno = aggno;
/* Fill in the peraggstate data */
peraggstate->percstate = percstate;
peraggstate->perc = perc;
/*
* numArguments = arg + ORDER BY + pc + tc
* See notes on percentile_cont_trans() and ExecInitExpr() for
* PercentileExpr.
*/
numArguments = list_length(perc->args) + list_length(perc->sortClause) + 2;
peraggstate->numArguments = numArguments;
/*
* Set up transfn. In general, we should use fmgr_info, but we don't
* have the catalog function (thus no oid for functions) due to the
* difficulity of changing the catalog at the moment. This should
* be cleaned when we can change the catalog.
*/
transfn = &peraggstate->transfn;
transfn->fn_nargs = list_length(perc->args) + 1;
transfn->fn_strict = false;
transfn->fn_retset = false;
transfn->fn_mcxt = CurrentMemoryContext;
transfn->fn_addr = perc->perckind == PERC_DISC ?
percentile_disc_trans : percentile_cont_trans;
transfn->fn_oid = InvalidOid;
/*
* trans type is the same as result type, as they don't have final func.
*/
trans_argtypes[0] = perc->perctype;
i = 1;
/*
* Literal arguments.
*/
foreach (lc, perc->args)
{
Node *arg = lfirst(lc);
trans_argtypes[i++] = exprType(arg);
}
/*
* ORDER BY arguments.
*/
foreach (lc, perc->sortTargets)
{
TargetEntry *tle = lfirst(lc);
trans_argtypes[i++] = exprType((Node *) tle->expr);
}
/*
* Peer count and total count.
*/
trans_argtypes[i++] = INT8OID;
trans_argtypes[i++] = INT8OID;
/*
* Build FuncExpr for the transition function.
*/
build_aggregate_fnexprs(trans_argtypes,
i,
perc->perctype,
perc->perctype,
InvalidOid,
InvalidOid,
InvalidOid,
InvalidOid,
InvalidOid,
(Expr **) &transfn->fn_expr,
&dummy_expr, NULL, NULL, NULL);
get_typlenbyval(perc->perctype,
&peraggstate->resulttypeLen,
&peraggstate->resulttypeByVal);
get_typlenbyval(perc->perctype,
&peraggstate->transtypeLen,
&peraggstate->transtypeByVal);
/*
* Hard code for the known information.
*/
peraggstate->initValueIsNull = true;
peraggstate->initValue = (Datum) 0;
peraggstate->finalfn_oid = InvalidOid;
peraggstate->prelimfn_oid = InvalidOid;
/*
* Get a tupledesc corresponding to the inputs (including sort
* expressions) of the agg.
*/
peraggstate->evaldesc = ExecTypeFromTL(percstate->tlist, false);
/* Create slot we're going to do argument evaluation in */
peraggstate->evalslot = ExecInitExtraTupleSlot(estate);
ExecSetSlotDescriptor(peraggstate->evalslot, peraggstate->evaldesc);
/* Set up projection info for evaluation */
peraggstate->evalproj = ExecBuildProjectionInfo(percstate->args,
aggstate->tmpcontext,
peraggstate->evalslot,
NULL);
}
/* Update numaggs to match number of unique aggregates found */
aggstate->numaggs = aggno + 1;
/* MPP */
aggstate->hhashtable = NULL;
/* ROLLUP */
aggstate->perpassthru = NULL;
if (node->inputHasGrouping)
{
AggStatePerGroup perpassthru;
perpassthru = (AggStatePerGroup) palloc0(sizeof(AggStatePerGroupData) * numaggs);
aggstate->perpassthru = perpassthru;
}
aggstate->num_attrs = 0;
aggstate->aggType = getAggType(node);
/* Set the default memory manager */
aggstate->mem_manager.alloc = cxt_alloc;
aggstate->mem_manager.free = cxt_free;
aggstate->mem_manager.manager = aggstate->aggcontext;
aggstate->mem_manager.realloc_ratio = 1;
initGpmonPktForAgg((Plan *)node, &aggstate->ss.ps.gpmon_pkt, estate);
return aggstate;
}
Datum
GetAggInitVal(Datum textInitVal, Oid transtype)
{
Oid typinput,
typioparam;
char *strInitVal;
Datum initVal;
getTypeInputInfo(transtype, &typinput, &typioparam);
strInitVal = DatumGetCString(DirectFunctionCall1(textout, textInitVal));
initVal = OidInputFunctionCall(typinput, strInitVal,
typioparam, -1);
pfree(strInitVal);
return initVal;
}
/*
* Standard API to count tuple table slots used by an execution
* instance of an Agg node.
*
* GPDB precomputes tuple table size, but use of projection means
* aggregates use a slot. Since the count is needed earlier, we
* than the determination of then number of different aggregate
* call that happens during initializaiton, we just count Aggref
* nodes. This may be an over count (in case some aggregate
* calls are duplicated), but shouldn't be too bad.
*/
int
ExecCountSlotsAgg(Agg *node)
{
int nextraslots = 0;
nextraslots += count_extra_agg_slots((Node*)node->plan.targetlist);
nextraslots += count_extra_agg_slots((Node*)node->plan.qual);
return ExecCountSlotsNode(outerPlan(node)) +
ExecCountSlotsNode(innerPlan(node)) +
nextraslots + /* may be high due to duplicate Aggref nodes. */
AGG_NSLOTS;
}
void
ExecEndAgg(AggState *node)
{
PlanState *outerPlan;
ExecEagerFreeAgg(node);
/*
* Free both the expr contexts.
*/
ExecFreeExprContext(&node->ss.ps);
node->ss.ps.ps_ExprContext = node->tmpcontext;
ExecFreeExprContext(&node->ss.ps);
/* clean up tuple table */
ExecClearTuple(node->ss.ss_ScanTupleSlot);
if (node->num_attrs > 0)
{
pfree(node->replValues);
pfree(node->replIsnull);
pfree(node->doReplace);
}
MemoryContextDelete(node->aggcontext);
outerPlan = outerPlanState(node);
ExecEndNode(outerPlan);
EndPlanStateGpmonPkt(&node->ss.ps);
}
void
ExecReScanAgg(AggState *node, ExprContext *exprCtxt)
{
ExprContext *econtext = node->ss.ps.ps_ExprContext;
ExecEagerFreeAgg(node);
/*
* Release all temp storage. Note that with AGG_HASHED, the hash table
* is allocated in a sub-context of the aggcontext. We're going to
* rebuild the hash table from scratch, so we need to use
* MemoryContextResetAndDeleteChildren() to avoid leaking the old hash
* table's memory context header.
*/
MemoryContextResetAndDeleteChildren(node->aggcontext);
/* Re-initialize some variables */
node->agg_done = false;
ExecClearTuple(node->ss.ss_ScanTupleSlot);
/* Forget current agg values */
MemSet(econtext->ecxt_aggvalues, 0, sizeof(Datum) * node->numaggs);
MemSet(econtext->ecxt_aggnulls, 0, sizeof(bool) * node->numaggs);
if (!IS_HASHAGG(node))
{
/*
* Reset the per-group state (in particular, mark transvalues null)
*/
MemSet(node->pergroup, 0,
sizeof(AggStatePerGroupData) * node->numaggs);
}
if (((Agg *) node->ss.ps.plan)->inputHasGrouping)
{
/*
* Reset the per-passthru state (in particular, mark transvalues null)
*/
MemSet(node->perpassthru, 0,
sizeof(AggStatePerGroupData) * node->numaggs);
}
/*
* if chgParam of subnode is not null then plan will be re-scanned by
* first ExecProcNode.
*/
if (((PlanState *) node)->lefttree->chgParam == NULL)
ExecReScan(((PlanState *) node)->lefttree, exprCtxt);
}
/*
* ExecAggExplainEnd
* Called before ExecutorEnd to finish EXPLAIN ANALYZE reporting.
*/
void
ExecAggExplainEnd(PlanState *planstate, struct StringInfoData *buf)
{
AggState *aggstate = (AggState *)planstate;
/* Report executor memory used by our memory context. */
planstate->instrument->execmemused +=
(double)MemoryContextGetPeakSpace(aggstate->aggcontext);
} /* ExecAggExplainEnd */
/*
* aggregate_dummy - dummy execution routine for aggregate functions
*
* This function is listed as the implementation (prosrc field) of pg_proc
* entries for aggregate functions. Its only purpose is to throw an error
* if someone mistakenly executes such a function in the normal way.
*
* Perhaps someday we could assign real meaning to the prosrc field of
* an aggregate?
*/
Datum
aggregate_dummy(PG_FUNCTION_ARGS)
{
elog(ERROR, "aggregate function %u called as normal function",
fcinfo->flinfo->fn_oid);
return (Datum) 0; /* keep compiler quiet */
}
/* resolve actual type of transition state, if polymorphic */
Oid
resolve_polymorphic_transtype(Oid aggtranstype, Oid aggfnoid,
Oid *inputTypes)
{
if (aggtranstype == ANYARRAYOID || aggtranstype == ANYELEMENTOID)
{
/* have to fetch the agg's declared input types... */
Oid *declaredArgTypes;
int agg_nargs;
(void) get_func_signature(aggfnoid, &declaredArgTypes, &agg_nargs);
aggtranstype = enforce_generic_type_consistency(inputTypes,
declaredArgTypes,
agg_nargs,
aggtranstype);
pfree(declaredArgTypes);
}
return aggtranstype;
}
/*
* tuple_grouping - return the GROUPING value for an input tuple.
*
* This is used for a ROLLUP.
*/
int64
tuple_grouping(TupleTableSlot *outerslot,
int input_grouping,
int grpingIdx)
{
Datum grping_datum;
int64 grping;
bool isnull;
/* Simple return 0 if input_grouping is 0. */
if (input_grouping == 0)
return 0;
grping_datum = slot_getattr(outerslot, grpingIdx, &isnull);
Assert(!isnull);
grping = DatumGetInt64(grping_datum);
return grping;
}
/*
* get_grouping_groupid() -- return either grouping or group_id
* as given in 'grping_attno'.
*/
uint64
get_grouping_groupid(TupleTableSlot *slot, int grping_attno)
{
bool isnull;
uint64 grouping;
/* Obtain grouping or group_id from input */
Datum grping_datum = slot_getattr(slot,
grping_attno,
&isnull);
Assert(!isnull);
grouping = DatumGetInt64(grping_datum);
return grouping;
}
void
initGpmonPktForAgg(Plan *planNode, gpmon_packet_t *gpmon_pkt, EState *estate)
{
Assert(planNode != NULL && gpmon_pkt != NULL && IsA(planNode, Agg));
{
PerfmonNodeType type = PMNT_Invalid;
switch(((Agg*)planNode)->aggstrategy)
{
case AGG_PLAIN:
type = PMNT_Aggregate;
break;
case AGG_SORTED:
type = PMNT_GroupAggregate;
break;
case AGG_HASHED:
type = PMNT_HashAggregate;
break;
}
Assert(type != PMNT_Invalid);
Assert(GPMON_AGG_TOTAL <= (int)GPMON_QEXEC_M_COUNT);
InitPlanNodeGpmonPkt(planNode, gpmon_pkt, estate, type,
(int64)planNode->plan_rows,
NULL);
}
}
/*
* Combine the argument and sortkey expressions of an Aggref
* node into a single target list (of TargetEntry*) and, if
* needed, an associated sort key list (of SortClause*).
*
* The explicit result is a palloc'd target list incorporating
* the underlying expressions by reference. (Everything but
* the expressions is newly allocated.)
*
* The implicit result, if requested by passing a non-null
* pointer in sort_clauses, is a palloc'd sort key list.
*/
List *
combineAggrefArgs(Aggref *aggref, List **sort_clauses)
{
ListCell *lc;
TargetEntry *tle;
List *inputTargets = NIL;
List *inputSorts = NIL;
int i = 0;
/* In GPDB, can't have it both ways. */
Assert( !aggref->aggdistinct || aggref->aggorder == NULL );
/* Target list for cataloged aggregate arguments. */
foreach(lc, aggref->args)
{
TargetEntry *tle = makeNode(TargetEntry);
tle->expr = (Expr*)lfirst(lc);
tle->resno = ++i;
inputTargets = lappend(inputTargets, tle);
}
if ( aggref->aggorder != NULL )
{
/* Add targets and sort clauses for call supplied ordering. */
inputSorts = aggref->aggorder->sortClause;
if ( sort_clauses != NULL )
inputSorts = (List*)copyObject(inputSorts);
foreach(lc, inputSorts)
{
SortClause *sc = (SortClause*)lfirst(lc);
TargetEntry *newtle;
tle = get_sortgroupclause_tle(sc, aggref->aggorder->sortTargets);
/* XXX Is it worth looking for tle->expr in the tlist so far to avoid copy? */
newtle = makeNode(TargetEntry);
newtle->expr = tle->expr; /* by reference */
newtle->resno = ++i;
newtle->resname = tle->resname ? pstrdup(tle->resname) : NULL;
newtle->ressortgroupref = tle->ressortgroupref;
inputTargets = lappend(inputTargets, newtle);
}
}
else if ( aggref->aggdistinct )
{
SortClause *sc;
/* In GPDB, DISTINCT implies single argument. */
Assert( list_length(inputTargets) == 1 );
/* Add targets and sort clauses for implied DISTINCT ordering. */
tle = (TargetEntry*)linitial(inputTargets);
tle->ressortgroupref = 1;
if ( sort_clauses != NULL )
{
sc = makeNode(SortClause);
sc->tleSortGroupRef = tle->ressortgroupref;
inputSorts = list_make1(sc);
}
}
if ( sort_clauses != NULL )
*sort_clauses = inputSorts;
return inputTargets;
}
/*
* Combine the argument and ordering expression with the peer count
* and the total count, to create the TupleTableSlot for this
* expression. This is similar to ordered aggregate Aggref,
* but the difference is that PercentileExpr will accept those
* additional values as the arguments to the transition function.
*/
List *
combinePercentileArgs(PercentileExpr *p)
{
List *tlist;
ListCell *l;
AttrNumber resno;
TargetEntry *tle;
tlist = NIL;
resno = 1;
foreach (l, p->args)
{
Expr *arg = lfirst(l);
tle = makeTargetEntry((Expr *) arg,
resno++,
NULL,
false);
tlist = lappend(tlist, tle);
}
/*
* Extract ordering expressions from sortTargets.
*/
foreach (l, p->sortClause)
{
SortClause *sc = lfirst(l);
TargetEntry *sc_tle;
sc_tle = get_sortgroupclause_tle(sc, p->sortTargets);
tle = flatCopyTargetEntry(sc_tle);
tle->resno = resno++;
tlist = lappend(tlist, tle);
}
/*
* peer count expresssion.
*/
Assert(p->pcExpr);
tle = makeTargetEntry((Expr *) p->pcExpr,
resno++,
"peer_count",
false);
tlist = lappend(tlist, tle);
/*
* total count expresssion.
*/
Assert(p->tcExpr);
tle = makeTargetEntry((Expr *) p->tcExpr,
resno++,
"total_count",
false);
tlist = lappend(tlist, tle);
return tlist;
}
/*
* Subroutines for ExecCountSlotsAgg.
*/
int
count_extra_agg_slots(Node *node)
{
int count = 0;
count_extra_agg_slots_walker(node, &count);
return count;
}
bool
count_extra_agg_slots_walker(Node *node, int *count)
{
if (node == NULL)
return false;
if (IsA(node, Aggref))
{
(*count)++;
}
else if (IsA(node, PercentileExpr))
{
(*count)++;
}
return expression_tree_walker(node, count_extra_agg_slots_walker, (void *) count);
}
void
ExecEagerFreeAgg(AggState *node)
{
/* Close any open tuplesorts */
for (int aggno = 0; aggno < node->numaggs; aggno++)
{
AggStatePerAgg peraggstate = &node->peragg[aggno];
if (!peraggstate->sortstate)
{
continue;
}
if (gp_enable_mk_sort)
{
tuplesort_end_mk((Tuplesortstate_mk *) peraggstate->sortstate);
}
else
{
tuplesort_end((Tuplesortstate *) peraggstate->sortstate);
}
peraggstate->sortstate = NULL;
}
if (IS_HASHAGG(node))
{
destroy_agg_hash_table(node);
/**
* Clean out the tuple descriptor.
*/
if (node->hashslot
&& node->hashslot->tts_tupleDescriptor)
{
ReleaseTupleDesc(node->hashslot->tts_tupleDescriptor);
node->hashslot->tts_tupleDescriptor = NULL;
}
}
/* Release first tuple of group, if we have made a copy. */
if (node->grp_firstTuple != NULL)
{
pfree(node->grp_firstTuple);
node->grp_firstTuple = NULL;
}
}