Rewind tuple store to fix scrollable with hold cursor fetches (#7014)

We need to rewind the tuplestorestate's tuple index to get correct
results on fetching scrollable with hold cursors.


`PersistHoldablePortal` is responsible for persisting out
tuplestorestate inside a with hold cursor before commiting a
transaction.

It rewinds the cursor like below (`ExecutorRewindcalls` calls `rescan`):
```c
if (portal->cursorOptions & CURSOR_OPT_SCROLL)
{
  ExecutorRewind(queryDesc);
}
```

At the end, it adjusts tuple index for holdStore in the portal properly.
```c
if (portal->cursorOptions & CURSOR_OPT_SCROLL)
{
         if (!tuplestore_skiptuples(portal->holdStore,
	                                         portal->portalPos,
	                                         true))
	    elog(ERROR, "unexpected end of tuple stream");
}
```

DESCRIPTION: Fixes incorrect results on fetching scrollable with hold
cursors.

Fixes https://github.com/citusdata/citus/issues/7010

src/backend/distributed/commands/utility_hook.c

@@ -186,7 +186,9 @@ multi_ProcessUtility(PlannedStmt *pstmt,
 		IsA(parsetree, ExecuteStmt) ||
 		IsA(parsetree, PrepareStmt) ||
 		IsA(parsetree, DiscardStmt) ||
-		IsA(parsetree, DeallocateStmt))
+		IsA(parsetree, DeallocateStmt) ||
+		IsA(parsetree, DeclareCursorStmt) ||
+		IsA(parsetree, FetchStmt))
 	{
 		/*
 		 * Skip additional checks for common commands that do not have any

src/backend/distributed/executor/citus_custom_scan.c

@@ -820,7 +820,19 @@ CitusEndScan(CustomScanState *node)
  */
 static void
 CitusReScan(CustomScanState *node)
-{ }
+{
+	if (node->ss.ps.ps_ResultTupleSlot)
+	{
+		ExecClearTuple(node->ss.ps.ps_ResultTupleSlot);
+	}
+	ExecScanReScan(&node->ss);
+
+	CitusScanState *scanState = (CitusScanState *) node;
+	if (scanState->tuplestorestate)
+	{
+		tuplestore_rescan(scanState->tuplestorestate);
+	}
+}
 
 
 /*

src/test/regress/expected/multi_utility_statements.out

@@ -254,6 +254,76 @@ FETCH FORWARD 3 FROM holdCursor;
  1 | 19
 (3 rows)
 
+CLOSE holdCursor;
+-- Test DECLARE CURSOR .. WITH HOLD inside transaction block
+BEGIN;
+DECLARE holdCursor CURSOR WITH HOLD FOR
+		SELECT * FROM cursor_me WHERE x = 1 ORDER BY y;
+FETCH 3 FROM holdCursor;
+ x | y
+---------------------------------------------------------------------
+ 1 | 10
+ 1 | 11
+ 1 | 12
+(3 rows)
+
+FETCH BACKWARD 3 FROM holdCursor;
+ x | y
+---------------------------------------------------------------------
+ 1 | 11
+ 1 | 10
+(2 rows)
+
+FETCH FORWARD 3 FROM holdCursor;
+ x | y
+---------------------------------------------------------------------
+ 1 | 10
+ 1 | 11
+ 1 | 12
+(3 rows)
+
+COMMIT;
+FETCH 3 FROM holdCursor;
+ x | y
+---------------------------------------------------------------------
+ 1 | 13
+ 1 | 14
+ 1 | 15
+(3 rows)
+
+CLOSE holdCursor;
+-- Test DECLARE NO SCROLL CURSOR .. WITH HOLD inside transaction block
+BEGIN;
+DECLARE holdCursor NO SCROLL CURSOR WITH HOLD FOR
+		SELECT * FROM cursor_me WHERE x = 1 ORDER BY y;
+FETCH 3 FROM holdCursor;
+ x | y
+---------------------------------------------------------------------
+ 1 | 10
+ 1 | 11
+ 1 | 12
+(3 rows)
+
+FETCH FORWARD 3 FROM holdCursor;
+ x | y
+---------------------------------------------------------------------
+ 1 | 13
+ 1 | 14
+ 1 | 15
+(3 rows)
+
+COMMIT;
+FETCH 3 FROM holdCursor;
+ x | y
+---------------------------------------------------------------------
+ 1 | 16
+ 1 | 17
+ 1 | 18
+(3 rows)
+
+FETCH BACKWARD 3 FROM holdCursor;
+ERROR:  cursor can only scan forward
+HINT:  Declare it with SCROLL option to enable backward scan.
 CLOSE holdCursor;
 -- Test DECLARE CURSOR .. WITH HOLD with parameter
 CREATE OR REPLACE FUNCTION declares_cursor(p int)

src/test/regress/sql/multi_utility_statements.sql

@@ -137,6 +137,30 @@ FETCH FORWARD 3 FROM holdCursor;
 
 CLOSE holdCursor;
 
+-- Test DECLARE CURSOR .. WITH HOLD inside transaction block
+BEGIN;
+DECLARE holdCursor CURSOR WITH HOLD FOR
+		SELECT * FROM cursor_me WHERE x = 1 ORDER BY y;
+FETCH 3 FROM holdCursor;
+FETCH BACKWARD 3 FROM holdCursor;
+FETCH FORWARD 3 FROM holdCursor;
+COMMIT;
+
+FETCH 3 FROM holdCursor;
+CLOSE holdCursor;
+
+-- Test DECLARE NO SCROLL CURSOR .. WITH HOLD inside transaction block
+BEGIN;
+DECLARE holdCursor NO SCROLL CURSOR WITH HOLD FOR
+		SELECT * FROM cursor_me WHERE x = 1 ORDER BY y;
+FETCH 3 FROM holdCursor;
+FETCH FORWARD 3 FROM holdCursor;
+COMMIT;
+
+FETCH 3 FROM holdCursor;
+FETCH BACKWARD 3 FROM holdCursor;
+CLOSE holdCursor;
+
 -- Test DECLARE CURSOR .. WITH HOLD with parameter
 CREATE OR REPLACE FUNCTION declares_cursor(p int)
 RETURNS void AS $$