Add a new column to pg_am to specify whether an index AM supports backward

scanning; GiST and GIN do not, and it seems like too much trouble to make
them do so.  By teaching ExecSupportsBackwardScan() about this restriction,
we ensure that the planner will protect a scroll cursor from the problem
by adding a Materialize node.

In passing, fix another longstanding bug in the same area: backwards scan of
a plan with set-returning functions in the targetlist did not work either,
since the TupFromTlist expansion code pays no attention to direction (and
has no way to run a SRF backwards anyway).  Again the fix is to make
ExecSupportsBackwardScan check this restriction.

Also adjust the index AM API specification to note that mark/restore support
is unnecessary if the AM can't produce ordered output.

doc/src/sgml/catalogs.sgml

-<!-- $PostgreSQL: pgsql/doc/src/sgml/catalogs.sgml,v 2.178 2008/10/06 13:59:37 tgl Exp $ -->
+<!-- $PostgreSQL: pgsql/doc/src/sgml/catalogs.sgml,v 2.179 2008/10/17 22:10:29 tgl Exp $ -->
 <!--
  Documentation of the system catalogs, directed toward PostgreSQL developers
  -->
@@ -401,6 +401,13 @@
       <entry>Does the access method support ordered scans?</entry>
      </row>
 
+     <row>
+      <entry><structfield>amcanbackward</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry></entry>
+      <entry>Does the access method support backward scanning?</entry>
+     </row>
+
      <row>
       <entry><structfield>amcanunique</structfield></entry>
       <entry><type>bool</type></entry>

doc/src/sgml/indexam.sgml

-<!-- $PostgreSQL: pgsql/doc/src/sgml/indexam.sgml,v 2.27 2008/08/14 18:47:58 tgl Exp $ -->
+<!-- $PostgreSQL: pgsql/doc/src/sgml/indexam.sgml,v 2.28 2008/10/17 22:10:29 tgl Exp $ -->
 
 <chapter id="indexam">
  <title>Index Access Method Interface Definition</title>
@@ -474,15 +474,20 @@ amrestrpos (IndexScanDesc scan);
    normally would.  (This will only occur for access
    methods that advertise they support ordered scans.)  After the
    first call, <function>amgettuple</> must be prepared to advance the scan in
-   either direction from the most recently returned entry.
+   either direction from the most recently returned entry.  (But if
+   <structname>pg_am</>.<structfield>amcanbackward</> is false, all subsequent
+   calls will have the same direction as the first one.)
   </para>
 
   <para>
-   The access method must support <quote>marking</> a position in a scan
-   and later returning to the marked position.  The same position might be
-   restored multiple times.  However, only one position need be remembered
-   per scan; a new <function>ammarkpos</> call overrides the previously
-   marked position.
+   Access methods that support ordered scans must support <quote>marking</> a
+   position in a scan and later returning to the marked position.  The same
+   position might be restored multiple times.  However, only one position need
+   be remembered per scan; a new <function>ammarkpos</> call overrides the
+   previously marked position.  An access method that does not support
+   ordered scans should still provide mark and restore functions in
+   <structname>pg_am</>, but it is sufficient to have them throw errors if
+   called.
   </para>
 
   <para>

src/backend/executor/execAmi.c

@@ -6,7 +6,7 @@
  * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- *	$PostgreSQL: pgsql/src/backend/executor/execAmi.c,v 1.99 2008/10/04 21:56:53 tgl Exp $
+ *	$PostgreSQL: pgsql/src/backend/executor/execAmi.c,v 1.100 2008/10/17 22:10:29 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -42,6 +42,12 @@
 #include "executor/nodeValuesscan.h"
 #include "executor/nodeCtescan.h"
 #include "executor/nodeWorktablescan.h"
+#include "nodes/nodeFuncs.h"
+#include "utils/syscache.h"
+
+
+static bool TargetListSupportsBackwardScan(List *targetlist);
+static bool IndexSupportsBackwardScan(Oid indexid);
 
 
 /*
@@ -390,7 +396,8 @@ ExecSupportsBackwardScan(Plan *node)
 	{
 		case T_Result:
 			if (outerPlan(node) != NULL)
-				return ExecSupportsBackwardScan(outerPlan(node));
+				return ExecSupportsBackwardScan(outerPlan(node)) &&
+					TargetListSupportsBackwardScan(node->targetlist);
 			else
 				return false;
 
@@ -403,29 +410,85 @@ ExecSupportsBackwardScan(Plan *node)
 					if (!ExecSupportsBackwardScan((Plan *) lfirst(l)))
 						return false;
 				}
+				/* need not check tlist because Append doesn't evaluate it */
 				return true;
 			}
 
 		case T_SeqScan:
