Fix ReScanForeignScan API to make the parameterized query work

correctly.

Sub-select or correlated queries that use a parameterized plan
uses ReScanForeignScan API.  However, this API is not correctly
rescanning the data.  In the case of parameters, we need to bind
the new param values after processing them and need to re-execute
the query.  The patch does the same.

In passing, refactor some code in this area and rename a few
functions and variables to match the context.

Reported on GitHub through issues #172 by JackWolfskind
and #201 by lukesilvia.

FDW-157, Suraj Kharage, reviewed by Jeevan Ladhe.

Author: jeevanchalke

expected/select.out

@@ -944,9 +944,86 @@ SELECT * FROM enum_t1 ORDER BY id;
 (3 rows)
 
 DROP FOREIGN TABLE enum_t1;
+-- Parameterized queries should work correctly.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT c1, c2 FROM f_test_tbl1
+  WHERE c8 = (SELECT c1 FROM f_test_tbl2 WHERE c1 = (SELECT 20))
+  ORDER BY c1;
+                                           QUERY PLAN                                            
+-------------------------------------------------------------------------------------------------
+ Sort
+   Output: f_test_tbl1.c1, f_test_tbl1.c2
+   Sort Key: f_test_tbl1.c1
+   InitPlan 2 (returns $1)
+     ->  Foreign Scan on public.f_test_tbl2
+           Output: f_test_tbl2.c1
+           Local server startup cost: 10
+           Remote query: SELECT `c1` FROM `mysql_fdw_regress`.`test_tbl2` WHERE ((`c1` = ?))
+           InitPlan 1 (returns $0)
+             ->  Result
+                   Output: 20
+   ->  Foreign Scan on public.f_test_tbl1
+         Output: f_test_tbl1.c1, f_test_tbl1.c2
+         Local server startup cost: 10
+         Remote query: SELECT `c1`, `c2` FROM `mysql_fdw_regress`.`test_tbl1` WHERE ((`c8` = ?))
+(15 rows)
+
+SELECT c1, c2 FROM f_test_tbl1
+  WHERE c8 = (SELECT c1 FROM f_test_tbl2 WHERE c1 = (SELECT 20))
+  ORDER BY c1;
+  c1  |  c2   
+------+-------
+  100 | EMP1
+  400 | EMP4
+  800 | EMP8
+ 1100 | EMP11
+ 1300 | EMP13
+(5 rows)
+
+SELECT * FROM f_test_tbl1
+  WHERE c8 NOT IN (SELECT c1 FROM f_test_tbl2 WHERE c1 = (SELECT 20))
+  ORDER BY c1;
+  c1  |  c2   |    c3    | c4  |     c5     |     c6     |  c7  | c8 
+------+-------+----------+-----+------------+------------+------+----
+  200 | EMP2  | SALESMAN | 600 | 1981-02-20 | 1600.00000 |  300 | 30
+  300 | EMP3  | SALESMAN | 600 | 1981-02-22 | 1250.00000 |  500 | 30
+  500 | EMP5  | SALESMAN | 600 | 1981-09-28 | 1250.00000 | 1400 | 30
+  600 | EMP6  | MANAGER  | 900 | 1981-05-01 | 2850.00000 |      | 30
+  700 | EMP7  | MANAGER  | 900 | 1981-06-09 | 2450.45000 |      | 10
+  900 | EMP9  | HEAD     |     | 1981-11-17 | 5000.00000 |      | 10
+ 1000 | EMP10 | SALESMAN | 600 | 1980-09-08 | 1500.00000 |    0 | 30
+ 1200 | EMP12 | ADMIN    | 600 | 1981-12-03 |  950.00000 |      | 30
+ 1400 | EMP14 | ADMIN    | 700 | 1982-01-23 | 1300.00000 |      | 10
+(9 rows)
+
+-- Check parameterized queries with text/varchar column, should not crash.
+CREATE FOREIGN TABLE f_test_tbl3 (c1 INTEGER, c2 text, c3 text)
+  SERVER mysql_svr OPTIONS (dbname 'mysql_fdw_regress', table_name 'test_tbl2');
+CREATE TABLE local_t1 (c1 INTEGER, c2 text);
+INSERT INTO local_t1 VALUES (1, 'SALES');
+SELECT c1, c2 FROM f_test_tbl3 WHERE c3 = (SELECT 'PUNE'::text) ORDER BY c1;
+ c1 |     c2      
+----+-------------
+ 10 | DEVELOPMENT
+(1 row)
+
+SELECT c1, c2 FROM f_test_tbl2 WHERE c3 = (SELECT 'PUNE'::varchar) ORDER BY c1;
+ c1 |     c2      
+----+-------------
+ 10 | DEVELOPMENT
+(1 row)
+
+SELECT * FROM local_t1 lt1 WHERE lt1.c1 =
+  (SELECT count(*) FROM f_test_tbl3 ft1 WHERE ft1.c2 = lt1.c2) ORDER BY lt1.c1;
+ c1 |  c2   
+----+-------
+  1 | SALES
+(1 row)
+
 -- Cleanup
 DROP TABLE l_test_tbl1;
 DROP TABLE l_test_tbl2;
