multiset-src-20110112.patch

diff --git a/src/backend/nodes/makefuncs.c b/src/backend/nodes/makefuncs.c
index 79da185..df95fee 100644
*** a/src/backend/nodes/makefuncs.c
--- b/src/backend/nodes/makefuncs.c
*************** makeFuncExpr(Oid funcid, Oid rettype, Li
*** 454,459 ****
--- 454,474 ----
  }
  
  /*
+  * makeFuncCall -
+  *	build a FuncCall node
+  */
+ FuncCall *
+ makeFuncCall(List *funcname, List *args, int location)
+ {
+ 	FuncCall   *n = makeNode(FuncCall);
+ 
+ 	n->funcname = funcname;
+ 	n->args = args;
+ 	n->location = location;
+ 	return n;
+ }
+ 
+ /*
   * makeDefElem -
   *	build a DefElem node
   *
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 660947c..77fb898 100644
*** a/src/backend/parser/gram.y
--- b/src/backend/parser/gram.y
*************** static RangeVar *makeRangeVarFromAnyName
*** 468,474 ****
   */
  
  /* ordinary key words in alphabetical order */
! %token <keyword> ABORT_P ABSOLUTE_P ACCESS ACTION ADD_P ADMIN AFTER
  	AGGREGATE ALL ALSO ALTER ALWAYS ANALYSE ANALYZE AND ANY ARRAY AS ASC
  	ASSERTION ASSIGNMENT ASYMMETRIC AT ATTRIBUTE AUTHORIZATION
  
--- 468,474 ----
   */
  
  /* ordinary key words in alphabetical order */
! %token <keyword> A ABORT_P ABSOLUTE_P ACCESS ACTION ADD_P ADMIN AFTER
  	AGGREGATE ALL ALSO ALTER ALWAYS ANALYSE ANALYZE AND ANY ARRAY AS ASC
  	ASSERTION ASSIGNMENT ASYMMETRIC AT ATTRIBUTE AUTHORIZATION
  
*************** static RangeVar *makeRangeVarFromAnyName
*** 511,517 ****
  	LEAST LEFT LEVEL LIKE LIMIT LISTEN LOAD LOCAL LOCALTIME LOCALTIMESTAMP
  	LOCATION LOCK_P LOGIN_P
  
! 	MAPPING MATCH MAXVALUE MINUTE_P MINVALUE MODE MONTH_P MOVE
  
  	NAME_P NAMES NATIONAL NATURAL NCHAR NEXT NO NOCREATEDB
  	NOCREATEROLE NOCREATEUSER NOINHERIT NOLOGIN_P NONE NOREPLICATION_P
--- 511,517 ----
  	LEAST LEFT LEVEL LIKE LIMIT LISTEN LOAD LOCAL LOCALTIME LOCALTIMESTAMP
  	LOCATION LOCK_P LOGIN_P
  
! 	MAPPING MATCH MAXVALUE MEMBER MINUTE_P MINVALUE MODE MONTH_P MOVE MULTISET
  
  	NAME_P NAMES NATIONAL NATURAL NCHAR NEXT NO NOCREATEDB
  	NOCREATEROLE NOCREATEUSER NOINHERIT NOLOGIN_P NONE NOREPLICATION_P
*************** static RangeVar *makeRangeVarFromAnyName
*** 535,542 ****
  	SAVEPOINT SCHEMA SCROLL SEARCH SECOND_P SECURITY SELECT SEQUENCE SEQUENCES
  	SERIALIZABLE SERVER SESSION SESSION_USER SET SETOF SHARE
  	SHOW SIMILAR SIMPLE SMALLINT SOME STABLE STANDALONE_P START STATEMENT
! 	STATISTICS STDIN STDOUT STORAGE STRICT_P STRIP_P SUBSTRING SUPERUSER_P
! 	SYMMETRIC SYSID SYSTEM_P
  
  	TABLE TABLES TABLESPACE TEMP TEMPLATE TEMPORARY TEXT_P THEN TIME TIMESTAMP
  	TO TRAILING TRANSACTION TREAT TRIGGER TRIM TRUE_P
--- 535,542 ----
  	SAVEPOINT SCHEMA SCROLL SEARCH SECOND_P SECURITY SELECT SEQUENCE SEQUENCES
  	SERIALIZABLE SERVER SESSION SESSION_USER SET SETOF SHARE
  	SHOW SIMILAR SIMPLE SMALLINT SOME STABLE STANDALONE_P START STATEMENT
! 	STATISTICS STDIN STDOUT STORAGE STRICT_P STRIP_P SUBMULTISET SUBSTRING
! 	SUPERUSER_P SYMMETRIC SYSID SYSTEM_P
  
  	TABLE TABLES TABLESPACE TEMP TEMPLATE TEMPORARY TEXT_P THEN TIME TIMESTAMP
  	TO TRAILING TRANSACTION TREAT TRIGGER TRIM TRUE_P
*************** static RangeVar *makeRangeVarFromAnyName
*** 605,610 ****
--- 605,611 ----
  %nonassoc	NOTNULL
  %nonassoc	ISNULL
  %nonassoc	IS NULL_P TRUE_P FALSE_P UNKNOWN /* sets precedence for IS NULL, etc */
+ %nonassoc	MEMBER MULTISET SUBMULTISET
  %left		'+' '-'
  %left		'*' '/' '%'
  %left		'^'
*************** a_expr:		c_expr									{ $$ = $1; }
*** 9584,9589 ****
--- 9585,9654 ----
  														 list_make1($1), @2),
  											 @2);
  				}
+ 			| a_expr IS A SET
+ 				{
+ 					$$ = (Node *) makeFuncCall(
+ 							SystemFuncName("is_a_set"),
+ 							list_make1($1), @2);
+ 				}
+ 			| a_expr IS NOT A SET
+ 				{
+ 					$$ = (Node *) makeA_Expr(AEXPR_NOT, NIL, NULL,
+ 							(Node *) makeFuncCall(
+ 								SystemFuncName("is_a_set"),
+ 								list_make1($1), @2), @2);
+ 				}
+ 			| a_expr MEMBER opt_of a_expr
+ 				{
+ 					$$ = (Node *) makeSimpleA_Expr(AEXPR_OP_ANY,
+ 							"=", $1, $4, @2);
+ 				}
+ 			| a_expr NOT MEMBER opt_of a_expr
+ 				{
+ 					$$ = (Node *) makeA_Expr(AEXPR_NOT, NIL, NULL,
+ 							(Node *) makeSimpleA_Expr(AEXPR_OP_ANY, "=",
+ 								$1, $5, @2), @2);
+ 				}
+ 			| a_expr SUBMULTISET opt_of a_expr
+ 				{
+ 					$$ = (Node *) makeFuncCall(
+ 							SystemFuncName("submultiset_of"),
+ 							list_make2($1, $4), @2);
+ 				}
+ 			| a_expr NOT SUBMULTISET opt_of a_expr
+ 				{
+ 					$$ = (Node *) makeA_Expr(AEXPR_NOT, NIL, NULL,
+ 							(Node *) makeFuncCall(
+ 								SystemFuncName("submultiset_of"),
+ 								list_make2($1, $5), @2), @2);
+ 				}
+ 			| a_expr MULTISET UNION opt_all a_expr
+ 				{
+ 					$$ = (Node *) makeFuncCall(
+ 							SystemFuncName("multiset_union"),
+ 							list_make3($1, $5, makeBoolAConst($4, -1)), @2);
+ 				}
+ 			| a_expr MULTISET INTERSECT opt_all a_expr
+ 				{
+ 					$$ = (Node *) makeFuncCall(
+ 							SystemFuncName("multiset_intersect"),
+ 							list_make3($1, $5, makeBoolAConst($4, -1)), @2);
+ 				}
+ 			| a_expr MULTISET EXCEPT opt_all a_expr
+ 				{
+ 					$$ = (Node *) makeFuncCall(
+ 							SystemFuncName("multiset_except"),
+ 							list_make3($1, $5, makeBoolAConst($4, -1)), @2);
+ 				}
+ 		;
+ 
+ opt_of:		OF				{}
+ /*	FIXME: OF is an option in the SQL standard, but I cannot solve
+ 	shift/reduce errors without OF. To solve the errors, we might need
+ 	to make OF, MEMBER, and/or SUBMULTISET to reserved keywords. They
+ 	are reserved keywords in the SQL standard.
+ 			|				{}
+ */
  		;
  
  /*
*************** ColLabel:	IDENT									{ $$ = $1; }
*** 11294,11300 ****
  /* "Unreserved" keywords --- available for use as any kind of name.
   */
  unreserved_keyword:
! 			  ABORT_P
  			| ABSOLUTE_P
  			| ACCESS
  			| ACTION
