Issue #487: Add TableSupportsBackwardScan() and TableAmRoutine->amcanbackward

OrioleDB table AM doesn't support BACKWARD SCAN in all cases. A user
needs to declare a cursor with SCROLL option.

Author: za-arthur

src/backend/access/heap/heapam_handler.c

@@ -2917,6 +2917,7 @@ heapam_reloptions(char relkind, Datum reloptions, bool validate)
 
 static const TableAmRoutine heapam_methods = {
 	.type = T_TableAmRoutine,
+	.amcanbackward = true,
 
 	.slot_callbacks = heapam_slot_callbacks,
 	.get_row_ref_type = heapam_get_row_ref_type,

src/backend/commands/portalcmds.c

@@ -134,7 +134,7 @@ PerformCursorOpen(ParseState *pstate, DeclareCursorStmt *cstmt, ParamListInfo pa
 	if (!(portal->cursorOptions & (CURSOR_OPT_SCROLL | CURSOR_OPT_NO_SCROLL)))
 	{
 		if (plan->rowMarks == NIL &&
-			ExecSupportsBackwardScan(plan->planTree))
+			ExecSupportsBackwardScan(plan->planTree, plan->rtable))
 			portal->cursorOptions |= CURSOR_OPT_SCROLL;
 		else
 			portal->cursorOptions |= CURSOR_OPT_NO_SCROLL;

src/backend/executor/execAmi.c

@@ -14,7 +14,9 @@
 
 #include "access/amapi.h"
 #include "access/htup_details.h"
+#include "access/tableam.h"
 #include "executor/execdebug.h"
+#include "catalog/pg_class.h"
 #include "executor/nodeAgg.h"
 #include "executor/nodeAppend.h"
 #include "executor/nodeBitmapAnd.h"
@@ -61,10 +63,12 @@
 #include "nodes/extensible.h"
 #include "nodes/nodeFuncs.h"
 #include "nodes/pathnodes.h"
+#include "parser/parsetree.h"
 #include "utils/rel.h"
 #include "utils/syscache.h"
 
 static bool IndexSupportsBackwardScan(Oid indexid);
+static bool TableSupportsBackwardScan(Oid tableid);
 
 
 /*
@@ -509,7 +513,7 @@ ExecSupportsMarkRestore(Path *pathnode)
  * children do.  Therefore, this routine must be passed a complete plan tree.
  */
 bool
-ExecSupportsBackwardScan(Plan *node)
+ExecSupportsBackwardScan(Plan *node, List *rtable)
 {
 	if (node == NULL)
 		return false;
@@ -526,7 +530,7 @@ ExecSupportsBackwardScan(Plan *node)
 	{
 		case T_Result:
 			if (outerPlan(node) != NULL)
-				return ExecSupportsBackwardScan(outerPlan(node));
+				return ExecSupportsBackwardScan(outerPlan(node), rtable);
 			else
 				return false;
 
@@ -540,7 +544,7 @@ ExecSupportsBackwardScan(Plan *node)
 
 				foreach(l, ((Append *) node)->appendplans)
 				{
-					if (!ExecSupportsBackwardScan((Plan *) lfirst(l)))
+					if (!ExecSupportsBackwardScan((Plan *) lfirst(l), rtable))
 						return false;
 				}
 				/* need not check tlist because Append doesn't evaluate it */
@@ -561,7 +565,7 @@ ExecSupportsBackwardScan(Plan *node)
 			return IndexSupportsBackwardScan(((IndexOnlyScan *) node)->indexid);
 
 		case T_SubqueryScan:
-			return ExecSupportsBackwardScan(((SubqueryScan *) node)->subplan);
+			return ExecSupportsBackwardScan(((SubqueryScan *) node)->subplan, rtable);
 
 		case T_CustomScan:
 			if (((CustomScan *) node)->flags & CUSTOMPATH_SUPPORT_BACKWARD_SCAN)
@@ -571,6 +575,16 @@ ExecSupportsBackwardScan(Plan *node)
 		case T_SeqScan:
 		case T_TidScan:
 		case T_TidRangeScan:
+			{
+				RangeTblEntry *rte;
+
+				Assert(((Scan *) node)->scanrelid > 0 &&
+					   ((Scan *) node)->scanrelid <= list_length(rtable));
+
+				rte = rt_fetch(((Scan *) node)->scanrelid, rtable);
+				return TableSupportsBackwardScan(rte->relid);
+			}
+
 		case T_FunctionScan:
 		case T_ValuesScan:
 		case T_CteScan:
@@ -589,7 +603,7 @@ ExecSupportsBackwardScan(Plan *node)
 
 		case T_LockRows:
 		case T_Limit:
-			return ExecSupportsBackwardScan(outerPlan(node));
+			return ExecSupportsBackwardScan(outerPlan(node), rtable);
 
 		default:
 			return false;
@@ -625,6 +639,34 @@ IndexSupportsBackwardScan(Oid indexid)
 	return result;
 }
 
+/*
+ * An SeqScan, TidScan or TidRangeScan node supports backward scan only if the
+ * table's AM does.
+ */
+static bool
+TableSupportsBackwardScan(Oid tableid)
+{
+	bool		result;
+	HeapTuple	ht_tabrel;
+	Form_pg_class tabrelrec;
+	const TableAmRoutine *amroutine;
+
+	/* Fetch the pg_class tuple of the index relation */
+	ht_tabrel = SearchSysCache1(RELOID, ObjectIdGetDatum(tableid));
+	if (!HeapTupleIsValid(ht_tabrel))
+		elog(ERROR, "cache lookup failed for relation %u", tableid);
+	tabrelrec = (Form_pg_class) GETSTRUCT(ht_tabrel);
+
+	/* Fetch the table AM's API struct */
+	amroutine = GetTableAmRoutineByAmOid(tabrelrec->relam);
+
+	result = amroutine->amcanbackward;
+
+	ReleaseSysCache(ht_tabrel);
+
+	return result;
+}
+
 /*
  * ExecMaterializesOutput - does a plan type materialize its output?
  *

src/backend/executor/spi.c

@@ -1698,7 +1698,8 @@ SPI_cursor_open_internal(const char *name, SPIPlanPtr plan,
 		if (list_length(stmt_list) == 1 &&
 			linitial_node(PlannedStmt, stmt_list)->commandType != CMD_UTILITY &&
 			linitial_node(PlannedStmt, stmt_list)->rowMarks == NIL &&
-			ExecSupportsBackwardScan(linitial_node(PlannedStmt, stmt_list)->planTree))
+			ExecSupportsBackwardScan(linitial_node(PlannedStmt, stmt_list)->planTree,
+									 linitial_node(PlannedStmt, stmt_list)->rtable))
 			portal->cursorOptions |= CURSOR_OPT_SCROLL;
 		else
 			portal->cursorOptions |= CURSOR_OPT_NO_SCROLL;

src/backend/optimizer/plan/planner.c

@@ -426,7 +426,7 @@ standard_planner(Query *parse, const char *query_string, int cursorOptions,
 	 */
 	if (cursorOptions & CURSOR_OPT_SCROLL)
 	{
-		if (!ExecSupportsBackwardScan(top_plan))
+		if (!ExecSupportsBackwardScan(top_plan, root->parse->rtable))
 			top_plan = materialize_finished_plan(top_plan);
 	}
 

src/include/access/tableam.h

@@ -304,6 +304,8 @@ typedef struct TableAmRoutine
 	/* this must be set to T_TableAmRoutine */
 	NodeTag		type;
 
+	/* does AM support backward scanning? */
+	bool		amcanbackward;
 
 	/* ------------------------------------------------------------------------
 	 * Slot related callbacks.

src/include/executor/executor.h

@@ -106,7 +106,7 @@ extern void ExecReScan(PlanState *node);
 extern void ExecMarkPos(PlanState *node);
 extern void ExecRestrPos(PlanState *node);
 extern bool ExecSupportsMarkRestore(struct Path *pathnode);
-extern bool ExecSupportsBackwardScan(Plan *node);
+extern bool ExecSupportsBackwardScan(Plan *node, List *rtable);
 extern bool ExecMaterializesOutput(NodeTag plantype);
 
 /*