pgpool2.git / summary / log / commit / refs
commit 7ba993921304db93e82ee0ca4ccd72fb9ef3e7a7
Author: Yoshiharu Mori <y-mori at sraoss.co.jp>
Date: Wed Nov 14 03:59:58 2007 +0000
fix:
-When the setting of replication_timeout was effective,
SELECT that the execution time is long did the time-out
in only a parallel mode.
-A wrong number was displayed when replicated table was
updated or deleted in only a parallel mode.
pool.h | 1 +
pool_process_query.c | 44 +++++++++++++++++++++--
pool_stream.c | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 140 insertions(+), 3 deletions(-)
diff --git a/pool.h b/pool.h
index 3d2ab7419..2664507bd 100644
--- a/pool.h
+++ b/pool.h
@@ -423,6 +423,7 @@ extern POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
extern POOL_CONNECTION *pool_open(int fd);
extern void pool_close(POOL_CONNECTION *cp);
extern int pool_read(POOL_CONNECTION *cp, void *buf, int len);
+extern int pool_read_parallel(POOL_CONNECTION *cp, void *buf, int len);
extern char *pool_read2(POOL_CONNECTION *cp, int len);
extern int pool_write(POOL_CONNECTION *cp, void *buf, int len);
extern int pool_flush(POOL_CONNECTION *cp);
diff --git a/pool_process_query.c b/pool_process_query.c
index 363b57014..561ce36b9 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -206,6 +206,7 @@ static void query_ps_status(char *query, POOL_CONNECTION_POOL *backend); /* sho
static int is_select_pgcatalog = 0;
static int is_select_for_update = 0; /* also for SELECT ... INTO */
+static bool is_parallel_table = false;
static char *parsed_query = NULL;
static POOL_MEMORY_POOL *prepare_memory_context = NULL;
static int receive_sync = 0;
@@ -217,6 +218,7 @@ static int extract_ntuples(char *message);
static int detect_error(POOL_CONNECTION *master, char *error_code, int major, bool unread);
static int detect_deadlock_error(POOL_CONNECTION *master, int major);
static int detect_postmaster_down_error(POOL_CONNECTION *master, int major);
+static bool is_partition_table(POOL_CONNECTION_POOL *backend, Node *node);
POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend,
@@ -924,6 +926,8 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
{
node = (Node *) lfirst(list_head(parse_tree_list));
+ is_parallel_table = is_partition_table(backend,node);
+
if (pool_config->enable_query_cache &&
SYSDB_STATUS == CON_UP &&
IsA(node, SelectStmt) &&
@@ -3665,7 +3669,7 @@ POOL_STATUS SimpleForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_C
* if we are in the parallel mode, we have to sum up the number
* of affected rows
*/
- if (PARALLEL_MODE &&
+ if (PARALLEL_MODE && is_parallel_table &&
(strstr(p, "UPDATE") || strstr(p, "DELETE")))
{
delete_or_update = 1;
@@ -4570,7 +4574,7 @@ POOL_STATUS OneNode_do_command(POOL_CONNECTION *frontend, POOL_CONNECTION *backe
for(;;)
{
- status = pool_read(backend, &kind, sizeof(kind));
+ status = pool_read_parallel(backend, &kind, sizeof(kind));
if (status < 0)
{
pool_error("OneNode_do_command: error while reading message kind");
@@ -4586,7 +4590,7 @@ POOL_STATUS OneNode_do_command(POOL_CONNECTION *frontend, POOL_CONNECTION *backe
/*
* * Expecting ReadyForQuery
* */
- status = pool_read(backend, &kind, sizeof(kind));
+ status = pool_read_parallel(backend, &kind, sizeof(kind));
if (status < 0)
{
pool_error("OneNode_do_command: error while reading message kind");
@@ -4699,6 +4703,40 @@ static POOL_STATUS insert_lock(POOL_CONNECTION_POOL *backend, char *query, Inser
return POOL_CONTINUE;
}
+static bool is_partition_table(POOL_CONNECTION_POOL *backend, Node *node)
+{
+ DistDefInfo *info = NULL;
+ RangeVar *var = NULL;;
+
+ if (IsA(node, UpdateStmt))
+ {
+ UpdateStmt *update = (UpdateStmt*) node;
+
+ if(!IsA(update->relation,RangeVar))
+ return false;
+
+ var = (RangeVar *) update->relation;
+ }
+ else if (IsA(node, DeleteStmt))
+ {
+ DeleteStmt *delete = (DeleteStmt*) node;
+
+ if(!IsA(delete->relation,RangeVar))
+ return false;
+
+ var = (RangeVar *) delete->relation;
+ } else
+ return false;
+
+ info = pool_get_dist_def_info(MASTER_CONNECTION(backend)->sp->database,
+ var->schemaname,
+ var->relname);
+ if(info)
+ return true;
+ else
+ return false;
+}
+
/*
* obtain table name in INSERT statement
*/
diff --git a/pool_stream.c b/pool_stream.c
index a3715bf31..76b57313c 100644
--- a/pool_stream.c
+++ b/pool_stream.c
@@ -203,6 +203,104 @@ int pool_read(POOL_CONNECTION *cp, void *buf, int len)
return 0;
}
+/*
+* this function is used by SELECT QUERY at parallel query
+* read len bytes from cp
+* returns 0 on success otherwise -1.
+*/
+int pool_read_parallel(POOL_CONNECTION *cp, void *buf, int len)
+{
+ static char readbuf[READBUFSZ];
+
+ int consume_size;
+ int readlen;
+ int notimeout;
+
+ consume_size = consume_pending_data(cp, buf, len);
+ len -= consume_size;
+ buf += consume_size;
+
+ /* if this is not the master backend, then set timeout */
+ notimeout = 1;
+
+ while (len > 0)
+ {
+ if (pool_check_fd(cp, notimeout))
+ {
+ if (!IS_MASTER_NODE_ID(cp->db_node_id))
+ {
+ pool_log("pool_read: data is not ready in DB node: %d. abort this session",
+ cp->db_node_id);
+ exit(1);
+ }
+ else
+ {
+ pool_error("pool_read: pool_check_fd failed (%s)", strerror(errno));
+ return -1;
+ }
+ }
+
+ readlen = read(cp->fd, readbuf, READBUFSZ);
+ if (readlen == -1)
+ {
+ if (errno == EINTR || errno == EAGAIN)
+ {
+ pool_debug("pool_read: retrying due to %s", strerror(errno));
+ continue;
+ }
+
+ pool_error("pool_read: read failed (%s)", strerror(errno));
+
+ if (cp->isbackend)
+ {
+ /* fatal error, notice to parent and exit */
+ notice_backend_error(cp->db_node_id);
+ child_exit(1);
+ }
+ else
+ {
+ return -1;
+ }
+ }
+ else if (readlen == 0)
+ {
+ if (cp->isbackend)
+ {
+ pool_error("pool_read2: EOF encountered with backend");
+ return -1;
+
+#ifdef NOT_USED
+ /* fatal error, notice to parent and exit */
+ notice_backend_error(IS_MASTER_NODE_ID(cp->db_node_id));
+ child_exit(1);
+#endif
+ }
+ else
+ {
+ /*
+ * if backend offers authentication method, frontend could close connection
+ */
+ return -1;
+ }
+ }
+
+ if (len < readlen)
+ {
+ /* overrun. we need to save remaining data to pending buffer */
+ if (save_pending_data(cp, readbuf+len, readlen-len))
+ return -1;
+ memmove(buf, readbuf, len);
+ break;
+ }
+
+ memmove(buf, readbuf, readlen);
+ buf += readlen;
+ len -= readlen;
+ }
+
+ return 0;
+}
+
/*
* read exactly len bytes from cp
* returns buffer address on success otherwise NULL.
[parent: a4c302d6f72a]