WIP: plpgsql - foreach in

*** a/doc/src/sgml/plpgsql.sgml
--- b/doc/src/sgml/plpgsql.sgml
***************
*** 2238,2243 **** END LOOP <optional> <replaceable>label</replaceable> </optional>;
--- 2238,2268 ----
      </para>
     </sect2>
  
+    <sect2 id="plpgsql-array-iterating">
+     <title>Looping Through Array</title>
+ <synopsis>
+ <optional> &lt;&lt;<replaceable>label</replaceable>&gt;&gt; </optional>
+ FOREACH <replaceable>target</replaceable> <optional> SCALE <replaceable>number</replaceable> </optional> IN <replaceable>expression</replaceable> LOOP
+     <replaceable>statements</replaceable>
+ END LOOP <optional> <replaceable>label</replaceable> </optional>;
+ </synopsis>
+ 
+ <programlisting>
+ CREATE FUNCTION sum(VARIADIC int[]) RETURNS int8 AS $$
+ DECLARE
+   s int8; x int;
+ BEGIN
+   FOREACH x IN $1
+   LOOP
+     s := s + x;
+   END LOOP;
+   RETURN s;
+ END;
+ $$ LANGUAGE plpgsql;
+ </programlisting>
+ 
+    </sect2>
+ 
     <sect2 id="plpgsql-error-trapping">
      <title>Trapping Errors</title>
  
*** a/src/pl/plpgsql/src/gram.y
--- b/src/pl/plpgsql/src/gram.y
***************
*** 21,26 ****
--- 21,27 ----
  #include "parser/parse_type.h"
  #include "parser/scanner.h"
  #include "parser/scansup.h"
+ #include "utils/array.h"
  
  
  /* Location tracking support --- simpler than bison's default */
***************
*** 134,139 **** static	List			*read_raise_options(void);
--- 135,141 ----
  			PLpgSQL_datum   *scalar;
  			PLpgSQL_rec     *rec;
  			PLpgSQL_row     *row;
+ 			int	slice;
  		}						forvariable;
  		struct
  		{
***************
*** 178,184 **** static	List			*read_raise_options(void);
  %type <ival>	assign_var
  %type <var>		cursor_variable
  %type <datum>	decl_cursor_arg
! %type <forvariable>	for_variable
  %type <stmt>	for_control
  
  %type <str>		any_identifier opt_block_label opt_label
--- 180,186 ----
  %type <ival>	assign_var
  %type <var>		cursor_variable
  %type <datum>	decl_cursor_arg
! %type <forvariable>	for_variable foreach_control
  %type <stmt>	for_control
  
  %type <str>		any_identifier opt_block_label opt_label
***************
*** 190,196 **** static	List			*read_raise_options(void);
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
--- 192,198 ----
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case stmt_foreach_a
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
***************
*** 264,269 **** static	List			*read_raise_options(void);
--- 266,272 ----
  %token <keyword>	K_FETCH
  %token <keyword>	K_FIRST
  %token <keyword>	K_FOR
+ %token <keyword>	K_FOREACH
  %token <keyword>	K_FORWARD
  %token <keyword>	K_FROM
  %token <keyword>	K_GET
***************
*** 298,303 **** static	List			*read_raise_options(void);
--- 301,307 ----
  %token <keyword>	K_ROWTYPE
  %token <keyword>	K_ROW_COUNT
  %token <keyword>	K_SCROLL
+ %token <keyword>	K_SLICE
  %token <keyword>	K_SQLSTATE
  %token <keyword>	K_STRICT
  %token <keyword>	K_THEN
***************
*** 739,744 **** proc_stmt		: pl_block ';'
--- 743,750 ----
  						{ $$ = $1; }
  				| stmt_for
  						{ $$ = $1; }
+ 				| stmt_foreach_a
+ 						{ $$ = $1; }
  				| stmt_exit
  						{ $$ = $1; }
  				| stmt_return
***************
*** 1386,1391 **** for_variable	: T_DATUM
--- 1392,1455 ----
  					}
  				;
  
