sourceTag missed in plan cache

We were not storing sourceTag in CachedPlanSource, so if a cached plan
is used later, that field is broken, and would break resource queue
concurrency limit functionality for example.

Reported by issue #2308 when using JDBC/ODBC

src/backend/commands/prepare.c

@@ -489,6 +489,7 @@ StorePreparedStatement(const char *stmt_name,
 	/* Create a plancache entry */
 	plansource = CreateCachedPlan(raw_parse_tree,
 								  query_string,
+								  sourceTag,
 								  commandTag,
 								  param_types,
 								  num_params,

src/backend/executor/spi.c

@@ -2550,6 +2550,7 @@ _SPI_save_plan(SPIPlanPtr plan)
 
 		newsource = CreateCachedPlan(plansource->raw_parse_tree,
 									 plansource->query_string,
+									 plansource->sourceTag,
 									 plansource->commandTag,
 									 newplan->argtypes,
 									 newplan->nargs,

src/backend/tcop/postgres.c

@@ -2145,6 +2145,7 @@ exec_parse_message(const char *query_string,	/* string to execute */
 
 		unnamed_stmt_psrc = FastCreateCachedPlan(raw_parse_tree,
 												 pstrdup(query_string),
+												 sourceTag,
 												 commandTag,
 												 newParamTypes,
 												 numParams,

src/backend/utils/cache/plancache.c

@@ -109,6 +109,7 @@ InitPlanCache(void)
 CachedPlanSource *
 CreateCachedPlan(Node *raw_parse_tree,
 				 const char *query_string,
+				 NodeTag sourceTag,
 				 const char *commandTag,
 				 Oid *param_types,
 				 int num_params,
@@ -145,6 +146,7 @@ CreateCachedPlan(Node *raw_parse_tree,
 	plansource = (CachedPlanSource *) palloc(sizeof(CachedPlanSource));
 	plansource->raw_parse_tree = copyObject(raw_parse_tree);
 	plansource->query_string = query_string ? pstrdup(query_string) : NULL;
+	plansource->sourceTag = sourceTag;
 	plansource->commandTag = commandTag;		/* no copying needed */
 	if (num_params > 0)
 	{
@@ -207,6 +209,7 @@ CreateCachedPlan(Node *raw_parse_tree,
 CachedPlanSource *
 FastCreateCachedPlan(Node *raw_parse_tree,
 					 char *query_string,
+					 NodeTag sourceTag,
 					 const char *commandTag,
 					 Oid *param_types,
 					 int num_params,
@@ -233,6 +236,7 @@ FastCreateCachedPlan(Node *raw_parse_tree,
 	plansource = (CachedPlanSource *) palloc(sizeof(CachedPlanSource));
 	plansource->raw_parse_tree = raw_parse_tree;
 	plansource->query_string = query_string;
+	plansource->sourceTag = sourceTag;
 	plansource->commandTag = commandTag;		/* no copying needed */
 	plansource->param_types = param_types;
 	plansource->num_params = num_params;

src/include/utils/plancache.h

@@ -92,6 +92,7 @@ typedef struct CachedPlan
 extern void InitPlanCache(void);
 extern CachedPlanSource *CreateCachedPlan(Node *raw_parse_tree,
 				 const char *query_string,
+				 NodeTag sourceTag,
 				 const char *commandTag,
 				 Oid *param_types,
 				 int num_params,
@@ -101,6 +102,7 @@ extern CachedPlanSource *CreateCachedPlan(Node *raw_parse_tree,
 				 bool fixed_result);
 extern CachedPlanSource *FastCreateCachedPlan(Node *raw_parse_tree,
 					 char *query_string,
+					 NodeTag sourceTag,
 					 const char *commandTag,
 					 Oid *param_types,
 					 int num_params,

src/test/isolation2/expected/resource_queue.out

+0:CREATE RESOURCE QUEUE rq_concurrency_test WITH (active_statements = 1);
+CREATE
+0:CREATE role role_concurrency_test RESOURCE QUEUE rq_concurrency_test;
+CREATE
+
+1:SET role role_concurrency_test;
+SET
+1:BEGIN;
+BEGIN
+1:DECLARE c1 CURSOR FOR SELECT 1;
+DECLARE
+
+2:SET role role_concurrency_test;
+SET
+2:PREPARE fooplan AS SELECT 1;
+PREPARE
+2&:EXECUTE fooplan;  <waiting ...>
+
+-- EXECUTE statement(cached plan) will be blocked when the concurrency limit of the resource queue is reached.
+0:SELECT rsqcountvalue FROM gp_toolkit.gp_resqueue_status WHERE rsqname='rq_concurrency_test';
+rsqcountvalue
+-------------
+1            
+(1 row)
+0:SELECT waiting_reason from pg_stat_activity where current_query = 'EXECUTE fooplan;';
+waiting_reason
+--------------
+lock          
+(1 row)
+
+1:END;
+END
+
+2<:  <... completed>
+?column?
+--------
+1       
+(1 row)
+2:END;
+END
+
+0:DROP role role_concurrency_test;
+DROP
+0:DROP RESOURCE QUEUE rq_concurrency_test;
+DROP

src/test/isolation2/isolation2_schedule

 test: commit_transaction_block_checkpoint
 test: pg_views_concurrent_drop
 test: resource_group
+test: resource_queue
 
 test: setup
 # Tests on Append-Optimized tables (row-oriented).

src/test/isolation2/sql/resource_queue.sql

+0:CREATE RESOURCE QUEUE rq_concurrency_test WITH (active_statements = 1);
+0:CREATE role role_concurrency_test RESOURCE QUEUE rq_concurrency_test;
+
+1:SET role role_concurrency_test;
+1:BEGIN;
+1:DECLARE c1 CURSOR FOR SELECT 1;
+
+2:SET role role_concurrency_test;
+2:PREPARE fooplan AS SELECT 1;
+2&:EXECUTE fooplan;
+
+-- EXECUTE statement(cached plan) will be blocked when the concurrency limit of the resource queue is reached.
+0:SELECT rsqcountvalue FROM gp_toolkit.gp_resqueue_status WHERE rsqname='rq_concurrency_test';
+0:SELECT waiting_reason from pg_stat_activity where current_query = 'EXECUTE fooplan;';
+
+1:END;
+
+2<:
+2:END;
+
+0:DROP role role_concurrency_test;
+0:DROP RESOURCE QUEUE rq_concurrency_test;