/*-------------------------------------------------------------------------
 *
 * nodeAgg.c--
 *	  Routines to handle aggregate nodes.
 *
 * Copyright (c) 1994, Regents of the University of California
 *
 *
 * NOTE
 *	  The implementation of Agg node has been reworked to handle legal
 *	  SQL aggregates. (Do not expect POSTQUEL semantics.)	 -- ay 2/95
 *
 * IDENTIFICATION
 *	  /usr/local/devel/pglite/cvs/src/backend/executor/nodeAgg.c,v 1.13 1995/08/01 20:19:07 jolly Exp
 *
 *-------------------------------------------------------------------------
 */
#include <string.h>

#include "postgres.h"
#include "fmgr.h"

#include "access/heapam.h"
#include "catalog/pg_aggregate.h"
#include "catalog/catalog.h"
#include "parser/parse_type.h"
#include "executor/executor.h"
#include "executor/nodeAgg.h"
#include "storage/bufmgr.h"
#include "utils/palloc.h"
#include "utils/syscache.h"
#include "optimizer/clauses.h"

/*
 * AggFuncInfo -
 *	  keeps the transition functions information around
 */
typedef struct AggFuncInfo
{
	Oid			xfn1_oid;
	Oid			xfn2_oid;
	Oid			finalfn_oid;
	FmgrInfo	xfn1;
	FmgrInfo	xfn2;
	FmgrInfo	finalfn;
} AggFuncInfo;

static Datum aggGetAttr(TupleTableSlot *tuple, Aggreg *agg, bool *isNull);


/* ---------------------------------------
 *
 * ExecAgg -
 *
 *	  ExecAgg receives tuples from its outer subplan and aggregates over
 *	  the appropriate attribute for each (unique) aggregate in the target
 *	  list. (The number of tuples to aggregate over depends on whether a
 *	  GROUP BY clause is present. It might be the number of tuples in a
 *	  group or all the tuples that satisfy the qualifications.) The value of
 *	  each aggregate is stored in the expression context for ExecProject to
 *	  evaluate the result tuple.
 *
 *	  ExecAgg evaluates each aggregate in the following steps: (initcond1,
 *	  initcond2 are the initial values and sfunc1, sfunc2, and finalfunc are
 *	  the transition functions.)
 *
 *		 value1[i] = initcond1
 *		 value2[i] = initcond2
 *		 forall tuples do
 *			value1[i] = sfunc1(aggregate_attribute, value1[i])
 *			value2[i] = sfunc2(value2[i])
 *		 value1[i] = finalfunc(value1[i], value2[i])
 *
 *	  If the outer subplan is a Group node, ExecAgg returns as many tuples
 *	  as there are groups.
 *
 *	  XXX handling of NULL doesn't work
 *
 *	  OLD COMMENTS
 *
 *		XXX Aggregates should probably have another option: what to do
 *		with transfn2 if we hit a null value.  "count" (transfn1 = null,
 *		transfn2 = increment) will want to have transfn2 called; "avg"
 *		(transfn1 = add, transfn2 = increment) will not. -pma 1/3/93
 *
 * ------------------------------------------
 */