+ stmt_foreach_a		: opt_block_label K_FOREACH foreach_control K_IN expr_until_loop loop_body
+ 					{
+ 						PLpgSQL_stmt_foreach_a *new = palloc0(sizeof(PLpgSQL_stmt_foreach_a));
+ 						new->cmd_type = PLPGSQL_STMT_FOREACH_A;
+ 						new->lineno = plpgsql_location_to_lineno(@2);
+ 						new->label = $1;
+ 						new->expr = $5;
+ 						new->slice = $3.slice;
+ 						new->body = $6.stmts;
+ 
+ 						if ($3.rec)
+ 						{
+ 							new->rec = $3.rec;
+ 							check_assignable((PLpgSQL_datum *) new->rec, @3);
+ 						}
+ 						else if ($3.row)
+ 						{
+ 							new->row = $3.row;
+ 							check_assignable((PLpgSQL_datum *) new->row, @3);
+ 						}
+ 						else if ($3.scalar)
+ 						{
+ 							Assert($3.scalar->dtype == PLPGSQL_DTYPE_VAR);
+ 							new->var = (PLpgSQL_var *) $3.scalar;
+ 							check_assignable($3.scalar, @3);
+ 
+ 						}
+ 						else
+ 						{
+ 							ereport(ERROR,
+ 									(errcode(ERRCODE_SYNTAX_ERROR),
+ 									 errmsg("loop variable of loop over array must be a record, row variable, scalar variable or list of scalar variables"),
+ 											 parser_errposition(@3)));
+ 						}
+ 
+ 						check_labels($1, $6.end_label, $6.end_label_location);
+ 						$$ = (PLpgSQL_stmt *) new;
+ 					}
+ 				;
+ 
+ foreach_control		: for_variable
+ 					{
+ 						$$ = $1;
+ 						$$.slice = 0;
+ 					}
+ 				| for_variable K_SLICE ICONST
+ 					{
+ 						$$ = $1;
+ 						$$.slice = $3;
+ 						if ($3 < 0 || $3 >= MAXDIM)
+ 							ereport(ERROR,
+ 									(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+ 									 errmsg("number of slicing array dimensions (%d) exceeds the maximum allowed (%d)",
+ 												$3, MAXDIM),
+ 											 parser_errposition(@3)));
+ 					}
+ 				;
+ 
  stmt_exit		: exit_type opt_label opt_exitcond
  					{
  						PLpgSQL_stmt_exit *new;
***************
*** 2063,2068 **** unreserved_keyword	:
--- 2127,2133 ----
  				| K_ROW_COUNT
  				| K_ROWTYPE
  				| K_SCROLL
+ 				| K_SLICE
  				| K_SQLSTATE
  				| K_TYPE
  				| K_USE_COLUMN
*** a/src/pl/plpgsql/src/pl_exec.c
--- b/src/pl/plpgsql/src/pl_exec.c
***************
*** 107,112 **** static int exec_stmt_fors(PLpgSQL_execstate *estate,
--- 107,114 ----
  			   PLpgSQL_stmt_fors *stmt);
  static int exec_stmt_forc(PLpgSQL_execstate *estate,
  			   PLpgSQL_stmt_forc *stmt);
+ static int exec_stmt_foreach_a(PLpgSQL_execstate *estate,
+ 				    PLpgSQL_stmt_foreach_a *stmt);
  static int exec_stmt_open(PLpgSQL_execstate *estate,
  			   PLpgSQL_stmt_open *stmt);
  static int exec_stmt_fetch(PLpgSQL_execstate *estate,
***************
*** 1312,1317 **** exec_stmt(PLpgSQL_execstate *estate, PLpgSQL_stmt *stmt)
--- 1314,1323 ----
  			rc = exec_stmt_forc(estate, (PLpgSQL_stmt_forc *) stmt);
  			break;
  
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			rc = exec_stmt_foreach_a(estate, (PLpgSQL_stmt_foreach_a *) stmt);
+ 			break;
+ 
  		case PLPGSQL_STMT_EXIT:
  			rc = exec_stmt_exit(estate, (PLpgSQL_stmt_exit *) stmt);
  			break;
***************
*** 2028,2033 **** exec_stmt_forc(PLpgSQL_execstate *estate, PLpgSQL_stmt_forc *stmt)
--- 2034,2290 ----
  
  
  /* ----------
+  * exec_stmt_foreach_a			Implements loop over array
+  * ----------
+  */
+ static int
+ exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
+ {
+ 	Datum		value;
+ 	bool		isnull;
+ 	Oid			valtype;
+ 	int			numelems = 0;
+ 	Oid			array_typelem;
+ 	int			idx;
+ 	ArrayType  *arr;
+ 	char	   *ptr;
+ 	bits8	   *arraynullsptr;
+ 	int16		elmlen;
+ 	bool		elmbyval;
+ 	char		elmalign;
+ 	PLpgSQL_datum *ctrl_var;
+ 	bool		found = false;
+ 	int			rc = PLPGSQL_RC_OK;
+ 	int			nitems = 1;
+ 
+ 	/* get the value of the array expression using array_expr */
+ 	value = exec_eval_expr(estate, stmt->expr, &isnull, &valtype);
+ 	if (isnull)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
+ 				 errmsg("NULL value isn't allowed as parameter of FOREACH-IN")));
+ 
+ 	/* check the type of the expression - must be an array */
+ 	array_typelem = get_element_type(valtype);
+ 
+ 	if (!OidIsValid(array_typelem))
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 				 errmsg("result of expression isn't an array"),
+ 				 errdetail("result of expression is %s",
+ 									format_type_be(valtype))));
+ 
+ 	/* make a copy of the result */
+ 	arr = DatumGetArrayTypePCopy(value);
+ 
+ 	/* Calculate the number of elements we're going to loop through */
+ 	numelems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));
+ 
+ 	/* Gather information about the array */
+ 	ptr = ARR_DATA_PTR(arr);
+ 	arraynullsptr = ARR_NULLBITMAP(arr);
+ 	get_typlenbyvalalign(ARR_ELEMTYPE(arr),
+ 						&elmlen,
+ 						&elmbyval,
+ 						&elmalign);
+ 
+ 	/* Clean up any leftover temporary memory */
+ 	exec_eval_cleanup(estate);
+ 
+ 	/*
+ 	 * If the target needs to be an array, check that it actually is,
+ 	 * and that it has a valid dimension for the array we're looping
+ 	 * through.
+ 	 *
+ 	 * XXX: Is this redundant?  Could it be merged with the below
+ 	 * set of conditionals?
+ 	 */
+ 	if (stmt->slice > 0)
+ 	{
+ 		if (stmt->rec != NULL || stmt->row != NULL)
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 					 errmsg("target variable \"%s\" isn't an array type",
+ 							    stmt->row ? stmt->row->refname : stmt->rec->refname),
+ 					 errhint("Value assigned will be of an array type.")));
+ 
+ 		if (stmt->slice > ARR_NDIM(arr))
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ 					 errmsg("slice level %d is higher than array dimension",
+ 								    stmt->slice)));
+ 	}
+ 
+ 	/* Set up the target variable */
+ 	if (stmt->var != NULL)
+ 	{
+ 		int	typoid = stmt->var->datatype->typoid;
+ 		int	elmoid = get_element_type(typoid);
+ 
+ 		if (stmt->slice > 0)
+ 		{
+ 			/* target variable has to be an array */
+ 			if (elmoid == InvalidOid)
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 						 errmsg("target variable \"%s\" for sliced array should be an array type",
+ 										stmt->var->refname)));
+ 
+ 			/* Determine the number of items in the slice */
+ 			nitems = ArrayGetNItems(stmt->slice, ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice);
+ 		}
+ 		else
+ 		{
+ 			/* target variable should be a scalar */
+ 			if (elmoid != InvalidOid)
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 						 errmsg("target variable \"%s\" is an array",
+ 										stmt->var->refname),
+ 						 errhint("Value assigned will be of a scalar type")));
+ 		}
+ 		ctrl_var = estate->datums[stmt->var->dno];
+ 	}
+ 	else if (stmt->row != NULL)
+ 	{
+ 		ctrl_var = estate->datums[stmt->row->dno];
+ 	}
+ 	else
+ 	{
+ 		ctrl_var = estate->datums[stmt->rec->dno];
+ 	}
+ 
+ 	/*
+ 	 * Now do the loop
+ 	 *
+ 	 * XXX: These variable names are.. unfortunate.  Could we come up
+ 	 * with soe which are more descriptive?
+ 	 */
+ 	idx = 0;
+ 	while (idx < numelems)
+ 	{
+ 		int		j;
+ 		ArrayBuildState *astate = NULL;
+ 
+ 		found = true;				/* looped at least once */
+ 
+ 		/* Loop through the items in this slice to build up the
+ 		 * array slice to be used through the exec statements ... ?
+ 		 * XXX: This really needs some better commenting.. */
+ 		for (j = 0; j < nitems; j++)
+ 		{
+ 			/* XXX: Do we have a better way to check arraynullsptr?
+ 			 * Maybe a macro of some kind ..? */
+ 			if (arraynullsptr != NULL && !(arraynullsptr[idx / 8] & (1 << (idx % 8))))
+ 			{
+ 				isnull = true;
+ 				value = (Datum) 0;
+ 			}
+ 			else
+ 			{
+ 				isnull = false;
+ 				value = fetch_att(ptr, elmbyval, elmlen);
+ 
+ 				ptr = att_addlength_pointer(ptr, elmlen, ptr);
+ 				ptr = (char *) att_align_nominal(ptr, elmalign);
+ 			}
+ 
+ 			if (stmt->slice > 0)
+ 				astate = accumArrayResult(astate, value, isnull,
+ 									array_typelem,
+ 									CurrentMemoryContext);
+ 
+ 			/* XXX: Why are we updating this here..? */
+ 			idx++;
+ 		}
+ 
+ 		/*
+ 		 * store an item in to the variable - we are expecting almost all
+ 		 * iterations will be over scalar values, so we don't create
+ 		 * a fake tuple over scalar and we don't use exec_move_row for
+ 		 * scalars. This is about four times faster.
+ 		 */
+ 		if (astate != NULL)
+ 		{
+ 			Datum sliced_arr;
+ 			bool isnull = false;
+ 
+ 			sliced_arr = makeMdArrayResult(astate, stmt->slice,
+ 									    ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice,
+ 									    ARR_LBOUND(arr) ? ARR_LBOUND(arr) + ARR_NDIM(arr) - stmt->slice : NULL,
+ 										    CurrentMemoryContext, true);
+ 			exec_assign_value(estate, ctrl_var, sliced_arr, valtype, &isnull);
+ 		}
+ 		else
+ 			exec_assign_value(estate, ctrl_var,
+ 						    value, array_typelem, &isnull);
+ 
+ 		/*
+ 		 * Execute the statements
+ 		 */
+ 		rc = exec_stmts(estate, stmt->body);
+ 
+ 		if (rc == PLPGSQL_RC_RETURN)
+ 			break;				/* break out of the loop */
+ 		else if (rc == PLPGSQL_RC_EXIT)
+ 		{
+ 			if (estate->exitlabel == NULL)
+ 				/* unlabelled exit, finish the current loop */
+ 				rc = PLPGSQL_RC_OK;
+ 			else if (stmt->label != NULL &&
+ 					 strcmp(stmt->label, estate->exitlabel) == 0)
+ 			{
+ 				/* labelled exit, matches the current stmt's label */
+ 				estate->exitlabel = NULL;
+ 				rc = PLPGSQL_RC_OK;
+ 			}
+ 
+ 			/*
+ 			 * otherwise, this is a labelled exit that does not match the
+ 			 * current statement's label, if any: return RC_EXIT so that the
+ 			 * EXIT continues to propagate up the stack.
+ 			 */
+ 			break;
+ 		}
+ 		else if (rc == PLPGSQL_RC_CONTINUE)
+ 		{
+ 			if (estate->exitlabel == NULL)
+ 				/* unlabelled continue, so re-run the current loop */
+ 				rc = PLPGSQL_RC_OK;
+ 			else if (stmt->label != NULL &&
+ 					 strcmp(stmt->label, estate->exitlabel) == 0)
+ 			{
+ 				/* label matches named continue, so re-run loop */
+ 				estate->exitlabel = NULL;
+ 				rc = PLPGSQL_RC_OK;
+ 			}
+ 			else
+ 			{
+ 				/*
+ 				 * otherwise, this is a named continue that does not match the
+ 				 * current statement's label, if any: return RC_CONTINUE so
+ 				 * that the CONTINUE will propagate up the stack.
+ 				 */
+ 				break;
+ 			}
+ 		}
+ 	}
+ 
+ 	pfree(arr);
+ 
+ 	/*
+ 	 * Set the FOUND variable to indicate the result of executing the loop
+ 	 * (namely, whether we looped one or more times). This must be set here so
+ 	 * that it does not interfere with the value of the FOUND variable inside
+ 	 * the loop processing itself.
+ 	 */
+ 	exec_set_found(estate, found);
+ 
+ 	return rc;
+ }
+ 
+ 
+ /* ----------
   * exec_stmt_exit			Implements EXIT and CONTINUE
   *
   * This begins the process of exiting / restarting a loop.
*** a/src/pl/plpgsql/src/pl_funcs.c
--- b/src/pl/plpgsql/src/pl_funcs.c
***************
*** 230,235 **** plpgsql_stmt_typename(PLpgSQL_stmt *stmt)
--- 230,237 ----
  			return _("FOR over SELECT rows");
  		case PLPGSQL_STMT_FORC:
  			return _("FOR over cursor");
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			return _("FOREACH over array");
  		case PLPGSQL_STMT_EXIT:
  			return "EXIT";
  		case PLPGSQL_STMT_RETURN:
***************
*** 293,298 **** static void dump_cursor_direction(PLpgSQL_stmt_fetch *stmt);
--- 295,301 ----
  static void dump_close(PLpgSQL_stmt_close *stmt);
  static void dump_perform(PLpgSQL_stmt_perform *stmt);
  static void dump_expr(PLpgSQL_expr *expr);
+ static void dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt);
  
  
  static void
***************
*** 376,381 **** dump_stmt(PLpgSQL_stmt *stmt)
--- 379,387 ----
  		case PLPGSQL_STMT_PERFORM:
  			dump_perform((PLpgSQL_stmt_perform *) stmt);
  			break;
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			dump_foreach_a((PLpgSQL_stmt_foreach_a *) stmt);
+ 			break;
  		default:
  			elog(ERROR, "unrecognized cmd_type: %d", stmt->cmd_type);
  			break;
***************
*** 596,601 **** dump_forc(PLpgSQL_stmt_forc *stmt)
--- 602,625 ----
  }
  
  static void
+ dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt)
+ {
+ 	dump_ind();
+ 	printf("FOREACHA %s", (stmt->rec != NULL) ? stmt->rec->refname :
+ 						    (stmt->row != NULL) ? stmt->row->refname : stmt->var->refname);
+ 	if (stmt->slice != 0)
+ 		printf("SLICE %d ", stmt->slice);
+ 	printf("IN ");
+ 	dump_expr(stmt->expr);
+ 	printf("\n");
+ 
+ 	dump_stmts(stmt->body);
+ 
+ 	dump_ind();
+ 	printf("    ENDOFOREACHA");
+ }
+ 
+ static void
  dump_open(PLpgSQL_stmt_open *stmt)
  {
  	dump_ind();
*** a/src/pl/plpgsql/src/pl_scanner.c
--- b/src/pl/plpgsql/src/pl_scanner.c
***************
*** 77,82 **** static const ScanKeyword reserved_keywords[] = {
--- 77,83 ----
  	PG_KEYWORD("exit", K_EXIT, RESERVED_KEYWORD)
  	PG_KEYWORD("fetch", K_FETCH, RESERVED_KEYWORD)
  	PG_KEYWORD("for", K_FOR, RESERVED_KEYWORD)
+ 	PG_KEYWORD("foreach", K_FOREACH, RESERVED_KEYWORD)
  	PG_KEYWORD("from", K_FROM, RESERVED_KEYWORD)
  	PG_KEYWORD("get", K_GET, RESERVED_KEYWORD)
  	PG_KEYWORD("if", K_IF, RESERVED_KEYWORD)
***************
*** 133,138 **** static const ScanKeyword unreserved_keywords[] = {
--- 134,140 ----
  	PG_KEYWORD("row_count", K_ROW_COUNT, UNRESERVED_KEYWORD)
  	PG_KEYWORD("rowtype", K_ROWTYPE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("scroll", K_SCROLL, UNRESERVED_KEYWORD)
+ 	PG_KEYWORD("slice", K_SLICE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("sqlstate", K_SQLSTATE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("type", K_TYPE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("use_column", K_USE_COLUMN, UNRESERVED_KEYWORD)
*** a/src/pl/plpgsql/src/plpgsql.h
--- b/src/pl/plpgsql/src/plpgsql.h
***************
*** 87,92 **** enum PLpgSQL_stmt_types
--- 87,93 ----
  	PLPGSQL_STMT_CASE,
  	PLPGSQL_STMT_LOOP,
  	PLPGSQL_STMT_WHILE,
+ 	PLPGSQL_STMT_FOREACH_A,
  	PLPGSQL_STMT_FORI,
  	PLPGSQL_STMT_FORS,
  	PLPGSQL_STMT_FORC,
***************
*** 427,432 **** typedef struct
--- 428,447 ----
  
  
  typedef struct
+ {								/* FOREACH item in array loop */
+ 	int			cmd_type;
+ 	int			lineno;
+ 	char	   *label;
+ 	PLpgSQL_var *var;
+ 	PLpgSQL_rec *rec;
+ 	PLpgSQL_row *row;
+ 	PLpgSQL_expr	*expr;
+ 	int			slice;
+ 	List	   *body;			/* List of statements */
+ } PLpgSQL_stmt_foreach_a;
+ 
+ 
+ typedef struct
  {								/* FOR statement with integer loopvar	*/
  	int			cmd_type;
  	int			lineno;
*** a/src/test/regress/expected/plpgsql.out
--- b/src/test/regress/expected/plpgsql.out
***************
*** 4240,4242 **** select unreserved_test();
--- 4240,4434 ----
  (1 row)
  
  drop function unreserved_test();
+ -- Checking a FOREACH statement
+ create function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[1,2,3,4]);
+ NOTICE:  1
+ NOTICE:  2
+ NOTICE:  3
+ NOTICE:  4
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ NOTICE:  1
+ NOTICE:  2
+ NOTICE:  3
+ NOTICE:  4
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ ERROR:  target variable "x" for sliced array should be an array type
+ CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ ERROR:  target variable "x" for sliced array should be an array type
+ CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[1,2,3,4]);
+ NOTICE:  {1,2,3,4}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ NOTICE:  {1,2}
+ NOTICE:  {3,4}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ -- higher level of slicing
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 2 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ ERROR:  slice level 2 is higher than array dimension
+ CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
+ -- ok
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ NOTICE:  {{1,2},{3,4}}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[[1,2]],[[3,4]]]);
+ NOTICE:  {{1,2}}
+ NOTICE:  {{3,4}}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ create type xy_tuple AS (x int, y int);
+ -- iteration over array of records
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare r record;
+ begin
+   foreach r in $1
+   loop
+     raise notice '%', r;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ NOTICE:  (10,20)
+ NOTICE:  (40,69)
+ NOTICE:  (35,78)
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ NOTICE:  (10,20)
+ NOTICE:  (40,69)
+ NOTICE:  (35,78)
+ NOTICE:  (88,76)
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int; y int;
+ begin
+   foreach x, y in $1
+   loop
+     raise notice 'x = %, y = %', x, y;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ NOTICE:  x = 10, y = 20
+ NOTICE:  x = 40, y = 69
+ NOTICE:  x = 35, y = 78
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ NOTICE:  x = 10, y = 20
+ NOTICE:  x = 40, y = 69
+ NOTICE:  x = 35, y = 78
+ NOTICE:  x = 88, y = 76
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ -- slicing over array of composite types
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x xy_tuple[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ NOTICE:  {"(10,20)","(40,69)","(35,78)"}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ NOTICE:  {"(10,20)","(40,69)"}
+ NOTICE:  {"(35,78)","(88,76)"}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ drop function foreach_test(anyarray);
+ drop type xy_tuple;
*** a/src/test/regress/sql/plpgsql.sql
--- b/src/test/regress/sql/plpgsql.sql
***************
*** 3375,3377 **** $$ language plpgsql;
--- 3375,3488 ----
  select unreserved_test();
  
  drop function unreserved_test();
+ 
+ -- Checking a FOREACH statement
+ create function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[1,2,3,4]);
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[1,2,3,4]);
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ 
+ -- higher level of slicing
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 2 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ -- ok
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ select foreach_test(ARRAY[[[1,2]],[[3,4]]]);
+ 
+ create type xy_tuple AS (x int, y int);
+ 
+ -- iteration over array of records
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare r record;
+ begin
+   foreach r in $1
+   loop
+     raise notice '%', r;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int; y int;
+ begin
+   foreach x, y in $1
+   loop
+     raise notice 'x = %, y = %', x, y;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ 
+ -- slicing over array of composite types
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x xy_tuple[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ 
+ drop function foreach_test(anyarray);
+ drop type xy_tuple;

*** a/src/pl/plpgsql/src/gram.y
--- b/src/pl/plpgsql/src/gram.y
***************
*** 192,198 **** static	List			*read_raise_options(void);
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case stmt_foreach_a 
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
--- 192,198 ----
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case stmt_foreach_a
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
***************
*** 1392,1398 **** for_variable	: T_DATUM
  					}
  				;
  
