Support scrollable cursors (ie, 'direction' clause in FETCH) in plpgsql.

Pavel Stehule, reworked a bit by Tom.

doc/src/sgml/plpgsql.sgml

-<!-- $PostgreSQL: pgsql/doc/src/sgml/plpgsql.sgml,v 1.106 2007/04/02 03:49:37 tgl Exp $ -->
+<!-- $PostgreSQL: pgsql/doc/src/sgml/plpgsql.sgml,v 1.107 2007/04/16 17:21:22 tgl Exp $ -->
 
 <chapter id="plpgsql"> 
   <title><application>PL/pgSQL</application> - <acronym>SQL</acronym> Procedural Language</title>
@@ -2364,10 +2364,14 @@ SELECT merge_db(1, 'dennis');
      Another way is to use the cursor declaration syntax,
      which in general is:
 <synopsis>
-<replaceable>name</replaceable> CURSOR <optional> ( <replaceable>arguments</replaceable> ) </optional> FOR <replaceable>query</replaceable>;
+<replaceable>name</replaceable> <optional> <optional> NO </optional> SCROLL </optional> CURSOR <optional> ( <replaceable>arguments</replaceable> ) </optional> FOR <replaceable>query</replaceable>;
 </synopsis>
      (<literal>FOR</> can be replaced by <literal>IS</> for
      <productname>Oracle</productname> compatibility.)
+     If <literal>SCROLL</> is specified, the cursor will be capable of
+     scrolling backward; if <literal>NO SCROLL</> is specified, backward
+     fetches will be rejected; if neither specification appears, it is
+     query-dependent whether backward fetches will be allowed.
      <replaceable>arguments</replaceable>, if specified, is a
      comma-separated list of pairs <literal><replaceable>name</replaceable>
      <replaceable>datatype</replaceable></literal> that define names to be
@@ -2409,7 +2413,7 @@ DECLARE
      <title><command>OPEN FOR</command> <replaceable>query</replaceable></title>
 
 <synopsis>
-OPEN <replaceable>unbound_cursor</replaceable> FOR <replaceable>query</replaceable>;
+OPEN <replaceable>unbound_cursor</replaceable> <optional> <optional> NO </optional> SCROLL </optional> FOR <replaceable>query</replaceable>;
 </synopsis>
 
        <para>
@@ -2422,7 +2426,8 @@ OPEN <replaceable>unbound_cursor</replaceable> FOR <replaceable>query</replaceab
         is treated in the same way as other SQL commands in
         <application>PL/pgSQL</>: <application>PL/pgSQL</>
         variable names are substituted, and the query plan is cached for
-        possible reuse.
+        possible reuse.  The <literal>SCROLL</> and <literal>NO SCROLL</>
+        options have the same meanings as for a bound cursor.
        </para>
 
        <para>
@@ -2437,7 +2442,7 @@ OPEN curs1 FOR SELECT * FROM foo WHERE key = mykey;
      <title><command>OPEN FOR EXECUTE</command></title>
 
 <synopsis>
-OPEN <replaceable>unbound_cursor</replaceable> FOR EXECUTE <replaceable class="command">query_string</replaceable>;
+OPEN <replaceable>unbound_cursor</replaceable> <optional> <optional> NO </optional> SCROLL </optional> FOR EXECUTE <replaceable class="command">query_string</replaceable>;
 </synopsis>
 
          <para>
@@ -2447,8 +2452,10 @@ OPEN <replaceable>unbound_cursor</replaceable> FOR EXECUTE <replaceable class="c
           <type>refcursor</> variable).  The query is specified as a string
           expression, in the same way as in the <command>EXECUTE</command>
           command.  As usual, this gives flexibility so the query can vary
-          from one run to the next.
-       </para>
+          from one run to the next.  The <literal>SCROLL</> and
+          <literal>NO SCROLL</> options have the same meanings as for a bound
+          cursor.
+         </para>
 
        <para>
         An example:
