postgresql/src/backend/catalog/pg_aggregate.c

/*-------------------------------------------------------------------------
 *
 * pg_aggregate.c
 *	  routines to support manipulation of the pg_aggregate relation
 *
 * Portions Copyright (c) 1996-2001, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  $Header: /cvsroot/pgsql/src/backend/catalog/pg_aggregate.c,v 1.44 2002/04/11 19:59:57 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/heapam.h"
#include "catalog/catname.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/pg_aggregate.h"
#include "catalog/pg_language.h"
#include "catalog/pg_proc.h"
#include "optimizer/cost.h"
#include "parser/parse_coerce.h"
#include "parser/parse_func.h"
#include "parser/parse_type.h"
#include "utils/builtins.h"
#include "utils/syscache.h"


/*
 * AggregateCreate
 */
void
AggregateCreate(const char *aggName,
				Oid aggNamespace,
				List *aggtransfnName,
				List *aggfinalfnName,
				Oid aggBaseType,
				Oid aggTransType,
				const char *agginitval)
{
	Relation	aggdesc;
	HeapTuple	tup;
	char		nulls[Natts_pg_aggregate];
	Datum		values[Natts_pg_aggregate];
	Form_pg_proc proc;
	Oid			transfn;
	Oid			finalfn = InvalidOid;	/* can be omitted */
	Oid			finaltype;
	Oid			fnArgs[FUNC_MAX_ARGS];
	int			nargs;
	Oid			procOid;
	TupleDesc	tupDesc;
	int			i;

	/* sanity checks */
	if (!aggName)
		elog(ERROR, "no aggregate name supplied");

	if (!aggtransfnName)
		elog(ERROR, "aggregate must have a transition function");

	/* handle transfn */
	MemSet(fnArgs, 0, FUNC_MAX_ARGS * sizeof(Oid));
	fnArgs[0] = aggTransType;
	if (OidIsValid(aggBaseType))
	{
		fnArgs[1] = aggBaseType;
		nargs = 2;
	}
	else
		nargs = 1;
	transfn = LookupFuncName(aggtransfnName, nargs, fnArgs);
	if (!OidIsValid(transfn))
		func_error("AggregateCreate", aggtransfnName, nargs, fnArgs, NULL);
	tup = SearchSysCache(PROCOID,
						 ObjectIdGetDatum(transfn),
						 0, 0, 0);
	if (!HeapTupleIsValid(tup))
		func_error("AggregateCreate", aggtransfnName, nargs, fnArgs, NULL);
	proc = (Form_pg_proc) GETSTRUCT(tup);
	if (proc->prorettype != aggTransType)
		elog(ERROR, "return type of transition function %s is not %s",
			 NameListToString(aggtransfnName), typeidTypeName(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.
	 */
	if (proc->proisstrict && agginitval == NULL)
	{
		if (!IsBinaryCompatible(aggBaseType, aggTransType))
			elog(ERROR, "must not omit initval when transfn is strict and transtype is not compatible with input type");
	}
	ReleaseSysCache(tup);

	/* handle finalfn, if supplied */
	if (aggfinalfnName)
	{
		MemSet(fnArgs, 0, FUNC_MAX_ARGS * sizeof(Oid));
		fnArgs[0] = aggTransType;
		finalfn = LookupFuncName(aggfinalfnName, 1, fnArgs);
		if (!OidIsValid(finalfn))
			func_error("AggregateCreate", aggfinalfnName, 1, fnArgs, NULL);
		tup = SearchSysCache(PROCOID,
							 ObjectIdGetDatum(finalfn),
							 0, 0, 0);
		if (!HeapTupleIsValid(tup))
			func_error("AggregateCreate", aggfinalfnName, 1, fnArgs, NULL);
		proc = (Form_pg_proc) GETSTRUCT(tup);
		finaltype = proc->prorettype;
		ReleaseSysCache(tup);
	}
	else
	{
		/*
		 * If no finalfn, aggregate result type is type of the state value
		 */
		finaltype = aggTransType;
	}
	Assert(OidIsValid(finaltype));

	/*
	 * Everything looks okay.  Try to create the pg_proc entry for the
	 * aggregate.  (This could fail if there's already a conflicting entry.)
	 */
	MemSet(fnArgs, 0, FUNC_MAX_ARGS * sizeof(Oid));
	fnArgs[0] = aggBaseType;

	procOid = ProcedureCreate(aggName,
							  aggNamespace,
							  false,		/* no replacement */
							  false,		/* doesn't return a set */
							  finaltype,	/* returnType */
							  INTERNALlanguageId,	/* languageObjectId */
							  "aggregate_dummy",	/* placeholder proc */
							  "-",			/* probin */
							  true,			/* isAgg */
							  true,			/* (obsolete "trusted") */
							  false,		/* isImplicit */
							  false,		/* isStrict (not needed for agg) */
							  PROVOLATILE_IMMUTABLE,	/* volatility (not needed for agg) */
							  BYTE_PCT,		/* default cost values */
							  PERBYTE_CPU,
							  PERCALL_CPU,
							  OUTIN_RATIO,
							  1,			/* parameterCount */
							  fnArgs);		/* parameterTypes */

	/*
	 * Okay to create the pg_aggregate entry.
	 */

	/* initialize nulls and values */
	for (i = 0; i < Natts_pg_aggregate; i++)
	{
		nulls[i] = ' ';
		values[i] = (Datum) NULL;
	}
	values[Anum_pg_aggregate_aggfnoid - 1] = ObjectIdGetDatum(procOid);
	values[Anum_pg_aggregate_aggtransfn - 1] = ObjectIdGetDatum(transfn);
	values[Anum_pg_aggregate_aggfinalfn - 1] = ObjectIdGetDatum(finalfn);
	values[Anum_pg_aggregate_aggtranstype - 1] = ObjectIdGetDatum(aggTransType);
	if (agginitval)
		values[Anum_pg_aggregate_agginitval - 1] =
			DirectFunctionCall1(textin, CStringGetDatum(agginitval));
	else
		nulls[Anum_pg_aggregate_agginitval - 1] = 'n';

	aggdesc = heap_openr(AggregateRelationName, RowExclusiveLock);
	tupDesc = aggdesc->rd_att;

	tup = heap_formtuple(tupDesc, values, nulls);
	heap_insert(aggdesc, tup);

	if (RelationGetForm(aggdesc)->relhasindex)
	{
		Relation	idescs[Num_pg_aggregate_indices];

		CatalogOpenIndices(Num_pg_aggregate_indices, Name_pg_aggregate_indices, idescs);
		CatalogIndexInsert(idescs, Num_pg_aggregate_indices, aggdesc, tup);
		CatalogCloseIndices(Num_pg_aggregate_indices, idescs);
	}

	heap_close(aggdesc, RowExclusiveLock);
}