! stmt_foreach_a		: opt_block_label K_FOREACH foreach_control K_IN  expr_until_loop loop_body
  					{
  						PLpgSQL_stmt_foreach_a *new = palloc0(sizeof(PLpgSQL_stmt_foreach_a));
  						new->cmd_type = PLPGSQL_STMT_FOREACH_A;
--- 1392,1398 ----
  					}
  				;
  
! stmt_foreach_a		: opt_block_label K_FOREACH foreach_control K_IN expr_until_loop loop_body
  					{
  						PLpgSQL_stmt_foreach_a *new = palloc0(sizeof(PLpgSQL_stmt_foreach_a));
  						new->cmd_type = PLPGSQL_STMT_FOREACH_A;
***************
*** 1423,1429 **** stmt_foreach_a		: opt_block_label K_FOREACH foreach_control K_IN  expr_until_loo
  						{
  							ereport(ERROR,
  									(errcode(ERRCODE_SYNTAX_ERROR),
! 									 errmsg("loop variable of loop over arrat must be a record, row variable, scalar variable or list of scalar variables"),
  											 parser_errposition(@3)));
  						}
  
--- 1423,1429 ----
  						{
  							ereport(ERROR,
  									(errcode(ERRCODE_SYNTAX_ERROR),
! 									 errmsg("loop variable of loop over array must be a record, row variable, scalar variable or list of scalar variables"),
  											 parser_errposition(@3)));
  						}
  
*** a/src/pl/plpgsql/src/pl_exec.c
--- b/src/pl/plpgsql/src/pl_exec.c
***************
*** 2035,2083 **** exec_stmt_forc(PLpgSQL_execstate *estate, PLpgSQL_stmt_forc *stmt)
  
  /* ----------
   * exec_stmt_foreach_a			Implements loop over array
-  *
   * ----------
   */
! static int 
  exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  {
  	Datum		value;
  	bool		isnull;
  	Oid			valtype;
! 	int	numelems = 0;
! 	Oid 	array_typelem;
! 	int	idx;
! 	ArrayType	*arr;
! 	char *ptr;
! 	bits8	*arraynullsptr;
! 	int16	elmlen;
! 	bool	elmbyval;
! 	char	elmalign;
  	PLpgSQL_datum *ctrl_var;
  	bool		found = false;
  	int			rc = PLPGSQL_RC_OK;
! 	int		nitems = 1;
  
! 	/* get a result of array_expr */
  	value = exec_eval_expr(estate, stmt->expr, &isnull, &valtype);
  	if (isnull)
  		ereport(ERROR,
  				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
  				 errmsg("NULL value isn't allowed as parameter of FOREACH-IN")));
  
! 	/* check a result of expression - must be a array */
  	array_typelem = get_element_type(valtype);
  
  	if (!OidIsValid(array_typelem))
  		ereport(ERROR,
  				(errcode(ERRCODE_DATATYPE_MISMATCH),
! 				 errmsg("result of expression isn't array"),
! 				 errdetail("result of expression is %s", 
  									format_type_be(valtype))));
!  
! 	/* copy a result and takes some infos */
  	arr = DatumGetArrayTypePCopy(value);
  	numelems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));
  	ptr = ARR_DATA_PTR(arr);
  	arraynullsptr = ARR_NULLBITMAP(arr);
  	get_typlenbyvalalign(ARR_ELEMTYPE(arr),
--- 2035,2086 ----
  
  /* ----------
   * exec_stmt_foreach_a			Implements loop over array
   * ----------
   */
! static int
  exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  {
  	Datum		value;
  	bool		isnull;
  	Oid			valtype;
! 	int			numelems = 0;
! 	Oid			array_typelem;
! 	int			idx;
! 	ArrayType  *arr;
! 	char	   *ptr;
! 	bits8	   *arraynullsptr;
! 	int16		elmlen;
! 	bool		elmbyval;
! 	char		elmalign;
  	PLpgSQL_datum *ctrl_var;
  	bool		found = false;
  	int			rc = PLPGSQL_RC_OK;
! 	int			nitems = 1;
  
! 	/* get the value of the array expression using array_expr */
  	value = exec_eval_expr(estate, stmt->expr, &isnull, &valtype);
  	if (isnull)
  		ereport(ERROR,
  				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
  				 errmsg("NULL value isn't allowed as parameter of FOREACH-IN")));
  
! 	/* check the type of the expression - must be an array */
  	array_typelem = get_element_type(valtype);
  
  	if (!OidIsValid(array_typelem))
  		ereport(ERROR,
  				(errcode(ERRCODE_DATATYPE_MISMATCH),
! 				 errmsg("result of expression isn't an array"),
! 				 errdetail("result of expression is %s",
  									format_type_be(valtype))));
! 
! 	/* make a copy of the result */
  	arr = DatumGetArrayTypePCopy(value);
+ 
+ 	/* Calculate the number of elements we're going to loop through */
  	numelems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));
+ 
+ 	/* Gather information about the array */
  	ptr = ARR_DATA_PTR(arr);
  	arraynullsptr = ARR_NULLBITMAP(arr);
  	get_typlenbyvalalign(ARR_ELEMTYPE(arr),
***************
*** 2085,2101 **** exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  						&elmbyval,
  						&elmalign);
  
! 	/* clean a stack */
  	exec_eval_cleanup(estate);
  
  	if (stmt->slice > 0)
  	{
  		if (stmt->rec != NULL || stmt->row != NULL)
  			ereport(ERROR,
  					(errcode(ERRCODE_DATATYPE_MISMATCH),
! 					 errmsg("target variable \"%s\" isn't a of array type",
  							    stmt->row ? stmt->row->refname : stmt->rec->refname),
! 					 errhint("Assigned value will be a value of array type.")));
  
  		if (stmt->slice > ARR_NDIM(arr))
  			ereport(ERROR,
--- 2088,2112 ----
  						&elmbyval,
  						&elmalign);
  
! 	/* Clean up any leftover temporary memory */
  	exec_eval_cleanup(estate);
  
+ 	/*
+ 	 * If the target needs to be an array, check that it actually is,
+ 	 * and that it has a valid dimension for the array we're looping
+ 	 * through.
+ 	 *
+ 	 * XXX: Is this redundant?  Could it be merged with the below
+ 	 * set of conditionals?
+ 	 */
  	if (stmt->slice > 0)
  	{
  		if (stmt->rec != NULL || stmt->row != NULL)
  			ereport(ERROR,
  					(errcode(ERRCODE_DATATYPE_MISMATCH),
! 					 errmsg("target variable \"%s\" isn't an array type",
  							    stmt->row ? stmt->row->refname : stmt->rec->refname),
! 					 errhint("Value assigned will be of an array type.")));
  
  		if (stmt->slice > ARR_NDIM(arr))
  			ereport(ERROR,
***************
*** 2104,2110 **** exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  								    stmt->slice)));
  	}
  
! 	/* get a target variable */
  	if (stmt->var != NULL)
  	{
  		int	typoid = stmt->var->datatype->typoid;
--- 2115,2121 ----
  								    stmt->slice)));
  	}
  
! 	/* Set up the target variable */
  	if (stmt->var != NULL)
  	{
  		int	typoid = stmt->var->datatype->typoid;
***************
*** 2112,2123 **** exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  
  		if (stmt->slice > 0)
  		{
! 			/* target variable have to be a array type */
  			if (elmoid == InvalidOid)
  				ereport(ERROR,
  						(errcode(ERRCODE_DATATYPE_MISMATCH),
! 						 errmsg("target variable \"%s\" for sliced array should be a array",
  										stmt->var->refname)));
  			nitems = ArrayGetNItems(stmt->slice, ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice);
  		}
  		else
--- 2123,2136 ----
  
  		if (stmt->slice > 0)
  		{
! 			/* target variable has to be an array */
  			if (elmoid == InvalidOid)
  				ereport(ERROR,
  						(errcode(ERRCODE_DATATYPE_MISMATCH),
! 						 errmsg("target variable \"%s\" for sliced array should be an array type",
  										stmt->var->refname)));
+ 
+ 			/* Determine the number of items in the slice */
  			nitems = ArrayGetNItems(stmt->slice, ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice);
  		}
  		else
***************
*** 2126,2133 **** exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  			if (elmoid != InvalidOid)
  				ereport(ERROR,
  						(errcode(ERRCODE_DATATYPE_MISMATCH),
! 						 errmsg("target variable \"%s\" is a array", 
! 										stmt->var->refname)));
  		}
  		ctrl_var = estate->datums[stmt->var->dno];
  	}