+DROP TABLE local_t1;
 DROP VIEW smpl_vw;
 DROP VIEW comp_vw;
 DROP VIEW ttest_tbl1_vw;
@@ -959,6 +1036,7 @@ DROP FOREIGN TABLE f_test_tbl2;
 DROP FOREIGN TABLE f_numbers;
 DROP FOREIGN TABLE f_mysql_test;
 DROP FOREIGN TABLE f_enum_t1;
+DROP FOREIGN TABLE f_test_tbl3;
 DROP TYPE size_t;
 DROP FUNCTION test_param_where();
 DROP FUNCTION test_param_where2(int, text);

mysql_fdw.c

@@ -207,7 +207,7 @@ static void process_query_params(ExprContext *econtext,
 								 MYSQL_BIND **mysql_bind_buf,
 								 Oid *param_types);
 
-static void create_cursor(ForeignScanState *node);
+static void bind_stmt_params_and_exec(ForeignScanState *node);
 
 void *mysql_dll_handle = NULL;
 static int wait_timeout = WAIT_TIMEOUT;
@@ -475,7 +475,7 @@ mysqlBeginForeignScan(ForeignScanState *node, int eflags)
 	festate->query = strVal(list_nth(fsplan->fdw_private, 0));
 	festate->retrieved_attrs = list_nth(fsplan->fdw_private, 1);
 	festate->conn = conn;
-	festate->cursor_exists = false;
+	festate->query_executed = false;
 
 #if PG_VERSION_NUM >= 110000
 	festate->temp_cxt = AllocSetContextCreate(estate->es_query_cxt,
@@ -530,13 +530,6 @@ mysqlBeginForeignScan(ForeignScanState *node, int eflags)
 							 &festate->param_values,
 							 &festate->param_types);
 
-	/*
-	 * If this is the first call after Begin or ReScan, we need to create the
-	 * cursor on the remote side.
-	 */
-	if (!festate->cursor_exists)
-		create_cursor(node);
-
 	/* int column_count = mysql_num_fields(festate->meta); */
 
 	/* Set the statement as cursor type */
@@ -579,13 +572,6 @@ mysqlBeginForeignScan(ForeignScanState *node, int eflags)
 	/* Bind the results pointers for the prepare statements */
 	if (mysql_stmt_bind_result(festate->stmt, festate->table->mysql_bind) != 0)
 		mysql_stmt_error_print(festate, "failed to bind the MySQL query");
-
-	/*
-	 * Finally execute the query and result will be placed in the array we
-	 * already bind
-	 */
-	if (mysql_stmt_execute(festate->stmt) != 0)
-		mysql_stmt_error_print(festate, "failed to execute the MySQL query");
 }
 
 /*
@@ -608,6 +594,13 @@ mysqlIterateForeignScan(ForeignScanState *node)
 
 	ExecClearTuple(tupleSlot);
 
+	/*
+	 * If this is the first call after Begin or ReScan, we need to bind the
+	 * params and execute the query.
+	 */
+	if (!festate->query_executed)
+		bind_stmt_params_and_exec(node);
+
 	attid = 0;
 	rc = mysql_stmt_fetch(festate->stmt);
 	if (rc == 0)
@@ -723,12 +716,12 @@ mysqlReScanForeignScan(ForeignScanState *node)
 {
 	MySQLFdwExecState *festate = (MySQLFdwExecState *) node->fdw_state;
 
-	/* If we haven't created the cursor yet, nothing to do. */
-	if (!festate->cursor_exists)
-		return;
+	/*
+	 * Set the query_executed flag to false so that the query will be executed
+	 * in mysqlIterateForeignScan().
+	 */
+	festate->query_executed = false;
 
-	if (mysql_stmt_execute(festate->stmt) != 0)
-		mysql_stmt_error_print(festate, "failed to execute the MySQL query");
 }
 
 /*
@@ -1908,10 +1901,11 @@ process_query_params(ExprContext *econtext,
 }
 
 /*
- * Create cursor for node's query with current parameter values.
+ * Process the query params and bind the same with the statement, if any.
+ * Also, execute the statement.
  */
 static void