@@ -2473,6 +2480,9 @@ OPEN <replaceable>bound_cursor</replaceable> <optional> ( <replaceable>argument_
           take arguments.  These values will be substituted in the query.
           The query plan for a bound cursor is always considered cacheable;
           there is no equivalent of <command>EXECUTE</command> in this case.
+          Notice that <literal>SCROLL</> and
+          <literal>NO SCROLL</> cannot be specified, as the cursor's scrolling
+          behavior was already determined.
          </para>
 
     <para>
@@ -2513,23 +2523,45 @@ OPEN curs3(42);
      <title><literal>FETCH</></title>
 
 <synopsis>
-FETCH <replaceable>cursor</replaceable> INTO <replaceable>target</replaceable>;
+FETCH <optional> <replaceable>direction</replaceable> FROM </optional> <replaceable>cursor</replaceable> INTO <replaceable>target</replaceable>;
 </synopsis>
 
-         <para>
-          <command>FETCH</command> retrieves the next row from the
-          cursor into a target, which might be a row variable, a record
-          variable, or a comma-separated list of simple variables, just like
-          <command>SELECT INTO</command>.  As with <command>SELECT
-          INTO</command>, the special variable <literal>FOUND</literal> can
-          be checked to see whether a row was obtained or not.
-         </para>
+    <para>
+     <command>FETCH</command> retrieves the next row from the
+     cursor into a target, which might be a row variable, a record
+     variable, or a comma-separated list of simple variables, just like
+     <command>SELECT INTO</command>.  As with <command>SELECT
+     INTO</command>, the special variable <literal>FOUND</literal> can
+     be checked to see whether a row was obtained or not.
+    </para>
 
     <para>
-     An example:
+     The <replaceable>direction</replaceable> clause can be any of the
+     variants allowed in the SQL <xref linkend="sql-fetch"
+     endterm="sql-fetch-title"> command except the ones that can fetch
+     more than one row; namely, it can be
+     <literal>NEXT</>,
+     <literal>PRIOR</>,
+     <literal>FIRST</>,
+     <literal>LAST</>,
+     <literal>ABSOLUTE</> <replaceable>count</replaceable>,
+     <literal>RELATIVE</> <replaceable>count</replaceable>,
+     <literal>FORWARD</>, or
+     <literal>BACKWARD</>.
+     Omitting <replaceable>direction</replaceable> is the same
+     as specifying <literal>NEXT</>.
+     <replaceable>direction</replaceable> values that require moving
+     backward are likely to fail unless the cursor was declared or opened
+     with the <literal>SCROLL</> option.
+    </para>
+
+    <para>
+     Examples:
 <programlisting>
 FETCH curs1 INTO rowvar;
 FETCH curs2 INTO foo, bar, baz;
+FETCH LAST FROM curs3 INTO x, y;
+FETCH RELATIVE -2 FROM curs4 INTO x;
 </programlisting>
        </para>
      </sect3>

src/backend/executor/spi.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/executor/spi.c,v 1.176 2007/04/16 01:14:56 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/executor/spi.c,v 1.177 2007/04/16 17:21:23 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -34,8 +34,7 @@ static int	_SPI_stack_depth = 0;		/* allocated size of _SPI_stack */
 static int	_SPI_connected = -1;
 static int	_SPI_curid = -1;
 
-static void _SPI_prepare_plan(const char *src, SPIPlanPtr plan,
-							  int cursorOptions);
+static void _SPI_prepare_plan(const char *src, SPIPlanPtr plan);
 
 static int _SPI_execute_plan(SPIPlanPtr plan,
 				  Datum *Values, const char *Nulls,
@@ -311,8 +310,9 @@ SPI_execute(const char *src, bool read_only, long tcount)
 
 	memset(&plan, 0, sizeof(_SPI_plan));
 	plan.magic = _SPI_PLAN_MAGIC;
+	plan.cursor_options = 0;
 
-	_SPI_prepare_plan(src, &plan, 0);
+	_SPI_prepare_plan(src, &plan);
 
 	res = _SPI_execute_plan(&plan, NULL, NULL,
 							InvalidSnapshot, InvalidSnapshot,
@@ -423,10 +423,11 @@ SPI_prepare_cursor(const char *src, int nargs, Oid *argtypes,
 
 	memset(&plan, 0, sizeof(_SPI_plan));
 	plan.magic = _SPI_PLAN_MAGIC;
+	plan.cursor_options = cursorOptions;
 	plan.nargs = nargs;
 	plan.argtypes = argtypes;
 
-	_SPI_prepare_plan(src, &plan, cursorOptions);
+	_SPI_prepare_plan(src, &plan);
 
 	/* copy plan to procedure context */
 	result = _SPI_copy_plan(&plan, _SPI_current->procCxt);
@@ -963,15 +964,19 @@ SPI_cursor_open(const char *name, SPIPlanPtr plan,
 					  cplan);
 
 	/*
-	 * Set up options for portal.
+	 * Set up options for portal.  Default SCROLL type is chosen the same
+	 * way as PerformCursorOpen does it.
 	 */
-	portal->cursorOptions &= ~(CURSOR_OPT_SCROLL | CURSOR_OPT_NO_SCROLL);
-	if (list_length(stmt_list) == 1 &&
-		IsA((Node *) linitial(stmt_list), PlannedStmt) &&
-		ExecSupportsBackwardScan(((PlannedStmt *) linitial(stmt_list))->planTree))
-		portal->cursorOptions |= CURSOR_OPT_SCROLL;
-	else
-		portal->cursorOptions |= CURSOR_OPT_NO_SCROLL;
+	portal->cursorOptions = plan->cursor_options;
+	if (!(portal->cursorOptions & (CURSOR_OPT_SCROLL | CURSOR_OPT_NO_SCROLL)))
+	{
+		if (list_length(stmt_list) == 1 &&
+			IsA((Node *) linitial(stmt_list), PlannedStmt) &&
+			ExecSupportsBackwardScan(((PlannedStmt *) linitial(stmt_list))->planTree))
+			portal->cursorOptions |= CURSOR_OPT_SCROLL;
+		else
+			portal->cursorOptions |= CURSOR_OPT_NO_SCROLL;
+	}
 
 	/*
 	 * If told to be read-only, we'd better check for read-only queries.
@@ -1331,14 +1336,15 @@ spi_printtup(TupleTableSlot *slot, DestReceiver *self)
 /*
  * Parse and plan a querystring.
  *
- * At entry, plan->argtypes and plan->nargs must be valid.
+ * At entry, plan->argtypes, plan->nargs, and plan->cursor_options must be
+ * valid.
  *
  * Results are stored into *plan (specifically, plan->plancache_list).
  * Note however that the result trees are all in CurrentMemoryContext
  * and need to be copied somewhere to survive.
  */
 static void
-_SPI_prepare_plan(const char *src, SPIPlanPtr plan, int cursorOptions)
+_SPI_prepare_plan(const char *src, SPIPlanPtr plan)
 {
 	List	   *raw_parsetree_list;
 	List	   *plancache_list;
@@ -1346,6 +1352,7 @@ _SPI_prepare_plan(const char *src, SPIPlanPtr plan, int cursorOptions)
 	ErrorContextCallback spierrcontext;
 	Oid		   *argtypes = plan->argtypes;
 	int			nargs = plan->nargs;
+	int			cursor_options = plan->cursor_options;
 
 	/*
 	 * Increment CommandCounter to see changes made by now.  We must do this
@@ -1384,7 +1391,7 @@ _SPI_prepare_plan(const char *src, SPIPlanPtr plan, int cursorOptions)
 		/* Need a copyObject here to keep parser from modifying raw tree */
 		stmt_list = pg_analyze_and_rewrite(copyObject(parsetree),
 										   src, argtypes, nargs);
-		stmt_list = pg_plan_queries(stmt_list, cursorOptions, NULL, false);
+		stmt_list = pg_plan_queries(stmt_list, cursor_options, NULL, false);
 
 		plansource = (CachedPlanSource *) palloc0(sizeof(CachedPlanSource));
 		cplan = (CachedPlan *) palloc0(sizeof(CachedPlan));
@@ -1926,6 +1933,7 @@ _SPI_copy_plan(SPIPlanPtr plan, MemoryContext parentcxt)
 	newplan->saved = false;
 	newplan->plancache_list = NIL;
 	newplan->plancxt = plancxt;
+	newplan->cursor_options = plan->cursor_options;
 	newplan->nargs = plan->nargs;
 	if (plan->nargs > 0)
 	{
@@ -2000,6 +2008,7 @@ _SPI_save_plan(SPIPlanPtr plan)
 	newplan->saved = true;
 	newplan->plancache_list = NIL;
 	newplan->plancxt = plancxt;
+	newplan->cursor_options = plan->cursor_options;
 	newplan->nargs = plan->nargs;
 	if (plan->nargs > 0)
 	{

src/include/executor/spi_priv.h

@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2007, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/executor/spi_priv.h,v 1.28 2007/03/15 23:12:07 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/executor/spi_priv.h,v 1.29 2007/04/16 17:21:23 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -65,6 +65,7 @@ typedef struct _SPI_plan
 	bool		saved;			/* saved or unsaved plan? */
 	List	   *plancache_list;	/* one CachedPlanSource per parsetree */
 	MemoryContext plancxt;		/* Context containing _SPI_plan and data */
+	int			cursor_options;	/* Cursor options used for planning */
 	int			nargs;			/* number of plan arguments */
 	Oid		   *argtypes;		/* Argument types (NULL if nargs is 0) */
 } _SPI_plan;

src/pl/plpgsql/src/gram.y

@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/gram.y,v 1.99 2007/02/19 03:18:51 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/gram.y,v 1.100 2007/04/16 17:21:23 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -29,7 +29,7 @@ static PLpgSQL_expr		*read_sql_construct(int until,
 static	PLpgSQL_expr	*read_sql_stmt(const char *sqlstart);
 static	PLpgSQL_type	*read_datatype(int tok);
 static	PLpgSQL_stmt	*make_execsql_stmt(const char *sqlstart, int lineno);
-static	PLpgSQL_stmt	*make_fetch_stmt(int lineno, int curvar);
+static	PLpgSQL_stmt_fetch *read_fetch_direction(void);
 static	PLpgSQL_stmt	*make_return_stmt(int lineno);
 static	PLpgSQL_stmt	*make_return_next_stmt(int lineno);
 static	void			 check_assignable(PLpgSQL_datum *datum);
@@ -92,6 +92,7 @@ static	void			 check_labels(const char *start_label,
 		PLpgSQL_exception_block	*exception_block;
 		PLpgSQL_nsitem			*nsitem;
 		PLpgSQL_diag_item		*diagitem;
+		PLpgSQL_stmt_fetch		*fetch;
 }
 
 %type <declhdr> decl_sect
@@ -109,8 +110,8 @@ static	void			 check_labels(const char *start_label,
 %type <expr>	expr_until_then expr_until_loop
 %type <expr>	opt_exitcond
 
-%type <ival>	assign_var cursor_variable
-%type <var>		cursor_varptr
+%type <ival>	assign_var
+%type <var>		cursor_variable
 %type <variable>	decl_cursor_arg
 %type <forvariable>	for_variable
 %type <stmt>	for_control
@@ -139,6 +140,9 @@ static	void			 check_labels(const char *start_label,
 %type <diagitem> getdiag_list_item
 %type <ival>	getdiag_kind getdiag_target
 
+%type <ival>	opt_scrollable
+%type <fetch>   opt_fetch_direction
+
 %type <ival>	lno
 
 		/*
@@ -176,6 +180,7 @@ static	void			 check_labels(const char *start_label,
 %token	K_LOG
 %token	K_LOOP
 %token	K_NEXT
+%token	K_NOSCROLL
 %token	K_NOT
 %token	K_NOTICE
 %token	K_NULL
@@ -188,6 +193,7 @@ static	void			 check_labels(const char *start_label,
 %token	K_RESULT_OID
 %token	K_RETURN
 %token	K_REVERSE
+%token	K_SCROLL
 %token	K_STRICT
 %token	K_THEN
 %token	K_TO
@@ -358,9 +364,9 @@ decl_statement	: decl_varname decl_const decl_datatype decl_notnull decl_defval
 					{
 						plpgsql_ns_rename($2, $4);
 					}
-				| decl_varname K_CURSOR
+				| decl_varname opt_scrollable K_CURSOR
 					{ plpgsql_ns_push(NULL); }
-				  decl_cursor_args decl_is_from decl_cursor_query
+				  decl_cursor_args decl_is_for decl_cursor_query
 					{
 						PLpgSQL_var *new;
 						PLpgSQL_expr *curname_def;
@@ -400,11 +406,26 @@ decl_statement	: decl_varname decl_const decl_datatype decl_notnull decl_defval
 						curname_def->query = pstrdup(buf);
 						new->default_val = curname_def;
 
-						new->cursor_explicit_expr = $6;
-						if ($4 == NULL)
+						new->cursor_explicit_expr = $7;
+						if ($5 == NULL)
 							new->cursor_explicit_argrow = -1;
 						else
-							new->cursor_explicit_argrow = $4->rowno;
+							new->cursor_explicit_argrow = $5->rowno;
+						new->cursor_options = CURSOR_OPT_FAST_PLAN | $2;
+					}
+				;
+
+opt_scrollable :
+					{
+						$$ = 0;
+					}
+				| K_NOSCROLL
+					{
+						$$ = CURSOR_OPT_NO_SCROLL;
+					}
+				| K_SCROLL
+					{
+						$$ = CURSOR_OPT_SCROLL;
 					}
 				;
 
@@ -470,7 +491,7 @@ decl_cursor_arg : decl_varname decl_datatype
 					}
 				;
 
-decl_is_from	:	K_IS |		/* Oracle */
+decl_is_for		:	K_IS |		/* Oracle */
 					K_FOR;		/* ANSI */
 
 decl_aliasitem	: T_WORD
@@ -956,7 +977,7 @@ for_control		:
 														   false,
 														   &tok);
 
-								if (tok == K_BY) 
+								if (tok == K_BY)
 									expr_by = plpgsql_read_expression(K_LOOP, "LOOP");
 								else
 								{
@@ -1311,7 +1332,7 @@ stmt_dynexecute : K_EXECUTE lno
 				;
 
 
-stmt_open		: K_OPEN lno cursor_varptr
+stmt_open		: K_OPEN lno cursor_variable
 					{
 						PLpgSQL_stmt_open *new;
 						int				  tok;
@@ -1320,10 +1341,23 @@ stmt_open		: K_OPEN lno cursor_varptr
 						new->cmd_type = PLPGSQL_STMT_OPEN;
 						new->lineno = $2;
 						new->curvar = $3->varno;
+						new->cursor_options = CURSOR_OPT_FAST_PLAN;
 
 						if ($3->cursor_explicit_expr == NULL)
 						{
+							/* be nice if we could use opt_scrollable here */
 						    tok = yylex();
+							if (tok == K_NOSCROLL)
+							{
+								new->cursor_options |= CURSOR_OPT_NO_SCROLL;
+								tok = yylex();
+							}
+							else if (tok == K_SCROLL)
+							{
+								new->cursor_options |= CURSOR_OPT_SCROLL;
+								tok = yylex();
+							}
+
 							if (tok != K_FOR)
 							{
 								plpgsql_error_lineno = $2;
@@ -1428,9 +1462,30 @@ stmt_open		: K_OPEN lno cursor_varptr
 					}
 				;
 
-stmt_fetch		: K_FETCH lno cursor_variable K_INTO
+stmt_fetch		: K_FETCH lno opt_fetch_direction cursor_variable K_INTO
 					{
-						$$ = make_fetch_stmt($2, $3);
+						PLpgSQL_stmt_fetch *fetch = $3;
+						PLpgSQL_rec	   *rec;
+						PLpgSQL_row	   *row;
+
+						/* We have already parsed everything through the INTO keyword */
+						read_into_target(&rec, &row, NULL);
+
+						if (yylex() != ';')
+							yyerror("syntax error");
+
+						fetch->lineno = $2;
+						fetch->rec		= rec;
+						fetch->row		= row;
+						fetch->curvar	= $4->varno;
+
+						$$ = (PLpgSQL_stmt *)fetch;
+					}
+				;
+
+opt_fetch_direction	:
+					{
+						$$ = read_fetch_direction();
 					}
 				;
 
@@ -1441,7 +1496,7 @@ stmt_close		: K_CLOSE lno cursor_variable ';'
 						new = palloc(sizeof(PLpgSQL_stmt_close));
 						new->cmd_type = PLPGSQL_STMT_CLOSE;
 						new->lineno = $2;
-						new->curvar = $3;
+						new->curvar = $3->varno;
 
 						$$ = (PLpgSQL_stmt *)new;
 					}
@@ -1454,7 +1509,7 @@ stmt_null		: K_NULL ';'
 					}
 				;
 
-cursor_varptr	: T_SCALAR
+cursor_variable	: T_SCALAR
 					{
 						if (yylval.scalar->dtype != PLPGSQL_DTYPE_VAR)
 							yyerror("cursor variable must be a simple variable");
@@ -1471,23 +1526,6 @@ cursor_varptr	: T_SCALAR
 					}
 				;
 
-cursor_variable	: T_SCALAR
-					{
-						if (yylval.scalar->dtype != PLPGSQL_DTYPE_VAR)
-							yyerror("cursor variable must be a simple variable");
-
-						if (((PLpgSQL_var *) yylval.scalar)->datatype->typoid != REFCURSOROID)
-						{
-							plpgsql_error_lineno = plpgsql_scanner_lineno();
-							ereport(ERROR,
-									(errcode(ERRCODE_DATATYPE_MISMATCH),
-									 errmsg("\"%s\" must be of type refcursor",
-											((PLpgSQL_var *) yylval.scalar)->refname)));
-						}
-						$$ = yylval.scalar->dno;
-					}
-				;
-
 exception_sect	:
 					{ $$ = NULL; }
 				| K_EXCEPTION lno
@@ -1959,28 +1997,81 @@ make_execsql_stmt(const char *sqlstart, int lineno)
 }
 
 
-static PLpgSQL_stmt *
-make_fetch_stmt(int lineno, int curvar)
+static PLpgSQL_stmt_fetch *
+read_fetch_direction(void)
 {
-	int					tok;
-	PLpgSQL_rec		   *rec;
-	PLpgSQL_row		   *row;
 	PLpgSQL_stmt_fetch *fetch;
+	int			tok;
+	bool		check_FROM = true;
 
-	/* We have already parsed everything through the INTO keyword */
-	read_into_target(&rec, &row, NULL);
+	/*
+	 * We create the PLpgSQL_stmt_fetch struct here, but only fill in
+	 * the fields arising from the optional direction clause
+	 */
+	fetch = (PLpgSQL_stmt_fetch *) palloc0(sizeof(PLpgSQL_stmt_fetch));
+	fetch->cmd_type = PLPGSQL_STMT_FETCH;
+	/* set direction defaults: */
+	fetch->direction = FETCH_FORWARD;
+	fetch->how_many  = 1;
+	fetch->expr      = NULL;
+
+	/*
+	 * Most of the direction keywords are not plpgsql keywords, so we
+	 * rely on examining yytext ...
+	 */
 	tok = yylex();
-	if (tok != ';')
-		yyerror("syntax error");
+	if (tok == 0)
+		yyerror("unexpected end of function definition");
 
-	fetch = palloc0(sizeof(PLpgSQL_stmt_fetch));
-	fetch->cmd_type = PLPGSQL_STMT_FETCH;
-	fetch->lineno	= lineno;
-	fetch->rec		= rec;
-	fetch->row		= row;
-	fetch->curvar	= curvar;
+	if (pg_strcasecmp(yytext, "next") == 0)
+	{
+		/* use defaults */
+	}
+	else if (pg_strcasecmp(yytext, "prior") == 0)
+	{
+		fetch->direction = FETCH_BACKWARD;
+	}
+	else if (pg_strcasecmp(yytext, "first") == 0)
+	{
+		fetch->direction = FETCH_ABSOLUTE;
+	}
+	else if (pg_strcasecmp(yytext, "last") == 0)
+	{
+		fetch->direction = FETCH_ABSOLUTE;
+		fetch->how_many  = -1;
+	}
+	else if (pg_strcasecmp(yytext, "absolute") == 0)
+	{
+		fetch->direction = FETCH_ABSOLUTE;
+		fetch->expr = plpgsql_read_expression(K_FROM, "FROM");
+		check_FROM = false;
+	}
+	else if (pg_strcasecmp(yytext, "relative") == 0)
+	{
+		fetch->direction = FETCH_RELATIVE;
+		fetch->expr = plpgsql_read_expression(K_FROM, "FROM");
+		check_FROM = false;
+	}
+	else if (pg_strcasecmp(yytext, "forward") == 0)
+	{
+		/* use defaults */
+	}
+	else if (pg_strcasecmp(yytext, "backward") == 0)
+	{
+		fetch->direction = FETCH_BACKWARD;
+	}
+	else
+	{
+		/* Assume there's no direction clause */
+		plpgsql_push_back_token(tok);
+		check_FROM = false;
+	}
+
+	/* check FROM keyword after direction's specification */
+	if (check_FROM && yylex() != K_FROM)
+		yyerror("expected \"FROM\"");
 
-	return (PLpgSQL_stmt *) fetch;
+	return fetch;
 }
 
 

src/pl/plpgsql/src/pl_exec.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/pl_exec.c,v 1.193 2007/04/02 03:49:42 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/pl_exec.c,v 1.194 2007/04/16 17:21:23 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -120,7 +120,7 @@ static void plpgsql_estate_setup(PLpgSQL_execstate *estate,
 static void exec_eval_cleanup(PLpgSQL_execstate *estate);
 
 static void exec_prepare_plan(PLpgSQL_execstate *estate,
-				  PLpgSQL_expr *expr);
+				  PLpgSQL_expr *expr, int cursorOptions);
 static bool exec_simple_check_node(Node *node);
 static void exec_simple_check_plan(PLpgSQL_expr *expr);
 static bool exec_eval_simple_expr(PLpgSQL_execstate *estate,
@@ -2292,7 +2292,7 @@ exec_eval_cleanup(PLpgSQL_execstate *estate)
  */
 static void
 exec_prepare_plan(PLpgSQL_execstate *estate,
-				  PLpgSQL_expr *expr)
+				  PLpgSQL_expr *expr, int cursorOptions)
 {
 	int			i;
 	SPIPlanPtr	plan;
@@ -2317,7 +2317,8 @@ exec_prepare_plan(PLpgSQL_execstate *estate,
 	/*
 	 * Generate and save the plan
 	 */
-	plan = SPI_prepare(expr->query, expr->nparams, argtypes);
+	plan = SPI_prepare_cursor(expr->query, expr->nparams, argtypes,
+							  cursorOptions);
 	if (plan == NULL)
 	{
 		/* Some SPI errors deserve specific error messages */
@@ -2333,7 +2334,7 @@ exec_prepare_plan(PLpgSQL_execstate *estate,
 						 errmsg("cannot begin/end transactions in PL/pgSQL"),
 						 errhint("Use a BEGIN block with an EXCEPTION clause instead.")));
 			default:
-				elog(ERROR, "SPI_prepare failed for \"%s\": %s",
+				elog(ERROR, "SPI_prepare_cursor failed for \"%s\": %s",
 					 expr->query, SPI_result_code_string(SPI_result));
 		}
 	}
@@ -2370,7 +2371,7 @@ exec_stmt_execsql(PLpgSQL_execstate *estate,
 	{
 		ListCell   *l;
 
-		exec_prepare_plan(estate, expr);
+		exec_prepare_plan(estate, expr, 0);
 		stmt->mod_stmt = false;
 		foreach(l, expr->plan->plancache_list)
 		{
@@ -2936,7 +2937,7 @@ exec_stmt_open(PLpgSQL_execstate *estate, PLpgSQL_stmt_open *stmt)
 		 */
 		query = stmt->query;
 		if (query->plan == NULL)
-			exec_prepare_plan(estate, query);
+			exec_prepare_plan(estate, query, stmt->cursor_options);
 	}
 	else if (stmt->dynquery != NULL)
 	{
@@ -2970,9 +2971,9 @@ exec_stmt_open(PLpgSQL_execstate *estate, PLpgSQL_stmt_open *stmt)
 		 * Now we prepare a query plan for it and open a cursor
 		 * ----------
 		 */
-		curplan = SPI_prepare(querystr, 0, NULL);
+		curplan = SPI_prepare_cursor(querystr, 0, NULL, stmt->cursor_options);
 		if (curplan == NULL)
-			elog(ERROR, "SPI_prepare failed for \"%s\": %s",
+			elog(ERROR, "SPI_prepare_cursor failed for \"%s\": %s",
 				 querystr, SPI_result_code_string(SPI_result));
 		portal = SPI_cursor_open(curname, curplan, NULL, NULL,
 								 estate->readonly_func);
@@ -3039,7 +3040,7 @@ exec_stmt_open(PLpgSQL_execstate *estate, PLpgSQL_stmt_open *stmt)
 
 		query = curvar->cursor_explicit_expr;
 		if (query->plan == NULL)
-			exec_prepare_plan(estate, query);
+			exec_prepare_plan(estate, query, curvar->cursor_options);
 	}
 
 	/* ----------
@@ -3103,6 +3104,7 @@ exec_stmt_fetch(PLpgSQL_execstate *estate, PLpgSQL_stmt_fetch *stmt)
 	PLpgSQL_var *curvar = NULL;
 	PLpgSQL_rec *rec = NULL;
 	PLpgSQL_row *row = NULL;
+	long		how_many = stmt->how_many;
 	SPITupleTable *tuptab;
 	Portal		portal;
 	char	   *curname;
@@ -3126,6 +3128,22 @@ exec_stmt_fetch(PLpgSQL_execstate *estate, PLpgSQL_stmt_fetch *stmt)
 				 errmsg("cursor \"%s\" does not exist", curname)));
 	pfree(curname);
 
+	/* Calculate position for FETCH_RELATIVE or FETCH_ABSOLUTE */
+	if (stmt->expr)
+	{
+		bool isnull;
+
+		/* XXX should be doing this in LONG not INT width */
+		how_many = exec_eval_integer(estate, stmt->expr, &isnull);
+
+		if (isnull)
+			ereport(ERROR,
+					(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
+					 errmsg("relative or absolute cursor position is NULL")));
+
+		exec_eval_cleanup(estate);
+	}
+
 	/* ----------
 	 * Determine if we fetch into a record or a row
 	 * ----------
@@ -3141,7 +3159,7 @@ exec_stmt_fetch(PLpgSQL_execstate *estate, PLpgSQL_stmt_fetch *stmt)
 	 * Fetch 1 tuple from the cursor
 	 * ----------
 	 */
-	SPI_cursor_fetch(portal, true, 1);
+	SPI_scroll_cursor_fetch(portal, stmt->direction, how_many);
 	tuptab = SPI_tuptable;
 	n = SPI_processed;
 
@@ -3853,7 +3871,7 @@ exec_eval_expr(PLpgSQL_execstate *estate,
 	 * If first time through, create a plan for this expression.
 	 */
 	if (expr->plan == NULL)
-		exec_prepare_plan(estate, expr);
+		exec_prepare_plan(estate, expr, 0);
 
 	/*
 	 * If this is a simple expression, bypass SPI and use the executor
@@ -3920,7 +3938,7 @@ exec_run_select(PLpgSQL_execstate *estate,
 	 * On the first call for this expression generate the plan
 	 */
 	if (expr->plan == NULL)
-		exec_prepare_plan(estate, expr);
+		exec_prepare_plan(estate, expr, 0);
 
 	/*
 	 * Now build up the values and nulls arguments for SPI_execute_plan()

src/pl/plpgsql/src/plpgsql.h

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/plpgsql.h,v 1.86 2007/03/15 23:12:07 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/plpgsql.h,v 1.87 2007/04/16 17:21:23 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -207,6 +207,7 @@ typedef struct
 	PLpgSQL_expr *default_val;
 	PLpgSQL_expr *cursor_explicit_expr;
 	int			cursor_explicit_argrow;
+	int			cursor_options;
 
 	Datum		value;
 	bool		isnull;
@@ -436,6 +437,7 @@ typedef struct
 	int			cmd_type;
 	int			lineno;
 	int			curvar;
+	int			cursor_options;
 	PLpgSQL_row *returntype;
 	PLpgSQL_expr *argquery;
 	PLpgSQL_expr *query;
@@ -444,12 +446,15 @@ typedef struct
 
 
 typedef struct
-{								/* FETCH curvar INTO statement		*/
+{								/* FETCH statement */
 	int			cmd_type;
 	int			lineno;
-	PLpgSQL_rec *rec;
+	PLpgSQL_rec *rec;			/* target, as record or row */
 	PLpgSQL_row *row;
-	int			curvar;
+	int			curvar;			/* cursor variable to fetch from */
+	FetchDirection direction;	/* fetch direction */
+	int			how_many;		/* count, if constant (expr is NULL) */
+	PLpgSQL_expr *expr;			/* count, if expression */
 } PLpgSQL_stmt_fetch;
 
 

src/pl/plpgsql/src/scan.l

@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/scan.l,v 1.55 2007/01/05 22:20:02 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/pl/plpgsql/src/scan.l,v 1.56 2007/04/16 17:21:23 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -143,6 +143,7 @@ is				{ return K_IS;				}
 log				{ return K_LOG;				}
 loop			{ return K_LOOP;			}
 next			{ return K_NEXT;			}
+no{space}+scroll { return K_NOSCROLL;		}
 not				{ return K_NOT;				}
 notice			{ return K_NOTICE;			}
 null			{ return K_NULL;			}
@@ -155,6 +156,7 @@ result_oid		{ return K_RESULT_OID;		}
 return			{ return K_RETURN;			}
 reverse			{ return K_REVERSE;			}
 row_count		{ return K_ROW_COUNT;		}
+scroll			{ return K_SCROLL;			}
 strict			{ return K_STRICT;	    	}
 then			{ return K_THEN;			}
 to				{ return K_TO;				}

src/test/regress/expected/plpgsql.out

@@ -2934,3 +2934,93 @@ select footest();
 ERROR:  query returned more than one row
 CONTEXT:  PL/pgSQL function "footest" line 4 at execute statement
 drop function footest();
+-- test scrollable cursor support
+create function sc_test() returns setof integer as $$
+declare 
+  c scroll cursor for select f1 from int4_tbl;
+  x integer;
+begin
+  open c;
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch prior from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+select * from sc_test();
+   sc_test   
+-------------
+ -2147483647
+  2147483647
+     -123456
+      123456
+           0
+(5 rows)
+
+create or replace function sc_test() returns setof integer as $$
+declare 
+  c no scroll cursor for select f1 from int4_tbl;
+  x integer;
+begin
+  open c;
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch prior from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+select * from sc_test();  -- fails because of NO SCROLL specification
+ERROR:  cursor can only scan forward
+HINT:  Declare it with SCROLL option to enable backward scan.
+CONTEXT:  PL/pgSQL function "sc_test" line 6 at fetch
+create or replace function sc_test() returns setof integer as $$
+declare 
+  c refcursor;
+  x integer;
+begin
+  open c scroll for select f1 from int4_tbl;
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch prior from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+select * from sc_test();
+   sc_test   
+-------------
+ -2147483647
+  2147483647
+     -123456
+      123456
+           0
+(5 rows)
+
+create or replace function sc_test() returns setof integer as $$
+declare 
+  c refcursor;
+  x integer;
+begin
+  open c scroll for execute 'select f1 from int4_tbl';
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch relative -2 from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+select * from sc_test();
+   sc_test   
+-------------
+ -2147483647
+     -123456
+           0
+(3 rows)
+
+drop function sc_test();

src/test/regress/sql/plpgsql.sql

@@ -2440,3 +2440,75 @@ end$$ language plpgsql;
 select footest();
 
 drop function footest();
+
+-- test scrollable cursor support
+
+create function sc_test() returns setof integer as $$
+declare 
+  c scroll cursor for select f1 from int4_tbl;
+  x integer;
+begin
+  open c;
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch prior from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+
+select * from sc_test();
+
+create or replace function sc_test() returns setof integer as $$
+declare 
+  c no scroll cursor for select f1 from int4_tbl;
+  x integer;
+begin
+  open c;
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch prior from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+
+select * from sc_test();  -- fails because of NO SCROLL specification
+
+create or replace function sc_test() returns setof integer as $$
+declare 
+  c refcursor;
+  x integer;
+begin
+  open c scroll for select f1 from int4_tbl;
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch prior from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+
+select * from sc_test();
+
+create or replace function sc_test() returns setof integer as $$
+declare 
+  c refcursor;
+  x integer;
+begin
+  open c scroll for execute 'select f1 from int4_tbl';
+  fetch last from c into x;
+  while found loop
+    return next x;
+    fetch relative -2 from c into x;
+  end loop;
+  close c;
+end;
+$$ language plpgsql;
+
+select * from sc_test();
+
+drop function sc_test();