--- 2139,2147 ----
  			if (elmoid != InvalidOid)
  				ereport(ERROR,
  						(errcode(ERRCODE_DATATYPE_MISMATCH),
! 						 errmsg("target variable \"%s\" is an array",
! 										stmt->var->refname),
! 						 errhint("Value assigned will be of a scalar type")));
  		}
  		ctrl_var = estate->datums[stmt->var->dno];
  	}
***************
*** 2135,2147 **** exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  	{
  		ctrl_var = estate->datums[stmt->row->dno];
  	}
! 	else 
  	{
  		ctrl_var = estate->datums[stmt->rec->dno];
  	}
  
  	/*
  	 * Now do the loop
  	 */
  	idx = 0;
  	while (idx < numelems)
--- 2149,2164 ----
  	{
  		ctrl_var = estate->datums[stmt->row->dno];
  	}
! 	else
  	{
  		ctrl_var = estate->datums[stmt->rec->dno];
  	}
  
  	/*
  	 * Now do the loop
+ 	 *
+ 	 * XXX: These variable names are.. unfortunate.  Could we come up
+ 	 * with soe which are more descriptive?
  	 */
  	idx = 0;
  	while (idx < numelems)
***************
*** 2151,2158 **** exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
--- 2168,2180 ----
  
  		found = true;				/* looped at least once */
  
+ 		/* Loop through the items in this slice to build up the
+ 		 * array slice to be used through the exec statements ... ?
+ 		 * XXX: This really needs some better commenting.. */
  		for (j = 0; j < nitems; j++)
  		{
+ 			/* XXX: Do we have a better way to check arraynullsptr?
+ 			 * Maybe a macro of some kind ..? */
  			if (arraynullsptr != NULL && !(arraynullsptr[idx / 8] & (1 << (idx % 8))))
  			{
  				isnull = true;
***************
*** 2168,2193 **** exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
  			}
  
  			if (stmt->slice > 0)
! 				astate = accumArrayResult(astate, value, isnull, 
! 									array_typelem, 
  									CurrentMemoryContext);
  			idx++;
  		}
  
! 		/* 
! 		 * store a item to variable - we expecting so almost all
! 		 * iteration will be over scalar values, so it is reason
! 		 * why we don't create a fake tuple over scalar and we 
! 		 * don't use a exec_move_row for scalars. This is about 
! 		 * four times faster.
  		 */
  		if (astate != NULL)
  		{
  			Datum sliced_arr;
  			bool isnull = false;
  
! 			sliced_arr = makeMdArrayResult(astate, stmt->slice, 
! 									    ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice, 
  									    ARR_LBOUND(arr) ? ARR_LBOUND(arr) + ARR_NDIM(arr) - stmt->slice : NULL,
  										    CurrentMemoryContext, true);
  			exec_assign_value(estate, ctrl_var, sliced_arr, valtype, &isnull);
--- 2190,2216 ----
  			}
  
  			if (stmt->slice > 0)
! 				astate = accumArrayResult(astate, value, isnull,
! 									array_typelem,
  									CurrentMemoryContext);
+ 
+ 			/* XXX: Why are we updating this here..? */
  			idx++;
  		}
  
! 		/*
! 		 * store an item in to the variable - we are expecting almost all
! 		 * iterations will be over scalar values, so we don't create
! 		 * a fake tuple over scalar and we don't use exec_move_row for
! 		 * scalars. This is about four times faster.
  		 */
  		if (astate != NULL)
  		{
  			Datum sliced_arr;
  			bool isnull = false;
  
! 			sliced_arr = makeMdArrayResult(astate, stmt->slice,
! 									    ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice,
  									    ARR_LBOUND(arr) ? ARR_LBOUND(arr) + ARR_NDIM(arr) - stmt->slice : NULL,
  										    CurrentMemoryContext, true);
  			exec_assign_value(estate, ctrl_var, sliced_arr, valtype, &isnull);
*** a/src/pl/plpgsql/src/pl_funcs.c
--- b/src/pl/plpgsql/src/pl_funcs.c
***************
*** 605,611 **** static void
  dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt)
  {
  	dump_ind();
! 	printf("FOREACHA %s", (stmt->rec != NULL) ? stmt->rec->refname : 
  						    (stmt->row != NULL) ? stmt->row->refname : stmt->var->refname);
  	if (stmt->slice != 0)
  		printf("SLICE %d ", stmt->slice);
--- 605,611 ----
  dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt)
  {
  	dump_ind();
! 	printf("FOREACHA %s", (stmt->rec != NULL) ? stmt->rec->refname :
  						    (stmt->row != NULL) ? stmt->row->refname : stmt->var->refname);
  	if (stmt->slice != 0)
  		printf("SLICE %d ", stmt->slice);
*** a/src/pl/plpgsql/src/plpgsql.h
--- b/src/pl/plpgsql/src/plpgsql.h
***************
*** 436,442 **** typedef struct
  	PLpgSQL_rec *rec;
  	PLpgSQL_row *row;
  	PLpgSQL_expr	*expr;
! 	int		slice;
  	List	   *body;			/* List of statements */
  } PLpgSQL_stmt_foreach_a;
  
--- 436,442 ----
  	PLpgSQL_rec *rec;
  	PLpgSQL_row *row;
  	PLpgSQL_expr	*expr;
! 	int			slice;
  	List	   *body;			/* List of statements */
  } PLpgSQL_stmt_foreach_a;
  
*** a/src/test/regress/expected/plpgsql.out
--- b/src/test/regress/expected/plpgsql.out
***************
*** 4283,4292 **** begin
  $$ language plpgsql;
  -- should fail
  select foreach_test(ARRAY[1,2,3,4]);
! ERROR:  target variable "x" for sliced array should be a array
  CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
  select foreach_test(ARRAY[[1,2],[3,4]]);
! ERROR:  target variable "x" for sliced array should be a array
  CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
  create or replace function foreach_test(anyarray)
  returns void as $$
--- 4283,4292 ----
  $$ language plpgsql;
  -- should fail
  select foreach_test(ARRAY[1,2,3,4]);
! ERROR:  target variable "x" for sliced array should be an array type
  CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
  select foreach_test(ARRAY[[1,2],[3,4]]);
! ERROR:  target variable "x" for sliced array should be an array type
  CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
  create or replace function foreach_test(anyarray)
  returns void as $$

*** ./doc/src/sgml/plpgsql.sgml.orig	2011-01-16 14:18:58.000000000 +0100
--- ./doc/src/sgml/plpgsql.sgml	2011-01-17 11:31:54.086217514 +0100
***************
*** 2238,2243 ****
--- 2238,2268 ----
      </para>
     </sect2>
  
+    <sect2 id="plpgsql-array-iterating">
+     <title>Looping Through Array</title>
+ <synopsis>
+ <optional> &lt;&lt;<replaceable>label</replaceable>&gt;&gt; </optional>
+ FOREACH <replaceable>target</replaceable> <optional> SCALE <replaceable>number</replaceable> </optional> IN <replaceable>expression</replaceable> LOOP
+     <replaceable>statements</replaceable>
+ END LOOP <optional> <replaceable>label</replaceable> </optional>;
+ </synopsis>
+ 
+ <programlisting>
+ CREATE FUNCTION sum(VARIADIC int[]) RETURNS int8 AS $$
+ DECLARE
+   s int8; x int;
+ BEGIN
+   FOREACH x IN $1
+   LOOP
+     s := s + x;
+   END LOOP;
+   RETURN s;
+ END;
+ $$ LANGUAGE plpgsql;
+ </programlisting>
+ 
+    </sect2>
+ 
     <sect2 id="plpgsql-error-trapping">
      <title>Trapping Errors</title>
  
*** ./src/backend/utils/adt/arrayfuncs.c.orig	2011-01-16 14:18:58.000000000 +0100
--- ./src/backend/utils/adt/arrayfuncs.c	2011-01-21 22:59:00.315522289 +0100
***************
*** 68,74 ****
  			 Datum *values, bool *nulls, int nitems,
  			 int typlen, bool typbyval, char typalign,
  			 bool freedata);