TupleTableSlot *
ExecAgg(Agg *node)
{
	AggState   *aggstate;
	EState	   *estate;
	Aggreg	  **aggregates;
	Plan	   *outerPlan;
	int			i,
				nagg;
	Datum	   *value1,
			   *value2;
	int		   *noInitValue;
	AggFuncInfo *aggFuncInfo;
	long		nTuplesAgged = 0;
	ExprContext *econtext;
	ProjectionInfo *projInfo;
	TupleTableSlot *resultSlot;
	HeapTuple	oneTuple;
	List	   *alist;
	char	   *nulls;
	bool		isDone;
	bool		isNull = FALSE,
				isNull1 = FALSE,
				isNull2 = FALSE;
	bool		qual_result;


	/* ---------------------
	 *	get state info from node
	 * ---------------------
	 */

	/*
	 * We loop retrieving groups until we find one matching
	 * node->plan.qual
	 */
	do
	{

		aggstate = node->aggstate;
		if (aggstate->agg_done)
			return NULL;

		estate = node->plan.state;
		econtext = aggstate->csstate.cstate.cs_ExprContext;

		nagg = length(node->aggs);

		aggregates = (Aggreg **) palloc(sizeof(Aggreg *) * nagg);

		/* take List* and make it an array that can be quickly indexed */
		alist = node->aggs;
		for (i = 0; i < nagg; i++)
		{
			aggregates[i] = lfirst(alist);
			aggregates[i]->aggno = i;
			alist = lnext(alist);
		}

		value1 = node->aggstate->csstate.cstate.cs_ExprContext->ecxt_values;
		nulls = node->aggstate->csstate.cstate.cs_ExprContext->ecxt_nulls;

		value2 = (Datum *) palloc(sizeof(Datum) * nagg);
		MemSet(value2, 0, sizeof(Datum) * nagg);

		aggFuncInfo = (AggFuncInfo *) palloc(sizeof(AggFuncInfo) * nagg);
		MemSet(aggFuncInfo, 0, sizeof(AggFuncInfo) * nagg);

		noInitValue = (int *) palloc(sizeof(int) * nagg);
		MemSet(noInitValue, 0, sizeof(noInitValue) * nagg);

		outerPlan = outerPlan(node);
		oneTuple = NULL;

		projInfo = aggstate->csstate.cstate.cs_ProjInfo;

		for (i = 0; i < nagg; i++)
		{
			Aggreg	   *agg;
			char	   *aggname;
			HeapTuple	aggTuple;
			Form_pg_aggregate aggp;
			Oid			xfn1_oid,
						xfn2_oid,
						finalfn_oid;

			agg = aggregates[i];

			/* ---------------------
			 *	find transfer functions of all the aggregates and initialize
			 *	their initial values
			 * ---------------------
			 */
			aggname = agg->aggname;
			aggTuple = SearchSysCacheTuple(AGGNAME,
										   PointerGetDatum(aggname),
										 ObjectIdGetDatum(agg->basetype),
										   0, 0);
			if (!HeapTupleIsValid(aggTuple))
				elog(ERROR, "ExecAgg: cache lookup failed for aggregate \"%s\"(%s)",
					 aggname,
					 typeidTypeName(agg->basetype));
			aggp = (Form_pg_aggregate) GETSTRUCT(aggTuple);

			xfn1_oid = aggp->aggtransfn1;
			xfn2_oid = aggp->aggtransfn2;
			finalfn_oid = aggp->aggfinalfn;

			if (OidIsValid(finalfn_oid))
			{
				fmgr_info(finalfn_oid, &aggFuncInfo[i].finalfn);
				aggFuncInfo[i].finalfn_oid = finalfn_oid;
			}

			if (OidIsValid(xfn2_oid))
			{
				fmgr_info(xfn2_oid, &aggFuncInfo[i].xfn2);
				aggFuncInfo[i].xfn2_oid = xfn2_oid;
				value2[i] = (Datum) AggNameGetInitVal((char *) aggname,
													  aggp->aggbasetype,
													  2,
													  &isNull2);
				/* ------------------------------------------
				 * If there is a second transition function, its initial
				 * value must exist -- as it does not depend on data values,
				 * we have no other way of determining an initial value.
				 * ------------------------------------------
				 */
				if (isNull2)
					elog(ERROR, "ExecAgg: agginitval2 is null");
			}

			if (OidIsValid(xfn1_oid))
			{
				fmgr_info(xfn1_oid, &aggFuncInfo[i].xfn1);
				aggFuncInfo[i].xfn1_oid = xfn1_oid;
				value1[i] = (Datum) AggNameGetInitVal((char *) aggname,
													  aggp->aggbasetype,
													  1,
													  &isNull1);

				/* ------------------------------------------
				 * If the initial value for the first transition function
				 * doesn't exist in the pg_aggregate table then we let
				 * the first value returned from the outer procNode become
				 * the initial value. (This is useful for aggregates like
				 * max{} and min{}.)
				 * ------------------------------------------
				 */
				if (isNull1)
				{
					noInitValue[i] = 1;
					nulls[i] = 1;
				}
			}
		}

		/* ----------------
		 *	 for each tuple from the the outer plan, apply all the aggregates
		 * ----------------
		 */
		for (;;)
		{
			HeapTuple	outerTuple = NULL;
			TupleTableSlot *outerslot;

			isNull = isNull1 = isNull2 = 0;
			outerslot = ExecProcNode(outerPlan, (Plan *) node);
			if (outerslot)
				outerTuple = outerslot->val;
			if (!HeapTupleIsValid(outerTuple))
			{

				/*
				 * when the outerplan doesn't return a single tuple,
				 * create a dummy heaptuple anyway because we still need
				 * to return a valid aggregate value. The value returned
				 * will be the initial values of the transition functions
				 */
				if (nTuplesAgged == 0)
				{
					TupleDesc	tupType;
					Datum	   *tupValue;
					char	   *null_array;

					tupType = aggstate->csstate.css_ScanTupleSlot->ttc_tupleDescriptor;
					tupValue = projInfo->pi_tupValue;

					/* initially, set all the values to NULL */
					null_array = palloc(tupType->natts);
					for (i = 0; i < tupType->natts; i++)
						null_array[i] = 'n';
					oneTuple = heap_formtuple(tupType, tupValue, null_array);
					pfree(null_array);
				}
				break;
			}

			for (i = 0; i < nagg; i++)
			{
				AttrNumber	attnum;
				int2		attlen = 0;
				Datum		newVal = (Datum) NULL;
				AggFuncInfo *aggfns = &aggFuncInfo[i];
				Datum		args[2];
				Node	   *tagnode = NULL;

				switch (nodeTag(aggregates[i]->target))
				{
					case T_Var:
						tagnode = NULL;
						newVal = aggGetAttr(outerslot,
											aggregates[i],
											&isNull);
						break;
					case T_Expr:
						tagnode = ((Expr *) aggregates[i]->target)->oper;
						econtext->ecxt_scantuple = outerslot;
						newVal = ExecEvalExpr(aggregates[i]->target, econtext,
											  &isNull, &isDone);
						break;
					case T_Const:
						tagnode = NULL;
						econtext->ecxt_scantuple = outerslot;
						newVal = ExecEvalExpr(aggregates[i]->target, econtext,
											  &isNull, &isDone);
						break;
					default:
						elog(ERROR, "ExecAgg: Bad Agg->Target for Agg %d", i);
				}

				if (isNull && !aggregates[i]->usenulls)
					continue;	/* ignore this tuple for this agg */

				if (aggfns->xfn1.fn_addr != NULL)
				{
					if (noInitValue[i])
					{
						int			byVal = 0;

						/*
						 * value1 and value2 has not been initialized.
						 * This is the first non-NULL value. We use it as
						 * the initial value.
						 */

						/*
						 * but we can't just use it straight, we have to
						 * make a copy of it since the tuple from which it
						 * came will be freed on the next iteration of the
						 * scan
						 */
						switch (nodeTag(aggregates[i]->target))
						{
							case T_Var:
								attnum = ((Var *) aggregates[i]->target)->varattno;
								attlen = outerslot->ttc_tupleDescriptor->attrs[attnum - 1]->attlen;
								byVal = outerslot->ttc_tupleDescriptor->attrs[attnum - 1]->attbyval;

								break;
							case T_Expr:
								{
									FunctionCachePtr fcache_ptr;

									if (nodeTag(tagnode) == T_Func)
										fcache_ptr = ((Func *) tagnode)->func_fcache;
									else
										fcache_ptr = ((Oper *) tagnode)->op_fcache;
									attlen = fcache_ptr->typlen;
									byVal = fcache_ptr->typbyval;

									break;
								}
							case T_Const:
								attlen = ((Const *) aggregates[i]->target)->constlen;
								byVal = ((Const *) aggregates[i]->target)->constbyval;

								break;
							default:
								elog(ERROR, "ExecAgg: Bad Agg->Target for Agg %d", i);
						}
						if (attlen == -1)
						{
							/* variable length */
							attlen = VARSIZE((struct varlena *) newVal);
						}
						value1[i] = (Datum) palloc(attlen);
						if (byVal)
							value1[i] = newVal;
						else
							memmove((char *) (value1[i]), (char *) newVal, attlen);
						noInitValue[i] = 0;
						nulls[i] = 0;
					}
					else
					{

						/*
						 * apply the transition functions.
						 */
						args[0] = value1[i];
						args[1] = newVal;
						value1[i] =
							(Datum) fmgr_c(&aggfns->xfn1,
										   (FmgrValues *) args,
										   &isNull1);
						Assert(!isNull1);
					}
				}

				if (aggfns->xfn2.fn_addr != NULL)
				{
					Datum		xfn2_val = value2[i];

					value2[i] =
						(Datum) fmgr_c(&aggfns->xfn2,
									 (FmgrValues *) &xfn2_val, &isNull2);
					Assert(!isNull2);
				}
			}

			/*
			 * keep this for the projection (we only need one of these -
			 * all the tuples we aggregate over share the same group
			 * column)
			 */
			if (!oneTuple)
				oneTuple = heap_copytuple(outerslot->val);

			nTuplesAgged++;
		}

		/* --------------
		 * finalize the aggregate (if necessary), and get the resultant value
		 * --------------
		 */
		for (i = 0; i < nagg; i++)
		{
			char	   *args[2];
			AggFuncInfo *aggfns = &aggFuncInfo[i];

			if (noInitValue[i])
			{

				/*
				 * No values found for this agg; return current state.
				 * This seems to fix behavior for avg() aggregate. -tgl
				 * 12/96
				 */
			}
			else if (aggfns->finalfn.fn_addr != NULL && nTuplesAgged > 0)
			{
				if (aggfns->finalfn.fn_nargs > 1)
				{
					args[0] = (char *) value1[i];
					args[1] = (char *) value2[i];
				}
				else if (aggfns->xfn1.fn_addr != NULL)
					args[0] = (char *) value1[i];
				else if (aggfns->xfn2.fn_addr != NULL)
					args[0] = (char *) value2[i];
				else
					elog(NOTICE, "ExecAgg: no valid transition functions??");
				value1[i] = (Datum) fmgr_c(&aggfns->finalfn,
									   (FmgrValues *) args, &(nulls[i]));
			}
			else if (aggfns->xfn1.fn_addr != NULL)
			{

				/*
				 * value in the right place, ignore. (If you remove this
				 * case, fix the else part. -ay 2/95)
				 */
			}
			else if (aggfns->xfn2.fn_addr != NULL)
				value1[i] = value2[i];
			else
				elog(ERROR, "ExecAgg: no valid transition functions??");
		}

		/*
		 * whether the aggregation is done depends on whether we are doing
		 * aggregation over groups or the entire table
		 */
		if (nodeTag(outerPlan) == T_Group)
		{
			/* aggregation over groups */
			aggstate->agg_done = ((Group *) outerPlan)->grpstate->grp_done;
		}
		else
			aggstate->agg_done = TRUE;

		/* ----------------
		 *	form a projection tuple, store it in the result tuple
		 *	slot and return it.
		 * ----------------
		 */

		ExecStoreTuple(oneTuple,
					   aggstate->csstate.css_ScanTupleSlot,
					   InvalidBuffer,
					   false);
		econtext->ecxt_scantuple = aggstate->csstate.css_ScanTupleSlot;

		resultSlot = ExecProject(projInfo, &isDone);

		/*
		 * As long as the retrieved group does not match the
		 * qualifications it is ignored and the next group is fetched
		 */
		qual_result = ExecQual(fix_opids(node->plan.qual), econtext);
		if (oneTuple)
			pfree(oneTuple);
	}
	while ((node->plan.qual != NULL) && (qual_result != true));

	return resultSlot;
}

