pgpool2.git / summary / log / commit / refs
commit 4bd655c9db360c2afb99dc32577f578ec095f09c
Author: Yoshiharu Mori <y-mori at sraoss.co.jp>
Date: Thu Jan 22 09:16:37 2009 +0000
add the comment related to parallel query
pool_process_query.c | 21 +++++++++++++++++---
pool_proto_modules.c | 23 +++++++++++++++++----
pool_rewrite_query.c | 56 ++++++++++++++++++++++++++++++++++++++++++++++------
pool_rewrite_query.h | 11 ++++++++++-
4 files changed, 97 insertions(+), 14 deletions(-)
diff --git a/pool_process_query.c b/pool_process_query.c
index 14a667013..bcc48e6c3 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -468,6 +468,11 @@ POOL_STATUS pool_process_query(POOL_CONNECTION *frontend,
}
+/*
+ * set_fd,isset_fs,zero_fd are used
+ * for check fd in parallel mode
+ */
+
/* used only in pool_parallel_exec */
#define BITS (8 * sizeof(long int))
@@ -499,7 +504,10 @@ static void zero_fd(unsigned long *setp)
}
}
-
+/*
+ * This function transmits to a parallel Query, and does processing
+ * that receives the result to each back end.
+ */
POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,
POOL_CONNECTION_POOL *backend, char *string,
Node *node,bool send_to_frontend)
@@ -547,7 +555,7 @@ POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,
return POOL_CONTINUE;
}
- /* In this loop,forward the query to the backend */
+ /* In this loop,forward the query to the all backends */
for (i=0;i<NUM_BACKENDS;i++)
{
if (!VALID_BACKEND(i))
@@ -584,7 +592,8 @@ POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,
}
zero_fd(donemask);
- /* In this loop,get data from backend */
+
+ /* In this loop, receive data from the all backends and send data to frontend */
for (;;)
{
FD_ZERO(&readmask);
@@ -1414,6 +1423,11 @@ void pool_send_frontend_exits(POOL_CONNECTION_POOL *backend)
* -------------------------------------------------------
*/
+/*
+ * This function transmits to a parallel Query to each backend,
+ * and receives the results from backends .
+ *
+ */
static POOL_STATUS ParallelForwardToFrontend(char kind, POOL_CONNECTION *frontend, POOL_CONNECTION *backend, char *database, bool send_to_frontend)
{
int len;
@@ -2687,6 +2701,7 @@ static POOL_STATUS do_error_execute_command(POOL_CONNECTION_POOL *backend, int n
}
/*
+ * Transmit an arbitrary Query to a specific node.
* This function is only used in parallel mode
*/
POOL_STATUS OneNode_do_command(POOL_CONNECTION *frontend, POOL_CONNECTION *backend, char *query, char *database)
diff --git a/pool_proto_modules.c b/pool_proto_modules.c
index 1c7d3f252..bc8a17abe 100644
--- a/pool_proto_modules.c
+++ b/pool_proto_modules.c
@@ -226,11 +226,20 @@ POOL_STATUS NotificationResponse(POOL_CONNECTION *frontend,
if (pool_config->parallel_mode)
{
- /* analyze query */
+ /* The Query is analyzed first in a parallel mode(in_parallel_query),
+ * and, next, the Query is rewritten(rewrite_query_stmt).
+ */
+
+ /* analyze the query */
RewriteQuery *r_query = is_parallel_query(node,backend);
if(r_query->is_loadbalance)
{
+ /* Usual processing of pgpool is done by using the rewritten Query
+ * if judged a possible load-balancing as a result of analyzing
+ * the Query.
+ * Of course, the load is distributed only for load_balance_mode=true.
+ */
if(r_query->r_code == SEND_LOADBALANCE_ENGINE)
{
/* use rewritten query */
@@ -241,8 +250,11 @@ POOL_STATUS NotificationResponse(POOL_CONNECTION *frontend,
pool_debug("SimpleQuery: loadbalance_query =%s",string);
}
else if (r_query->is_parallel)
- {
- /* call parallel exe engine */
+ {
+ /*
+ * For the Query that the parallel processing is possible.
+ * Call parallel exe engine and return status to the upper layer.
+ */
POOL_STATUS stats = pool_parallel_exec(frontend,backend,r_query->rewrite_query, node,true);
free_parser();
in_progress = 0;
@@ -250,8 +262,8 @@ POOL_STATUS NotificationResponse(POOL_CONNECTION *frontend,
}
else if(!r_query->is_pg_catalog)
{
+ /* rewrite query and execute */
r_query = rewrite_query_stmt(node,frontend,backend,r_query);
- /* rewrite query phase */
if(r_query->type == T_InsertStmt)
{
free_parser();
@@ -268,6 +280,9 @@ POOL_STATUS NotificationResponse(POOL_CONNECTION *frontend,
return r_query->status;
}
}
+ /*
+ * The same processing as usual pgpool is done to other Query type.
+ */
}
/* check COPY FROM STDIN
diff --git a/pool_rewrite_query.c b/pool_rewrite_query.c
index f0b7c273d..f8a0ca811 100644
--- a/pool_rewrite_query.c
+++ b/pool_rewrite_query.c
@@ -112,6 +112,9 @@ static int getInsertRule(ListCell *lc,List *list_t ,DistDefInfo *info,int div_ke
return node_number;
}
+/*
+ * This function processes the decision whether to distribute the insert sentence to the node.
+ */
static void examInsertStmt(Node *node,POOL_CONNECTION_POOL *backend, RewriteQuery *message)
{
RangeVar *table;
@@ -338,6 +341,10 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
case T_SelectStmt:
{
SelectStmt *stmt = (SelectStmt *)node;
+
+ /* Because "SELECT INTO" cannot be used in a parallel mode,
+ * the error message is generated and send "ready for query" to frontend.
+ */
if(stmt->intoClause)
{
pool_send_error_message(frontend, MAJOR(backend), "XX000",
@@ -351,11 +358,18 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
break;
}
+ /*
+ * The Query is actually rewritten based on analytical information on the Query.
+ */
examSelectStmt(node,backend,message);
if (message->r_code != SELECT_PGCATALOG &&
message->r_code != SELECT_RELATION_ERROR)
{
+ /*
+ * The rewritten Query is transmitted to system db,
+ * and execution status is received.
+ */
POOL_CONNECTION_POOL_SLOT *system_db = pool_system_db_connection();
message->status = OneNode_do_command(frontend,
system_db->con,
@@ -367,12 +381,21 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
if(TSTATE(backend) == 'T' &&
message->r_code == SELECT_RELATION_ERROR)
{
- pool_debug("pool_rewrite_stmt(select): inside transaction");
+ /*
+ * In the case of message->r_code == SELECT_RELATION_ERROR and in the transaction,
+ * Transmit the Query to all back ends, and to abort transaction.
+ */
+ pool_debug("pool_rewrite_stmt(select): Inside transaction. abort transaction");
message->rewrite_query = nodeToString(node);
message->status = pool_parallel_exec(frontend,backend,message->rewrite_query,node,true);
}
else
{
+ /*
+ * Ohter cases of message->r_code == SELECT_RELATION_ERROR
+ * or SELECT_PG_CATALOG,
+ * Transmit the Query to Master node and receive status.
+ */
pool_debug("pool_rewrite_stmt: executed by Master");
message->rewrite_query = nodeToString(node);
message->status = OneNode_do_command(frontend,
@@ -386,10 +409,13 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
break;
case T_InsertStmt:
+
+ /* The distribution of the INSERT sentence. */
examInsertStmt(node,backend,message);
if(message->r_code == 0 )
{
+ /* send the INSERT sentence */
message->status = OneNode_do_command(frontend,
CONNECTION(backend,message->r_node),
message->rewrite_query,
@@ -397,6 +423,7 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
}
else if (message->r_code == INSERT_SQL_RESTRICTION)
{
+ /* Restriction case of INSERT sentence */
pool_send_error_message(frontend, MAJOR(backend), "XX000",
"pgpool2 sql restriction",
message->rewrite_query, "", __FILE__,
@@ -404,11 +431,13 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
if(TSTATE(backend) == 'T')
{
- pool_debug("rewrite_query_stmt(insert): inside transaction");
+ /* In Transaction, send the invalid message to backend to abort this transaction */
+ pool_debug("rewrite_query_stmt(insert): Inside transaction. Abort transaction");
message->status = pool_parallel_exec(frontend,backend, "POOL_RESET_TSTATE",node,false);
}
else
{
+ /* return "ready for query" to frontend */
pool_send_readyforquery(frontend);
message->status=POOL_CONTINUE;
}
@@ -462,6 +491,11 @@ static int check_whereClause(Node *where)
}
}
+/*
+ * After analyzing query, check the analyze[0]->state.
+ * if the analyze[0]->state ==`P`, this query can be executed
+ * on parallel engine.
+ */
static int direct_parallel_query(RewriteQuery *message)
{
if(message && message->analyze[0] && message->analyze[0]->state == 'P')
@@ -508,6 +542,12 @@ void analyze_debug(RewriteQuery *message)
}
}
+/*
+ * This function checks the KEYWORD(POOL_PARALLEL,POOL_LOADBALANCE)
+ * if the special function(like pool_parallel() or pool_loadbalance())
+ * is used, mark the r_code,is_parallel and is_loadbalance.
+ * In othe cases, It is necessary to analyze the Query.
+ */
RewriteQuery *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
{
static RewriteQuery message;
@@ -523,7 +563,7 @@ RewriteQuery *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
stmt = (SelectStmt *) node;
- /* LOADBALANCE OR PARALLEL */
+ /* Check the special function is used in this query*/
if (!(stmt->distinctClause || stmt->intoClause ||
stmt->fromClause || stmt->groupClause || stmt->havingClause ||
stmt->sortClause || stmt->limitOffset || stmt->limitCount ||
@@ -540,7 +580,7 @@ RewriteQuery *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
Node *func_args = (Node *) lfirst(list_head(func->args));
message.rewrite_query = delimistr(nodeToString(func_args));
- /* PARALLEL */
+ /* pool_parallel() is used in this query */
if(strcmp(strVal(lfirst(list_head(func->funcname))),
POOL_PARALLEL) == 0)
{
@@ -550,7 +590,7 @@ RewriteQuery *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
pool_debug("can pool_parallel_exec %s",message.rewrite_query);
return &message;
}
- else /* LOADBALANCE */
+ else /* pool_loadbalance() is used in this query */
if(strcmp(strVal(lfirst(list_head(func->funcname))),
POOL_LOADBALANCE) == 0)
{
@@ -586,9 +626,12 @@ RewriteQuery *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
return &message;
}
+ /* Analyzing Query Start */
analyze_debug(&message);
- /* PARALLEL*/
+ /* After the analyzing query,
+ * this query can be executed as parallel exec, is_parallel flag is turned on
+ */
direct_ok = direct_parallel_query(&message);
if(direct_ok == 1)
{
@@ -601,6 +644,7 @@ RewriteQuery *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
}
else if (IsA(node, CopyStmt))
{
+ /* For Copy Statement, check the table name, mark the is_parallel flag. */
CopyStmt *stmt = (CopyStmt *)node;
if (stmt->is_from == FALSE && stmt->filename == NULL)
diff --git a/pool_rewrite_query.h b/pool_rewrite_query.h
index e6cb43ca4..09f103ea8 100644
--- a/pool_rewrite_query.h
+++ b/pool_rewrite_query.h
@@ -51,6 +51,7 @@
#define SELECT_ANALYZE 19
#define SELECT_DEFAULT_PREP 20
+/* build TARGET-LIST */
typedef struct {
char **col_list; /* column list */
char **type_list; /* type list */
@@ -59,6 +60,7 @@ typedef struct {
bool valid;
}SelectDefInfo;
+/* This struct is used as TARGET-LIST of Sub-Select */
typedef struct {
DistDefInfo *distinfo;
RepliDefInfo *repliinfo;
@@ -68,6 +70,7 @@ typedef struct {
int ret_num;
} RangeInfo;
+/* build Virtual Table of FROM-Cluase */
typedef struct {
char **col_list; /* column list */
char **type_list; /* type list */
@@ -78,6 +81,7 @@ typedef struct {
int col_num;
} VirtualTable;
+/* this struct is used by JOIN Expr */
typedef struct {
char **col_list; /* column list */
char **type_list; /* type list */
@@ -89,6 +93,7 @@ typedef struct {
int using_length;
} JoinTable;
+/* this struct is used in optimization of aggregate opr */
typedef struct {
ColumnRef **usec_p; /* targetlist columns */
FuncCall **tfunc_p; /* targetlist funcs */
@@ -105,7 +110,7 @@ typedef struct {
bool opt;
} Aggexpr;
-
+/* main struct of alanyzing query */
typedef struct {
int now_select;
int part;
@@ -130,6 +135,9 @@ typedef struct {
SelectDefInfo *select_ret;
} AnalyzeSelect;
+/* This struct is used as Information that relates
+ * to distribution processing of parallel query
+ */
typedef struct {
int r_code;
int r_node;
@@ -156,6 +164,7 @@ typedef struct {
AnalyzeSelect **analyze;
} RewriteQuery;
+/* This info is used in dblink */
typedef struct {
char *hostaddr;
char *dbname;
[parent: 034476f4ba2c]