- static bool array_get_isnull(const bits8 *nullbitmap, int offset);
  static void array_set_isnull(bits8 *nullbitmap, int offset, bool isNull);
  static Datum ArrayCast(char *value, bool byval, int len);
  static int ArrayCastAndSet(Datum src,
--- 68,73 ----
***************
*** 3836,3842 ****
   * nullbitmap: pointer to array's null bitmap (NULL if none)
   * offset: 0-based linear element number of array element
   */
! static bool
  array_get_isnull(const bits8 *nullbitmap, int offset)
  {
  	if (nullbitmap == NULL)
--- 3835,3841 ----
   * nullbitmap: pointer to array's null bitmap (NULL if none)
   * offset: 0-based linear element number of array element
   */
! bool
  array_get_isnull(const bits8 *nullbitmap, int offset)
  {
  	if (nullbitmap == NULL)
*** ./src/include/utils/array.h.orig	2011-01-16 14:18:59.000000000 +0100
--- ./src/include/utils/array.h	2011-01-21 23:03:07.294898398 +0100
***************
*** 215,220 ****
--- 215,221 ----
  extern ArrayType *array_set(ArrayType *array, int nSubscripts, int *indx,
  		  Datum dataValue, bool isNull,
  		  int arraytyplen, int elmlen, bool elmbyval, char elmalign);
+ extern bool array_get_isnull(const bits8 *nullbitmap, int offset);
  extern ArrayType *array_get_slice(ArrayType *array, int nSubscripts,
  				int *upperIndx, int *lowerIndx,
  				int arraytyplen, int elmlen, bool elmbyval, char elmalign);
*** ./src/pl/plpgsql/src/gram.y.orig	2011-01-16 14:18:59.000000000 +0100
--- ./src/pl/plpgsql/src/gram.y	2011-01-21 21:35:21.959579809 +0100
***************
*** 21,26 ****
--- 21,27 ----
  #include "parser/parse_type.h"
  #include "parser/scanner.h"
  #include "parser/scansup.h"
+ #include "utils/array.h"
  
  
  /* Location tracking support --- simpler than bison's default */
***************
*** 134,139 ****
--- 135,141 ----
  			PLpgSQL_datum   *scalar;
  			PLpgSQL_rec     *rec;
  			PLpgSQL_row     *row;
+ 			int	slice;
  		}						forvariable;
  		struct
  		{
***************
*** 178,184 ****
  %type <ival>	assign_var
  %type <var>		cursor_variable
  %type <datum>	decl_cursor_arg
! %type <forvariable>	for_variable
  %type <stmt>	for_control
  
  %type <str>		any_identifier opt_block_label opt_label
--- 180,186 ----
  %type <ival>	assign_var
  %type <var>		cursor_variable
  %type <datum>	decl_cursor_arg
! %type <forvariable>	for_variable foreach_control
  %type <stmt>	for_control
  
  %type <str>		any_identifier opt_block_label opt_label
***************
*** 190,196 ****
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
--- 192,198 ----
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case stmt_foreach_a
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
***************
*** 264,269 ****
--- 266,272 ----
  %token <keyword>	K_FETCH
  %token <keyword>	K_FIRST
  %token <keyword>	K_FOR
+ %token <keyword>	K_FOREACH
  %token <keyword>	K_FORWARD
  %token <keyword>	K_FROM
  %token <keyword>	K_GET
***************
*** 298,303 ****
--- 301,307 ----
  %token <keyword>	K_ROWTYPE
  %token <keyword>	K_ROW_COUNT
  %token <keyword>	K_SCROLL
+ %token <keyword>	K_SLICE
  %token <keyword>	K_SQLSTATE
  %token <keyword>	K_STRICT
  %token <keyword>	K_THEN
***************
*** 739,744 ****
--- 743,750 ----
  						{ $$ = $1; }
  				| stmt_for
  						{ $$ = $1; }
+ 				| stmt_foreach_a
+ 						{ $$ = $1; }
  				| stmt_exit
  						{ $$ = $1; }
  				| stmt_return
***************
*** 1386,1391 ****
--- 1392,1455 ----
  					}
  				;
  
+ stmt_foreach_a		: opt_block_label K_FOREACH foreach_control K_IN expr_until_loop loop_body
+ 					{
+ 						PLpgSQL_stmt_foreach_a *new = palloc0(sizeof(PLpgSQL_stmt_foreach_a));
+ 						new->cmd_type = PLPGSQL_STMT_FOREACH_A;
+ 						new->lineno = plpgsql_location_to_lineno(@2);
+ 						new->label = $1;
+ 						new->expr = $5;
+ 						new->slice = $3.slice;
+ 						new->body = $6.stmts;
+ 
+ 						if ($3.rec)
+ 						{
+ 							new->rec = $3.rec;
+ 							check_assignable((PLpgSQL_datum *) new->rec, @3);
+ 						}
+ 						else if ($3.row)
+ 						{
+ 							new->row = $3.row;
+ 							check_assignable((PLpgSQL_datum *) new->row, @3);
+ 						}
+ 						else if ($3.scalar)
+ 						{
+ 							Assert($3.scalar->dtype == PLPGSQL_DTYPE_VAR);
+ 							new->var = (PLpgSQL_var *) $3.scalar;
+ 							check_assignable($3.scalar, @3);
+ 
+ 						}
+ 						else
+ 						{
+ 							ereport(ERROR,
+ 									(errcode(ERRCODE_SYNTAX_ERROR),
+ 									 errmsg("loop variable of loop over array must be a record, row variable, scalar variable or list of scalar variables"),
+ 											 parser_errposition(@3)));
+ 						}
+ 
+ 						check_labels($1, $6.end_label, $6.end_label_location);
+ 						$$ = (PLpgSQL_stmt *) new;
+ 					}
+ 				;
+ 
+ foreach_control		: for_variable
+ 					{
+ 						$$ = $1;
+ 						$$.slice = 0;
+ 					}
+ 				| for_variable K_SLICE ICONST
+ 					{
+ 						$$ = $1;
+ 						$$.slice = $3;
+ 						if ($3 < 0 || $3 >= MAXDIM)
+ 							ereport(ERROR,
+ 									(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+ 									 errmsg("number of slicing array dimensions (%d) exceeds the maximum allowed (%d)",
+ 												$3, MAXDIM),
+ 											 parser_errposition(@3)));
+ 					}
+ 				;
+ 
  stmt_exit		: exit_type opt_label opt_exitcond
  					{
  						PLpgSQL_stmt_exit *new;
***************
*** 2063,2068 ****
--- 2127,2133 ----
  				| K_ROW_COUNT
  				| K_ROWTYPE
  				| K_SCROLL
+ 				| K_SLICE
  				| K_SQLSTATE
  				| K_TYPE
  				| K_USE_COLUMN
*** ./src/pl/plpgsql/src/pl_exec.c.orig	2011-01-16 14:18:59.000000000 +0100
--- ./src/pl/plpgsql/src/pl_exec.c	2011-01-21 22:54:46.406350001 +0100
***************
*** 107,112 ****
--- 107,114 ----
  			   PLpgSQL_stmt_fors *stmt);
  static int exec_stmt_forc(PLpgSQL_execstate *estate,
  			   PLpgSQL_stmt_forc *stmt);
+ static int exec_stmt_foreach_a(PLpgSQL_execstate *estate,
+ 				    PLpgSQL_stmt_foreach_a *stmt);
  static int exec_stmt_open(PLpgSQL_execstate *estate,
  			   PLpgSQL_stmt_open *stmt);
  static int exec_stmt_fetch(PLpgSQL_execstate *estate,
***************
*** 1312,1317 ****
--- 1314,1323 ----
  			rc = exec_stmt_forc(estate, (PLpgSQL_stmt_forc *) stmt);
  			break;
  
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			rc = exec_stmt_foreach_a(estate, (PLpgSQL_stmt_foreach_a *) stmt);
+ 			break;
+ 
  		case PLPGSQL_STMT_EXIT:
  			rc = exec_stmt_exit(estate, (PLpgSQL_stmt_exit *) stmt);
  			break;
***************
*** 2028,2033 ****
--- 2034,2295 ----
  
  
  /* ----------
+  * exec_stmt_foreach_a			Implements loop over array
+  *
+  * ----------
+  */
+ static int
+ exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
+ {
+ 	Datum		value;
+ 	bool		isnull;
+ 	Oid		valtype;
+ 	int		nitems;
+ 	int		slice_len = 1;
+ 	Oid 		array_typelem;
+ 	int		i;
+ 	ArrayType	*arr;
+ 	char		*ptr;
+ 	bits8		*arraynullsptr;
+ 	int16		elmlen;
+ 	bool		elmbyval;
+ 	char		elmalign;
+ 	PLpgSQL_datum	*ctrl_var = NULL;		/* be compiler quite */
+ 	bool		found = false;
+ 	int		rc = PLPGSQL_RC_OK;
+ 
+ 	/* get the value of the array expression using array_expr */
+ 	value = exec_eval_expr(estate, stmt->expr, &isnull, &valtype);
+ 	if (isnull)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
+ 				 errmsg("NULL value isn't allowed as parameter of FOREACH-IN")));
+ 
+ 	/* check the type of the expression - must be an array */
+ 	array_typelem = get_element_type(valtype);
+ 
+ 	if (!OidIsValid(array_typelem))
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 				 errmsg("result of expression isn't an array"),
+ 				 errdetail("result of expression is %s",
+ 									format_type_be(valtype))));
+ 
+ 	/* make a copy of the result */
+ 	arr = DatumGetArrayTypePCopy(value);
+ 
+ 	/* Calculate the number of elements we're going to loop through */
+ 	nitems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));
+ 
+ 	/* Gather information about the array */
+ 	ptr = ARR_DATA_PTR(arr);
+ 	arraynullsptr = ARR_NULLBITMAP(arr);
+ 	get_typlenbyvalalign(ARR_ELEMTYPE(arr),
+ 						&elmlen,
+ 						&elmbyval,
+ 						&elmalign);
+ 
+ 	/* Clean up any leftover temporary memory */
+ 	exec_eval_cleanup(estate);
+ 
+ 	/* slice dimension should be less or equal to array dimension */
+ 	if (stmt->slice > ARR_NDIM(arr))
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ 					 errmsg("slice level %d is higher than array dimension",
+ 								    stmt->slice)));
+ 
+  	/*
+ 	 * If the target needs to be an array, check that it actually is,
+ 	 * and that it has a valid dimension for the array we're looping
+ 	 * through.
+ 	 */
+ 	if (stmt->slice > 0)
+ 	{
+ 		if (stmt->var != NULL)
+ 		{
+ 			int	typoid = stmt->var->datatype->typoid;
+ 			int	elmoid = get_element_type(typoid);
+ 
+ 			/* target variable has to be an array */
+ 			if (elmoid == InvalidOid)
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 						 errmsg("target variable \"%s\" for sliced array should be an array type",
+ 									stmt->var->refname),
+ 						 errhint("Value assigned will be of an array type.")));
+ 			/* Determine the number of items in the slice */
+ 			slice_len = ArrayGetNItems(stmt->slice, ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice);
+ 			ctrl_var = estate->datums[stmt->var->dno];
+ 		}
+ 		else
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 					 errmsg("target variable \"%s\" for  sliced array should be an array type",
+ 								stmt->row ? stmt->row->refname : stmt->rec->refname),
+ 					 errhint("Value assigned will be of an array type.")));
+ 	}
+ 	else
+ 	{
+ 		/* target variable has to be a scalar */
+ 		if (stmt->var != NULL)
+ 		{
+ 			int	typoid = stmt->var->datatype->typoid;
+ 			int	elmoid = get_element_type(typoid);
+ 
+ 			/* target variable should be a scalar */
+ 			if (elmoid != InvalidOid)
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 						 errmsg("target variable \"%s\" is  an array",
+ 									stmt->var->refname),
+ 						errhint("Value assigned will be of a scalar type")));
+ 
+ 			ctrl_var = estate->datums[stmt->var->dno];
+ 		}
+ 		else if (stmt->row != NULL)
+ 		{
+ 			ctrl_var = estate->datums[stmt->row->dno];
+ 		}
+ 		else
+ 		{
+ 			ctrl_var = estate->datums[stmt->rec->dno];
+ 		}
+ 	}
+ 
+ 	i = 0;
+ 	/*
+ 	 * process all items in an array
+ 	 */
+ 	while (i < nitems)
+ 	{
+ 		int		j;
+ 		ArrayBuildState *astate = NULL;
+ 
+ 		found = true;				/* looped at least once */
+ 
+ 		/*
+ 		 * Loop through the items in this slice to build up the
+ 		 * array slice to be used through the exec statements.
+ 		 * When slice_len is one, then arr[i] value is stored in
+ 		 * variable "value", else astate is filled.
+ 		 */
+ 		for (j = 0; j < slice_len; j++)
+ 		{
+ 			if (array_get_isnull(arraynullsptr, i))
+ 			{
+ 				isnull = true;
+ 				value = (Datum) 0;
+ 			}
+ 			else
+ 			{
+ 				isnull = false;
+ 				value = fetch_att(ptr, elmbyval, elmlen);
+ 
+ 				ptr = att_addlength_pointer(ptr, elmlen, ptr);
+ 				ptr = (char *) att_align_nominal(ptr, elmalign);
+ 			}
+ 
+ 			/* build an array when is slice higher then one */
+ 			if (stmt->slice > 0)
+ 				astate = accumArrayResult(astate, value, isnull,
+ 									array_typelem,
+ 									CurrentMemoryContext);
+ 			/*
+ 			 * One item was processed. Leave cycle or add next item
+ 			 * to astate (items in slice). Increase number of processed
+ 			 * items per any item in slice.
+ 			 */
+ 			i++;
+ 		}
+ 
+  		/*
+ 		 * store an item in to the variable - we are expecting almost all
+ 		 * iterations will be over scalar values, so we don't create
+ 		 * a fake tuple over scalar and we don't use exec_move_row for
+ 		 * scalars. This is about four times faster.
+ 		 */
+ 		if (astate != NULL)
+ 		{
+ 			Datum sliced_arr;
+ 			bool isnull = false;
+ 
+ 			sliced_arr = makeMdArrayResult(astate, stmt->slice,
+ 									    ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice,
+ 									    ARR_LBOUND(arr) ? ARR_LBOUND(arr) + ARR_NDIM(arr) - stmt->slice : NULL,
+ 										    CurrentMemoryContext, true);
+ 			exec_assign_value(estate, ctrl_var, sliced_arr, valtype, &isnull);
+ 		}
+ 		else
+ 			exec_assign_value(estate, ctrl_var,
+ 						    value, array_typelem, &isnull);
+ 
+ 		/*
+ 		 * Execute the statements
+ 		 */
+ 		rc = exec_stmts(estate, stmt->body);
+ 
+ 		if (rc == PLPGSQL_RC_RETURN)
+ 			break;				/* break out of the loop */
+ 		else if (rc == PLPGSQL_RC_EXIT)
+ 		{
+ 			if (estate->exitlabel == NULL)
+ 				/* unlabelled exit, finish the current loop */
+ 				rc = PLPGSQL_RC_OK;
+ 			else if (stmt->label != NULL &&
+ 					 strcmp(stmt->label, estate->exitlabel) == 0)
+ 			{
+ 				/* labelled exit, matches the current stmt's label */
+ 				estate->exitlabel = NULL;
+ 				rc = PLPGSQL_RC_OK;
+ 			}
+ 
+ 			/*
+ 			 * otherwise, this is a labelled exit that does not match the
+ 			 * current statement's label, if any: return RC_EXIT so that the
+ 			 * EXIT continues to propagate up the stack.
+ 			 */
+ 			break;
+ 		}
+ 		else if (rc == PLPGSQL_RC_CONTINUE)
+ 		{
+ 			if (estate->exitlabel == NULL)
+ 				/* unlabelled continue, so re-run the current loop */
+ 				rc = PLPGSQL_RC_OK;
+ 			else if (stmt->label != NULL &&
+ 					 strcmp(stmt->label, estate->exitlabel) == 0)
+ 			{
+ 				/* label matches named continue, so re-run loop */
+ 				estate->exitlabel = NULL;
+ 				rc = PLPGSQL_RC_OK;
+ 			}
+ 			else
+ 			{
+ 				/*
+ 				 * otherwise, this is a named continue that does not match the
+ 				 * current statement's label, if any: return RC_CONTINUE so
+ 				 * that the CONTINUE will propagate up the stack.
+ 				 */
+ 				break;
+ 			}
+ 		}
+ 	}
+ 
+ 	pfree(arr);
+ 
+ 	/*
+ 	 * Set the FOUND variable to indicate the result of executing the loop
+ 	 * (namely, whether we looped one or more times). This must be set here so
+ 	 * that it does not interfere with the value of the FOUND variable inside
+ 	 * the loop processing itself.
+ 	 */
+ 	exec_set_found(estate, found);
+ 
+ 	return rc;
+ }
+ 
+ 
+ /* ----------
   * exec_stmt_exit			Implements EXIT and CONTINUE
   *
   * This begins the process of exiting / restarting a loop.
***************
*** 3179,3185 ****
  	SPI_cursor_close(portal);
  
  	return rc;
! }
  
  
  /* ----------
--- 3441,3447 ----
  	SPI_cursor_close(portal);
  
  	return rc;
! } 
  
  
  /* ----------
*** ./src/pl/plpgsql/src/pl_funcs.c.orig	2011-01-16 14:18:59.265620438 +0100
--- ./src/pl/plpgsql/src/pl_funcs.c	2011-01-21 22:55:19.266125346 +0100
***************
*** 230,235 ****
--- 230,237 ----
  			return _("FOR over SELECT rows");
  		case PLPGSQL_STMT_FORC:
  			return _("FOR over cursor");
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			return _("FOREACH over array");
  		case PLPGSQL_STMT_EXIT:
  			return "EXIT";
  		case PLPGSQL_STMT_RETURN:
***************
*** 293,298 ****
--- 295,301 ----
  static void dump_close(PLpgSQL_stmt_close *stmt);
  static void dump_perform(PLpgSQL_stmt_perform *stmt);
  static void dump_expr(PLpgSQL_expr *expr);
+ static void dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt);
  
  
  static void
***************
*** 376,381 ****
--- 379,387 ----
  		case PLPGSQL_STMT_PERFORM:
  			dump_perform((PLpgSQL_stmt_perform *) stmt);
  			break;
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			dump_foreach_a((PLpgSQL_stmt_foreach_a *) stmt);
+ 			break;
  		default:
  			elog(ERROR, "unrecognized cmd_type: %d", stmt->cmd_type);
  			break;
***************
*** 596,601 ****
--- 602,625 ----
  }
  
  static void
+ dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt)
+ {
+ 	dump_ind();
+ 	printf("FOREACHA %s", (stmt->rec != NULL) ? stmt->rec->refname :
+ 						    (stmt->row != NULL) ? stmt->row->refname : stmt->var->refname);
+ 	if (stmt->slice != 0)
+ 		printf("SLICE %d ", stmt->slice);
+ 	printf("IN ");
+ 	dump_expr(stmt->expr);
+ 	printf("\n");
+ 
+ 	dump_stmts(stmt->body);
+ 
+ 	dump_ind();
+ 	printf("    ENDOFOREACHA");
+ }
+ 
+ static void
  dump_open(PLpgSQL_stmt_open *stmt)
  {
  	dump_ind();
*** ./src/pl/plpgsql/src/plpgsql.h.orig	2011-01-16 14:18:59.000000000 +0100
--- ./src/pl/plpgsql/src/plpgsql.h	2011-01-21 22:56:26.879893675 +0100
***************
*** 87,92 ****
--- 87,93 ----
  	PLPGSQL_STMT_CASE,
  	PLPGSQL_STMT_LOOP,
  	PLPGSQL_STMT_WHILE,
+ 	PLPGSQL_STMT_FOREACH_A,
  	PLPGSQL_STMT_FORI,
  	PLPGSQL_STMT_FORS,
  	PLPGSQL_STMT_FORC,
***************
*** 427,432 ****
--- 428,447 ----
  
  
  typedef struct
+ {								/* FOREACH item in array loop */
+ 	int			cmd_type;
+ 	int			lineno;
+ 	char	   *label;
+ 	PLpgSQL_var *var;
+ 	PLpgSQL_rec *rec;
+ 	PLpgSQL_row *row;
+ 	PLpgSQL_expr	*expr;
+ 	int	   slice;
+ 	List	   *body;			/* List of statements */
+ } PLpgSQL_stmt_foreach_a;
+ 
+ 
+ typedef struct
  {								/* FOR statement with integer loopvar	*/
  	int			cmd_type;
  	int			lineno;
*** ./src/pl/plpgsql/src/pl_scanner.c.orig	2011-01-16 14:18:59.000000000 +0100
--- ./src/pl/plpgsql/src/pl_scanner.c	2011-01-17 11:31:54.114221247 +0100
***************
*** 77,82 ****
--- 77,83 ----
  	PG_KEYWORD("exit", K_EXIT, RESERVED_KEYWORD)
  	PG_KEYWORD("fetch", K_FETCH, RESERVED_KEYWORD)
  	PG_KEYWORD("for", K_FOR, RESERVED_KEYWORD)
+ 	PG_KEYWORD("foreach", K_FOREACH, RESERVED_KEYWORD)
  	PG_KEYWORD("from", K_FROM, RESERVED_KEYWORD)
  	PG_KEYWORD("get", K_GET, RESERVED_KEYWORD)
  	PG_KEYWORD("if", K_IF, RESERVED_KEYWORD)
***************
*** 133,138 ****
--- 134,140 ----
  	PG_KEYWORD("row_count", K_ROW_COUNT, UNRESERVED_KEYWORD)
  	PG_KEYWORD("rowtype", K_ROWTYPE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("scroll", K_SCROLL, UNRESERVED_KEYWORD)
+ 	PG_KEYWORD("slice", K_SLICE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("sqlstate", K_SQLSTATE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("type", K_TYPE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("use_column", K_USE_COLUMN, UNRESERVED_KEYWORD)
*** ./src/test/regress/expected/plpgsql.out.orig	2011-01-16 14:18:59.000000000 +0100
--- ./src/test/regress/expected/plpgsql.out	2011-01-21 23:05:24.000000000 +0100
***************
*** 4240,4242 ****
--- 4240,4436 ----
  (1 row)
  
  drop function unreserved_test();
+ -- Checking a FOREACH statement
+ create function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[1,2,3,4]);
+ NOTICE:  1
+ NOTICE:  2
+ NOTICE:  3
+ NOTICE:  4
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ NOTICE:  1
+ NOTICE:  2
+ NOTICE:  3
+ NOTICE:  4
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ ERROR:  target variable "x" for sliced array should be an array type
+ HINT:  Value assigned will be of an array type.
+ CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ ERROR:  target variable "x" for sliced array should be an array type
+ HINT:  Value assigned will be of an array type.
+ CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[1,2,3,4]);
+ NOTICE:  {1,2,3,4}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ NOTICE:  {1,2}
+ NOTICE:  {3,4}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ -- higher level of slicing
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 2 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ ERROR:  slice level 2 is higher than array dimension
+ CONTEXT:  PL/pgSQL function "foreach_test" line 4 at FOREACH over array
+ -- ok
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ NOTICE:  {{1,2},{3,4}}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[[1,2]],[[3,4]]]);
+ NOTICE:  {{1,2}}
+ NOTICE:  {{3,4}}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ create type xy_tuple AS (x int, y int);
+ -- iteration over array of records
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare r record;
+ begin
+   foreach r in $1
+   loop
+     raise notice '%', r;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ NOTICE:  (10,20)
+ NOTICE:  (40,69)
+ NOTICE:  (35,78)
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ NOTICE:  (10,20)
+ NOTICE:  (40,69)
+ NOTICE:  (35,78)
+ NOTICE:  (88,76)
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int; y int;
+ begin
+   foreach x, y in $1
+   loop
+     raise notice 'x = %, y = %', x, y;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ NOTICE:  x = 10, y = 20
+ NOTICE:  x = 40, y = 69
+ NOTICE:  x = 35, y = 78
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ NOTICE:  x = 10, y = 20
+ NOTICE:  x = 40, y = 69
+ NOTICE:  x = 35, y = 78
+ NOTICE:  x = 88, y = 76
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ -- slicing over array of composite types
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x xy_tuple[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ NOTICE:  {"(10,20)","(40,69)","(35,78)"}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ NOTICE:  {"(10,20)","(40,69)"}
+ NOTICE:  {"(35,78)","(88,76)"}
+  foreach_test 
+ --------------
+  
+ (1 row)
+ 
+ drop function foreach_test(anyarray);
+ drop type xy_tuple;
*** ./src/test/regress/sql/plpgsql.sql.orig	2011-01-16 14:18:59.000000000 +0100
--- ./src/test/regress/sql/plpgsql.sql	2011-01-17 11:31:54.123222448 +0100
***************
*** 3375,3377 ****
--- 3375,3488 ----
  select unreserved_test();
  
  drop function unreserved_test();
+ 
+ -- Checking a FOREACH statement
+ create function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[1,2,3,4]);
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int;
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[1,2,3,4]);
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ 
+ -- higher level of slicing
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int[];
+ begin
+   foreach x slice 2 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ -- should fail
+ select foreach_test(ARRAY[1,2,3,4]);
+ -- ok
+ select foreach_test(ARRAY[[1,2],[3,4]]);
+ select foreach_test(ARRAY[[[1,2]],[[3,4]]]);
+ 
+ create type xy_tuple AS (x int, y int);
+ 
+ -- iteration over array of records
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare r record;
+ begin
+   foreach r in $1
+   loop
+     raise notice '%', r;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ 
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x int; y int;
+ begin
+   foreach x, y in $1
+   loop
+     raise notice 'x = %, y = %', x, y;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ 
+ -- slicing over array of composite types
+ create or replace function foreach_test(anyarray)
+ returns void as $$
+ declare x xy_tuple[];
+ begin
+   foreach x slice 1 in $1
+   loop
+     raise notice '%', x;
+   end loop;
+   end;
+ $$ language plpgsql;
+ 
+ select foreach_test(ARRAY[(10,20),(40,69),(35,78)]::xy_tuple[]);
+ select foreach_test(ARRAY[[(10,20),(40,69)],[(35,78),(88,76)]]::xy_tuple[]);
+ 
+ drop function foreach_test(anyarray);
+ drop type xy_tuple;