-create_cursor(ForeignScanState *node)
+bind_stmt_params_and_exec(ForeignScanState *node)
 {
 	MySQLFdwExecState *festate = (MySQLFdwExecState *) node->fdw_state;
 	ExprContext *econtext = node->ss.ps.ps_ExprContext;
@@ -1941,11 +1935,18 @@ create_cursor(ForeignScanState *node)
 
 		mysql_stmt_bind_param(festate->stmt, mysql_bind_buffer);
 
-		/* Mark the cursor as created, and show no tuples have been retrieved */
-		festate->cursor_exists = true;
-
 		MemoryContextSwitchTo(oldcontext);
 	}
+
+	/*
+	 * Finally, execute the query. The result will be placed in the array we
+	 * already bind.
+	 */
+	if (mysql_stmt_execute(festate->stmt) != 0)
+		mysql_stmt_error_print(festate, "failed to execute the MySQL query");
+
+	/* Mark the query as executed */
+	festate->query_executed = true;
 }
 
 Datum

mysql_fdw.h

@@ -135,7 +135,7 @@ typedef struct MySQLFdwExecState
 	char	   *query;			/* Query string */
 	Relation	rel;			/* relcache entry for the foreign table */
 	List	   *retrieved_attrs;	/* list of target attribute numbers */
-	bool		cursor_exists;	/* have we created the cursor? */
+	bool		query_executed;	/* have we executed the query? */
 	int			numParams;		/* number of parameters passed to query */
 	FmgrInfo   *param_flinfo;	/* output conversion functions for them */
 	List	   *param_exprs;	/* executable expressions for param values */

sql/select.sql

@@ -237,9 +237,35 @@ IMPORT FOREIGN SCHEMA mysql_fdw_regress LIMIT TO (enum_t1) FROM SERVER mysql_svr
 SELECT * FROM enum_t1 ORDER BY id;
 DROP FOREIGN TABLE enum_t1;
 
+-- Parameterized queries should work correctly.
+EXPLAIN (VERBOSE, COSTS OFF)
+SELECT c1, c2 FROM f_test_tbl1
+  WHERE c8 = (SELECT c1 FROM f_test_tbl2 WHERE c1 = (SELECT 20))
+  ORDER BY c1;
+SELECT c1, c2 FROM f_test_tbl1
+  WHERE c8 = (SELECT c1 FROM f_test_tbl2 WHERE c1 = (SELECT 20))
+  ORDER BY c1;
+
+SELECT * FROM f_test_tbl1
+  WHERE c8 NOT IN (SELECT c1 FROM f_test_tbl2 WHERE c1 = (SELECT 20))
+  ORDER BY c1;
+
+-- Check parameterized queries with text/varchar column, should not crash.
+CREATE FOREIGN TABLE f_test_tbl3 (c1 INTEGER, c2 text, c3 text)
+  SERVER mysql_svr OPTIONS (dbname 'mysql_fdw_regress', table_name 'test_tbl2');
+CREATE TABLE local_t1 (c1 INTEGER, c2 text);
+INSERT INTO local_t1 VALUES (1, 'SALES');
+
+SELECT c1, c2 FROM f_test_tbl3 WHERE c3 = (SELECT 'PUNE'::text) ORDER BY c1;
+SELECT c1, c2 FROM f_test_tbl2 WHERE c3 = (SELECT 'PUNE'::varchar) ORDER BY c1;
+
+SELECT * FROM local_t1 lt1 WHERE lt1.c1 =
+  (SELECT count(*) FROM f_test_tbl3 ft1 WHERE ft1.c2 = lt1.c2) ORDER BY lt1.c1;
+
 -- Cleanup
 DROP TABLE l_test_tbl1;
 DROP TABLE l_test_tbl2;
+DROP TABLE local_t1;
 DROP VIEW smpl_vw;
 DROP VIEW comp_vw;
 DROP VIEW ttest_tbl1_vw;
@@ -252,6 +278,7 @@ DROP FOREIGN TABLE f_test_tbl2;
 DROP FOREIGN TABLE f_numbers;
 DROP FOREIGN TABLE f_mysql_test;
 DROP FOREIGN TABLE f_enum_t1;
+DROP FOREIGN TABLE f_test_tbl3;
 DROP TYPE size_t;
 DROP FUNCTION test_param_where();
 DROP FUNCTION test_param_where2(int, text);