-		case T_IndexScan:
 		case T_TidScan:
 		case T_FunctionScan:
 		case T_ValuesScan:
 		case T_CteScan:
 		case T_WorkTableScan:
-			return true;
+			return TargetListSupportsBackwardScan(node->targetlist);
+
+		case T_IndexScan:
+			return IndexSupportsBackwardScan(((IndexScan *) node)->indexid) &&
+				TargetListSupportsBackwardScan(node->targetlist);
 
 		case T_SubqueryScan:
-			return ExecSupportsBackwardScan(((SubqueryScan *) node)->subplan);
+			return ExecSupportsBackwardScan(((SubqueryScan *) node)->subplan) &&
+				TargetListSupportsBackwardScan(node->targetlist);
 
 		case T_Material:
 		case T_Sort:
+			/* these don't evaluate tlist */
 			return true;
 
 		case T_Limit:
+			/* doesn't evaluate tlist */
 			return ExecSupportsBackwardScan(outerPlan(node));
 
 		default:
 			return false;
 	}
 }
+
+/*
+ * If the tlist contains set-returning functions, we can't support backward
+ * scan, because the TupFromTlist code is direction-ignorant.
+ */
+static bool
+TargetListSupportsBackwardScan(List *targetlist)
+{
+	if (expression_returns_set((Node *) targetlist))
+		return false;
+	return true;
+}
+
+/*
+ * An IndexScan node supports backward scan only if the index's AM does.
+ */
+static bool
+IndexSupportsBackwardScan(Oid indexid)
+{
+	bool		result;
+	HeapTuple	ht_idxrel;
+	HeapTuple	ht_am;
+	Form_pg_class idxrelrec;
+	Form_pg_am	amrec;
+
+	/* Fetch the pg_class tuple of the index relation */
+	ht_idxrel = SearchSysCache(RELOID,
+							   ObjectIdGetDatum(indexid),
+							   0, 0, 0);
+	if (!HeapTupleIsValid(ht_idxrel))
+		elog(ERROR, "cache lookup failed for relation %u", indexid);
+	idxrelrec = (Form_pg_class) GETSTRUCT(ht_idxrel);
+
+	/* Fetch the pg_am tuple of the index' access method */
+	ht_am = SearchSysCache(AMOID,
+						   ObjectIdGetDatum(idxrelrec->relam),
+						   0, 0, 0);
+	if (!HeapTupleIsValid(ht_am))
+		elog(ERROR, "cache lookup failed for access method %u",
+			 idxrelrec->relam);
+	amrec = (Form_pg_am) GETSTRUCT(ht_am);
+
+	result = amrec->amcanbackward;
+
+	ReleaseSysCache(ht_idxrel);
+	ReleaseSysCache(ht_am);
+
+	return result;
+}

src/include/catalog/catversion.h

@@ -37,7 +37,7 @@
  * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.498 2008/10/14 17:12:33 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.499 2008/10/17 22:10:30 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	200810141
+#define CATALOG_VERSION_NO	200810171
 
 #endif

src/include/catalog/pg_am.h

@@ -8,7 +8,7 @@
  * Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $PostgreSQL: pgsql/src/include/catalog/pg_am.h,v 1.58 2008/09/15 18:43:41 tgl Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_am.h,v 1.59 2008/10/17 22:10:30 tgl Exp $
  *
  * NOTES
  *		the genbki.sh script reads this file and generates .bki
@@ -41,6 +41,7 @@ CATALOG(pg_am,2601)
 	int2		amsupport;		/* total number of support functions that this
 								 * AM uses */
 	bool		amcanorder;		/* does AM support ordered scan results? */
+	bool		amcanbackward;	/* does AM support backward scan? */
 	bool		amcanunique;	/* does AM support UNIQUE indexes? */
 	bool		amcanmulticol;	/* does AM support multi-column indexes? */
 	bool		amoptionalkey;	/* can query omit key for the first column? */
@@ -75,48 +76,49 @@ typedef FormData_pg_am *Form_pg_am;
  *		compiler constants for pg_am
  * ----------------
  */
-#define Natts_pg_am						25
+#define Natts_pg_am						26
 #define Anum_pg_am_amname				1
 #define Anum_pg_am_amstrategies			2
 #define Anum_pg_am_amsupport			3
 #define Anum_pg_am_amcanorder			4
