Fix the following bugs in master_slave mode && load_balance mode.

We force to replicate PREPARE and DEALLOCATE statement because of the case.

  PREPARE xx AS SELECT 1;
  EXECUTE xx;

If EXECUTE statement was load balanced, it occured the error.

  ERROR:  prepared statement "xx" does not exist

1 files changed, 29 insertions, 1 deletions

diff --git a/pool_process_query.c b/pool_process_query.c
index e7b26f6..f3750a7 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -156,6 +156,9 @@ static PreparedStatement *pending_prepared_stmt = NULL;
 static PreparedStatementList prepared_list; /* prepared statement name list */
 static PreparedStatement *unnamed_statement = NULL;
 static PreparedStatement *unnamed_portal = NULL;
+static int force_replication = 0; /* non 0 if force to replicate query */
+static int prepare_in_session = 0;
+
 static int is_drop_database(char *query);		/* returns non 0 if this is a DROP DATABASE command */
 static void query_ps_status(char *query, POOL_CONNECTION_POOL *backend);		/* show ps status */
 static int detect_deadlock_error(POOL_CONNECTION *master, int major);
@@ -218,6 +221,7 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
 			else	/* no more query(st == 2) */
 			{
 				frontend->no_forward = 0;
+				prepare_in_session = 0;
 				return POOL_CONTINUE;
 			}
 
@@ -611,6 +615,8 @@ static POOL_STATUS Query(POOL_CONNECTION *frontend,
 		{
 			pending_function = add_prepared_list;
 			pending_prepared_stmt = stmt;
+			force_replication = 1;
+			prepare_in_session = 1;
 		}
 	}
 	else if (frontend &&
@@ -647,6 +653,7 @@ static POOL_STATUS Query(POOL_CONNECTION *frontend,
 			return POOL_END;
 		}
 		free(buf);
+		force_replication = 1;
 	}
 
 	if (frontend &&
@@ -675,9 +682,18 @@ static POOL_STATUS Query(POOL_CONNECTION *frontend,
 		string1 = string;
 	}
 
+	/* reset query have to be replicated */
+	if (frontend == NULL && prepare_in_session)
+		force_replication = 1;
+
 	/* load balance trick */
 	if (load_balance_enabled(backend, string1))
 		start_load_balance(backend);
+	else if (force_replication)
+	{
+		replication_was_enabled = REPLICATION;
+		REPLICATION = 1;
+	}
 	else if (MASTER_SLAVE)
 	{
 		master_slave_was_enabled = 1;
@@ -1084,6 +1100,13 @@ static POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
 		master_slave_dml = 0;
 	}
 
+	if (force_replication)
+	{
+		force_replication = 0;
+		REPLICATION = replication_was_enabled;
+		replication_was_enabled = 0;
+	}
+
 #ifdef NOT_USED
 	return ProcessFrontendResponse(frontend, backend);
 #endif
@@ -1958,7 +1981,6 @@ static POOL_STATUS ProcessFrontendResponse(POOL_CONNECTION *frontend,
 					MASTER_SLAVE = 0;
 					master_slave_dml = 1;
 				}
-
 				status = SimpleForwardToBackend(fkind, frontend, backend);
 				if (pool_flush(MASTER(backend)))
 					status = POOL_ERROR;
@@ -3538,6 +3560,12 @@ static PreparedStatement *get_prepared_command_portal_and_statement(char *query)
 	/* skip data type list */
 	while (*query && *query != ')')
 		query++;
+
+	if (!*query)
+	{
+		pool_debug("get_prepared_command_portal_and_statement: could not get statement");
+		return NULL;
+	}
 	query++;
 
 	/* skip spaces */