*** ./gram.y.orig	2011-01-16 14:18:59.000000000 +0100
--- ./gram.y	2011-01-21 21:35:21.000000000 +0100
***************
*** 21,26 ****
--- 21,27 ----
  #include "parser/parse_type.h"
  #include "parser/scanner.h"
  #include "parser/scansup.h"
+ #include "utils/array.h"
  
  
  /* Location tracking support --- simpler than bison's default */
***************
*** 134,139 ****
--- 135,141 ----
  			PLpgSQL_datum   *scalar;
  			PLpgSQL_rec     *rec;
  			PLpgSQL_row     *row;
+ 			int	slice;
  		}						forvariable;
  		struct
  		{
***************
*** 178,184 ****
  %type <ival>	assign_var
  %type <var>		cursor_variable
  %type <datum>	decl_cursor_arg
! %type <forvariable>	for_variable
  %type <stmt>	for_control
  
  %type <str>		any_identifier opt_block_label opt_label
--- 180,186 ----
  %type <ival>	assign_var
  %type <var>		cursor_variable
  %type <datum>	decl_cursor_arg
! %type <forvariable>	for_variable foreach_control
  %type <stmt>	for_control
  
  %type <str>		any_identifier opt_block_label opt_label
***************
*** 190,196 ****
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
--- 192,198 ----
  %type <stmt>	stmt_return stmt_raise stmt_execsql
  %type <stmt>	stmt_dynexecute stmt_for stmt_perform stmt_getdiag
  %type <stmt>	stmt_open stmt_fetch stmt_move stmt_close stmt_null
! %type <stmt>	stmt_case stmt_foreach_a
  
  %type <list>	proc_exceptions
  %type <exception_block> exception_sect
***************
*** 264,269 ****
--- 266,272 ----
  %token <keyword>	K_FETCH
  %token <keyword>	K_FIRST
  %token <keyword>	K_FOR
+ %token <keyword>	K_FOREACH
  %token <keyword>	K_FORWARD
  %token <keyword>	K_FROM
  %token <keyword>	K_GET
***************
*** 298,303 ****
--- 301,307 ----
  %token <keyword>	K_ROWTYPE
  %token <keyword>	K_ROW_COUNT
  %token <keyword>	K_SCROLL
+ %token <keyword>	K_SLICE
  %token <keyword>	K_SQLSTATE
  %token <keyword>	K_STRICT
  %token <keyword>	K_THEN
***************
*** 739,744 ****
--- 743,750 ----
  						{ $$ = $1; }
  				| stmt_for
  						{ $$ = $1; }
+ 				| stmt_foreach_a
+ 						{ $$ = $1; }
  				| stmt_exit
  						{ $$ = $1; }
  				| stmt_return
***************
*** 1386,1391 ****
--- 1392,1455 ----
  					}
  				;
  