-#define Anum_pg_am_amcanunique			5
-#define Anum_pg_am_amcanmulticol		6
-#define Anum_pg_am_amoptionalkey		7
-#define Anum_pg_am_amindexnulls			8
-#define Anum_pg_am_amsearchnulls		9
-#define Anum_pg_am_amstorage			10
-#define Anum_pg_am_amclusterable		11
-#define Anum_pg_am_amkeytype			12
-#define Anum_pg_am_aminsert				13
-#define Anum_pg_am_ambeginscan			14
-#define Anum_pg_am_amgettuple			15
-#define Anum_pg_am_amgetbitmap			16
-#define Anum_pg_am_amrescan				17
-#define Anum_pg_am_amendscan			18
-#define Anum_pg_am_ammarkpos			19
-#define Anum_pg_am_amrestrpos			20
-#define Anum_pg_am_ambuild				21
-#define Anum_pg_am_ambulkdelete			22
-#define Anum_pg_am_amvacuumcleanup		23
-#define Anum_pg_am_amcostestimate		24
-#define Anum_pg_am_amoptions			25
+#define Anum_pg_am_amcanbackward		5
+#define Anum_pg_am_amcanunique			6
+#define Anum_pg_am_amcanmulticol		7
+#define Anum_pg_am_amoptionalkey		8
+#define Anum_pg_am_amindexnulls			9
+#define Anum_pg_am_amsearchnulls		10
+#define Anum_pg_am_amstorage			11
+#define Anum_pg_am_amclusterable		12
+#define Anum_pg_am_amkeytype			13
+#define Anum_pg_am_aminsert				14
+#define Anum_pg_am_ambeginscan			15
+#define Anum_pg_am_amgettuple			16
+#define Anum_pg_am_amgetbitmap			17
+#define Anum_pg_am_amrescan				18
+#define Anum_pg_am_amendscan			19
+#define Anum_pg_am_ammarkpos			20
+#define Anum_pg_am_amrestrpos			21
+#define Anum_pg_am_ambuild				22
+#define Anum_pg_am_ambulkdelete			23
+#define Anum_pg_am_amvacuumcleanup		24
+#define Anum_pg_am_amcostestimate		25
+#define Anum_pg_am_amoptions			26
 
 /* ----------------
  *		initial contents of pg_am
  * ----------------
  */
 
-DATA(insert OID = 403 (  btree	5 1 t t t t t t f t 0 btinsert btbeginscan btgettuple btgetbitmap btrescan btendscan btmarkpos btrestrpos btbuild btbulkdelete btvacuumcleanup btcostestimate btoptions ));
+DATA(insert OID = 403 (  btree	5 1 t t t t t t t f t 0 btinsert btbeginscan btgettuple btgetbitmap btrescan btendscan btmarkpos btrestrpos btbuild btbulkdelete btvacuumcleanup btcostestimate btoptions ));
 DESCR("b-tree index access method");
 #define BTREE_AM_OID 403
-DATA(insert OID = 405 (  hash	1 1 f f f f f f f f 23 hashinsert hashbeginscan hashgettuple hashgetbitmap hashrescan hashendscan hashmarkpos hashrestrpos hashbuild hashbulkdelete hashvacuumcleanup hashcostestimate hashoptions ));
+DATA(insert OID = 405 (  hash	1 1 f t f f f f f f f 23 hashinsert hashbeginscan hashgettuple hashgetbitmap hashrescan hashendscan hashmarkpos hashrestrpos hashbuild hashbulkdelete hashvacuumcleanup hashcostestimate hashoptions ));
 DESCR("hash index access method");
 #define HASH_AM_OID 405
-DATA(insert OID = 783 (  gist	0 7 f f t t t t t t 0 gistinsert gistbeginscan gistgettuple gistgetbitmap gistrescan gistendscan gistmarkpos gistrestrpos gistbuild gistbulkdelete gistvacuumcleanup gistcostestimate gistoptions ));
+DATA(insert OID = 783 (  gist	0 7 f f f t t t t t t 0 gistinsert gistbeginscan gistgettuple gistgetbitmap gistrescan gistendscan gistmarkpos gistrestrpos gistbuild gistbulkdelete gistvacuumcleanup gistcostestimate gistoptions ));
 DESCR("GiST index access method");
 #define GIST_AM_OID 783
-DATA(insert OID = 2742 (  gin	0 5 f f t t f f t f 0 gininsert ginbeginscan gingettuple gingetbitmap ginrescan ginendscan ginmarkpos ginrestrpos ginbuild ginbulkdelete ginvacuumcleanup gincostestimate ginoptions ));
+DATA(insert OID = 2742 (  gin	0 5 f f f t t f f t f 0 gininsert ginbeginscan gingettuple gingetbitmap ginrescan ginendscan ginmarkpos ginrestrpos ginbuild ginbulkdelete ginvacuumcleanup gincostestimate ginoptions ));
 DESCR("GIN index access method");
 #define GIN_AM_OID 2742