pgpool2.git / summary / log / commit / refs
commit ebd5b622169ca3ea4a9a6d72f88c52703a7eabe3
Author: Yoshiharu Mori <y-mori at sraoss.co.jp>
Date: Fri Jul 04 05:27:30 2008 +0000
FIX: join partitioned tables
Delete: pool_read_parallel
pool.h | 1 -
pool_process_query.c | 119 ++++++++++++++++++++++++++----------------------
pool_rewrite_outfuncs.c | 30 ++++++++----
pool_rewrite_query.c | 4 +-
pool_stream.c | 94 --------------------------------------
5 files changed, 88 insertions(+), 160 deletions(-)
diff --git a/pool.h b/pool.h
index 6739b8652..9bde7752d 100644
--- a/pool.h
+++ b/pool.h
@@ -417,7 +417,6 @@ 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 0e5f7f466..5840ca915 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -635,6 +635,7 @@ POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,
int num_fds;
int used_count = 0;
int error_flag = 0;
+ unsigned long datacount = 0;
timeout.tv_sec = 1;
timeout.tv_usec = 0;
@@ -811,6 +812,7 @@ POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,
if(error_flag ==0)
{
pool_debug("pool_parallel_exec: kind from backend: %c", kind);
+
status = ParallelForwardToFrontend(kind,
frontend,
CONNECTION(backend, i),
@@ -846,6 +848,8 @@ POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,
if((kind == 'C' || kind == 'c' || kind == 'E') &&
used_count == NUM_BACKENDS -1)
{
+ pool_debug("pool_parallel_exec: kind from backend: D %ul", datacount);
+
if(error_flag == 0)
{
pool_debug("pool_parallel_exec: kind from backend: %c", kind);
@@ -864,8 +868,12 @@ POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,
}
return POOL_CONTINUE;
}
+
+ if(kind == 'D')
+ datacount++;
+ else
+ pool_debug("pool_parallel_exec: kind from backend: %c", kind);
- pool_debug("pool_parallel_exec: kind from backend: %c", kind);
status = ParallelForwardToFrontend(kind,
frontend,
CONNECTION(backend, i),
@@ -1004,6 +1012,7 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
/* call parallel exe engine */
POOL_STATUS stats = pool_parallel_exec(frontend,backend,r_query->rewrite_query, node,true);
free_parser();
+ in_progress = 0;
return stats;
}
else if(!r_query->is_pg_catalog)
@@ -1022,6 +1031,7 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
else if(r_query->type == T_SelectStmt)
{
free_parser();
+ in_progress = 0;
return r_query->status;
}
}
@@ -4728,59 +4738,60 @@ static POOL_STATUS do_error_execute_command(POOL_CONNECTION_POOL *backend, int n
POOL_STATUS OneNode_do_command(POOL_CONNECTION *frontend, POOL_CONNECTION *backend, char *query, char *database)
{
- int len,sendlen;
- int status;
- char kind;
-
- pool_debug("OneNode_do_command: Query: %s", query);
-
- /* send the query to the backend */
- pool_write(backend, "Q", 1);
- len = strlen(query)+1;
-
- sendlen = htonl(len + 4);
- pool_write(backend, &sendlen, sizeof(sendlen));
-
- if (pool_write_and_flush(backend, query, len) < 0)
- {
- return POOL_END;
- }
-
- for(;;)
- {
- status = pool_read_parallel(backend, &kind, sizeof(kind));
- if (status < 0)
- {
- pool_error("OneNode_do_command: error while reading message kind");
- return POOL_END;
- }
-
- status = ParallelForwardToFrontend(kind, frontend, backend, database, true);
- if (kind == 'C' || kind =='E')
- {
- break;
- }
- }
- /*
- * * Expecting ReadyForQuery
- * */
- status = pool_read_parallel(backend, &kind, sizeof(kind));
- if (status < 0)
- {
- pool_error("OneNode_do_command: error while reading message kind");
- return POOL_END;
- }
-
- if (kind != 'Z')
- {
- pool_error("OneNode_do_command: backend does not return ReadyForQuery");
- return POOL_END;
- }
-
- status = ParallelForwardToFrontend(kind, frontend, backend, database, true);
- pool_flush(frontend);
-
- return status;
+ int len,sendlen;
+ int status;
+ char kind;
+
+ pool_debug("OneNode_do_command: Query: %s", query);
+
+ /* send the query to the backend */
+ pool_write(backend, "Q", 1);
+ len = strlen(query)+1;
+
+ sendlen = htonl(len + 4);
+ pool_write(backend, &sendlen, sizeof(sendlen));
+
+ if (pool_write_and_flush(backend, query, len) < 0)
+ {
+ return POOL_END;
+ }
+
+ for(;;)
+ {
+ status = pool_read(backend, &kind, sizeof(kind));
+ if (status < 0)
+ {
+ pool_error("OneNode_do_command: error while reading message kind");
+ return POOL_END;
+ }
+
+ status = ParallelForwardToFrontend(kind, frontend, backend, database, true);
+ if (kind == 'C' || kind =='E')
+ {
+ break;
+ }
+ }
+ /*
+ * Expecting ReadyForQuery
+ *
+ */
+ status = pool_read(backend, &kind, sizeof(kind));
+ if (status < 0)
+ {
+ pool_error("OneNode_do_command: error while reading message kind");
+ return POOL_END;
+ }
+
+ if (kind != 'Z')
+ {
+ pool_error("OneNode_do_command: backend does not return ReadyForQuery");
+ return POOL_END;
+ }
+
+ status = ParallelForwardToFrontend(kind, frontend, backend, database, true);
+ pool_flush(frontend);
+
+ return status;
}
diff --git a/pool_rewrite_outfuncs.c b/pool_rewrite_outfuncs.c
index 1d23637e4..326610396 100644
--- a/pool_rewrite_outfuncs.c
+++ b/pool_rewrite_outfuncs.c
@@ -255,7 +255,12 @@ delay_string_append_char(RewriteQuery *message,String *str, char *parts)
if(message->r_code == SELECT_DEFAULT && message->ignore_rewrite == -1)
{
if(parts)
+ {
string_append_char( str, parts);
+#ifdef DEBUG
+ pool_debug("debug Rewrite Query %s", str->data);
+#endif
+ }
else
message->r_code = SELECT_RELATION_ERROR;
}
@@ -2162,6 +2167,7 @@ ConvertFromUsingToON(RewriteQuery *message, String *str, JoinExpr *node, int sel
lname = GetTableName(node->larg);
rname = GetTableName(node->rarg);
+ delay_string_append_char(message, str, " ON");
foreach (lc, node->using)
{
Value *value;
@@ -2293,11 +2299,10 @@ _rewriteJoinExpr(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
using_length= list_length(node->using);
using_list = (char **) palloc(sizeof(char *) * using_length);
- if(message->r_code == SELECT_DEFAULT
+ if(message->r_code == SELECT_DEFAULT && message->rewritelock == -1
&& message->analyze[select_num]->state == 'S')
{
/* Rewrite Using Cluase to On Clause */
- delay_string_append_char(message, str, " ON");
ConvertFromUsingToON(message, str, node, select_num);
} else {
delay_string_append_char(message, str, " USING(");
@@ -2312,9 +2317,7 @@ _rewriteJoinExpr(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
delay_string_append_char(message, str, ",");
value = lfirst(lc);
- delay_string_append_char(message, str, "\"");
delay_string_append_char(message, str, value->val.str);
- delay_string_append_char(message, str, "\"");
using_list[count] = value->val.str;
count++;
}
@@ -2746,7 +2749,7 @@ static void writeSelectAggHeader(RewriteQuery *message,ConInfoTodblink *dblink,
ret_count = agg->t_num + agg->c_num + agg->h_num;
sprintf(port,"%d",dblink->port);
- pool_debug("writeSelectHeader select_no=%d state=%d",message->current_select,state);
+ pool_debug("writeSelectAggHeader select_no=%d state=%d",message->current_select,state);
delay_string_append_char(message, str, "dblink(");
delay_string_append_char(message, str, "'");
@@ -3283,6 +3286,7 @@ _rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
if(message->r_code == SELECT_DEFAULT && message->rewritelock == -1
&& message->ignore_rewrite == -1)
{
+#if 0
if(analyze->state == 'P')
{
if(analyze->call_part != SELECT_FROMCLAUSE || !CheckUnionFromClause(message))
@@ -3294,6 +3298,7 @@ _rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
else
lock =true;
}
+#endif
if(analyze->state == 'L')
{
if(analyze->call_part != SELECT_FROMCLAUSE && !CheckUnionFromClause(message))
@@ -3743,6 +3748,9 @@ _rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
pool_debug("_rewriteSelectStmt select_no=%d state=%s",message->current_select,analyze->partstate);
+ if(strstr(analyze->partstate,"E"))
+ message->is_loadbalance = true;
+
/* change state */
if(count != 0)
{
@@ -4842,12 +4850,16 @@ static bool GetPoolColumn(RewriteQuery *message,String *str,char *table_name,cha
char buf[16];
snprintf(buf, 16, "%d", virtual->column_no[i]);
- if(analyze_now->partstate[SELECT_FROMCLAUSE] != 'S')
+#if 0
+ if(analyze_now->partstate[SELECT_FROMCLAUSE] != 'S') {
delay_string_append_char(message, str, analyze->table_name);
- else
+ delay_string_append_char(message, str, ".");
+ } else {
delay_string_append_char(message, str, virtual->table_list[i]);
-
- delay_string_append_char(message, str, ".pool_c$");
+ delay_string_append_char(message, str, ".");
+ }
+#endif
+ delay_string_append_char(message, str, "pool_c$");
delay_string_append_char(message, str, buf);
return true;
diff --git a/pool_rewrite_query.c b/pool_rewrite_query.c
index 6e81a2bc8..f0b7c273d 100644
--- a/pool_rewrite_query.c
+++ b/pool_rewrite_query.c
@@ -381,7 +381,7 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
backend->info->database);
}
}
- pool_debug("pool_rewrite_stmt: XXX message_code %d",message->r_code);
+ pool_debug("pool_rewrite_stmt: select message_code %d",message->r_code);
}
break;
@@ -425,7 +425,7 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
break;
}
- pool_debug("pool_rewrite_stmt: XXX rule %d",node->type);
+ pool_debug("pool_rewrite_stmt: query rule %d",node->type);
return message;
}
diff --git a/pool_stream.c b/pool_stream.c
index da37881aa..1cbe37bcb 100644
--- a/pool_stream.c
+++ b/pool_stream.c
@@ -199,100 +199,6 @@ 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;
-
- consume_size = consume_pending_data(cp, buf, len);
- len -= consume_size;
- buf += consume_size;
-
- while (len > 0)
- {
- if (pool_check_fd(cp))
- {
- 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: 6f6d26798300]