pgpool2.git / summary / log / commit / refs
commit a64bca180cae5ecced4e9f790eeb0a9e459acac2
Author: Toshihiro Kitagawa <kitagawa at sraoss.co.jp>
Date: Fri Jul 09 01:08:51 2010 +0000
Change pool_process_query.c, pool_proto_modules.c and child.c so that
they use session_context and query_context.
child.c | 2 -
pool_process_query.c | 271 ++++++++++++------------------------
pool_proto_modules.c | 380 ++++++++++++++++++++-------------------------------
pool_proto_modules.h | 30 ----
4 files changed, 233 insertions(+), 450 deletions(-)
diff --git a/child.c b/child.c
index c2553b3fb..c3a21ea1f 100644
--- a/child.c
+++ b/child.c
@@ -148,8 +148,6 @@ void do_child(int unix_fd, int inet_fd)
timeout.tv_sec = pool_config->child_life_time;
timeout.tv_usec = 0;
- init_prepared_list();
-
for (;;)
{
StartupPacket *sp;
diff --git a/pool_process_query.c b/pool_process_query.c
index feb47e8bb..8d5b66cce 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -56,8 +56,6 @@
#define FD_SETSIZE 512
#endif
-#define INIT_STATEMENT_LIST_SIZE 8
-
#define ACTIVE_SQL_TRANSACTION_ERROR_CODE "25001" /* SET TRANSACTION ISOLATION LEVEL must be called before any query */
#define DEADLOCK_ERROR_CODE "40P01"
#define SERIALIZATION_FAIL_ERROR_CODE "40001"
@@ -68,7 +66,6 @@
static int reset_backend(POOL_CONNECTION_POOL *backend, int qcnt);
static POOL_STATUS do_error_execute_command(POOL_CONNECTION_POOL *backend, int node_id, int major);
static char *get_insert_command_table_name(InsertStmt *node);
-static void reset_prepared_list(PreparedStatementList *p);
static int send_deallocate(POOL_CONNECTION_POOL *backend, PreparedStatementList *p, int n);
static int is_cache_empty(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend);
static POOL_STATUS ParallelForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_CONNECTION *backend, char *database, bool send_to_frontend);
@@ -991,11 +988,13 @@ POOL_STATUS send_extended_protocol_message(POOL_CONNECTION_POOL *backend,
return POOL_CONTINUE;
}
+#ifdef NOT_USED
POOL_STATUS send_execute_message(POOL_CONNECTION_POOL *backend,
int node_id, int len, char *string)
{
return send_extended_protocol_message(backend, node_id, "E", len, string);
}
+#endif
/*
* wait until read data is ready
@@ -1190,38 +1189,46 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
int delete_or_update = 0;
char kind1;
POOL_STATUS ret;
+ POOL_SESSION_CONTEXT *session_context;
+
+ /* Get session context */
+ session_context = pool_get_session_context();
+ if (!session_context)
+ {
+ pool_error("SimpleForwardToFrontend: cannot get session context");
+ return POOL_END;
+ }
/*
* Check if packet kind == 'C'(Command complete), '1'(Parse
* complete), '3'(Close complete). If so, then register or
* unregister pending prepared statement.
*/
- if ((kind == 'C' || kind == '1' || kind == '3') &&
- pending_function)
+ if ((kind == 'C' || kind == '1' || kind == '2' || kind == '3') &&
+ session_context->pending_function)
{
- pending_function(&prepared_list, pending_prepared_portal);
- if (pending_prepared_portal &&
- pending_prepared_portal->stmt &&
- IsA(pending_prepared_portal->stmt, DeallocateStmt))
+ Node *parse_tree;
+
+ session_context->pending_function();
+
+ if (session_context->pending_pstmt)
{
- free(pending_prepared_portal->portal_name);
- pending_prepared_portal->portal_name = NULL;
- pool_memory_delete(pending_prepared_portal->prepare_ctxt, 0);
- free(pending_prepared_portal);
+ parse_tree = session_context->pending_pstmt->qctxt->parse_tree;
+
+ if (IsA(parse_tree, DeallocateStmt))
+ pool_remove_pending_objects();
}
+
+ session_context->pending_function = NULL;
+ session_context->pending_pstmt = NULL;
+ session_context->pending_portal = NULL;
}
- else if (kind == 'E' && pending_function)
+ else if (kind == 'E' && session_context->pending_function)
{
/* An error occurred with PREPARE or DEALLOCATE command.
* Free pending portal object.
*/
- if (pending_prepared_portal)
- {
- free(pending_prepared_portal->portal_name);
- pending_prepared_portal->portal_name = NULL;
- pool_memory_delete(pending_prepared_portal->prepare_ctxt, 0);
- free(pending_prepared_portal);
- }
+ pool_remove_pending_objects();
}
/*
* The response of Execute command will be EmptyQueryResponse(I),
@@ -1239,8 +1246,7 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
*/
if (kind != 'N')
{
- pending_function = NULL;
- pending_prepared_portal = NULL;
+ pool_remove_pending_objects();
}
status = pool_read(MASTER(backend), &len, sizeof(len));
@@ -1470,6 +1476,9 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
}
}
}
+
+ pool_unset_query_in_progress();
+
return POOL_CONTINUE;
}
@@ -1479,10 +1488,20 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
int sendlen;
char *p;
int i;
- char *name;
char *rewrite_msg = NULL;
POOL_STATUS ret;
char msgbuf[128];
+ PreparedStatement *pstmt;
+ Portal *portal;
+ POOL_SESSION_CONTEXT *session_context;
+
+ /* Get session context */
+ session_context = pool_get_session_context();
+ if (!session_context)
+ {
+ pool_error("SimpleForwardToBackend: cannot get session context");
+ return POOL_END;
+ }
/* Read message length */
if (pool_read(frontend, &sendlen, sizeof(sendlen)))
@@ -1547,25 +1566,24 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
*/
else if (kind == 'B')
{
- Portal *portal = NULL;
char *stmt_name, *portal_name;
portal_name = p;
stmt_name = p + strlen(portal_name) + 1;
+
+ pstmt = pool_get_prepared_statement_by_pstmt_name(stmt_name);
- if (*stmt_name == '\0')
- portal = unnamed_statement;
- else
+ portal = pool_create_portal(portal_name, pstmt->num_tsparams, pstmt);
+ if (portal == NULL)
{
- portal = lookup_prepared_statement_by_statement(&prepared_list, stmt_name);
- if (portal)
- {
- portal->portal_name = strdup(portal_name);
- }
+ pool_error("SimpleForwardToBackend: creating portal failed: %s", strerror(errno));
+ return POOL_END;
}
+ session_context->pending_portal = portal;
+ session_context->pending_function = pool_add_portal;
/* rewrite bind message */
- if (REPLICATION && portal && portal->num_tsparams > 0)
+ if (REPLICATION && portal->num_tsparams > 0)
{
rewrite_msg = bind_rewrite_timestamp(backend, portal, p, &len);
if (rewrite_msg != NULL)
@@ -1640,6 +1658,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
*/
if (kind == 'B')
{
+#ifdef NOT_USED
Portal *portal = NULL;
char *stmt_name, *portal_name;
@@ -1664,6 +1683,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
free(portal->portal_name);
portal->portal_name = strdup(portal_name);
}
+#endif
if (rewrite_msg)
free(rewrite_msg);
}
@@ -1671,6 +1691,7 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
/* Close message with prepared statement name. */
else if (kind == 'C' && *p == 'S' && *(p + 1))
{
+#ifdef NOT_USED
POOL_MEMORY_POOL *old_context = pool_memory;
DeallocateStmt *deallocate_stmt;
@@ -1703,6 +1724,10 @@ POOL_STATUS SimpleForwardToBackend(char kind, POOL_CONNECTION *frontend, POOL_CO
pending_prepared_portal->portal_name = NULL;
pending_function = del_prepared_list;
pool_memory = old_context;
+#endif
+ pstmt = pool_get_prepared_statement_by_pstmt_name(p+1);
+ session_context->pending_pstmt = pstmt;
+ session_context->pending_function = pool_remove_prepared_statement;
}
/* Bind, Describe or Close message? */
@@ -1915,7 +1940,7 @@ void reset_variables(void)
void reset_connection(void)
{
reset_variables();
- reset_prepared_list(&prepared_list);
+ pool_clear_prepared_statement_list();
}
@@ -1927,6 +1952,15 @@ static int reset_backend(POOL_CONNECTION_POOL *backend, int qcnt)
{
char *query;
int qn;
+ POOL_SESSION_CONTEXT *session_context;
+
+ /* Get session context */
+ session_context = pool_get_session_context();
+ if (!session_context)
+ {
+ pool_error("reset_backend: cannot get session context");
+ return POOL_END;
+ }
/*
* Reset all state variables
@@ -1941,7 +1975,7 @@ static int reset_backend(POOL_CONNECTION_POOL *backend, int qcnt)
*/
if (qcnt >= qn)
{
- if (prepared_list.cnt == 0)
+ if (session_context->pstmt_list->size == 0)
{
/*
* Either no prepared objects were created or DISCARD ALL
@@ -1949,17 +1983,20 @@ static int reset_backend(POOL_CONNECTION_POOL *backend, int qcnt)
* were executed. The latter causes call to
* reset_prepared_list which removes all prepared objects.
*/
- reset_prepared_list(&prepared_list);
+// reset_prepared_list(&prepared_list);
return 2;
}
+ char *name = session_context->pstmt_list->pstmts[0]->name;
+
/* Delete from prepared list */
- if (send_deallocate(backend, &prepared_list, 0))
+ if (send_deallocate(backend, session_context->pstmt_list, 0))
{
/* Deallocate failed. We are in unknown state. Ask caller
* to reset backend connection.
*/
- reset_prepared_list(&prepared_list);
+// reset_prepared_list(&prepared_list);
+ pool_remove_prepared_statement_by_pstmt_name(name);
return -1;
}
/*
@@ -1970,7 +2007,8 @@ static int reset_backend(POOL_CONNECTION_POOL *backend, int qcnt)
* del_prepared_list() again. This is harmless since trying to
* remove same prepared object will be ignored.
*/
- del_prepared_list(&prepared_list, prepared_list.portal_list[0]);
+// del_prepared_list(&prepared_list, prepared_list.portal_list[0]);
+ pool_remove_prepared_statement_by_pstmt_name(name);
return 1;
}
@@ -2629,7 +2667,7 @@ static POOL_STATUS do_error_execute_command(POOL_CONNECTION_POOL *backend, int n
p = buf;
memset(msg + len, 0, sizeof(int));
- if (send_execute_message(backend, node_id, len + 5, msg))
+ if (send_extended_protocol_message(backend, node_id, "E", len + 5, msg))
{
return POOL_END;
}
@@ -3727,14 +3765,14 @@ POOL_STATUS read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_PO
if (IsA(node, DeallocateStmt))
{
- Portal *portal;
+ PreparedStatement *ps;
DeallocateStmt *d = (DeallocateStmt *)node;
- portal = lookup_prepared_statement_by_statement(&prepared_list, d->name);
- if (portal && portal->sql_string)
+ ps = pool_get_prepared_statement_by_pstmt_name(d->name);
+ if (ps && ps->qctxt->original_query)
{
string_append_char(msg, "[");
- string_append_char(msg, portal->sql_string);
+ string_append_char(msg, ps->qctxt->original_query);
string_append_char(msg, "]");
}
}
@@ -3774,6 +3812,7 @@ POOL_STATUS read_kind_from_backend(POOL_CONNECTION *frontend, POOL_CONNECTION_PO
return POOL_CONTINUE;
}
+#ifdef NOT_USED
/*
* Create portal object
* Return object is allocated from heap memory.
@@ -3793,143 +3832,7 @@ Portal *create_portal(void)
}
return p;
}
-
-void init_prepared_list(void)
-{
- prepared_list.cnt = 0;
- prepared_list.size = INIT_STATEMENT_LIST_SIZE;
- prepared_list.portal_list = malloc(sizeof(Portal *) * prepared_list.size);
- if (prepared_list.portal_list == NULL)
- {
- pool_error("init_prepared_list: malloc failed: %s", strerror(errno));
- exit(1);
- }
-}
-
-void add_prepared_list(PreparedStatementList *p, Portal *portal)
-{
- if (p->cnt == p->size)
- {
- p->size *= 2;
- p->portal_list = realloc(p->portal_list, sizeof(Portal *) * p->size);
- if (p->portal_list == NULL)
- {
- pool_error("add_prepared_list: realloc failed: %s", strerror(errno));
- exit(1);
- }
- }
- p->portal_list[p->cnt++] = portal;
-}
-
-void add_unnamed_portal(PreparedStatementList *p, Portal *portal)
-{
- if (unnamed_statement)
- {
- pool_memory_delete(unnamed_statement->prepare_ctxt, 0);
- free(unnamed_statement);
- }
-
- unnamed_portal = NULL;
- unnamed_statement = portal;
-}
-
-void del_prepared_list(PreparedStatementList *p, Portal *portal)
-{
- int i;
- DeallocateStmt *s = (DeallocateStmt *)portal->stmt;
-
- /* DEALLOCATE ALL? */
- if (s->name == NULL)
- {
- reset_prepared_list(p);
- }
- else
- {
- for (i = 0; i < p->cnt; i++)
- {
- PrepareStmt *p_stmt = (PrepareStmt *)p->portal_list[i]->stmt;
- if (strcmp(p_stmt->name, s->name) == 0)
- break;
- }
-
- if (i == p->cnt)
- return;
-
- pool_memory_delete(p->portal_list[i]->prepare_ctxt, 0);
- free(p->portal_list[i]->portal_name);
- free(p->portal_list[i]);
- if (i != p->cnt - 1)
- {
- memmove(&p->portal_list[i], &p->portal_list[i+1],
- sizeof(Portal *) * (p->cnt - i - 1));
- }
- p->cnt--;
- }
-}
-
-void delete_all_prepared_list(PreparedStatementList *p, Portal *portal)
-{
- reset_prepared_list(p);
-}
-
-static void reset_prepared_list(PreparedStatementList *p)
-{
- int i;
-
- if (p)
- {
- for (i = 0; i < p->cnt; i++)
- {
- pool_memory_delete(p->portal_list[i]->prepare_ctxt, 0);
- free(p->portal_list[i]->portal_name);
- free(p->portal_list[i]);
- }
- if (unnamed_statement)
- {
- pool_memory_delete(unnamed_statement->prepare_ctxt, 0);
- free(unnamed_statement);
- }
- unnamed_portal = NULL;
- unnamed_statement = NULL;
- p->cnt = 0;
- }
-}
-
-Portal *lookup_prepared_statement_by_statement(PreparedStatementList *p, const char *name)
-{
- int i;
-
- /* unnamed portal? */
- if (name == NULL || name[0] == '\0' || (name[0] == '\"' && name[1] == '\"'))
- return unnamed_statement;
-
- for (i = 0; i < p->cnt; i++)
- {
- PrepareStmt *p_stmt = (PrepareStmt *)p->portal_list[i]->stmt;
- if (strcmp(p_stmt->name, name) == 0)
- return p->portal_list[i];
- }
-
- return NULL;
-}
-
-Portal *lookup_prepared_statement_by_portal(PreparedStatementList *p, const char *name)
-{
- int i;
-
- /* unnamed portal? */
- if (name == NULL || name[0] == '\0' || (name[0] == '\"' && name[1] == '\"'))
- return unnamed_portal;
-
- for (i = 0; i < p->cnt; i++)
- {
- if (p->portal_list[i]->portal_name &&
- strcmp(p->portal_list[i]->portal_name, name) == 0)
- return p->portal_list[i];
- }
-
- return NULL;
-}
+#endif
/*
* Send DEALLOCATE message to backend by using SimpleQuery.
@@ -3941,10 +3844,10 @@ static int send_deallocate(POOL_CONNECTION_POOL *backend, PreparedStatementList
int len;
PrepareStmt *p_stmt;
- if (p->cnt <= n)
+ if (p->size <= n)
return 1;
- p_stmt = (PrepareStmt *)p->portal_list[n]->stmt;
+ p_stmt = (PrepareStmt *)p->pstmts[n]->qctxt->parse_tree;
len = strlen(p_stmt->name) + 14; /* "DEALLOCATE \"" + "\"" + '\0' */
query = malloc(len);
if (query == NULL)
diff --git a/pool_proto_modules.c b/pool_proto_modules.c
index 885f3839c..318ed8b82 100644
--- a/pool_proto_modules.c
+++ b/pool_proto_modules.c
@@ -65,10 +65,6 @@ char *copy_table = NULL; /* copy table name */
char *copy_schema = NULL; /* copy table name */
char copy_delimiter; /* copy delimiter char */
char *copy_null = NULL; /* copy null string */
-void (*pending_function)(PreparedStatementList *p, Portal *portal) = NULL;
-Portal *pending_prepared_portal = NULL;
-Portal *unnamed_statement = NULL;
-Portal *unnamed_portal = NULL;
int select_in_transaction = 0; /* non 0 if select query is in transaction */
int execute_select = 0; /* non 0 if select query is in transaction */
@@ -83,8 +79,6 @@ int receive_extended_begin = 0;
*/
static int allow_close_transaction = 1;
-PreparedStatementList prepared_list; /* prepared statement name list */
-
int is_select_pgcatalog = 0;
int is_select_for_update = 0; /* 1 if SELECT INTO or SELECT FOR UPDATE */
bool is_parallel_table = false;
@@ -122,9 +116,6 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
POOL_SESSION_CONTEXT *session_context;
POOL_QUERY_CONTEXT *query_context;
- POOL_MEMORY_POOL *old_context = NULL;
- Portal *portal;
-
/* Get session context */
session_context = pool_get_session_context();
if (!session_context)
@@ -279,77 +270,66 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
*/
if (IsA(node, PrepareStmt))
{
- pending_function = add_prepared_list;
- portal = create_portal();
- if (portal == NULL)
+ PreparedStatement *ps;
+
+ ps = pool_create_prepared_statement(((PrepareStmt *)node)->name,
+ 0, query_context);
+ if (ps == NULL)
{
- pool_error("SimpleQuery: create_portal() failed");
+ pool_error("SimpleQuery: failed to create prepared statement: %s", strerror(errno));
return POOL_END;
}
- /* switch memory context */
- old_context = pool_memory;
- pool_memory = portal->prepare_ctxt;
-
- portal->portal_name = NULL;
- portal->stmt = copyObject(node);
- portal->sql_string = NULL;
- pending_prepared_portal = portal;
+ session_context->pending_pstmt = ps;
+ session_context->pending_function = pool_add_prepared_statement;
}
else if (IsA(node, DeallocateStmt))
{
- pending_function = del_prepared_list;
- portal = create_portal();
- if (portal == NULL)
+ char *name;
+ PreparedStatement *ps;
+
+ name = ((DeallocateStmt *)node)->name;
+ ps = pool_create_prepared_statement(name, 0, query_context);
+ if (ps == NULL)
{
- pool_error("SimpleQuery: create_portal() failed");
+ pool_error("SimpleQuery: failed to create prepared statement: %s", strerror(errno));
return POOL_END;
}
- /* switch memory context */
- old_context = pool_memory;
- pool_memory = portal->prepare_ctxt;
+ session_context->pending_pstmt = ps;
- portal->portal_name = NULL;
- portal->stmt = copyObject(node);
- portal->sql_string = NULL;
- pending_prepared_portal = portal;
+ if (name == NULL)
+ session_context->pending_function = pool_clear_prepared_statement_list;
+ else
+ session_context->pending_function = pool_remove_prepared_statement;
}
else if (IsA(node, DiscardStmt))
{
DiscardStmt *stmt = (DiscardStmt *)node;
if (stmt->target == DISCARD_ALL || stmt->target == DISCARD_PLANS)
{
- pending_function = delete_all_prepared_list;
- pending_prepared_portal = NULL;
+ session_context->pending_pstmt = NULL;
+ session_context->pending_portal = NULL;
+ session_context->pending_function = pool_clear_prepared_statement_list;
}
}
-
- /* switch old memory context */
- if (old_context)
- pool_memory = old_context;
-
- /* end of wrong if (see 2009/4/3 comment above) */
}
if (frontend && IsA(node, ExecuteStmt))
{
- Portal *portal;
- PrepareStmt *p_stmt;
- ExecuteStmt *e_stmt = (ExecuteStmt *)node;
+ PreparedStatement *ps;
- portal = lookup_prepared_statement_by_statement(&prepared_list,
- e_stmt->name);
- if (!portal)
+ ps = pool_get_prepared_statement_by_pstmt_name(((ExecuteStmt *)node)->name);
+
+ if (ps)
{
- string1 = string;
- node1 = node;
+ string1 = ps->qctxt->original_query;
+ node1 = ps->qctxt->parse_tree;
}
else
{
- p_stmt = (PrepareStmt *)portal->stmt;
- string1 = nodeToString(p_stmt->query);
- node1 = (Node *)p_stmt->query;
+ string1 = string;
+ node1 = node;
}
}
else
@@ -435,19 +415,18 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
if (node)
{
- Portal *portal = NULL;
+ PreparedStatement *ps = NULL;
if (IsA(node, PrepareStmt))
{
- portal = pending_prepared_portal;
- portal->num_tsparams = 0;
+ ps = session_context->pending_pstmt;
+ ps->num_tsparams = 0;
}
else if (IsA(node, ExecuteStmt))
- portal = lookup_prepared_statement_by_statement(
- &prepared_list, ((ExecuteStmt *) node)->name);
+ ps = pool_get_prepared_statement_by_pstmt_name(((ExecuteStmt *) node)->name);
/* rewrite `now()' to timestamp literal */
- rewrite_query = rewrite_timestamp(backend, node, false, portal);
+ rewrite_query = rewrite_timestamp(backend, node, false, ps);
if (rewrite_query != NULL)
{
string = rewrite_query;
@@ -464,13 +443,13 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
if (pool_config->num_init_children == 1)
{
/* Send query to all DB nodes at once */
- status = pool_send_and_wait(query_context, string, len, 0, 0);
+ status = pool_send_and_wait(query_context, string, len, 0, 0, "");
free_parser();
return status;
}
/* Send the query to master node */
- if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID, "") != POOL_CONTINUE)
{
free_parser();
return POOL_END;
@@ -478,7 +457,7 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
}
/* Send query to other than master node */
- if (pool_send_and_wait(query_context, string, len, -1, MASTER_NODE_ID) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, -1, MASTER_NODE_ID, "") != POOL_CONTINUE)
{
free_parser();
return POOL_END;
@@ -487,7 +466,7 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
/* Send "COMMIT" or "ROLLBACK" to only master node if query is "COMMIT" or "ROLLBACK" */
if (commit)
{
- if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID, "") != POOL_CONTINUE)
{
free_parser();
return POOL_END;
@@ -497,7 +476,7 @@ static int is_temp_table(POOL_CONNECTION_POOL *backend, Node *node);
}
else
{
- if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID, "") != POOL_CONTINUE)
{
free_parser();
return POOL_END;
@@ -515,14 +494,24 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend,
{
char *string; /* portal name + null terminate + max_tobe_returned_rows */
int len;
- int i;
char kind;
int status, commit = 0;
+ PreparedStatement *ps;
Portal *portal;
char *string1 = NULL;
PrepareStmt *p_stmt;
POOL_STATUS ret;
int specific_error = 0;
+ POOL_SESSION_CONTEXT *session_context;
+ POOL_QUERY_CONTEXT *query_context;
+
+ /* Get session context */
+ session_context = pool_get_session_context();
+ if (!session_context)
+ {
+ pool_error("Execute: cannot get session context");
+ return POOL_END;
+ }
/* read Execute packet */
if (pool_read(frontend, &len, sizeof(len)) < 0)
@@ -533,21 +522,32 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend,
pool_debug("Execute: portal name <%s>", string);
- portal = lookup_prepared_statement_by_portal(&prepared_list,
- string);
+ portal = pool_get_portal_by_portal_name(string);
+ ps = pool_get_prepared_statement_by_pstmt_name(portal->pstmt->name);
+ query_context = ps->qctxt;
/* load balance trick */
if (portal)
{
Node *node;
- p_stmt = (PrepareStmt *)portal->stmt;
+ p_stmt = (PrepareStmt *)portal->pstmt->qctxt->parse_tree;
- string1 = portal->sql_string;
+ string1 = portal->pstmt->qctxt->original_query;
pool_debug("Execute: query: %s", string1);
- node = (Node *)p_stmt->query;
+ node = (Node *)p_stmt;
strncpy(query_string_buffer, string1, sizeof(query_string_buffer));
+ /*
+ * Start query context
+ */
+ pool_start_query(query_context, string1, node);
+
+ /*
+ * Decide where to send query
+ */
+ pool_where_to_send(query_context, query_context->original_query, node);
+
if ((IsA(node, PrepareStmt) || IsA(node, DeallocateStmt) ||
IsA(node, VariableSetStmt)) &&
MASTER_SLAVE && TSTATE(backend) != 'E')
@@ -592,20 +592,8 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend,
if (!commit)
{
/* Send the query to master node */
- per_node_statement_log(backend, MASTER_NODE_ID, string1);
- if (send_execute_message(backend, MASTER_NODE_ID, len, string) != POOL_CONTINUE)
- return POOL_END;
-
- if (wait_for_query_response(frontend, MASTER(backend), string, MAJOR(backend)) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID, "E") != POOL_CONTINUE)
{
- /* Cancel current transaction */
- CancelPacket cancel_packet;
-
- cancel_packet.protoVersion = htonl(PROTO_CANCEL);
- cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
- cancel_packet.key= MASTER_CONNECTION(backend)->key;
- cancel_request(&cancel_packet);
-
return POOL_END;
}
@@ -619,89 +607,37 @@ POOL_STATUS Execute(POOL_CONNECTION *frontend,
}
}
- /* send query to other nodes */
- for (i=0;i<NUM_BACKENDS;i++)
+ if (specific_error)
{
- if (!VALID_BACKEND(i) || IS_MASTER_NODE_ID(i))
- continue;
+ char msg[1024] = "pgpool_error_portal"; /* large enough */
+ int len = strlen(msg);
- if (specific_error)
- {
- char msg[1024] = "pgpoool_error_portal"; /* large enough */
- int len = strlen(msg);
+ memset(msg + len, 0, sizeof(int));
- memset(msg + len, 0, sizeof(int));
- if (send_execute_message(backend, i, len + 5, msg))
- return POOL_END;
- }
- else
- {
- per_node_statement_log(backend, i, string1);
- if (send_execute_message(backend, i, len, string) != POOL_CONTINUE)
- return POOL_END;
- }
+ /* send query to other nodes */
+ if (pool_send_and_wait(query_context, msg, len, -1, MASTER_NODE_ID, "E") != POOL_CONTINUE)
+ return POOL_END;
}
-
- /* Wait for nodes other than the master node */
- for (i=0;i<NUM_BACKENDS;i++)
+ else
{
- if (!VALID_BACKEND(i) || IS_MASTER_NODE_ID(i))
- continue;
-
- if (wait_for_query_response(frontend, CONNECTION(backend, i), string, MAJOR(backend)) != POOL_CONTINUE)
- {
- /* Cancel current transaction */
- CancelPacket cancel_packet;
-
- cancel_packet.protoVersion = htonl(PROTO_CANCEL);
- cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
- cancel_packet.key= MASTER_CONNECTION(backend)->key;
- cancel_request(&cancel_packet);
-
+ if (pool_send_and_wait(query_context, string, len, -1, MASTER_NODE_ID, "E") != POOL_CONTINUE)
return POOL_END;
- }
}
-
+
/* send "COMMIT" or "ROLLBACK" to only master node if query is "COMMIT" or "ROLLBACK" */
if (commit)
{
- per_node_statement_log(backend, MASTER_NODE_ID, string1);
-
- if (send_execute_message(backend, MASTER_NODE_ID, len, string) != POOL_CONTINUE)
- return POOL_END;
-
- if (wait_for_query_response(frontend, MASTER(backend), string, MAJOR(backend)) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID, "E") != POOL_CONTINUE)
{
- /* Cancel current transaction */
- CancelPacket cancel_packet;
-
- cancel_packet.protoVersion = htonl(PROTO_CANCEL);
- cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
- cancel_packet.key= MASTER_CONNECTION(backend)->key;
- cancel_request(&cancel_packet);
-
return POOL_END;
}
}
}
else
{
- per_node_statement_log(backend, MASTER_NODE_ID, string1);
-
- if (send_execute_message(backend, MASTER_NODE_ID, len, string) != POOL_CONTINUE)
- return POOL_END;
-
- if (wait_for_query_response(frontend, MASTER(backend), string, MAJOR(backend)) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, -1, MASTER_NODE_ID, "E") != POOL_CONTINUE)
{
- /* Cancel current transaction */
- CancelPacket cancel_packet;
-
- cancel_packet.protoVersion = htonl(PROTO_CANCEL);
- cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
- cancel_packet.key= MASTER_CONNECTION(backend)->key;
- cancel_request(&cancel_packet);
-
- return POOL_END;
+ return POOL_END;
}
}
@@ -738,7 +674,6 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
int len;
char *string;
int i;
- Portal *portal;
POOL_MEMORY_POOL *old_context;
PrepareStmt *p_stmt;
char *name, *stmt;
@@ -748,6 +683,20 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
int insert_stmt_with_lock = 0;
POOL_STATUS status;
char per_node_statement_log_buffer[1024];
+ PreparedStatement *ps;
+ POOL_SESSION_CONTEXT *session_context;
+ POOL_QUERY_CONTEXT *query_context;
+
+ /* Get session context */
+ session_context = pool_get_session_context();
+ if (!session_context)
+ {
+ pool_error("Parse: cannot get session context");
+ return POOL_END;
+ }
+
+ /* Create query context */
+ query_context = pool_init_query_context();
/* read Parse packet */
if (pool_read(frontend, &len, sizeof(len)) < 0)
@@ -795,17 +744,9 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
}
}
- portal = create_portal();
- if (portal == NULL)
- {
- pool_error("Parse: create_portal() failed");
- free_parser();
- return POOL_END;
- }
-
/* switch memory context */
old_context = pool_memory;
- pool_memory = portal->prepare_ctxt;
+ pool_memory = session_context->memory_context;
/* translate Parse message to PrepareStmt */
p_stmt = palloc(sizeof(PrepareStmt));
@@ -817,21 +758,29 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
*/
p_stmt->name = pstrdup(name);
p_stmt->query = copyObject(node);
- portal->stmt = (Node *)p_stmt;
- portal->portal_name = NULL;
- portal->sql_string = pstrdup(stmt);
+ ps = pool_create_prepared_statement(name, 0, query_context);
+ /*
+ * Start query context
+ */
+ pool_start_query(query_context, pstrdup(stmt), (Node *)p_stmt);
+
+ /*
+ * Decide where to send query
+ */
+ pool_where_to_send(query_context, query_context->original_query, (Node *)p_stmt);
+
if (*name)
{
- pending_function = add_prepared_list;
- pending_prepared_portal = portal;
+ session_context->pending_pstmt = ps;
+ session_context->pending_function = pool_add_prepared_statement;
}
else /* unnamed statement */
{
- pending_function = add_unnamed_portal;
pfree(p_stmt->name);
- p_stmt->name = NULL;
- pending_prepared_portal = portal;
+ ps->name = "";
+ session_context->pending_pstmt = ps;
+ session_context->pending_function = pool_add_prepared_statement;
}
/*
@@ -854,8 +803,8 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
*/
if (*name == '\0')
rewrite_to_params = false;
- portal->num_tsparams = 0;
- rewrite_query = rewrite_timestamp(backend, node, rewrite_to_params, portal);
+ ps->num_tsparams = 0;
+ rewrite_query = rewrite_timestamp(backend, node, rewrite_to_params, ps);
if (rewrite_query != NULL)
{
int alloc_len = len - strlen(stmt) + strlen(rewrite_query);
@@ -921,13 +870,6 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
/* send to master node */
snprintf(per_node_statement_log_buffer, sizeof(per_node_statement_log_buffer), "Parse: %s", stmt);
- per_node_statement_log(backend, MASTER_NODE_ID, per_node_statement_log_buffer);
- if (send_extended_protocol_message(backend, MASTER_NODE_ID,
- "P", len, string))
- {
- free_parser();
- return POOL_END;
- }
/*
* Cannot call free_parser() here. Since "string" might be allocated in parser context.
@@ -941,15 +883,8 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
* locks.
*/
pool_debug("Parse: waiting for master completing the query");
- if (wait_for_query_response(frontend, MASTER(backend), string, MAJOR(backend)) != POOL_CONTINUE)
+ if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID, "P") != POOL_CONTINUE)
{
- /* Cancel current transaction */
- CancelPacket cancel_packet;
-
- cancel_packet.protoVersion = htonl(PROTO_CANCEL);
- cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
- cancel_packet.key= MASTER_CONNECTION(backend)->key;
- cancel_request(&cancel_packet);
free_parser();
return POOL_END;
}
@@ -977,67 +912,33 @@ POOL_STATUS Parse(POOL_CONNECTION *frontend,
per_node_error_log(backend, MASTER_NODE_ID, stmt, "Parse(): Error or notice message from backend: ", true);
}
- for (i=0;i<NUM_BACKENDS;i++)
+ if (deadlock_detected)
{
- if (VALID_BACKEND(i) && !IS_MASTER_NODE_ID(i))
+ pool_log("Parse: received deadlock error message from master node");
+ if (pool_send_and_wait(query_context, POOL_ERROR_QUERY,
+ strlen(POOL_ERROR_QUERY)+1, -1,
+ MASTER_NODE_ID, "") != POOL_CONTINUE)
{
- if (deadlock_detected)
- {
- pool_log("Parse: received deadlock error message from master node");
-
- per_node_statement_log(backend, i, POOL_ERROR_QUERY);
-
- if (send_simplequery_message(CONNECTION(backend, i),
- strlen(POOL_ERROR_QUERY)+1,
- POOL_ERROR_QUERY,
- MAJOR(backend)))
- {
- free_parser();
- return POOL_END;
- }
- }
- else
- {
- snprintf(per_node_statement_log_buffer, sizeof(per_node_statement_log_buffer), "Parse: %s", stmt);
- per_node_statement_log(backend, i, per_node_statement_log_buffer);
-
- if (send_extended_protocol_message(backend, i,"P", len, string))
- {
- free_parser();
- return POOL_END;
- }
- }
+ free_parser();
+ return POOL_END;
}
}
-
- /* wait for DB nodes completing query except master node */
- for (i=0;i<NUM_BACKENDS;i++)
+ else
{
- if (!VALID_BACKEND(i) || IS_MASTER_NODE_ID(i))
- continue;
-
- pool_debug("Parse: waiting for %dth backend completing the query", i);
- if (wait_for_query_response(frontend, CONNECTION(backend, i), string, MAJOR(backend)) != POOL_CONTINUE)
+ snprintf(per_node_statement_log_buffer, sizeof(per_node_statement_log_buffer), "Parse: %s", stmt);
+ if (pool_send_and_wait(query_context, string, len, -1, MASTER_NODE_ID, "P") != POOL_CONTINUE)
{
- /* Cancel current transaction */
- CancelPacket cancel_packet;
-
- cancel_packet.protoVersion = htonl(PROTO_CANCEL);
- cancel_packet.pid = MASTER_CONNECTION(backend)->pid;
- cancel_packet.key= MASTER_CONNECTION(backend)->key;
- cancel_request(&cancel_packet);
free_parser();
return POOL_END;
}
-
- /*
- * Check if error (or notice response) from backend is
- * detected. If so, emit log. This is usefull when
- * invalid encoding error occurs. In this case, PostgreSQL
- * does not report what statement caused that error and
- * make users confused.
- */
- per_node_error_log(backend, i, stmt, "Parse(): Error or notice message from backend: ", true);
+ }
+ }
+ else
+ {
+ if (pool_send_and_wait(query_context, string, len, 1, MASTER_NODE_ID, "P") != POOL_CONTINUE)
+ {
+ free_parser();
+ return POOL_END;
}
}
@@ -1079,6 +980,15 @@ POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
int i;
int len;
signed char state;
+ POOL_SESSION_CONTEXT *session_context;
+
+ /* Get session context */
+ session_context = pool_get_session_context();
+ if (!session_context)
+ {
+ pool_error("ReadyForQuery: cannot get session context");
+ return POOL_END;
+ }
/*
* If the numbers of update tuples are differ, we need to abort transaction
@@ -1232,6 +1142,8 @@ POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
pool_flush(frontend);
}
+ pool_unset_query_in_progress();
+
sp = MASTER_CONNECTION(backend)->sp;
if (MASTER(backend)->tstate == 'T')
snprintf(psbuf, sizeof(psbuf), "%s %s %s idle in transaction",
@@ -1241,7 +1153,7 @@ POOL_STATUS ReadyForQuery(POOL_CONNECTION *frontend,
sp->user, sp->database, remote_ps_data);
set_ps_display(psbuf, false);
- pool_query_context_destroy(pool_get_session_context()->query_context);
+// pool_query_context_destroy(pool_get_session_context()->query_context);
return POOL_CONTINUE;
}
diff --git a/pool_proto_modules.h b/pool_proto_modules.h
index 7d880283e..b38565f8f 100644
--- a/pool_proto_modules.h
+++ b/pool_proto_modules.h
@@ -37,24 +37,6 @@
#define POOL_ERROR_QUERY "send invalid query from pgpool to abort transaction"
-/* Prepared statement information */
-typedef struct {
- char *portal_name; /* portal name*/
- Node *stmt; /* parse tree for prepared statement */
- char *sql_string; /* original SQL statement */
- POOL_MEMORY_POOL *prepare_ctxt; /* memory context for parse tree */
- int num_tsparams;
-} Portal;
-
-/*
- * prepared statement list
- */
-typedef struct {
- int size;
- int cnt;
- Portal **portal_list;
-} PreparedStatementList;
-
extern int force_replication;
extern int replication_was_enabled; /* replication mode was enabled */
extern int master_slave_was_enabled; /* master/slave mode was enabled */
@@ -65,10 +47,6 @@ extern char *copy_table; /* copy table name */
extern char *copy_schema; /* copy table name */
extern char copy_delimiter; /* copy delimiter char */
extern char *copy_null; /* copy null string */
-extern void (*pending_function)(PreparedStatementList *p, Portal *portal);
-extern Portal *pending_prepared_portal;
-extern Portal *unnamed_statement;
-extern Portal *unnamed_portal;
extern int select_in_transaction; /* non 0 if select query is in transaction */
extern int execute_select; /* non 0 if select query is in transaction */
@@ -80,8 +58,6 @@ extern int is_select_for_update; /* also for SELECT ... INTO */
extern bool is_parallel_table;
extern char *parsed_query;
-extern PreparedStatementList prepared_list; /* prepared statement name list */
-
/*
* modules defined in pool_proto_modules.c
*/
@@ -154,11 +130,7 @@ extern void start_load_balance(POOL_CONNECTION_POOL *backend);
extern void end_load_balance(void);
extern int need_insert_lock(POOL_CONNECTION_POOL *backend, char *query, Node *node);
extern POOL_STATUS insert_lock(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, char *query, InsertStmt *node);
-extern void add_prepared_list(PreparedStatementList *p, Portal *portal);
-extern void add_unnamed_portal(PreparedStatementList *p, Portal *portal);
-extern void delete_all_prepared_list(PreparedStatementList *p, Portal *portal);
extern char *parse_copy_data(char *buf, int len, char delimiter, int col_id);
-extern Portal *lookup_prepared_statement_by_portal(PreparedStatementList *p, const char *name);extern Portal *lookup_prepared_statement_by_statement(PreparedStatementList *p, const char *name);
extern int check_copy_from_stdin(Node *node); /* returns non 0 if this is a COPY FROM STDIN */
extern void query_ps_status(char *query, POOL_CONNECTION_POOL *backend); /* show ps status */
extern POOL_STATUS start_internal_transaction(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend, Node *node);
@@ -172,8 +144,6 @@ extern POOL_STATUS pool_discard_packet(POOL_CONNECTION_POOL *cp);
extern int is_drop_database(Node *node); /* returns non 0 if this is a DROP DATABASE command */
extern void process_reporting(POOL_CONNECTION *frontend, POOL_CONNECTION_POOL *backend);
-extern Portal *create_portal(void);
-extern void del_prepared_list(PreparedStatementList *p, Portal *portal);
extern POOL_STATUS send_simplequery_message(POOL_CONNECTION *backend, int len, char *string, int major);
extern POOL_STATUS send_extended_protocol_message(POOL_CONNECTION_POOL *backend,
[parent: 47d85d7b90eb]