--- 11359,11366 ----
  /* "Unreserved" keywords --- available for use as any kind of name.
   */
  unreserved_keyword:
! 			  A
! 			| ABORT_P
  			| ABSOLUTE_P
  			| ACCESS
  			| ACTION
*************** unreserved_keyword:
*** 11421,11431 ****
--- 11487,11499 ----
  			| MAPPING
  			| MATCH
  			| MAXVALUE
+ 			| MEMBER
  			| MINUTE_P
  			| MINVALUE
  			| MODE
  			| MONTH_P
  			| MOVE
+ 			| MULTISET
  			| NAME_P
  			| NAMES
  			| NEXT
*************** unreserved_keyword:
*** 11513,11518 ****
--- 11581,11587 ----
  			| STORAGE
  			| STRICT_P
  			| STRIP_P
+ 			| SUBMULTISET
  			| SUPERUSER_P
  			| SYSID
  			| SYSTEM_P
diff --git a/src/backend/utils/adt/array_userfuncs.c b/src/backend/utils/adt/array_userfuncs.c
index 499d357..e99c0f2 100644
*** a/src/backend/utils/adt/array_userfuncs.c
--- b/src/backend/utils/adt/array_userfuncs.c
***************
*** 15,21 ****
--- 15,26 ----
  #include "utils/array.h"
  #include "utils/builtins.h"
  #include "utils/lsyscache.h"
+ #include "utils/typcache.h"
  
+ static Datum array_cat_internal(PG_FUNCTION_ARGS, bool flatten);
+ static ArrayType *array_flatten(ArrayType *array);
+ static void check_concatinatable(Oid element_type1, Oid element_type2);
+ static void check_comparable(Oid element_type1, Oid element_type2);
  
  /*-----------------------------------------------------------------------------
   * array_push :
*************** array_push(PG_FUNCTION_ARGS)
*** 168,173 ****
--- 173,184 ----
  Datum
  array_cat(PG_FUNCTION_ARGS)
  {
+ 	return array_cat_internal(fcinfo, false);
+ }
+ 
+ static Datum
+ array_cat_internal(PG_FUNCTION_ARGS, bool flatten)
+ {
  	ArrayType  *v1,
  			   *v2;
  	ArrayType  *result;
*************** array_cat(PG_FUNCTION_ARGS)
*** 203,213 ****
--- 214,228 ----
  		if (PG_ARGISNULL(1))
  			PG_RETURN_NULL();
  		result = PG_GETARG_ARRAYTYPE_P(1);
+ 		if (flatten)
+ 			result = array_flatten(result);
  		PG_RETURN_ARRAYTYPE_P(result);
  	}
  	if (PG_ARGISNULL(1))
  	{
  		result = PG_GETARG_ARRAYTYPE_P(0);
+ 		if (flatten)
+ 			result = array_flatten(result);
  		PG_RETURN_ARRAYTYPE_P(result);
  	}
  
*************** array_cat(PG_FUNCTION_ARGS)
*** 218,231 ****
  	element_type2 = ARR_ELEMTYPE(v2);
  
  	/* Check we have matching element types */
! 	if (element_type1 != element_type2)
! 		ereport(ERROR,
! 				(errcode(ERRCODE_DATATYPE_MISMATCH),
! 				 errmsg("cannot concatenate incompatible arrays"),
! 				 errdetail("Arrays with element types %s and %s are not "
! 						   "compatible for concatenation.",
! 						   format_type_be(element_type1),
! 						   format_type_be(element_type2))));
  
  	/* OK, use it */
  	element_type = element_type1;
--- 233,239 ----
  	element_type2 = ARR_ELEMTYPE(v2);
  
  	/* Check we have matching element types */
! 	check_concatinatable(element_type1, element_type2);
  
  	/* OK, use it */
  	element_type = element_type1;
*************** array_cat(PG_FUNCTION_ARGS)
*** 249,261 ****
  	 * if both are empty, return the first one
  	 */
  	if (ndims1 == 0 && ndims2 > 0)
  		PG_RETURN_ARRAYTYPE_P(v2);
  
  	if (ndims2 == 0)
  		PG_RETURN_ARRAYTYPE_P(v1);
  
  	/* the rest fall under rule 3, 4, or 5 */