/* -----------------
 * ExecInitAgg
 *
 *	Creates the run-time information for the agg node produced by the
 *	planner and initializes its outer subtree
 * -----------------
 */
bool
ExecInitAgg(Agg *node, EState *estate, Plan *parent)
{
	AggState   *aggstate;
	Plan	   *outerPlan;
	ExprContext *econtext;

	/*
	 * assign the node's execution state
	 */
	node->plan.state = estate;

	/*
	 * create state structure
	 */
	aggstate = makeNode(AggState);
	node->aggstate = aggstate;
	aggstate->agg_done = FALSE;

	/*
	 * assign node's base id and create expression context
	 */
	ExecAssignNodeBaseInfo(estate, &aggstate->csstate.cstate,
						   (Plan *) parent);
	ExecAssignExprContext(estate, &aggstate->csstate.cstate);

#define AGG_NSLOTS 2

	/*
	 * tuple table initialization
	 */
	ExecInitScanTupleSlot(estate, &aggstate->csstate);
	ExecInitResultTupleSlot(estate, &aggstate->csstate.cstate);

	econtext = aggstate->csstate.cstate.cs_ExprContext;
	econtext->ecxt_values =
		(Datum *) palloc(sizeof(Datum) * length(node->aggs));
	MemSet(econtext->ecxt_values, 0, sizeof(Datum) * length(node->aggs));
	econtext->ecxt_nulls = (char *) palloc(length(node->aggs));
	MemSet(econtext->ecxt_nulls, 0, length(node->aggs));

	/*
	 * initializes child nodes
	 */
	outerPlan = outerPlan(node);
	ExecInitNode(outerPlan, estate, (Plan *) node);

	/*
	 * Result runs in its own context, but make it use our aggregates fix
	 * for 'select sum(2+2)'
	 */
	if (nodeTag(outerPlan) == T_Result)
	{
		((Result *) outerPlan)->resstate->cstate.cs_ProjInfo->pi_exprContext->ecxt_values =
			econtext->ecxt_values;
		((Result *) outerPlan)->resstate->cstate.cs_ProjInfo->pi_exprContext->ecxt_nulls =
			econtext->ecxt_nulls;
	}


	/* ----------------
	 *	initialize tuple type.
	 * ----------------
	 */
	ExecAssignScanTypeFromOuterPlan((Plan *) node, &aggstate->csstate);

	/*
	 * Initialize tuple type for both result and scan. This node does no
	 * projection
	 */
	ExecAssignResultTypeFromTL((Plan *) node, &aggstate->csstate.cstate);
	ExecAssignProjectionInfo((Plan *) node, &aggstate->csstate.cstate);

	return TRUE;
}

