Fix assertion failure in cursor_fetcher_rewind

The cursor_fetcher_rewind method assumes that the data node cursor is
rewind either after eof or when there is an associated request. But the
rewind can also occur once the server has generated required number of
rows by joining the relation being scanned with another regular
relation. In this case, the fetch would not have reached eof and there
will be no associated requests as the rows would have been already
loaded into the cursor causing the assertion in cursor_fetcher_rewind
to fail. Fixed that by removing the Assert and updating
cursor_fetcher_rewind to discard the response only if there is an
associated request.

Fixes #5053

tsl/src/remote/cursor_fetcher.c

@@ -371,10 +371,9 @@ cursor_fetcher_rewind(DataFetcher *df)
 	{
 		char sql[64];
 
-		Assert(cursor->state.eof || cursor->state.data_req != NULL);
-
-		if (!cursor->state.eof)
+		if (!cursor->state.eof && cursor->state.data_req != NULL)
 			async_request_discard_response(cursor->state.data_req);
+
 		/* We are beyond the first fetch, so need to rewind the remote end */
 		snprintf(sql, sizeof(sql), "MOVE BACKWARD ALL IN c%u", cursor->id);
 		remote_cursor_exec_cmd(cursor, sql);

tsl/test/shared/expected/dist_fetcher_type-12.out

@@ -285,3 +285,19 @@ SELECT * FROM metrics_dist ORDER BY time, device_id LIMIT 11;
  Fri Dec 31 16:04:00 1999 PST |         1 |  2 |  3 | 1.5 |   
 (11 rows)
 
+-- Verify that cursor fetcher can be rewind before EOF due to an
+-- intermediate JOIN product reaching LIMIT
+SET timescaledb.remote_data_fetcher = 'cursor';
+SELECT * FROM metrics_dist as m
+WHERE EXISTS
+    (SELECT *
+     FROM (SELECT m.time as t
+           FROM metrics_dist a INNER JOIN devices b
+                ON a.device_id = b.device_id
+           LIMIT 50) as subq
+     WHERE subq.t is NULL)
+LIMIT 1;
+ time | device_id | v0 | v1 | v2 | v3 
+------+-----------+----+----+----+----
+(0 rows)
+

tsl/test/shared/expected/dist_fetcher_type-13.out

@@ -285,3 +285,19 @@ SELECT * FROM metrics_dist ORDER BY time, device_id LIMIT 11;
  Fri Dec 31 16:04:00 1999 PST |         1 |  2 |  3 | 1.5 |   
 (11 rows)
 
+-- Verify that cursor fetcher can be rewind before EOF due to an
+-- intermediate JOIN product reaching LIMIT
+SET timescaledb.remote_data_fetcher = 'cursor';
+SELECT * FROM metrics_dist as m
+WHERE EXISTS
+    (SELECT *
+     FROM (SELECT m.time as t
+           FROM metrics_dist a INNER JOIN devices b
+                ON a.device_id = b.device_id
+           LIMIT 50) as subq
+     WHERE subq.t is NULL)
+LIMIT 1;
+ time | device_id | v0 | v1 | v2 | v3 
+------+-----------+----+----+----+----
+(0 rows)
+

tsl/test/shared/expected/dist_fetcher_type-14.out

@@ -285,3 +285,19 @@ SELECT * FROM metrics_dist ORDER BY time, device_id LIMIT 11;
  Fri Dec 31 16:04:00 1999 PST |         1 |  2 |  3 | 1.5 |   
 (11 rows)
 
+-- Verify that cursor fetcher can be rewind before EOF due to an
+-- intermediate JOIN product reaching LIMIT
+SET timescaledb.remote_data_fetcher = 'cursor';
+SELECT * FROM metrics_dist as m
+WHERE EXISTS
+    (SELECT *
+     FROM (SELECT m.time as t
+           FROM metrics_dist a INNER JOIN devices b
+                ON a.device_id = b.device_id
+           LIMIT 50) as subq
+     WHERE subq.t is NULL)
+LIMIT 1;
+ time | device_id | v0 | v1 | v2 | v3 
+------+-----------+----+----+----+----
+(0 rows)
+

tsl/test/shared/expected/dist_fetcher_type-15.out

@@ -286,3 +286,19 @@ SELECT * FROM metrics_dist ORDER BY time, device_id LIMIT 11;
  Fri Dec 31 16:04:00 1999 PST |         1 |  2 |  3 | 1.5 |   
 (11 rows)
 
+-- Verify that cursor fetcher can be rewind before EOF due to an
+-- intermediate JOIN product reaching LIMIT
+SET timescaledb.remote_data_fetcher = 'cursor';
+SELECT * FROM metrics_dist as m
+WHERE EXISTS
+    (SELECT *
+     FROM (SELECT m.time as t
+           FROM metrics_dist a INNER JOIN devices b
+                ON a.device_id = b.device_id
+           LIMIT 50) as subq
+     WHERE subq.t is NULL)
+LIMIT 1;
+ time | device_id | v0 | v1 | v2 | v3 
+------+-----------+----+----+----+----
+(0 rows)
+

tsl/test/shared/sql/dist_fetcher_type.sql.in

@@ -177,3 +177,16 @@ ORDER BY 1 DESC LIMIT 1;
 -- Test copy fetcher when query is aborted before EOF due to LIMIT
 SET timescaledb.remote_data_fetcher = 'copy';
 SELECT * FROM metrics_dist ORDER BY time, device_id LIMIT 11;
+
+-- Verify that cursor fetcher can be rewind before EOF due to an
+-- intermediate JOIN product reaching LIMIT
+SET timescaledb.remote_data_fetcher = 'cursor';
+SELECT * FROM metrics_dist as m
+WHERE EXISTS
+    (SELECT *
+     FROM (SELECT m.time as t
+           FROM metrics_dist a INNER JOIN devices b
+                ON a.device_id = b.device_id
+           LIMIT 50) as subq
+     WHERE subq.t is NULL)
+LIMIT 1;