! 	if (ndims1 != ndims2 &&
  		ndims1 != ndims2 - 1 &&
  		ndims1 != ndims2 + 1)
  		ereport(ERROR,
--- 257,278 ----
  	 * if both are empty, return the first one
  	 */
  	if (ndims1 == 0 && ndims2 > 0)
+ 	{
+ 		if (flatten)
+ 			v2 = array_flatten(v2);
  		PG_RETURN_ARRAYTYPE_P(v2);
+ 	}
  
  	if (ndims2 == 0)
+ 	{
+ 		if (flatten)
+ 			v1 = array_flatten(v1);
  		PG_RETURN_ARRAYTYPE_P(v1);
+ 	}
  
  	/* the rest fall under rule 3, 4, or 5 */
! 	if (!flatten &&
! 		ndims1 != ndims2 &&
  		ndims1 != ndims2 - 1 &&
  		ndims1 != ndims2 + 1)
  		ereport(ERROR,
*************** array_cat(PG_FUNCTION_ARGS)
*** 279,285 ****
  	ndatabytes1 = ARR_SIZE(v1) - ARR_DATA_OFFSET(v1);
  	ndatabytes2 = ARR_SIZE(v2) - ARR_DATA_OFFSET(v2);
  
! 	if (ndims1 == ndims2)
  	{
  		/*
  		 * resulting array is made up of the elements (possibly arrays
--- 296,310 ----
  	ndatabytes1 = ARR_SIZE(v1) - ARR_DATA_OFFSET(v1);
  	ndatabytes2 = ARR_SIZE(v2) - ARR_DATA_OFFSET(v2);
  
! 	if (flatten)
! 	{
! 		ndims = 1;
! 		dims = (int *) palloc(sizeof(int));
! 		lbs = (int *) palloc(sizeof(int));
! 		dims[0] = nitems1 + nitems2;
! 		lbs[0] = 1;
! 	}
! 	else if (ndims1 == ndims2)
  	{
  		/*
  		 * resulting array is made up of the elements (possibly arrays
*************** array_agg_finalfn(PG_FUNCTION_ARGS)
*** 544,546 ****
--- 569,1413 ----
  
  	PG_RETURN_DATUM(result);
  }
+ 
+ /*
+  * array_cardinality :
+  *		Return the number of elements in an array.
+  */
+ Datum
+ array_cardinality(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *v = PG_GETARG_ARRAYTYPE_P(0);
+ 	int			nitems;
+ 
+ 	nitems = ArrayGetNItems(ARR_NDIM(v), ARR_DIMS(v));
+ 
+ 	PG_RETURN_INT32(nitems);
+ }
+ 
+ /*
+  * trim_array :
+  *		Remove elements at end of an array. Multi-dimensional array is
+  *		flattened into one-dimensional array.
+  */
+ Datum
+ trim_array(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *array;
+ 	ArrayType  *result;
+ 	ArrayType  *v;
+ 	int32		ntrimmed = PG_GETARG_INT32(1);
+ 	int			nitems;
+ 	int			arrtyplen;
+ 	Oid			elmtype;
+ 	int16		elmlen;
+ 	bool		elmbyval;
+ 	char		elmalign;
+ 	int			lower;
+ 	int			upper;
+ 
+ 	if (ntrimmed < 0)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_ARRAY_ELEMENT_ERROR),
+ 				 errmsg("number of trimmed elements (%d) must not be negative", ntrimmed)));
+ 
+ 	array = PG_GETARG_ARRAYTYPE_P(0);
+ 
+ 	nitems = ArrayGetNItems(ARR_NDIM(array), ARR_DIMS(array));
+ 	if (ntrimmed > nitems)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_ARRAY_ELEMENT_ERROR),
+ 				 errmsg("number of trimmed elements (%d) is greater than cardinality of collection (%d)", ntrimmed, nitems)));
+ 
+ 	v = array_flatten(array);
+ 	arrtyplen = get_typlen(get_fn_expr_argtype(fcinfo->flinfo, 0));
+ 	lower = ARR_LBOUND(v)[0];
+ 	upper = ARR_DIMS(v)[0] - ntrimmed;
+ 	elmtype = ARR_ELEMTYPE(v);
+ 	get_typlenbyvalalign(elmtype, &elmlen, &elmbyval, &elmalign);
+ 	result = array_get_slice(
+ 		v, 1, &upper, &lower, arrtyplen, elmlen, elmbyval, elmalign);
+ 
+ 	PG_FREE_IF_COPY(array, 0);
+ 	PG_RETURN_ARRAYTYPE_P(result);
+ }
+ 
+ /*
+  * Find TypeCacheEntry with comparison functions for element_type.
+  * We arrange to look up the compare functions only once per series of
+  * calls, assuming the element type doesn't change underneath us.
+  */
+ static TypeCacheEntry *
+ get_type_cache(Oid element_type, void **fn_extra)
+ {
+ 	TypeCacheEntry *type;
+ 
+ 	type = (TypeCacheEntry *) *fn_extra;
+ 	if (type == NULL ||
+ 		type->type_id != element_type)
+ 	{
+ 		type = lookup_type_cache(element_type,
+ 					TYPECACHE_EQ_OPR_FINFO | TYPECACHE_CMP_PROC_FINFO);
+ 		if (!OidIsValid(type->eq_opr_finfo.fn_oid) ||
+ 			!OidIsValid(type->cmp_proc_finfo.fn_oid))
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_UNDEFINED_FUNCTION),
+ 			   errmsg("could not identify comparison functions for type %s",
+ 					  format_type_be(element_type))));
+ 		*fn_extra = type;
+ 	}
+ 
+ 	return type;
+ }
+ 
+ static int
+ compare_elements(const void *a, const void *b, void *arg)
+ {
+ 	return DatumGetInt32(FunctionCall2(
+ 		(FmgrInfo *) arg, *(const Datum *) a, *(const Datum *) b));
+ }
+ 
+ /*
+  * Sort values and move nulls to the end.
+  * Returns number of non-null elements as an option.
+  */
+ static void
+ sort_elements(TypeCacheEntry *type, Datum *values, bool *nulls,
+ 			  int nitems, int *nonnulls)
+ {
+ 	int			n,
+ 				i;
+ 
+ 	if (nulls == NULL)
+ 		n = nitems;
+ 	else
+ 	{
+ 		/* move nulls to end of the array */
+ 		for (i = n = 0; i < nitems; i++)
+ 		{
+ 			if (!nulls[i])
+ 			{
+ 				values[n] = values[i];
+ 				nulls[n] = false;
+ 				n++;
+ 			}
+ 		}
+ 		for (i = n; i < nitems; i++)
+ 			nulls[i] = true;
+ 	}
+ 
+ 	/* sort non-null values */
+ 	qsort_arg(values, n, sizeof(Datum),
+ 			  compare_elements, &type->cmp_proc_finfo);
+ 
+ 	if (nonnulls)
+ 		*nonnulls = n;
+ }
+ 
+ /*
+  * Remove duplicated values in already sorted elements. The values, nulls,
+  * nitems, and nonnulls parameters are modified directly. Note that only
+  * one null value will be kept in the result when there are some nulls.
+  */
+ static void
+ unique_elements(Datum *values, bool *nulls, int *nitems, int *nonnulls,
+ 				TypeCacheEntry *type)
+ {
+ 	int		i,
+ 			n,
+ 			nvalues = *nonnulls;
+ 	bool	has_nulls = (*nonnulls < *nitems);
+ 
+ 	for (i = n = 1; i < nvalues; i++)
+ 	{
+ 		if (!DatumGetBool(FunctionCall2(
+ 			&type->eq_opr_finfo, values[i - 1], values[i])))
+ 		{
+ 			Assert(!nulls[n]);
+ 			values[n++] = values[i];
+ 		}
+ 	}
+ 	*nonnulls = n;
+ 	if (has_nulls)
+ 		nulls[n++] = true;
+ 	*nitems = n;
+ }
+ 
+ /*
+  * Deconstruct an array to a list of elements and sort them. Returns values,
+  * null, number of all elements and non-null elements as output parameters.
+  * nulls and nonnulls can be NULLs.
+  */
+ static void
+ deconstruct_and_sort(ArrayType *array, Datum **values, bool **nulls,
+ 					 int *nitems, int *nonnulls, TypeCacheEntry *type)
+ {
+ 	Oid			element_type = ARR_ELEMTYPE(array);
+ 	bool	   *tmp_nulls;
+ 
+ 	AssertArg(values != NULL);
+ 	AssertArg(nitems != NULL);
+ 
+ 	deconstruct_array(array,
+ 					  element_type,
+ 					  type->typlen,
+ 					  type->typbyval,
+ 					  type->typalign,
+ 					  values, &tmp_nulls, nitems);
+ 	sort_elements(type, *values, tmp_nulls, *nitems, nonnulls);
+ 
+ 	if (nulls)
+ 		*nulls = tmp_nulls;
+ }
+ 
+ /*
+  * A worker for array_sort, array_to_set, and higher-level functions.
+  */
+ static ArrayType *
+ sort_or_unique(ArrayType *array, bool unique, void **fn_extra)
+ {
+ 	TypeCacheEntry *type;
+ 	Datum		   *values;
+ 	bool		   *nulls;
+ 	int				nitems,
+ 					nonnulls;
+ 	int				lbs = 1;
+ 	Oid				element_type = ARR_ELEMTYPE(array);
+ 
+ 	type = get_type_cache(element_type, fn_extra);
+ 	deconstruct_and_sort(array, &values, &nulls, &nitems, &nonnulls, type);
+ 	if (unique)
+ 		unique_elements(values, nulls, &nitems, &nonnulls, type);
+ 
+ 	return construct_md_array(values, nulls, 1, &nitems, &lbs, element_type,
+ 				type->typlen, type->typbyval, type->typalign);
+ }
+ 
+ /*
+  * array_sort :
+  *		Sort an array in ascending order. Nulls are in the last.
+  */
+ Datum
+ array_sort(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *array = PG_GETARG_ARRAYTYPE_P(0);
+ 	ArrayType  *result;
+ 
+ 	result = sort_or_unique(array, false, &fcinfo->flinfo->fn_extra);
+ 	PG_FREE_IF_COPY(array, 0);
+ 	PG_RETURN_ARRAYTYPE_P(result);
+ }
+ 
+ /*
+  * array_to_set :
+  *		Remove duplicated elements in an array.
+  */
+ Datum
+ array_to_set(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *array = PG_GETARG_ARRAYTYPE_P(0);
+ 	ArrayType  *result;
+ 
+ 	result = sort_or_unique(array, true, &fcinfo->flinfo->fn_extra);
+ 	PG_FREE_IF_COPY(array, 0);
+ 	PG_RETURN_ARRAYTYPE_P(result);
+ }
+ 
+ /*
+  * array_is_set :
+  *		Return true iff an array has not duplicated values. Note that
+  *		only one null is allowed in a set.
+  */
+ Datum
+ array_is_set(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *array = PG_GETARG_ARRAYTYPE_P(0);
+ 	bool		result;
+ 	Datum	   *values;
+ 	int			nitems,
+ 				nonnulls;
+ 	int			i;
+ 	TypeCacheEntry *type;
+ 	
+ 	type = get_type_cache(ARR_ELEMTYPE(array), &fcinfo->flinfo->fn_extra);
+ 	deconstruct_and_sort(array, &values, NULL, &nitems, &nonnulls, type);
+ 	if (nitems > nonnulls + 1)
+ 	{
+ 		/* only one null is allowd */
+ 		result = false;
+ 	}
+ 	else
+ 	{
+ 		result = true;
+ 		/* compare for each adjacent */
+ 		for (i = 1; i < nonnulls; i++)
+ 		{
+ 			if (DatumGetBool(FunctionCall2(
+ 				&type->eq_opr_finfo, values[i - 1], values[i])))
+ 			{
+ 				result = false;
+ 				break;
+ 			}
+ 		}
+ 	}
+ 
+ 	PG_FREE_IF_COPY(array, 0);
+ 	PG_RETURN_BOOL(result);
+ }
+ 
+ /*
+  * submultiset_of : SUBMULTISET OF
+  *		Return true iff v1 is a subset of v2,
+  */
+ Datum
+ submultiset_of(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *v1 = NULL;
+ 	ArrayType  *v2 = NULL;
+ 	bool		result = false;
+ 	bool		result_null = false;
+ 	Oid			element_type;
+ 	Datum	   *values1,
+ 			   *values2;
+ 	int			nitems1,
+ 				nitems2,
+ 				nonnulls1,
+ 				nonnulls2,
+ 				n1,
+ 				n2,
+ 				unmatch;
+ 	TypeCacheEntry *type;
+ 
+ 	/* always null when v1 is null */
+ 	if (PG_ARGISNULL(0))
+ 	{
+ 		result_null = true;
+ 		goto ok;
+ 	}
+ 
+ 	v1 = PG_GETARG_ARRAYTYPE_P(0);
+ 	nitems1 = ArrayGetNItems(ARR_NDIM(v1), ARR_DIMS(v1));
+ 
+ 	/* null when v2 is null, but false when v1 is empty */
+ 	if (PG_ARGISNULL(1))
+ 	{
+ 		if (nitems1 == 0)
+ 			result = true;
+ 		else
+ 			result_null = true;
+ 		goto ok;
+ 	}
+ 
+ 	v2 = PG_GETARG_ARRAYTYPE_P(1);
+ 	element_type = ARR_ELEMTYPE(v1);
+ 	check_comparable(element_type, ARR_ELEMTYPE(v2));
+ 
+ 	/* true when v1 is empty whether v2 is null or not */
+ 	if (nitems1 == 0)
+ 	{
+ 		result = true;
+ 		goto ok;
+ 	}
+ 
+ 	/* false when v1 has more elements than v2 */
+ 	nitems2 = ArrayGetNItems(ARR_NDIM(v2), ARR_DIMS(v2));
+ 	if (nitems1 > nitems2)
+ 	{
+ 		result = false;
+ 		goto ok;
+ 	}
+ 
+ 	/* compare non-null elements */
+ 	type = get_type_cache(element_type, &fcinfo->flinfo->fn_extra);
+ 	deconstruct_and_sort(v1, &values1, NULL, &nitems1, &nonnulls1, type);
+ 	deconstruct_and_sort(v2, &values2, NULL, &nitems2, &nonnulls2, type);
+ 
+ 	unmatch = 0;
+ 	for (n1 = n2 = 0;
+ 		 n1 < nonnulls1 && unmatch <= nitems2 - nonnulls2 && n2 < nonnulls2;)
+ 	{
+ 		int		r = DatumGetInt32(FunctionCall2(&type->cmp_proc_finfo,
+ 												values1[n1], values2[n2]));
+ 		if (r < 0)
+ 			unmatch++;
+ 		if (r <= 0)
+ 			n1++;
+ 		if (r >= 0)
+ 			n2++;
+ 	}
+ 
+ 	unmatch += nitems1 - n1;
+ 	if (unmatch == 0)
+ 		result = true;
+ 	else if (unmatch > nitems2 - nonnulls2)
+ 		result = false;
+ 	else
+ 		result_null = true;	/* v2 has equal or more nulls than unmatches */
+ 
+ ok:
+ 	if (v1 != NULL)
+ 		PG_FREE_IF_COPY(v1, 0);
+ 	if (v2 != NULL)
+ 		PG_FREE_IF_COPY(v2, 1);
+ 
+ 	if (result_null)
+ 		PG_RETURN_NULL();
+ 	else
+ 		PG_RETURN_BOOL(result);
+ }
+ 
+ /*
+  * multiset_union : MULTISET UNION [ DISTINCT | ALL ]
+  *		Concatinate two arrays, and optionally remove duplicated values.
+  */
+ Datum
+ multiset_union(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *v1,
+ 			   *v2;
+ 	bool		all = PG_GETARG_BOOL(2);
+ 	ArrayType  *result;
+ 	Datum	   *values,
+ 			   *values1,
+ 			   *values2;
+ 	bool	   *nulls,
+ 			   *nulls1,
+ 			   *nulls2;
+ 	int			nitems,
+ 				nitems1,
+ 				nitems2,
+ 				nonnulls;
+ 	Oid			element_type1,
+ 				element_type2;
+ 	int			lbs = 1;
+ 	TypeCacheEntry *type;
+ 
+ 	if (PG_ARGISNULL(0) && PG_ARGISNULL(1))
+ 		PG_RETURN_NULL();
+ 
+ 	/* fast path for UNION ALL */
+ 	if (all)
+ 		return array_cat_internal(fcinfo, true);
+ 
+ 	/* Concatenating a null array is a no-op, just return the other input */
+ 	if (PG_ARGISNULL(0))
+ 	{
+ 		v2 = PG_GETARG_ARRAYTYPE_P(1);
+ 		result = sort_or_unique(v2, true, &fcinfo->flinfo->fn_extra);
+ 		PG_FREE_IF_COPY(v2, 1);
+ 		PG_RETURN_ARRAYTYPE_P(result);
+ 	}
+ 	if (PG_ARGISNULL(1))
+ 	{
+ 		v1 = PG_GETARG_ARRAYTYPE_P(0);
+ 		result = sort_or_unique(v1, true, &fcinfo->flinfo->fn_extra);
+ 		PG_FREE_IF_COPY(v1, 0);
+ 		PG_RETURN_ARRAYTYPE_P(result);
+ 	}
+ 
+ 	v1 = PG_GETARG_ARRAYTYPE_P(0);
+ 	v2 = PG_GETARG_ARRAYTYPE_P(1);
+ 	element_type1 = ARR_ELEMTYPE(v1);
+ 	element_type2 = ARR_ELEMTYPE(v2);
+ 
+ 	check_concatinatable(element_type1, element_type2);
+ 	type = get_type_cache(element_type1, &fcinfo->flinfo->fn_extra);
+ 	deconstruct_array(v1,
+ 					  element_type1,
+ 					  type->typlen,
+ 					  type->typbyval,
+ 					  type->typalign,
+ 					  &values1, &nulls1, &nitems1);
+ 	deconstruct_array(v2,
+ 					  element_type2,
+ 					  type->typlen,
+ 					  type->typbyval,
+ 					  type->typalign,
+ 					  &values2, &nulls2, &nitems2);
+ 
+ 	nitems = nitems1 + nitems2;
+ 	values = (Datum *) palloc(sizeof(Datum) * nitems);
+ 	nulls = (bool *) palloc(sizeof(bool) * nitems);
+ 
+ 	memcpy(values, values1, sizeof(Datum) * nitems1);
+ 	memcpy(values + nitems1, values2, sizeof(Datum) * nitems2);
+ 	memcpy(nulls, nulls1, sizeof(bool) * nitems1);
+ 	memcpy(nulls + nitems1, nulls2, sizeof(bool) * nitems2);
+ 
+ 	sort_elements(type, values, nulls, nitems, &nonnulls);
+ 	unique_elements(values, nulls, &nitems, &nonnulls, type);
+ 	result = construct_md_array(values, nulls, 1, &nitems, &lbs,
+ 								element_type1,
+ 								type->typlen,
+ 								type->typbyval,
+ 								type->typalign);
+ 
+ 	PG_FREE_IF_COPY(v1, 0);
+ 	PG_FREE_IF_COPY(v2, 1);
+ 	PG_RETURN_ARRAYTYPE_P(result);
+ }
+ 
+ /*
+  * Intersection of two sorted arrays. The first array is modified directly.
+  * Return length of the result.
+  */
+ static int
+ intersect_sorted_arrays(TypeCacheEntry *type,
+ 						Datum *values1, bool *nulls1, int nitems1,
+ 						const Datum *values2, const bool *nulls2, int nitems2)
+ {
+ 	int			n1,
+ 				n2,
+ 				n;
+ 
+ 	/* add non-nulls */
+ 	for (n = n1 = n2 = 0;
+ 		 n1 < nitems1 && !nulls1[n1] &&
+ 		 n2 < nitems2 && !nulls2[n2];)
+ 	{
+ 		int		r = DatumGetInt32(FunctionCall2(&type->cmp_proc_finfo,
+ 												values1[n1], values2[n2]));
+ 		if (r == 0)
+ 			values1[n++] = values1[n1];
+ 		if (r <= 0)
+ 			n1++;
+ 		if (r >= 0)
+ 			n2++;
+ 	}
+ 
+ 	/* skip non-nulls */
+ 	for (; n1 < nitems1 && !nulls1[n1]; n1++) {}
+ 	for (; n2 < nitems2 && !nulls2[n2]; n2++) {}
+ 
+ 	/* add nulls */
+ 	for (; n1 < nitems1 && n2 < nitems2; n1++, n2++, n++)
+ 		nulls1[n] = true;
+ 
+ 	return n;
+ }
+ 
+ /*
+  * multiset_intersect : MULTISET INTERSECT [ DISTINCT | ALL ]
+  *		Intersection of two arrays, and optionally remove duplicated values.
+  */
+ Datum
+ multiset_intersect(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *v1 = PG_GETARG_ARRAYTYPE_P(0);
+ 	ArrayType  *v2 = PG_GETARG_ARRAYTYPE_P(1);
+ 	bool		all = PG_GETARG_BOOL(2);
+ 
+ 	ArrayType  *result;
+ 	Oid			element_type = ARR_ELEMTYPE(v1);
+ 	Datum	   *values1,
+ 			   *values2;
+ 	bool	   *nulls1,
+ 			   *nulls2;
+ 	int			nitems1,
+ 				nitems2,
+ 				nonnulls1,
+ 				nonnulls2;
+ 	int			lbs = 1;
+ 	TypeCacheEntry *type;
+ 
+ 	check_comparable(element_type, ARR_ELEMTYPE(v2));
+ 	type = get_type_cache(element_type, &fcinfo->flinfo->fn_extra);
+ 
+ 	deconstruct_and_sort(v1, &values1, &nulls1, &nitems1, &nonnulls1, type);
+ 	if (!all)
+ 		unique_elements(values1, nulls1, &nitems1, &nonnulls1, type);
+ 	deconstruct_and_sort(v2, &values2, &nulls2, &nitems2, &nonnulls2, type);
+ 	if (!all)
+ 		unique_elements(values2, nulls2, &nitems2, &nonnulls2, type);
+ 
+ 	nitems1 = intersect_sorted_arrays(type,
+ 									  values1, nulls1, nitems1,
+ 									  values2, nulls2, nitems2);
+ 	result = construct_md_array(values1, nulls1, 1, &nitems1, &lbs,
+ 								element_type, type->typlen,
+ 								type->typbyval, type->typalign);
+ 
+ 	PG_FREE_IF_COPY(v1, 0);
+ 	PG_FREE_IF_COPY(v2, 1);
+ 	PG_RETURN_ARRAYTYPE_P(result);
+ }
+ 
+ /*
+  * multiset_except : MULTISET EXCEPT [ DISTINCT | ALL ]
+  *		Subtraction of two arrays, and optionally remove duplicated values.
+  */
+ Datum
+ multiset_except(PG_FUNCTION_ARGS)
+ {
+ 	ArrayType  *v1;
+ 	ArrayType  *v2;
+ 	bool		all;
+ 	ArrayType  *result;
+ 	Oid			element_type;
+ 	Datum	   *values1,
+ 			   *values2;
+ 	bool	   *nulls1,
+ 			   *nulls2;
+ 	int			nitems1,
+ 				nitems2,
+ 				nonnulls1,
+ 				nonnulls2;
+ 	int			n1,
+ 				n2,
+ 				n;
+ 	int			lbs = 1;
+ 	TypeCacheEntry *type;
+ 
+ 	if (PG_ARGISNULL(0))
+ 		PG_RETURN_NULL();
+ 
+ 	v1 = PG_GETARG_ARRAYTYPE_P(0);
+ 	all = PG_GETARG_BOOL(2);
+ 
+ 	/* fast path for except null */
+ 	if (PG_ARGISNULL(1))
+ 	{
+ 		if (all)
+ 			PG_RETURN_ARRAYTYPE_P(array_flatten(v1));
+ 
+ 		result = sort_or_unique(v1, false, &fcinfo->flinfo->fn_extra);
+ 		PG_FREE_IF_COPY(v1, 0);
+ 		PG_RETURN_ARRAYTYPE_P(result);
+ 	}
+ 
+ 	v2 = PG_GETARG_ARRAYTYPE_P(1);
+ 	element_type = ARR_ELEMTYPE(v1);
+ 	check_concatinatable(element_type, ARR_ELEMTYPE(v2));
+ 	type = get_type_cache(element_type, &fcinfo->flinfo->fn_extra);
+ 
+ 	deconstruct_and_sort(v1, &values1, &nulls1, &nitems1, &nonnulls1, type);
+ 	if (!all)
+ 		unique_elements(values1, nulls1, &nitems1, &nonnulls1, type);
+ 	deconstruct_and_sort(v2, &values2, &nulls2, &nitems2, &nonnulls2, type);
+ 	if (!all)
+ 		unique_elements(values2, nulls2, &nitems2, &nonnulls2, type);
+ 
+ 	/* add non-nulls */
+ 	for (n = n1 = n2 = 0; n1 < nonnulls1 && n2 < nonnulls2;)
+ 	{
+ 		int		r = DatumGetInt32(FunctionCall2(&type->cmp_proc_finfo,
+ 												values1[n1], values2[n2]));
+ 		if (r < 0)
+ 			values1[n++] = values1[n1++];
+ 		else
+ 			n2++;
+ 		if (r == 0)
+ 			n1++;
+ 	}
+ 	for (; n1 < nonnulls1; n1++, n++)
+ 		values1[n] = values1[n1];
+ 
+ 	/* add nulls */
+ 	for (n1 = nonnulls1; n1 < nitems1 - (nitems2 - nonnulls2); n1++, n++)
+ 		nulls1[n] = true;
+ 
+ 	result = construct_md_array(values1, nulls1, 1, &n, &lbs, element_type,
+ 								type->typlen, type->typbyval, type->typalign);
+ 
+ 	PG_FREE_IF_COPY(v1, 0);
+ 	PG_FREE_IF_COPY(v2, 1);
+ 	PG_RETURN_ARRAYTYPE_P(result);
+ }
+ 
+ /*
+  * fusion aggregate function :
+  *		Similar to array_agg, but the input values are arrays.
+  */
+ Datum
+ fusion_transfn(PG_FUNCTION_ARGS)
+ {
+ 	MemoryContext		aggcontext;
+ 	ArrayBuildState	   *state;
+ 
+ 	if (!AggCheckCallContext(fcinfo, &aggcontext))
+ 	{
+ 		/* cannot be called directly because of internal-type argument */
+ 		elog(ERROR, "fusion_transfn called in non-aggregate context");
+ 	}
+ 
+ 	state = PG_ARGISNULL(0) ? NULL : (ArrayBuildState *) PG_GETARG_POINTER(0);
+ 	if (!PG_ARGISNULL(1))
+ 	{
+ 		ArrayType  *v = PG_GETARG_ARRAYTYPE_P(1);
+ 		Oid			elmtype;
+ 		int16		elmlen;
+ 		bool		elmbyval;
+ 		char		elmalign;
+ 		Datum	   *elems;
+ 		bool	   *nulls;
+ 		int			nitems;
+ 		int			i;
+ 
+ 		elmtype = ARR_ELEMTYPE(v);
+ 		get_typlenbyvalalign(elmtype, &elmlen, &elmbyval, &elmalign);
+ 		deconstruct_array(v, elmtype, elmlen, elmbyval, elmalign,
+ 						  &elems, &nulls, &nitems);
+ 		for (i = 0; i < nitems; i++)
+ 			state = accumArrayResult(state, elems[i], nulls[i],
+ 									 elmtype, aggcontext);
+ 
+ 		PG_FREE_IF_COPY(v, 1);
+ 	}
+ 
+ 	PG_RETURN_POINTER(state);
+ }
+ 
+ /*
+  * intersection aggregate function :
+  *		Intersection of all input arrays.
+  */
+ typedef struct IntersectState
+ {
+ 	Oid		element_type;
+ 	int		nitems;
+ 	Datum  *values;
+ 	bool   *nulls;
+ } IntersectState;
+ 
+ Datum
+ intersection_transfn(PG_FUNCTION_ARGS)
+ {
+ 	MemoryContext		aggcontext;
+ 	IntersectState	   *state;
+ 
+ 	if (!AggCheckCallContext(fcinfo, &aggcontext))
+ 	{
+ 		/* cannot be called directly because of internal-type argument */
+ 		elog(ERROR, "intersection_transfn called in non-aggregate context");
+ 	}
+ 
+ 	state = PG_ARGISNULL(0) ? NULL : (IntersectState *) PG_GETARG_POINTER(0);
+ 	if (!PG_ARGISNULL(1))
+ 	{
+ 		ArrayType	   *v = PG_GETARG_ARRAYTYPE_P(1);
+ 		TypeCacheEntry *type;
+ 
+ 		type = get_type_cache(ARR_ELEMTYPE(v), &fcinfo->flinfo->fn_extra);
+ 
+ 		if (state == NULL)
+ 		{
+ 			MemoryContext	oldcontext;
+ 
+ 			oldcontext = MemoryContextSwitchTo(aggcontext);
+ 			state = (IntersectState *) palloc(sizeof(IntersectState));
+ 			state->element_type = ARR_ELEMTYPE(v);
+ 			deconstruct_and_sort(v, &state->values, &state->nulls,
+ 								 &state->nitems, NULL, type);
+ 			MemoryContextSwitchTo(oldcontext);
+ 		}
+ 		else
+ 		{
+ 			Datum	   *values;
+ 			bool	   *nulls;
+ 			int			nitems;
+ 
+ 			check_concatinatable(state->element_type, ARR_ELEMTYPE(v));
+ 			deconstruct_and_sort(v, &values, &nulls, &nitems, NULL, type);
+ 			state->nitems = intersect_sorted_arrays(type,
+ 								state->values, state->nulls, state->nitems,
+ 								values, nulls, nitems);
+ 		}
+ 
+ 		PG_FREE_IF_COPY(v, 1);
+ 	}
+ 
+ 	PG_RETURN_POINTER(state);
+ }
+ 
+ Datum
+ intersection_finalfn(PG_FUNCTION_ARGS)
+ {
+ 	IntersectState	   *state;
+ 	ArrayType		   *result;
+ 	int					lbs = 1;
+ 	TypeCacheEntry	   *type;
+ 
+ 	state = PG_ARGISNULL(0) ? NULL : (IntersectState *) PG_GETARG_POINTER(0);
+ 	if (state == NULL)
+ 		PG_RETURN_NULL();
+ 
+ 	type = get_type_cache(state->element_type, &fcinfo->flinfo->fn_extra);
+ 	result = construct_md_array(state->values, state->nulls,
+ 								1, &state->nitems, &lbs, state->element_type,
+ 								type->typlen, type->typbyval, type->typalign);
+ 
+ 	PG_RETURN_ARRAYTYPE_P(result);
+ }
+ 
+ /*
+  * Flatten multi-dimensional array into one-dimensional array.
+  * The lower bounds is adjusted to 1.
+  */
+ static ArrayType *
+ array_flatten(ArrayType *array)
+ {
+ 	ArrayType  *result;
+ 	int			ndims = ARR_NDIM(array);
+ 	int32		dataoffset;
+ 	int			ndatabytes,
+ 				nbytes;
+ 	int			nitems;
+ 
+ 	if (ndims < 1 || (ndims == 1 && ARR_LBOUND(array)[0] == 1))
+ 		return array;
+ 
+ 	nitems = ArrayGetNItems(ndims, ARR_DIMS(array));
+ 	ndatabytes = ARR_SIZE(array) - ARR_DATA_OFFSET(array);
+ 	if (ARR_HASNULL(array))
+ 	{
+ 		dataoffset = ARR_OVERHEAD_WITHNULLS(1, nitems);
+ 		nbytes = ndatabytes + dataoffset;
+ 	}
+ 	else
+ 	{
+ 		dataoffset = 0;			/* marker for no null bitmap */
+ 		nbytes = ndatabytes + ARR_OVERHEAD_NONULLS(1);
+ 	}
+ 
+ 	result = (ArrayType *) palloc(nbytes);
+ 	SET_VARSIZE(result, nbytes);
+ 	result->ndim = 1;
+ 	result->dataoffset = dataoffset;
+ 	result->elemtype = ARR_ELEMTYPE(array);
+ 	ARR_DIMS(result)[0] = nitems;
+ 	ARR_LBOUND(result)[0] = 1;
+ 	/* data area is arg1 then arg2 */
+ 	memcpy(ARR_DATA_PTR(result), ARR_DATA_PTR(array), ndatabytes);
+ 	/* handle the null bitmap if needed */
+ 	if (ARR_HASNULL(result))
+ 		array_bitmap_copy(ARR_NULLBITMAP(result), 0,
+ 						  ARR_NULLBITMAP(array), 0, nitems);
+ 
+ 	return result;
+ }
+ 
+ static void
+ check_concatinatable(Oid element_type1, Oid element_type2)
+ {
+ 	if (element_type1 != element_type2)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 				 errmsg("cannot concatenate incompatible arrays"),
+ 				 errdetail("Arrays with element types %s and %s are not "
+ 						   "compatible for concatenation.",
+ 						   format_type_be(element_type1),
+ 						   format_type_be(element_type2))));
+ }
+ 
+ static void
+ check_comparable(Oid element_type1, Oid element_type2)
+ {
+ 	if (element_type1 != element_type2)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 				 errmsg("cannot compare incompatible arrays"),
+ 				 errdetail("Arrays with element types %s and %s are not "
+ 						   "compatible for comparison.",
+ 						   format_type_be(element_type1),
+ 						   format_type_be(element_type2))));
+ }
diff --git a/src/include/catalog/pg_aggregate.h b/src/include/catalog/pg_aggregate.h
index 26966d2..2adedf3 100644
*** a/src/include/catalog/pg_aggregate.h
--- b/src/include/catalog/pg_aggregate.h
*************** DATA(insert ( 2901 xmlconcat2	  -					0	
*** 222,227 ****
--- 222,230 ----
  
  /* array */
  DATA(insert ( 2335	array_agg_transfn	array_agg_finalfn		0	2281	_null_ ));
+ DATA(insert ( 3088	array_agg_transfn	array_agg_finalfn		0	2281	_null_ ));
+ DATA(insert ( 3090	fusion_transfn		array_agg_finalfn		0	2281	_null_ ));
+ DATA(insert ( 3093	intersection_transfn intersection_finalfn	0	2281	_null_ ));
  
  /* text */
  DATA(insert ( 3538	string_agg_transfn	string_agg_finalfn		0	2281	_null_ ));
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index f8b5d4d..e405e22 100644
*** a/src/include/catalog/pg_proc.h
--- b/src/include/catalog/pg_proc.h
*************** DATA(insert OID = 2334 (  array_agg_fina
*** 1062,1067 ****
--- 1062,1097 ----
  DESCR("array_agg final function");
  DATA(insert OID = 2335 (  array_agg		   PGNSP PGUID 12 1 0 0 t f f f f i 1 0 2277 "2283" _null_ _null_ _null_ _null_ aggregate_dummy _null_ _null_ _null_ ));
  DESCR("concatenate aggregate input into an array");
+ DATA(insert OID = 3079 (  cardinality	   PGNSP PGUID 12 1 0 0 f f f t f i 1 0 23 "2277" _null_ _null_ _null_ _null_ array_cardinality _null_ _null_ _null_ ));
+ DESCR("number of elements in array");
+ DATA(insert OID = 3080 (  trim_array	   PGNSP PGUID 12 1 0 0 f f f t f i 2 0 2277 "2277 23" _null_ _null_ _null_ _null_ trim_array _null_ _null_ _null_ ));
+ DESCR("remove elements end of array");
+ DATA(insert OID = 3081 (  array_sort	   PGNSP PGUID 12 1 0 0 f f f t f i 1 0 2277 "2277" _null_ _null_ _null_ _null_ array_sort _null_ _null_ _null_ ));
+ DESCR("sort an array in ascending order");
+ DATA(insert OID = 3082 (  set			   PGNSP PGUID 12 1 0 0 f f f t f i 1 0 2277 "2277" _null_ _null_ _null_ _null_ array_to_set _null_ _null_ _null_ ));
+ DESCR("remove duplicated values in an array");
+ DATA(insert OID = 3083 (  is_a_set		   PGNSP PGUID 12 1 0 0 f f f t f i 1 0 16 "2277" _null_ _null_ _null_ _null_ array_is_set _null_ _null_ _null_ ));
+ DESCR("no duplicated elements?");
+ DATA(insert OID = 3084 (  submultiset_of   PGNSP PGUID 12 1 0 0 f f f f f i 2 0 16 "2277 2277" _null_ _null_ _null_ _null_ submultiset_of _null_ _null_ _null_ ));
+ DESCR("contained as subset?");
+ DATA(insert OID = 3085 (  multiset_union PGNSP PGUID 12 1 0 0 f f f f f i 3 0 2277 "2277 2277 16" _null_ _null_ _null_ _null_ multiset_union _null_ _null_ _null_ ));
+ DESCR("concatenate two arrays");
+ DATA(insert OID = 3086 (  multiset_intersect PGNSP PGUID 12 1 0 0 f f f t f i 3 0 2277 "2277 2277 16" _null_ _null_ _null_ _null_ multiset_intersect _null_ _null_ _null_ ));
+ DESCR("intersection of two arrays");
+ DATA(insert OID = 3087 (  multiset_except PGNSP PGUID 12 1 0 0 f f f f f i 3 0 2277 "2277 2277 16" _null_ _null_ _null_ _null_ multiset_except _null_ _null_ _null_ ));
+ DESCR("exception of two arrays");
+ DATA(insert OID = 3088 (  collect		   PGNSP PGUID 12 1 0 0 t f f f f i 1 0 2277 "2283" _null_ _null_ _null_ _null_ aggregate_dummy _null_ _null_ _null_ ));
+ DESCR("concatenate aggregate elements into an array");
+ DATA(insert OID = 3089 (  fusion_transfn   PGNSP PGUID 12 1 0 0 f f f f f i 2 0 2281 "2281 2277" _null_ _null_ _null_ _null_ fusion_transfn _null_ _null_ _null_ ));
+ DESCR("fusion transition function");
+ DATA(insert OID = 3090 (  fusion		   PGNSP PGUID 12 1 0 0 t f f f f i 1 0 2277 "2277" _null_ _null_ _null_ _null_ aggregate_dummy _null_ _null_ _null_ ));
+ DESCR("concatenate aggregate arrays into an array");
+ DATA(insert OID = 3091 (  intersection_transfn PGNSP PGUID 12 1 0 0 f f f f f i 2 0 2281 "2281 2277" _null_ _null_ _null_ _null_ intersection_transfn _null_ _null_ _null_ ));
+ DESCR("intersection transition function");
+ DATA(insert OID = 3092 (  intersection_finalfn PGNSP PGUID 12 1 0 0 f f f f f i 1 0 2277 "2281" _null_ _null_ _null_ _null_ intersection_finalfn _null_ _null_ _null_ ));
+ DESCR("intersection final function");
+ DATA(insert OID = 3093 (  intersection	   PGNSP PGUID 12 1 0 0 t f f f f i 1 0 2277 "2277" _null_ _null_ _null_ _null_ aggregate_dummy _null_ _null_ _null_ ));
+ DESCR("intersection of all inputs");
  
  DATA(insert OID = 760 (  smgrin			   PGNSP PGUID 12 1 0 0 f f f t f s 1 0 210 "2275" _null_ _null_ _null_ _null_	smgrin _null_ _null_ _null_ ));
  DESCR("I/O");
diff --git a/src/include/nodes/makefuncs.h b/src/include/nodes/makefuncs.h
index 7c41312..b5cd584 100644
*** a/src/include/nodes/makefuncs.h
--- b/src/include/nodes/makefuncs.h
*************** extern TypeName *makeTypeNameFromOid(Oid
*** 71,76 ****
--- 71,77 ----
  
  extern FuncExpr *makeFuncExpr(Oid funcid, Oid rettype,
  			 List *args, CoercionForm fformat);
+ extern FuncCall *makeFuncCall(List *funcname, List *args, int location);
  
  extern DefElem *makeDefElem(char *name, Node *arg);
  extern DefElem *makeDefElemExtended(char *nameSpace, char *name, Node *arg,
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index 578d3cd..5f59d22 100644
*** a/src/include/parser/kwlist.h
--- b/src/include/parser/kwlist.h
***************
*** 26,31 ****
--- 26,32 ----
   */
  
  /* name, value, category */
+ PG_KEYWORD("a", A, UNRESERVED_KEYWORD)
  PG_KEYWORD("abort", ABORT_P, UNRESERVED_KEYWORD)
  PG_KEYWORD("absolute", ABSOLUTE_P, UNRESERVED_KEYWORD)
  PG_KEYWORD("access", ACCESS, UNRESERVED_KEYWORD)
*************** PG_KEYWORD("login", LOGIN_P, UNRESERVED_
*** 232,242 ****
--- 233,245 ----
  PG_KEYWORD("mapping", MAPPING, UNRESERVED_KEYWORD)
  PG_KEYWORD("match", MATCH, UNRESERVED_KEYWORD)
  PG_KEYWORD("maxvalue", MAXVALUE, UNRESERVED_KEYWORD)
+ PG_KEYWORD("member", MEMBER, UNRESERVED_KEYWORD)
  PG_KEYWORD("minute", MINUTE_P, UNRESERVED_KEYWORD)
  PG_KEYWORD("minvalue", MINVALUE, UNRESERVED_KEYWORD)
  PG_KEYWORD("mode", MODE, UNRESERVED_KEYWORD)
  PG_KEYWORD("month", MONTH_P, UNRESERVED_KEYWORD)
  PG_KEYWORD("move", MOVE, UNRESERVED_KEYWORD)
+ PG_KEYWORD("multiset", MULTISET, UNRESERVED_KEYWORD)
  PG_KEYWORD("name", NAME_P, UNRESERVED_KEYWORD)
  PG_KEYWORD("names", NAMES, UNRESERVED_KEYWORD)
  PG_KEYWORD("national", NATIONAL, COL_NAME_KEYWORD)
*************** PG_KEYWORD("stdout", STDOUT, UNRESERVED_
*** 358,363 ****
--- 361,367 ----
  PG_KEYWORD("storage", STORAGE, UNRESERVED_KEYWORD)
  PG_KEYWORD("strict", STRICT_P, UNRESERVED_KEYWORD)
  PG_KEYWORD("strip", STRIP_P, UNRESERVED_KEYWORD)
+ PG_KEYWORD("submultiset", SUBMULTISET, UNRESERVED_KEYWORD)
  PG_KEYWORD("substring", SUBSTRING, COL_NAME_KEYWORD)
  PG_KEYWORD("superuser", SUPERUSER_P, UNRESERVED_KEYWORD)
  PG_KEYWORD("symmetric", SYMMETRIC, RESERVED_KEYWORD)
diff --git a/src/include/utils/array.h b/src/include/utils/array.h
index 7f7e744..cf09606 100644
*** a/src/include/utils/array.h
--- b/src/include/utils/array.h
*************** extern ArrayType *create_singleton_array
*** 281,285 ****
--- 281,297 ----
  
  extern Datum array_agg_transfn(PG_FUNCTION_ARGS);
  extern Datum array_agg_finalfn(PG_FUNCTION_ARGS);
+ extern Datum array_cardinality(PG_FUNCTION_ARGS);
+ extern Datum trim_array(PG_FUNCTION_ARGS);
+ extern Datum array_sort(PG_FUNCTION_ARGS);
+ extern Datum array_to_set(PG_FUNCTION_ARGS);
+ extern Datum array_is_set(PG_FUNCTION_ARGS);
+ extern Datum submultiset_of(PG_FUNCTION_ARGS);
+ extern Datum multiset_union(PG_FUNCTION_ARGS);
+ extern Datum multiset_intersect(PG_FUNCTION_ARGS);
+ extern Datum multiset_except(PG_FUNCTION_ARGS);
+ extern Datum fusion_transfn(PG_FUNCTION_ARGS);
+ extern Datum intersection_transfn(PG_FUNCTION_ARGS);
+ extern Datum intersection_finalfn(PG_FUNCTION_ARGS);
  
  #endif   /* ARRAY_H */
diff --git a/src/test/regress/expected/arrays.out b/src/test/regress/expected/arrays.out
index 7b05ce3..955b6ec 100644
*** a/src/test/regress/expected/arrays.out
--- b/src/test/regress/expected/arrays.out
*************** select * from t1;
*** 1558,1560 ****
--- 1558,1765 ----
   [5:5]={"(42,43)"}
  (1 row)
  
+ -- MULTISET support
+ SELECT cardinality(ARRAY[1, 2, 3]), cardinality(ARRAY[[1, 2], [3, 4]]);
+  cardinality | cardinality 
+ -------------+-------------
+            3 |           4
+ (1 row)
+ 
+ SELECT trim_array(ARRAY[1, 2, 3], 0),
+        trim_array(ARRAY[1, 2, 3], 2),
+ 	   trim_array(ARRAY[[1, 2], [3, 4]], 1);
+  trim_array | trim_array | trim_array 
+ ------------+------------+------------
+  {1,2,3}    | {1}        | {1,2,3}
+ (1 row)
+ 
+ SELECT trim_array(ARRAY[1, 2, 3], -1);
+ ERROR:  number of trimmed elements (-1) must not be negative
+ SELECT trim_array(ARRAY[1, 2, 3], 4);
+ ERROR:  number of trimmed elements (4) is greater than cardinality of collection (3)
+ SELECT 'B' MEMBER OF ARRAY['A', 'B', 'C'];
+  ?column? 
+ ----------
+  t
+ (1 row)
+ 
+ SELECT 3 MEMBER OF ARRAY[1, 2], 3 MEMBER OF ARRAY[[1, 2], [3, 4]];
+  ?column? | ?column? 
+ ----------+----------
+  f        | t
+ (1 row)
+ 
+ SELECT 3 NOT MEMBER OF ARRAY[1, 2];
+  ?column? 
+ ----------
+  t
+ (1 row)
+ 
+ SELECT ARRAY[1, 2] SUBMULTISET OF ARRAY[3, 2, 1];
+  submultiset_of 
+ ----------------
+  t
+ (1 row)
+ 
+ SELECT ARRAY[1, 2] SUBMULTISET OF ARRAY[[1, 3], [2, 4]];
+  submultiset_of 
+ ----------------
+  t
+ (1 row)
+ 
+ SELECT ARRAY[1, 1, 2] SUBMULTISET OF ARRAY[1, 2, 2];
+  submultiset_of 
+ ----------------
+  f
+ (1 row)
+ 
+ SELECT ARRAY['A', 'B', 'C'] SUBMULTISET OF ARRAY['A', 'B', 'D'];
+  submultiset_of 
+ ----------------
+  f
+ (1 row)
+ 
+ SELECT ARRAY['A', 'B', 'C'] NOT SUBMULTISET OF ARRAY['A', 'B', 'D'];
+  ?column? 
+ ----------
+  t
+ (1 row)
+ 
+ SELECT ARRAY[]::int[] SUBMULTISET OF NULL::int[];
+  submultiset_of 
+ ----------------
+  t
+ (1 row)
+ 
+ SELECT NULL::int[] SUBMULTISET OF ARRAY[]::int[];
+  submultiset_of 
+ ----------------
+  
+ (1 row)
+ 
+ SELECT NULL::int[] SUBMULTISET OF NULL::int[];
+  submultiset_of 
+ ----------------
+  
+ (1 row)
+ 
+ SELECT ARRAY[1, 2] SUBMULTISET OF ARRAY[1, NULL],
+        ARRAY[1, 2] SUBMULTISET OF ARRAY[3, NULL];
+  submultiset_of | submultiset_of 
+ ----------------+----------------
+                 | f
+ (1 row)
+ 
+ SELECT ARRAY[1, NULL] SUBMULTISET OF ARRAY[1, NULL];
+  submultiset_of 
+ ----------------
+  
+ (1 row)
+ 
+ SELECT ARRAY[1, 2, 3] SUBMULTISET OF ARRAY[NULL, 1, 2, 3];
+  submultiset_of 
+ ----------------
+  t
+ (1 row)
+ 
+ SELECT ARRAY[1, 2, 3] SUBMULTISET OF ARRAY[1, 2, NULL];
+  submultiset_of 
+ ----------------
+  
+ (1 row)
+ 
+ SELECT ARRAY[1, 2, 3] SUBMULTISET OF ARRAY[NULL, 4, NULL];
+  submultiset_of 
+ ----------------
+  f
+ (1 row)
+ 
+ SELECT ARRAY[1, 2, 3] IS A SET;
+  is_a_set 
+ ----------
+  t
+ (1 row)
+ 
+ SELECT ARRAY['A', 'A', 'B'] IS A SET, ARRAY['A', 'A', 'B'] IS NOT A SET;
+  is_a_set | ?column? 
+ ----------+----------
+  f        | t
+ (1 row)
+ 
+ SELECT ARRAY[1, NULL] IS A SET, ARRAY[1, NULL, NULL] IS NOT A SET;
+  is_a_set | ?column? 
+ ----------+----------
+  t        | t
+ (1 row)
+ 
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET UNION ALL ARRAY[2, NULL];
+       multiset_union      
+ --------------------------
+  {2,NULL,1,2,NULL,2,NULL}
+ (1 row)
+ 
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET UNION ARRAY[2, NULL];
+  multiset_union 
+ ----------------
+  {1,2,NULL}
+ (1 row)
+ 
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET INTERSECT ALL ARRAY[2, NULL];
+  multiset_intersect 
+ --------------------
+  {2,NULL}
+ (1 row)
+ 
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET INTERSECT ARRAY[2, NULL];
+  multiset_intersect 
+ --------------------
+  {2,NULL}
+ (1 row)
+ 
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET EXCEPT ALL ARRAY[2, NULL];
+  multiset_except 
+ -----------------
+  {1,2,NULL}
+ (1 row)
+ 
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET EXCEPT ARRAY[2, NULL];
+  multiset_except 
+ -----------------
+  {1}
+ (1 row)
+ 
+ SELECT collect(s), fusion(a), intersection(a)
+   FROM (VALUES
+   ('A', ARRAY[1, 2, 3, 2, 2]),
+   ('B', ARRAY[1, 2, 4, 2]),
+   ('C', ARRAY[[3, 2], [2, 1]])
+ ) AS t(s, a);
+  collect |           fusion            | intersection 
+ ---------+-----------------------------+--------------
+  {A,B,C} | {1,2,3,2,2,1,2,4,2,3,2,2,1} | {1,2,2}
+ (1 row)
+ 
+ SELECT array_sort(ARRAY[1, 3, 2, 3, NULL, 1, NULL]);
+       array_sort       
+ -----------------------
+  {1,1,2,3,3,NULL,NULL}
+ (1 row)
+ 
+ SELECT array_sort(ARRAY[['A', 'D'], ['B', 'E'], ['C', 'F']]);
+   array_sort   
+ ---------------
+  {A,B,C,D,E,F}
+ (1 row)
+ 
+ SELECT set(ARRAY[1, 3, 2, 3, NULL, 1, NULL]);
+      set      
+ --------------
+  {1,2,3,NULL}
+ (1 row)
+ 
+ SELECT set(ARRAY[['A', 'C'], ['B', 'C'], ['C', 'A']]);
+    set   
+ ---------
+  {A,B,C}
+ (1 row)
+ 
diff --git a/src/test/regress/sql/arrays.sql b/src/test/regress/sql/arrays.sql
index 9ea53b1..3c38900 100644
*** a/src/test/regress/sql/arrays.sql
--- b/src/test/regress/sql/arrays.sql
*************** insert into t1 (f1[5].q1) values(42);
*** 438,440 ****
--- 438,485 ----
  select * from t1;
  update t1 set f1[5].q2 = 43;
  select * from t1;
+ 
+ -- MULTISET support
+ 
+ SELECT cardinality(ARRAY[1, 2, 3]), cardinality(ARRAY[[1, 2], [3, 4]]);
+ SELECT trim_array(ARRAY[1, 2, 3], 0),
+        trim_array(ARRAY[1, 2, 3], 2),
+ 	   trim_array(ARRAY[[1, 2], [3, 4]], 1);
+ SELECT trim_array(ARRAY[1, 2, 3], -1);
+ SELECT trim_array(ARRAY[1, 2, 3], 4);
+ SELECT 'B' MEMBER OF ARRAY['A', 'B', 'C'];
+ SELECT 3 MEMBER OF ARRAY[1, 2], 3 MEMBER OF ARRAY[[1, 2], [3, 4]];
+ SELECT 3 NOT MEMBER OF ARRAY[1, 2];
+ SELECT ARRAY[1, 2] SUBMULTISET OF ARRAY[3, 2, 1];
+ SELECT ARRAY[1, 2] SUBMULTISET OF ARRAY[[1, 3], [2, 4]];
+ SELECT ARRAY[1, 1, 2] SUBMULTISET OF ARRAY[1, 2, 2];
+ SELECT ARRAY['A', 'B', 'C'] SUBMULTISET OF ARRAY['A', 'B', 'D'];
+ SELECT ARRAY['A', 'B', 'C'] NOT SUBMULTISET OF ARRAY['A', 'B', 'D'];
+ SELECT ARRAY[]::int[] SUBMULTISET OF NULL::int[];
+ SELECT NULL::int[] SUBMULTISET OF ARRAY[]::int[];
+ SELECT NULL::int[] SUBMULTISET OF NULL::int[];
+ SELECT ARRAY[1, 2] SUBMULTISET OF ARRAY[1, NULL],
+        ARRAY[1, 2] SUBMULTISET OF ARRAY[3, NULL];
+ SELECT ARRAY[1, NULL] SUBMULTISET OF ARRAY[1, NULL];
+ SELECT ARRAY[1, 2, 3] SUBMULTISET OF ARRAY[NULL, 1, 2, 3];
+ SELECT ARRAY[1, 2, 3] SUBMULTISET OF ARRAY[1, 2, NULL];
+ SELECT ARRAY[1, 2, 3] SUBMULTISET OF ARRAY[NULL, 4, NULL];
+ SELECT ARRAY[1, 2, 3] IS A SET;
+ SELECT ARRAY['A', 'A', 'B'] IS A SET, ARRAY['A', 'A', 'B'] IS NOT A SET;
+ SELECT ARRAY[1, NULL] IS A SET, ARRAY[1, NULL, NULL] IS NOT A SET;
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET UNION ALL ARRAY[2, NULL];
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET UNION ARRAY[2, NULL];
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET INTERSECT ALL ARRAY[2, NULL];
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET INTERSECT ARRAY[2, NULL];
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET EXCEPT ALL ARRAY[2, NULL];
+ SELECT ARRAY[2, NULL, 1, 2, NULL] MULTISET EXCEPT ARRAY[2, NULL];
+ SELECT collect(s), fusion(a), intersection(a)
+   FROM (VALUES
+   ('A', ARRAY[1, 2, 3, 2, 2]),
+   ('B', ARRAY[1, 2, 4, 2]),
+   ('C', ARRAY[[3, 2], [2, 1]])
+ ) AS t(s, a);
+ SELECT array_sort(ARRAY[1, 3, 2, 3, NULL, 1, NULL]);
+ SELECT array_sort(ARRAY[['A', 'D'], ['B', 'E'], ['C', 'F']]);
+ SELECT set(ARRAY[1, 3, 2, 3, NULL, 1, NULL]);
+ SELECT set(ARRAY[['A', 'C'], ['B', 'C'], ['C', 'A']]);