+ stmt_foreach_a		: opt_block_label K_FOREACH foreach_control K_IN expr_until_loop loop_body
+ 					{
+ 						PLpgSQL_stmt_foreach_a *new = palloc0(sizeof(PLpgSQL_stmt_foreach_a));
+ 						new->cmd_type = PLPGSQL_STMT_FOREACH_A;
+ 						new->lineno = plpgsql_location_to_lineno(@2);
+ 						new->label = $1;
+ 						new->expr = $5;
+ 						new->slice = $3.slice;
+ 						new->body = $6.stmts;
+ 
+ 						if ($3.rec)
+ 						{
+ 							new->rec = $3.rec;
+ 							check_assignable((PLpgSQL_datum *) new->rec, @3);
+ 						}
+ 						else if ($3.row)
+ 						{
+ 							new->row = $3.row;
+ 							check_assignable((PLpgSQL_datum *) new->row, @3);
+ 						}
+ 						else if ($3.scalar)
+ 						{
+ 							Assert($3.scalar->dtype == PLPGSQL_DTYPE_VAR);
+ 							new->var = (PLpgSQL_var *) $3.scalar;
+ 							check_assignable($3.scalar, @3);
+ 
+ 						}
+ 						else
+ 						{
+ 							ereport(ERROR,
+ 									(errcode(ERRCODE_SYNTAX_ERROR),
+ 									 errmsg("loop variable of loop over array must be a record, row variable, scalar variable or list of scalar variables"),
+ 											 parser_errposition(@3)));
+ 						}
+ 
+ 						check_labels($1, $6.end_label, $6.end_label_location);
+ 						$$ = (PLpgSQL_stmt *) new;
+ 					}
+ 				;
+ 
+ foreach_control		: for_variable
+ 					{
+ 						$$ = $1;
+ 						$$.slice = 0;
+ 					}
+ 				| for_variable K_SLICE ICONST
+ 					{
+ 						$$ = $1;
+ 						$$.slice = $3;
+ 						if ($3 < 0 || $3 >= MAXDIM)
+ 							ereport(ERROR,
+ 									(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+ 									 errmsg("number of slicing array dimensions (%d) exceeds the maximum allowed (%d)",
+ 												$3, MAXDIM),
+ 											 parser_errposition(@3)));
+ 					}
+ 				;
+ 
  stmt_exit		: exit_type opt_label opt_exitcond
  					{
  						PLpgSQL_stmt_exit *new;
***************
*** 2063,2068 ****
--- 2127,2133 ----
  				| K_ROW_COUNT
  				| K_ROWTYPE
  				| K_SCROLL
+ 				| K_SLICE
  				| K_SQLSTATE
  				| K_TYPE
  				| K_USE_COLUMN
*** ./pl_exec.c.orig	2011-01-16 14:18:59.000000000 +0100
--- ./pl_exec.c	2011-01-24 12:05:58.599741965 +0100
***************
*** 107,112 ****
--- 107,114 ----
  			   PLpgSQL_stmt_fors *stmt);
  static int exec_stmt_forc(PLpgSQL_execstate *estate,
  			   PLpgSQL_stmt_forc *stmt);