int
ExecCountSlotsAgg(Agg *node)
{
	return ExecCountSlotsNode(outerPlan(node)) +
	ExecCountSlotsNode(innerPlan(node)) +
	AGG_NSLOTS;
}

/* ------------------------
 *		ExecEndAgg(node)
 *
 * -----------------------
 */
void
ExecEndAgg(Agg *node)
{
	AggState   *aggstate;
	Plan	   *outerPlan;

	aggstate = node->aggstate;

	ExecFreeProjectionInfo(&aggstate->csstate.cstate);

	outerPlan = outerPlan(node);
	ExecEndNode(outerPlan, (Plan *) node);

	/* clean up tuple table */
	ExecClearTuple(aggstate->csstate.css_ScanTupleSlot);
}


/*****************************************************************************
 *	Support Routines
 *****************************************************************************/

/*
 * aggGetAttr -
 *	  get the attribute (specified in the Var node in agg) to aggregate
 *	  over from the tuple
 */
static Datum
aggGetAttr(TupleTableSlot *slot,
		   Aggreg *agg,
		   bool *isNull)
{
	Datum		result;
	AttrNumber	attnum;
	HeapTuple	heapTuple;
	TupleDesc	tuple_type;
	Buffer		buffer;

	/* ----------------
	 *	 extract tuple information from the slot
	 * ----------------
	 */
	heapTuple = slot->val;
	tuple_type = slot->ttc_tupleDescriptor;
	buffer = slot->ttc_buffer;

	attnum = ((Var *) agg->target)->varattno;

	/*
	 * If the attribute number is invalid, then we are supposed to return
	 * the entire tuple, we give back a whole slot so that callers know
	 * what the tuple looks like.
	 */
	if (attnum == InvalidAttrNumber)
	{
		TupleTableSlot *tempSlot;
		TupleDesc	td;
		HeapTuple	tup;

		tempSlot = makeNode(TupleTableSlot);
		tempSlot->ttc_shouldFree = false;
		tempSlot->ttc_descIsNew = true;
		tempSlot->ttc_tupleDescriptor = (TupleDesc) NULL,
			tempSlot->ttc_buffer = InvalidBuffer;
		tempSlot->ttc_whichplan = -1;

		tup = heap_copytuple(slot->val);
		td = CreateTupleDescCopy(slot->ttc_tupleDescriptor);

		ExecSetSlotDescriptor(tempSlot, td);

		ExecStoreTuple(tup, tempSlot, InvalidBuffer, true);
		return (Datum) tempSlot;
	}

	result =
		heap_getattr(heapTuple, /* tuple containing attribute */
					 attnum,	/* attribute number of desired attribute */
					 tuple_type,/* tuple descriptor of tuple */
					 isNull);	/* return: is attribute null? */

	/* ----------------
	 *	return null if att is null
	 * ----------------
	 */
	if (*isNull)
		return (Datum) NULL;

	return result;
}

void
ExecReScanAgg(Agg *node, ExprContext *exprCtxt, Plan *parent)
{
	AggState   *aggstate = node->aggstate;
	ExprContext *econtext = aggstate->csstate.cstate.cs_ExprContext;

	aggstate->agg_done = FALSE;
	MemSet(econtext->ecxt_values, 0, sizeof(Datum) * length(node->aggs));
	MemSet(econtext->ecxt_nulls, 0, length(node->aggs));

	/*
	 * if chgParam of subnode is not null then plan will be re-scanned by
	 * first ExecProcNode.
	 */
	if (((Plan *) node)->lefttree->chgParam == NULL)
		ExecReScan(((Plan *) node)->lefttree, exprCtxt, (Plan *) node);

}