+ static int exec_stmt_foreach_a(PLpgSQL_execstate *estate,
+ 				    PLpgSQL_stmt_foreach_a *stmt);
  static int exec_stmt_open(PLpgSQL_execstate *estate,
  			   PLpgSQL_stmt_open *stmt);
  static int exec_stmt_fetch(PLpgSQL_execstate *estate,
***************
*** 1312,1317 ****
--- 1314,1323 ----
  			rc = exec_stmt_forc(estate, (PLpgSQL_stmt_forc *) stmt);
  			break;
  
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			rc = exec_stmt_foreach_a(estate, (PLpgSQL_stmt_foreach_a *) stmt);
+ 			break;
+ 
  		case PLPGSQL_STMT_EXIT:
  			rc = exec_stmt_exit(estate, (PLpgSQL_stmt_exit *) stmt);
  			break;
***************
*** 2028,2033 ****
--- 2034,2295 ----
  
  
  /* ----------
+  * exec_stmt_foreach_a			Implements loop over array
+  *
+  * ----------
+  */
+ static int
+ exec_stmt_foreach_a(PLpgSQL_execstate *estate, PLpgSQL_stmt_foreach_a *stmt)
+ {
+ 	Datum		value;
+ 	bool		isnull;
+ 	Oid		valtype;
+ 	int		nitems;
+ 	int		slice_len = 0;			/* be compiler quite */
+ 	Oid 		array_typelem;
+ 	int		i;
+ 	ArrayType	*arr;
+ 	char		*ptr;
+ 	bits8		*arraynullsptr;
+ 	int16		elmlen;
+ 	bool		elmbyval;
+ 	char		elmalign;
+ 	PLpgSQL_datum	*ctrl_var = NULL;		/* be compiler quite */
+ 	bool		found = false;
+ 	int		rc = PLPGSQL_RC_OK;
+ 	int		niterations;
+ 	int		offset;
+ 
+ 	/* get the value of the array expression using array_expr */
+ 	value = exec_eval_expr(estate, stmt->expr, &isnull, &valtype);
+ 	if (isnull)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
+ 				 errmsg("NULL value isn't allowed as parameter of FOREACH-IN")));
+ 
+ 	/* check the type of the expression - must be an array */
+ 	array_typelem = get_element_type(valtype);
+ 
+ 	if (!OidIsValid(array_typelem))
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 				 errmsg("result of expression isn't an array"),
+ 				 errdetail("result of expression is %s",
+ 									format_type_be(valtype))));
+ 
+ 	/* make a copy of the result */
+ 	arr = DatumGetArrayTypePCopy(value);
+ 
+ 	/* Calculate the number of elements we're going to loop through */
+ 	nitems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));
+ 
+ 	/* Gather information about the array */
+ 	ptr = ARR_DATA_PTR(arr);
+ 	arraynullsptr = ARR_NULLBITMAP(arr);
+ 	get_typlenbyvalalign(ARR_ELEMTYPE(arr),
+ 						&elmlen,
+ 						&elmbyval,
+ 						&elmalign);
+ 
+ 	/* Clean up any leftover temporary memory */
+ 	exec_eval_cleanup(estate);
+ 
+ 	/* slice dimension should be less or equal to array dimension */
+ 	if (stmt->slice > ARR_NDIM(arr))
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ 					 errmsg("slice level %d is higher than array dimension",
+ 								    stmt->slice)));
+ 
+  	/*
+ 	 * If the target needs to be an array, check that it actually is,
+ 	 * and that it has a valid dimension for the array we're looping
+ 	 * through.
+ 	 */
+ 	if (stmt->slice > 0)
+ 	{
+ 		if (stmt->var != NULL)
+ 		{
+ 			int	typoid = stmt->var->datatype->typoid;
+ 			int	elmoid = get_element_type(typoid);
+ 
+ 			/* target variable has to be an array */
+ 			if (elmoid == InvalidOid)
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 						 errmsg("target variable \"%s\" for sliced array should be an array type",
+ 									stmt->var->refname),
+ 						 errhint("Value assigned will be of an array type.")));
+ 			/* Determine the number of items in the slice */
+ 			slice_len = ArrayGetNItems(stmt->slice, ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice);
+ 			ctrl_var = estate->datums[stmt->var->dno];
+ 		}
+ 		else
+ 			ereport(ERROR,
+ 					(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 					 errmsg("target variable \"%s\" for  sliced array should be an array type",
+ 								stmt->row ? stmt->row->refname : stmt->rec->refname),
+ 					 errhint("Value assigned will be of an array type.")));
+ 
+ 		niterations = nitems / slice_len;
+ 	}
+ 	else
+ 	{
+ 		/* target variable has to be a scalar */
+ 		if (stmt->var != NULL)
+ 		{
+ 			int	typoid = stmt->var->datatype->typoid;
+ 			int	elmoid = get_element_type(typoid);
+ 
+ 			/* target variable should be a scalar */
+ 			if (elmoid != InvalidOid)
+ 				ereport(ERROR,
+ 						(errcode(ERRCODE_DATATYPE_MISMATCH),
+ 						 errmsg("target variable \"%s\" is  an array", 
+ 									stmt->var->refname),
+ 						errhint("Value assigned will be of a scalar type")));
+ 
+ 			ctrl_var = estate->datums[stmt->var->dno];
+ 		}
+ 		else if (stmt->row != NULL)
+ 		{
+ 			ctrl_var = estate->datums[stmt->row->dno];
+ 		}
+ 		else
+ 		{
+ 			ctrl_var = estate->datums[stmt->rec->dno];
+ 		}
+ 
+ 		niterations = nitems;
+ 	}
+ 
+ 	/* 
+ 	 * main loop - iteration over array's items or array's slices 
+ 	 */
+ 	offset = 0;
+ 	for (i = 0; i < niterations; i++)
+ 	{
+ 		/* 
+ 		 * Set a target variable. Construct an subarray (slice) when it is requested.
+ 		 * When slice isn't requested, then target type is same as array element type,
+ 		 * otherwise target type is same as original array.
+ 		 */
+ 		if (stmt->slice > 0)
+ 		{
+ 			int	j;
+ 			ArrayBuildState *astate = NULL;
+ 			Datum slice;
+ 
+ 			for (j = 0; j < slice_len; j++)
+ 			{
+ 				if (array_get_isnull(arraynullsptr, offset++))
+ 				{
+ 					isnull = true;
+ 					value = (Datum) 0;
+ 				}
+ 				else
+ 				{
+ 					isnull = false;
+ 					value = fetch_att(ptr, elmbyval, elmlen);
+ 					ptr = att_addlength_pointer(ptr, elmlen, ptr);
+ 					ptr = (char *) att_align_nominal(ptr, elmalign);
+ 				}
+ 				astate = accumArrayResult(astate, value, isnull, 
+ 									array_typelem, 
+ 									CurrentMemoryContext);
+ 			}
+ 			isnull = false;
+ 			slice = makeMdArrayResult(astate, stmt->slice,
+ 								    ARR_DIMS(arr) + ARR_NDIM(arr) - stmt->slice,
+ 								    ARR_LBOUND(arr) ? ARR_LBOUND(arr) + ARR_NDIM(arr) - stmt->slice : NULL,
+ 									    CurrentMemoryContext, true);
+ 			exec_assign_value(estate, ctrl_var, slice, valtype, &isnull);
+ 		}
+ 		else
+ 		{
+ 			/* get a value */
+ 			if (array_get_isnull(arraynullsptr, offset++))
+ 			{
+ 				isnull = true;
+ 				value = (Datum) 0;
+ 			}
+ 			else
+ 			{
+ 				isnull = false;
+ 				value = fetch_att(ptr, elmbyval, elmlen);
+ 
+ 				ptr = att_addlength_pointer(ptr, elmlen, ptr);
+ 				ptr = (char *) att_align_nominal(ptr, elmalign);
+ 			}
+ 			exec_assign_value(estate, ctrl_var, value, array_typelem, &isnull);
+ 		}
+ 
+ 		/*
+ 		 * Execute the statements
+ 		 */
+ 		rc = exec_stmts(estate, stmt->body);
+ 
+ 		if (rc == PLPGSQL_RC_RETURN)
+ 			break;				/* break out of the loop */
+ 		else if (rc == PLPGSQL_RC_EXIT)
+ 		{
+ 			if (estate->exitlabel == NULL)
+ 				/* unlabelled exit, finish the current loop */
+ 				rc = PLPGSQL_RC_OK;
+ 			else if (stmt->label != NULL &&
+ 					 strcmp(stmt->label, estate->exitlabel) == 0)
+ 			{
+ 				/* labelled exit, matches the current stmt's label */
+ 				estate->exitlabel = NULL;
+ 				rc = PLPGSQL_RC_OK;
+ 			}
+ 
+ 			/*
+ 			 * otherwise, this is a labelled exit that does not match the
+ 			 * current statement's label, if any: return RC_EXIT so that the
+ 			 * EXIT continues to propagate up the stack.
+ 			 */
+ 			break;
+ 		}
+ 		else if (rc == PLPGSQL_RC_CONTINUE)
+ 		{
+ 			if (estate->exitlabel == NULL)
+ 				/* unlabelled continue, so re-run the current loop */
+ 				rc = PLPGSQL_RC_OK;
+ 			else if (stmt->label != NULL &&
+ 					 strcmp(stmt->label, estate->exitlabel) == 0)
+ 			{
+ 				/* label matches named continue, so re-run loop */
+ 				estate->exitlabel = NULL;
+ 				rc = PLPGSQL_RC_OK;
+ 			}
+ 			else
+ 			{
+ 				/*
+ 				 * otherwise, this is a named continue that does not match the
+ 				 * current statement's label, if any: return RC_CONTINUE so
+ 				 * that the CONTINUE will propagate up the stack.
+ 				 */
+ 				break;
+ 			}
+ 		}
+ 	}
+ 
+ 	pfree(arr);
+ 
+ 	/*
+ 	 * Set the FOUND variable to indicate the result of executing the loop
+ 	 * (namely, whether we looped one or more times). This must be set here so
+ 	 * that it does not interfere with the value of the FOUND variable inside
+ 	 * the loop processing itself.
+ 	 */
+ 	exec_set_found(estate, found);
+ 
+ 	return rc;
+ }
+ 
+ 
+ /* ----------
   * exec_stmt_exit			Implements EXIT and CONTINUE
   *
   * This begins the process of exiting / restarting a loop.
***************
*** 3179,3185 ****
  	SPI_cursor_close(portal);
  
  	return rc;
! }
  
  
  /* ----------
--- 3441,3447 ----
  	SPI_cursor_close(portal);
  
  	return rc;
! } 
  
  
  /* ----------
*** ./pl_funcs.c.orig	2011-01-16 14:18:59.000000000 +0100
--- ./pl_funcs.c	2011-01-21 22:55:19.000000000 +0100
***************
*** 230,235 ****
--- 230,237 ----
  			return _("FOR over SELECT rows");
  		case PLPGSQL_STMT_FORC:
  			return _("FOR over cursor");
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			return _("FOREACH over array");
  		case PLPGSQL_STMT_EXIT:
  			return "EXIT";
  		case PLPGSQL_STMT_RETURN:
***************
*** 293,298 ****
--- 295,301 ----
  static void dump_close(PLpgSQL_stmt_close *stmt);
  static void dump_perform(PLpgSQL_stmt_perform *stmt);
  static void dump_expr(PLpgSQL_expr *expr);
+ static void dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt);
  
  
  static void
***************
*** 376,381 ****
--- 379,387 ----
  		case PLPGSQL_STMT_PERFORM:
  			dump_perform((PLpgSQL_stmt_perform *) stmt);
  			break;
+ 		case PLPGSQL_STMT_FOREACH_A:
+ 			dump_foreach_a((PLpgSQL_stmt_foreach_a *) stmt);
+ 			break;
  		default:
  			elog(ERROR, "unrecognized cmd_type: %d", stmt->cmd_type);
  			break;
***************
*** 596,601 ****
--- 602,625 ----
  }
  
  static void
+ dump_foreach_a(PLpgSQL_stmt_foreach_a *stmt)
+ {
+ 	dump_ind();
+ 	printf("FOREACHA %s", (stmt->rec != NULL) ? stmt->rec->refname :
+ 						    (stmt->row != NULL) ? stmt->row->refname : stmt->var->refname);
+ 	if (stmt->slice != 0)
+ 		printf("SLICE %d ", stmt->slice);
+ 	printf("IN ");
+ 	dump_expr(stmt->expr);
+ 	printf("\n");
+ 
+ 	dump_stmts(stmt->body);
+ 
+ 	dump_ind();
+ 	printf("    ENDOFOREACHA");
+ }
+ 
+ static void
  dump_open(PLpgSQL_stmt_open *stmt)
  {
  	dump_ind();
*** ./plpgsql.h.orig	2011-01-16 14:18:59.000000000 +0100
--- ./plpgsql.h	2011-01-21 22:56:26.000000000 +0100
***************
*** 87,92 ****
--- 87,93 ----
  	PLPGSQL_STMT_CASE,
  	PLPGSQL_STMT_LOOP,
  	PLPGSQL_STMT_WHILE,
+ 	PLPGSQL_STMT_FOREACH_A,
  	PLPGSQL_STMT_FORI,
  	PLPGSQL_STMT_FORS,
  	PLPGSQL_STMT_FORC,
***************
*** 427,432 ****
--- 428,447 ----
  
  
  typedef struct
+ {								/* FOREACH item in array loop */
+ 	int			cmd_type;
+ 	int			lineno;
+ 	char	   *label;
+ 	PLpgSQL_var *var;
+ 	PLpgSQL_rec *rec;
+ 	PLpgSQL_row *row;
+ 	PLpgSQL_expr	*expr;
+ 	int	   slice;
+ 	List	   *body;			/* List of statements */
+ } PLpgSQL_stmt_foreach_a;
+ 
+ 
+ typedef struct
  {								/* FOR statement with integer loopvar	*/
  	int			cmd_type;
  	int			lineno;
*** ./pl_scanner.c.orig	2011-01-16 14:18:59.000000000 +0100
--- ./pl_scanner.c	2011-01-17 11:31:54.000000000 +0100
***************
*** 77,82 ****
--- 77,83 ----
  	PG_KEYWORD("exit", K_EXIT, RESERVED_KEYWORD)
  	PG_KEYWORD("fetch", K_FETCH, RESERVED_KEYWORD)
  	PG_KEYWORD("for", K_FOR, RESERVED_KEYWORD)
+ 	PG_KEYWORD("foreach", K_FOREACH, RESERVED_KEYWORD)
  	PG_KEYWORD("from", K_FROM, RESERVED_KEYWORD)
  	PG_KEYWORD("get", K_GET, RESERVED_KEYWORD)
  	PG_KEYWORD("if", K_IF, RESERVED_KEYWORD)
***************
*** 133,138 ****
--- 134,140 ----
  	PG_KEYWORD("row_count", K_ROW_COUNT, UNRESERVED_KEYWORD)
  	PG_KEYWORD("rowtype", K_ROWTYPE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("scroll", K_SCROLL, UNRESERVED_KEYWORD)
+ 	PG_KEYWORD("slice", K_SLICE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("sqlstate", K_SQLSTATE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("type", K_TYPE, UNRESERVED_KEYWORD)
  	PG_KEYWORD("use_column", K_USE_COLUMN, UNRESERVED_KEYWORD)