pgpool2.git / summary / log / commit / refs
commit 51edf6239e2cba7ea674e3fa2ff9ffbaa2539fc1
Author: Yoshiharu Mori <y-mori at sraoss.co.jp>
Date: Mon Oct 15 06:56:13 2007 +0000
ADD query rewriting to main trunc
pool_process_query.c | 31 +-
pool_rewrite_outfuncs.c | 3992 ++++++++++++++++++++++++++++++++++-------------
pool_rewrite_query.c | 317 ++--
pool_rewrite_query.h | 87 +-
pool_system.c | 220 ++-
pool_type.h | 27 +-
6 files changed, 3410 insertions(+), 1264 deletions(-)
diff --git a/pool_process_query.c b/pool_process_query.c
index 4d46657b3..f009eb47a 100644
--- a/pool_process_query.c
+++ b/pool_process_query.c
@@ -918,21 +918,32 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
if (pool_config->parallel_mode)
{
- char *parallel_query = NULL;
+ /* analze query */
+ RewriteQuery *r_query = is_parallel_query(node,backend);
- /* Do select pool_parallel ? */
- parallel_query = is_parallel_query(node,backend);
- if (parallel_query)
+ if(r_query->is_loadbalance)
{
- POOL_STATUS stats = pool_parallel_exec(frontend,backend,parallel_query, node,true);
+ if(r_query->r_code == SEND_LOADBALANCE_ENGINE)
+ {
+ /* use rewrited query */
+ string = r_query->rewrite_query;
+ /* change query length */
+ len = strlen(string)+1;
+ }
+ pool_debug("SimpleQuery: loadbalance_query =%s",string);
+ }
+ else if (r_query->is_parallel)
+ {
+ /* call parallel exe engine */
+ POOL_STATUS stats = pool_parallel_exec(frontend,backend,r_query->rewrite_query, node,true);
free_parser();
return stats;
}
-
- /* rewrite_query_phase */
+ else if(!r_query->is_pg_catalog)
{
- RewriteQuery *r_query = rewrite_query_stmt(node,frontend,backend);
- if(r_query->type == T_InsertStmt)
+ r_query = rewrite_query_stmt(node,frontend,backend,r_query);
+ /* rewrite query phase */
+ if(r_query->type == T_InsertStmt && r_query->r_code != INSERT_DIST_NO_RULE)
{
free_parser();
return r_query->status;
@@ -942,7 +953,7 @@ static POOL_STATUS SimpleQuery(POOL_CONNECTION *frontend,
free_parser();
return r_query->status;
}
- }
+ }
}
/* check COPY FROM STDIN
diff --git a/pool_rewrite_outfuncs.c b/pool_rewrite_outfuncs.c
index f167f84db..a9b1c0cc1 100644
--- a/pool_rewrite_outfuncs.c
+++ b/pool_rewrite_outfuncs.c
@@ -31,7 +31,6 @@
#define booltostr(x) ((x) ? "true" : "false")
-static void KeepRewriteQueryCode(RewriteQuery *message, int r_code);
static void _rewriteNode(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, void *obj);
static void _rewriteList(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *node);
@@ -173,21 +172,51 @@ static void _rewriteOperatorArgTypes(Node *BaseSelect, RewriteQuery *message, Co
static void _rewriteRangeFunction(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, RangeFunction *node);
static void _rewriteWithDefinition(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *def_list);
-/* use optimization of pool_parallel */
-static int checkFuncArgs(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *list);
-static int _rewriteWhereClause(Node *BaseSelect,RewriteQuery *message,ConInfoTodblink *dblink,String *str,int true_count);
-static int checkSelectStmtOneTable(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SelectStmt *node);
-static int checkIntensiveOneTable(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SelectStmt *node);
-static void compareGroupby(RewriteQuery *message,List *targetList,List *groupby,String *str,int record);
-static void _rewriteIntensivedblink(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *list);
-static void _rewriteReturnRecords(Node *BaseSelect,RewriteQuery *message,List *list,String *str);
-static void _rewriteIntensiveTargetList(RewriteQuery *message,List *list,String *str);
-static int _checkDistDefColumn(ColumnRef *col,RewriteQuery *message);
+/* analyze */
+static void KeepRewriteQueryCode(RewriteQuery *message, int current_select);
+static void KeepMessages(RewriteQuery *message,int current_select,int part);
+static int CheckWhereCaluse(Node *BaseSelect,RewriteQuery *message,ConInfoTodblink *dblink,String *str,int true_count);
static int _writewhereClause(A_Expr *expr,RewriteQuery *message,ConInfoTodblink *dblink, String *str,int true_count);
-static void _rewriteSelectStmtIntensive(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SelectStmt *node);
static char *escape_string(char *str);
static void delay_string_append_char(RewriteQuery *message,String *str, char *parts);
-static int checkGroupbyClause(List *list, char *colname);
+static void build_range_info(RewriteQuery *message,DistDefInfo *info,RepliDefInfo *info2,SelectDefInfo *info3,char *alias,int select_num,int i_num);
+static void AnalyzeReturnRecord(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *list);
+static void build_virtual_table(RewriteQuery *message,void *obj, int next);
+static char *search_type_from_virtual(VirtualTable *virtual,char *table,char *col);
+static int _checkVirtualColumn(ColumnRef *col,RewriteQuery *message);
+/* rewirte */
+static void writeSelectHeader(RewriteQuery *message,ConInfoTodblink *dblink, String *str,int parallel,int state);
+static void writeSelectFooter(RewriteQuery *message,String *str,AnalyzeSelect *analyze,int state);
+static void KeepRewriteQueryReturnCode(RewriteQuery *message, int r_code);
+static void writeRangeHeader(RewriteQuery *message,ConInfoTodblink *dblink, String *str,DistDefInfo *info, RepliDefInfo *info2,char *alias);
+static void writeRangeFooter(RewriteQuery *message,ConInfoTodblink *dblink, String *str,DistDefInfo *info, RepliDefInfo *info2,char *alias);
+
+/* under define is used in _rewritejoinExpr */
+#define JDEFAULT 0
+#define JNORMAL 1
+#define JNATURAL_INNER 2
+#define JNATURAL_LEFT 3
+#define JNATURAL_RIGHT 4
+#define JNATURAL_FULL 5
+#define JINNER 6
+#define JLEFT 7
+#define JRIGHT 8
+#define JFULL 9
+#define JUSING 10
+
+#define LOADBALANCE false
+#define PARALLEL true
+
+#define SELECT_START -1
+#define SELECT_FROMCLAUSE 0
+#define SELECT_TARGETLIST 1
+#define SELECT_WHERECLAUSE 2
+#define SELECT_GROUPBYCLAUSE 3
+#define SELECT_HAVINGCLAUSE 4
+#define SELECT_SORTCLAUSE 5
+#define SELECT_OFFSETCLAUSE 6
+#define SELECT_LIMITCLAUSE 7
+#define SELECT_OTHER 8
static char *escape_string(char *str)
{
@@ -208,7 +237,7 @@ static char *escape_string(char *str)
else if (str[i] == '\\')
{
es[j++] = '\\';
- }
+ }
es[j] = str[i];
}
@@ -218,1000 +247,1390 @@ static char *escape_string(char *str)
static void
delay_string_append_char(RewriteQuery *message,String *str, char *parts)
{
- if(message->r_code == SELECT_DEFAULT ||
- message->r_code == SELECT_DEFAULT_INSIDE_DBLINK ||
- message->r_code == SELECT_ONETABLE)
+ if(message->r_code == SELECT_DEFAULT && message->ignore_rewrite == -1)
string_append_char(str, parts);
}
+
/*
- * check Group by Clause
- * if return value is zero,
- * cannot do intensive optimization.
- */
+ * Is this column member of table (message->table_relname) ?
+ * return 1 -- member
+ * return 0 or -1 -- not mmeber
+ */
static int
-checkGroupbyClause(List *list, char *colname)
+_checkVirtualColumn(ColumnRef *col,RewriteQuery *message)
{
- ListCell *lc;
- int ok_group_by = 0;
- foreach(lc ,list)
- {
- Node *n = lfirst(lc);
- if(!IsA(n,ColumnRef))
- return 0;
-
- if (IsA(n, ColumnRef))
- {
- ColumnRef *col = NULL;
- col = (ColumnRef *) n;
- if(colname &&
- strcmp(colname, strVal(lfirst(list_head(col->fields)))) == 0)
- {
- ok_group_by++;
- return ok_group_by;
- } else if (!colname) {
- ok_group_by++;
- }
- }
- }
- return ok_group_by;
-}
+ ListCell *c;
+ List *list;
+ VirtualTable *virtual = NULL;
+ AnalyzeSelect *analyze = NULL;
+ char first = 0;
+ char *tmp_table = NULL;
+ char *colname = NULL;
+ char *table = NULL;
+ int check_col = 0;
+ int no = message->current_select;
+ int v_colnum;
+ int i;
-/*
- * Include function,sub-query
- * and other table member in function args?
- */
-int
-checkFuncArgs(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *list)
-{
- void *obj;
- int message_r_code = message->r_code;
- if(!list)
- return 0;
+ list = col->fields;
+ analyze = message->analyze[no];
- if(IsA(lfirst(list_head(list)), TypeCast))
+ if(list->length > 2 || list->length == 0)
{
- TypeCast *type = (TypeCast *) lfirst(list_head(list));
- obj = type->arg;
- } else {
- obj = lfirst(list_head(list));
+ /* send error message */
+ return -1;
}
-
- if(IsA(obj,ColumnRef))
+
+ foreach (c, col->fields)
{
- ColumnRef *col = (ColumnRef *) obj;
- KeepRewriteQueryCode(message, SELECT_AEXPR);
- _rewriteNode(NULL, message, NULL, str, col);
- if(message->r_code == SELECT_AEXPR)
+ Node *n = (Node *) lfirst(c);
+
+ if (IsA(n, String))
{
- KeepRewriteQueryCode(message, message_r_code);
- return 1;
- } else {
- KeepRewriteQueryCode(message, message_r_code);
- return 0;
+ Value *v = (Value *) lfirst(c);
+ if(list->length == 2 && first == 0)
+ {
+ first = 1;
+ tmp_table = v->val.str;
+ }
+ else
+ colname = v->val.str;
}
}
- else if(IsA(obj,A_Expr))
- {
- A_Expr *expr = (A_Expr *) obj;
- if(expr->kind != AEXPR_OP)
- return 0;
-
- if (list_length(expr->name) == 1)
+
+ if(analyze->partstate[SELECT_FROMCLAUSE] == 'S')
+ {
+ if (message->table_relname)
{
- KeepRewriteQueryCode(message, SELECT_AEXPR);
- _rewriteNode(NULL, message, NULL, str, expr->lexpr);
- if(message->r_code == SELECT_AEXPR)
+ if(tmp_table && strcmp(message->table_relname,tmp_table) != 0)
{
- _rewriteNode(NULL, message, NULL, str, expr->rexpr);
- if(message->r_code == SELECT_AEXPR)
- {
- _rewriteNode(NULL, message, NULL, str, expr->rexpr);
- KeepRewriteQueryCode(message, message_r_code);
- return 1;
- } else {
- KeepRewriteQueryCode(message, message_r_code);
- return 0;
- }
- } else {
- KeepRewriteQueryCode(message, message_r_code);
- return 0;
+ return check_col;
}
- }
+ table = message->table_relname;
+ }
else
{
- KeepRewriteQueryCode(message, message_r_code);
- return 0;
+ return check_col;
}
- } else {
- return 0;
- }
-}
-
-/*
- * if can Intensive Optimization,
- * return value is 1
- */
-int
-checkIntensiveOneTable(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SelectStmt *node)
-{
- RangeVar *var = NULL;
- int message_r_code = message->r_code;
-
- /* Intensive optimization can do when tree have only one table */
- if(!((list_length(node->fromClause) == 1)
- && IsA(lfirst(list_head(node->fromClause)),RangeVar)))
- {
- return 0;
- }
-
- /* check groupClause's tree has only Column's list */
- if(node->groupClause &&
- checkGroupbyClause(node->groupClause, NULL) == 0)
- {
- return 0;
- }
-
- /* TODO: can use intensive optimization if having Clause exits */
- if (node->havingClause)
- return 0;
-
- var = (RangeVar *) lfirst(list_head(node->fromClause));
-
- /* set dblink info to message */
- message->table_relname = var->relname;
- if(var->schemaname)
- message->schemaname = var->schemaname;
- else
- message->schemaname = NULL;
-
- if(var->alias)
- {
- Alias *alias_p = (Alias *) var->alias;
- message->table_alias = alias_p->aliasname;
- } else {
- message->table_alias = NULL;
- }
- message->dbname = dblink->dbname;
-
- /* search target list */
- if(node->targetList)
- {
- ListCell *lc;
- foreach (lc, node->targetList)
- {
- ResTarget *target = NULL;
- Node *n = lfirst(lc);
- void *obj;
- if (!IsA(n, ResTarget))
- {
- return 0;
- }
-
- target = (ResTarget *) n;
+ virtual = analyze->virtual;
+ v_colnum = virtual->col_num;
- if(!target->val)
- {
- return 0;
- }
- /* check type cast */
- if(IsA(target->val, TypeCast))
+ for(i = 0; i < v_colnum; i++)
+ {
+ if(!strcmp(virtual->table_list[i],table) &&
+ !strcmp(virtual->col_list[i],colname))
{
- TypeCast *type = (TypeCast *) target->val;
- obj = type->arg;
- } else {
- obj = target->val;
+ check_col = 1;
+ break;
}
+ }
+ } else {
+ virtual = analyze->virtual;
+ v_colnum = virtual->col_num;
- /* check function, max,min,count and sum can do */
- if (obj && (IsA(obj, FuncCall)))
+ for(i = 0; i < v_colnum; i++)
+ {
+ if(tmp_table)
{
- FuncCall *func = (FuncCall *) obj;
- char *funcname = strVal(lfirst(list_head(func->funcname)));
-
- if(strcmp(funcname,"max") == 0 || strcmp(funcname,"min") == 0)
+ if(!strcmp(virtual->table_list[i],tmp_table) &&
+ !strcmp(virtual->col_list[i],colname))
{
- /* check function args: args must not be function or sub-query */
- if(checkFuncArgs(BaseSelect,message,dblink,str,func->args) == 0)
- return 0;
- }
- else if (strcmp(funcname,"count") == 0 || strcmp(funcname,"sum") == 0)
- {
- /* check function args: args must not be function or sub-query */
- if(checkFuncArgs(BaseSelect,message,dblink,str,func->args) == 0 && func->agg_star != TRUE)
- {
- return 0;
- }
- } else {
- return 0;
+ check_col = 1;
+ break;
}
}
- /* check ColumnRef */
- else if (obj && (IsA(obj, ColumnRef)))
+ else
{
- ColumnRef *col = NULL;
- col = (ColumnRef *) obj;
- /* if ColumnRef is exits, groupCluase's tree must be exits */
- if(!node->groupClause)
+ if(!strcmp(virtual->col_list[i],colname))
{
- return 0;
- }
- else
- {
- /* Is this Column member of this table ? */
- KeepRewriteQueryCode(message, SELECT_AEXPR);
- _rewriteNode(NULL, message, NULL, str, col);
-
- if(message->r_code != SELECT_AEXPR)
- {
- KeepRewriteQueryCode(message, message_r_code);
- return 0;
- }
-
- KeepRewriteQueryCode(message, message_r_code);
-
- /* Is this Column member in Groupby ? */
- if(checkGroupbyClause(node->groupClause, strVal(lfirst(list_head(col->fields)))) == 0)
- return 0;
+ check_col = 1;
+ break;
}
- } else {
- return 0;
}
}
}
- return 1;
+ return check_col;
+}
+
+static void KeepMessages(RewriteQuery *message,int current_select,int part)
+{
+ message->current_select = current_select;
+ message->part = part;
+}
+
+static void KeepRewriteQueryCode(RewriteQuery *message, int current_select)
+{
+ message->current_select = current_select;
+}
+
+static void KeepRewriteQueryReturnCode(RewriteQuery *message, int r_code)
+{
+ if((message->r_code != SELECT_RELATION_ERROR)
+ && (message->r_code != SELECT_PGCATALOG))
+ message->r_code = r_code;
}
/*
- * compare targetList to groupbylist
+ * A_Expr check
+ * if A_Expr-tree use subquery,function,or other table member
+ * rewrite expression as "TRUE"
+ *
+ * return value is counter of rewriting query.
*/
-static void
-compareGroupby(RewriteQuery *message,List *targetList,List *groupby,String *str,int record)
-{
- ListCell *lc_groupby;
- int counter = 0;
-
- foreach(lc_groupby,groupby)
- {
- char *colname;
- Node *n = lfirst(lc_groupby);
- ColumnRef *col = NULL;
- ListCell *lc_targetList;
- int write_flag =0;
-
- col = (ColumnRef *) n;
- colname = strVal(lfirst(list_head(col->fields)));
- foreach(lc_targetList,targetList)
- {
- ResTarget *target = NULL;
- ColumnRef *col = NULL;
- Node *n = lfirst(lc_targetList);
- target = (ResTarget *) n;
- if (!(target->val && (IsA(target->val, ColumnRef))))
- continue;
- col = (ColumnRef *) target->val;
-
- if(strcmp(colname,strVal(lfirst(list_head(col->fields)))) == 0);
- {
- write_flag = 1;
- break;
- }
- }
-
- if(write_flag == 0)
- {
- if(record == 0)
- {
- string_append_char(str, " ,");
- string_append_char(str, colname);
- string_append_char(str, " ");
- } else {
- DistDefInfo *info = pool_get_dist_def_info(message->dbname, message->schemaname, message->table_relname);
- int exist_col = 0;
- int i;
- string_append_char(str, " ,");
- string_append_char(str, colname);
- string_append_char(str, " ");
-
- if(!info)
- {
- message->r_code = SELECT_RELATION_ERROR;
- return;
- }
+static int
+_writewhereClause(A_Expr *expr,RewriteQuery *message,ConInfoTodblink *dblink, String *str,int true_count)
+{
+ int message_r_code = message->r_code;
- for(i = 0; i < info->col_num; i++)
+ switch (expr->kind)
+ {
+ case AEXPR_OP:
+ if (list_length(expr->name) == 1)
+ {
+ KeepRewriteQueryReturnCode(message, SELECT_AEXPR);
+ _rewriteNode(NULL, message, dblink, str, expr->lexpr);
+ if(message->r_code == SELECT_AEXPR)
{
- if(strcmp(info->col_list[i],colname) == 0)
+ _rewriteNode(NULL, message, dblink, str, expr->rexpr);
+ if(message->r_code == SELECT_AEXPR)
{
- delay_string_append_char(message, str, " ");
- delay_string_append_char(message, str, info->type_list[i]);
- exist_col = 1;
+ Value *op = (Value *) lfirst(list_head(expr->name));
+ KeepRewriteQueryReturnCode(message, message_r_code);
+ _rewriteNode(NULL, message, dblink, str, expr->lexpr);
+ delay_string_append_char(message, str, op->val.str);
+ _rewriteNode(NULL, message, dblink, str, expr->rexpr);
+ KeepRewriteQueryReturnCode(message, message_r_code);
+ } else {
+ KeepRewriteQueryReturnCode(message, message_r_code);
+ delay_string_append_char(message, str,"TRUE");
+ true_count++;
break;
}
+ } else {
+ KeepRewriteQueryReturnCode(message, message_r_code);
+ delay_string_append_char(message, str,"TRUE");
+ true_count++;
}
- if(exist_col != 1)
- {
- message->r_code = SELECT_RELATION_ERROR;
- return;
- }
+ break;
}
- }
- counter++;
- }
-}
+ else
+ {
+ delay_string_append_char(message, str,"TRUE");
+ true_count++;
+ }
+ break;
-static void
-_rewriteIntensivedblink(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *list)
-{
- ListCell *lc;
- int list_num = 1;
- SelectStmt *select = (SelectStmt *)BaseSelect;
+ case AEXPR_AND:
+ delay_string_append_char(message, str, " (");
+ true_count = true_count + CheckWhereCaluse(expr->lexpr,message,dblink,str,true_count);
+ delay_string_append_char(message, str, " AND ");
+ true_count = true_count + CheckWhereCaluse(expr->rexpr,message,dblink,str,true_count);
+ delay_string_append_char(message, str, ")");
+ break;
- foreach(lc,list)
- {
- ResTarget *target = NULL;
- char *typecast = NULL;
- Node *n = lfirst(lc);
- void *obj;
+ case AEXPR_OR:
+ delay_string_append_char(message, str, " (");
+ true_count = true_count + CheckWhereCaluse(expr->lexpr,message,dblink,str,true_count);
+ delay_string_append_char(message, str, " OR ");
+ true_count = true_count + CheckWhereCaluse(expr->rexpr,message,dblink,str,true_count);
+ delay_string_append_char(message, str, ")");
+ break;
- target = (ResTarget *) n;
+ case AEXPR_OP_ANY:
+ /* not implemented yet */
+ break;
- if(IsA(target->val, TypeCast))
- {
- TypeCast *type = (TypeCast *) target->val;
- TypeName *typename = (TypeName *)type->typename;
- obj = type->arg;
- typecast = strVal(lfirst(list_head(typename->names)));
- } else {
- obj = target->val;
- }
+ case AEXPR_OP_ALL:
+ /* not implemented yet */
+ break;
+#if 0
- if (obj && (IsA(obj, FuncCall)))
- {
- FuncCall *func = (FuncCall *) obj;
- _rewriteNode(BaseSelect, message, dblink, str, func);
- }
- else if (target->val && (IsA(target->val, ColumnRef)))
- {
- ColumnRef *col = NULL;
- col = (ColumnRef *) target->val;
- _rewriteNode(BaseSelect, message, dblink, str, col);
- }
+ case AEXPR_NOT:
+ delay_string_append_char(message, str, " (NOT ");
+ CheckWhereCaluse(expr->rexpr,str,tablename,dbname, schemaname,aliasname);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ case AEXPR_DISTINCT:
+ delay_string_append_char(message, str, " (");
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ delay_string_append_char(message, str, " IS DISTINCT FROM ");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ case AEXPR_NULLIF:
+ delay_string_append_char(message, str, " NULLIF(");
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ delay_string_append_char(message, str, ", ");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
- if(list_num != list_length(list))
- delay_string_append_char(message,str, ",");
- list_num++;
+ case AEXPR_OF:
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ if (*(char *)lfirst(list_head(node->name)) == '!')
+ delay_string_append_char(message, str, " IS NOT OF (");
+ else
+ delay_string_append_char(message, str, " IS OF (");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+#endif
+ default:
+ delay_string_append_char(message, str,"TRUE");
+ true_count++;
+ break;
}
- delay_string_append_char(message,str, " ");
- if(select->groupClause)
- compareGroupby(message,list,select->groupClause,str,0);
+ return true_count;
}
/*
- * Write down records from dblink()'return
+ * Start whereClasue check
+ * call _writewherelause();
+ *
+ * if messgae->r_code is SELECT DEFAULT,
+ * write down whereClause
+ * else
+ * check whereClause
+ *
+ * return value is counter of rewriting query
+ * if return value is zero, all whereclause can insert dblink function.
*/
-static void
-_rewriteReturnRecords(Node *BaseSelect,RewriteQuery *message,List *list,String *str)
+static int
+CheckWhereCaluse(Node *BaseSelect,RewriteQuery *message,ConInfoTodblink *dblink,String *str,int true_count)
{
- ListCell *lc;
- int list_first = 0;
- int args_num = 0;
- DistDefInfo *info = pool_get_dist_def_info(message->dbname, message->schemaname, message->table_relname);
- SelectStmt *select = (SelectStmt *)BaseSelect;
- ColumnRef *col = NULL;
- char *colname = NULL;
+ A_Expr *expr = NULL;
- foreach(lc,list)
+ if(!IsA(BaseSelect, A_Expr))
{
- ResTarget *target = NULL;
- char *typecast = NULL;
- char number[16];
- Node *n = lfirst(lc);
- void *obj;
-
- target = (ResTarget *) n;
+ delay_string_append_char(message, str, "TRUE");
+ true_count++;
+ return true_count;
+ }
- if(list_first !=0)
- delay_string_append_char(message,str, ",");
-
- if(IsA(target->val, TypeCast))
+ expr = (A_Expr *) BaseSelect;
+
+ if(expr->kind == AEXPR_NOT)
+ {
+ delay_string_append_char(message, str, "TRUE");
+ true_count++;
+ return true_count;
+ }
+ true_count = _writewhereClause(expr,message,dblink,str,true_count);
+ return true_count;
+}
+
+static void _rewriteIdList(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *node)
+{
+ ListCell *lc;
+ char first = 0;
+
+ foreach(lc, node)
+ {
+ Value *v = lfirst(lc);
+
+ if (first == 0)
+ first = 1;
+ else
+ delay_string_append_char(message, str, ", ");
+
+ delay_string_append_char(message, str, "\"");
+ delay_string_append_char(message, str, v->val.str);
+ delay_string_append_char(message, str, "\"");
+ }
+}
+
+static void _rewriteList(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *node)
+{
+ ListCell *lc;
+ char first = 0;
+ char state;
+ int loop = 0;
+ bool from;
+ int current_select = message->current_select;
+ int next = -1;
+
+ foreach(lc, node)
+ {
+ next = message->analyze_num;
+ if (first == 0)
{
- TypeCast *type = (TypeCast *) target->val;
- TypeName *typename = (TypeName *)type->typename;
- obj = type->arg;
- typecast = strVal(lfirst(list_head(typename->names)));
- } else {
- obj = target->val;
+ first = 1;
}
-
- if (obj && (IsA(obj, FuncCall)))
+ else
{
- FuncCall *func = (FuncCall *) obj;
- char *funcname = strVal(lfirst(list_head(func->funcname)));
-
- sprintf(number,"%d",args_num);
+ if(lfirst(lc) && !IsA(lfirst(lc),A_Indices))
+ delay_string_append_char(message, str, ",");
+ }
- if(strcmp(funcname,"max") == 0)
- {
- delay_string_append_char(message,str, " args_$");
- delay_string_append_char(message,str,number);
- delay_string_append_char(message,str, " numeric");
- }
- else if(strcmp(funcname,"min") == 0)
- {
- delay_string_append_char(message,str, " args_$");
- delay_string_append_char(message,str,number);
- delay_string_append_char(message,str, " numeric");
- }
- else if(strcmp(funcname,"sum") == 0)
- {
- delay_string_append_char(message,str, " args_$");
- delay_string_append_char(message,str,number);
- delay_string_append_char(message,str, " numeric");
- }
- else if(strcmp(funcname,"count") == 0)
+ /* init JoinTable */
+ if(message->r_code == SELECT_ANALYZE && message->fromClause)
+ {
+ if(lfirst(lc) && IsA(lfirst(lc),JoinExpr))
{
- delay_string_append_char(message,str, " args_$");
- delay_string_append_char(message,str,number);
- delay_string_append_char(message,str, " bigint");
+ int select_num = message->current_select;
+ AnalyzeSelect *analyze = message->analyze[select_num];
+ if(!analyze->join)
+ {
+ analyze->join = (JoinTable *) palloc(sizeof(JoinTable));
+ }
+ analyze->join->col_num = 0;
+ analyze->join->col_list = NULL;
+ analyze->join->type_list = NULL;
+ analyze->join->table_list = NULL;
+ analyze->join->using_list = NULL;
+ analyze->join->using_length = 0;
+ analyze->join->state = (char)0;
}
- args_num++;
}
- else if (target->val && (IsA(target->val, ColumnRef)))
+
+ from = message->fromClause;
+ _rewriteNode(BaseSelect, message, dblink, str, lfirst(lc));
+ message->current_select = current_select;
+ message->fromClause = from;
+
+ if(message->r_code == SELECT_ANALYZE && message->fromClause)
{
- int i;
- int exist_col = 0;
- col = (ColumnRef *) target->val;
- colname = strVal(lfirst(list_head(col->fields)));
- delay_string_append_char(message,str,strVal(lfirst(list_head(col->fields))));
+ if(lfirst(lc))
+ if(IsA(lfirst(lc),JoinExpr) || IsA(lfirst(lc),RangeVar) || IsA(lfirst(lc),RangeSubselect))
+ build_virtual_table(message,lfirst(lc),next);
- if(!info)
+ if(loop == 0)
{
- message->r_code = SELECT_RELATION_ERROR;
- return;
+ state = message->table_state;
}
-
- for(i = 0; i < info->col_num; i++)
+ else
{
- if(strcmp(info->col_list[i],colname) == 0)
+ if(state =='L' && message->table_state == 'L')
+ message->table_state = 'L';
+ else
{
- delay_string_append_char(message, str, " ");
- delay_string_append_char(message, str, info->type_list[i]);
- exist_col = 1;
- break;
+ state = 'S';
+ message->table_state = 'S';
}
}
- if(exist_col != 1)
- {
- message->r_code = SELECT_RELATION_ERROR;
- return;
- }
- }
- ++list_first;
+ }
+ loop++;
+
+ if(message->r_code == SELECT_DEFAULT && message->ignore_rewrite == -1 &&
+ message->analyze[current_select]->part == SELECT_TARGETLIST)
+ {
+ pool_debug("_rewriteList select=%d,count=%d", current_select,message->analyze[current_select]->ret_count);
+ message->analyze[current_select]->ret_count++;
+ }
}
- if(select->groupClause)
- compareGroupby(message,list,select->groupClause,str,1);
+}
+
+
+/*****************************************************************************
+ *
+ * Stuff from primnodes.h.
+ *
+ *****************************************************************************/
+
+static void
+_rewriteAlias(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, Alias *node)
+{
+ delay_string_append_char(message, str, " AS ");
+ delay_string_append_char(message, str, node->aliasname);
+ delay_string_append_char(message, str, " ");
- delay_string_append_char(message,str, " ");
+ if (node->colnames)
+ {
+ delay_string_append_char(message, str, "(");
+ _rewriteNode(BaseSelect, message, dblink, str, node->colnames);
+ delay_string_append_char(message, str, ")");
+ }
}
-/*
- * Write down targetList in outer dblink.
- */
static void
-_rewriteIntensiveTargetList(RewriteQuery *message,List *list,String *str)
+append_all_virtual(AnalyzeSelect *analyze,char *table)
{
- ListCell *lc;
- int list_num = 1;
- int args_num = 0;
+ VirtualTable *virtual;
+ SelectDefInfo *select_ret;
+ int num;
+ int test_num;
+ int base;
+ int i;
+ int counter = 0;
+
+ virtual = analyze->virtual;
+ test_num = virtual->col_num;
+ select_ret = analyze->select_ret;
+ base = select_ret->col_num;
+
+ if(!table)
+ {
+ num = test_num;
+ }
+ else
+ {
+ num = 0;
+ for(i = 0; i< test_num; i++)
+ {
+ if(!strcmp(virtual->table_list[i],table))
+ num++;
+ }
+ }
+
+ if(base == 0)
+ {
+ select_ret->col_list = (char **) palloc(num * sizeof(char *));
+ select_ret->type_list = (char **) palloc(num * sizeof(char *));
+ select_ret->return_list = (int *) palloc(num * sizeof(int));
+ }
+ else
+ {
+ select_ret->col_list = (char **) repalloc(select_ret->col_list,(base + num) * sizeof(char *));
+ select_ret->type_list = (char **) repalloc(select_ret->type_list,(base + num) * sizeof(char *));
+ select_ret->return_list = (int *) repalloc(select_ret->return_list,(base + num) * sizeof(int));
+ }
+
+ for(i = 0; i< test_num;i++)
+ {
+ if(table && strcmp(virtual->table_list[i],table))
+ {
+ continue;
+ }
+ select_ret->col_list[base + counter] = virtual->col_list[i];
+ select_ret->type_list[base + counter] = virtual->type_list[i];
+ select_ret->return_list[base + counter] = -1;
+ pool_debug("append_select_def_info: col=%s,type=%s", select_ret->col_list[base + counter],select_ret->type_list[base + counter]);
+ counter++;
+ }
+ select_ret->col_num = base + num;
+}
+
+static void
+append_select_def_info(SelectDefInfo *select_ret,char *col,char *type)
+{
+ int base;
+
+ base = select_ret->col_num;
+ pool_debug("append_select_def_info: base=%d",base);
+
+ if(!type)
+ {
+ type = (char *)palloc(sizeof(char) * strlen("text") + 1);
+ strcpy(type,"text");
+ }
+
+ if(base == 0)
+ {
+ select_ret->col_list = (char **) palloc(sizeof(char *));
+ select_ret->type_list = (char **) palloc(sizeof(char *));
+ select_ret->return_list = (int *) palloc(sizeof(int));
+ }
+ else
+ {
+ select_ret->col_list = (char **) repalloc(select_ret->col_list,(base + 1) * sizeof(char *));
+ select_ret->type_list = (char **) repalloc(select_ret->type_list,(base + 1) * sizeof(char *));
+ select_ret->return_list = (int *) repalloc(select_ret->return_list,(base + 1) * sizeof(int));
+ }
+
+ select_ret->col_list[base] = col;
+ select_ret->type_list[base] = type;
+ select_ret->return_list[base] = -1;
+
+ select_ret->col_num++;
+ pool_debug("append_select_def_info: col=%s,type=%s base=%d",select_ret->col_list[base],select_ret->type_list[base],select_ret->col_num);
+}
+
+static char *search_type_from_virtual(VirtualTable *virtual,char *table,char *col)
+{
+ int num = virtual->col_num;
+ int i;
+
+ for(i = 0; i < num; i++)
+ {
+ if(table)
+ {
+ if(!strcmp(virtual->table_list[i],table)
+ && !strcmp(virtual->col_list[i],col))
+ return virtual->type_list[i];
+ }
+ else
+ {
+ if(!strcmp(virtual->col_list[i],col))
+ return virtual->type_list[i];
+ }
+ }
+ return NULL;
+}
- foreach(lc,list)
+
+static void
+AnalyzeReturnRecord(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *list)
+{
+ int select_num;
+ int range_num;
+ AnalyzeSelect *analyze;
+ RangeInfo **range;
+ VirtualTable *virtual;
+ ListCell *lc;
+ int n_sublink = 0;
+ int c_sublink = 0;
+
+ select_num = message->current_select;
+
+ analyze=message->analyze[select_num];
+
+ virtual = analyze->virtual;
+
+ /* get range table info */
+ range_num = analyze->rangeinfo_num;
+
+ /* "range = 0" means that this select stmt have no table */
+ range = analyze->range;
+
+ analyze->select_ret = (SelectDefInfo *) palloc(sizeof(SelectDefInfo));
+
+ analyze->select_ret->valid = false;
+ analyze->select_ret->col_num = 0;
+
+ pool_debug("AnalyzeReturnRecord: current_select=%d",select_num);
+
+ foreach(lc, list)
{
ResTarget *target = NULL;
- char *typecast = NULL;
- char number[16];
Node *n = lfirst(lc);
void *obj;
- char *funcname = NULL;
- TypeName *typename = NULL;
+ char *alias = NULL;
+ char *typecast = NULL;
+ char *colname = NULL;
+ char *table_name = NULL;
+ char *gettype = NULL;
target = (ResTarget *) n;
+ /* alias name */
+ if(target->name)
+ alias = target->name;
+
+ /* type name */
if(IsA(target->val, TypeCast))
{
TypeCast *type = (TypeCast *) target->val;
- typename = (TypeName *)type->typename;
+ TypeName *typename = (TypeName *)type->typename;
obj = type->arg;
typecast = strVal(lfirst(list_head(typename->names)));
} else {
obj = target->val;
}
-
- if (obj && (IsA(obj, FuncCall)))
+ if(typecast && alias)
{
- FuncCall *func = (FuncCall *) obj;
- funcname = strVal(lfirst(list_head(func->funcname)));
-
- sprintf(number,"%d",args_num);
+ append_select_def_info(analyze->select_ret,alias,typecast);
+ continue;
+ }
+
+ /* column name */
+ if (obj && (IsA(obj, ColumnRef)))
+ {
+ int first = 0;
+ ListCell *c;
+ ColumnRef *col = NULL;
+
+ col = (ColumnRef *) obj;
+ foreach (c, col->fields)
+ {
+ Node *n = (Node *) lfirst(c);
- if(strcmp(funcname,"max") == 0)
+ if (IsA(n, String))
+ {
+ Value *v = (Value *) lfirst(c);
+ if(col->fields->length == 2 && first == 0)
+ {
+ first = 1;
+ table_name = v->val.str;
+ }
+ else
+ colname = v->val.str;
+ }
+ }
+ }
+ else if(obj && (IsA(obj, A_Expr)))
+ {
+ if(alias)
+ append_select_def_info(analyze->select_ret,alias,typecast);
+ else
+ {
+ char *colname;
+ colname = (char *)palloc(sizeof(char) * strlen("\"?column?\"") + 1);
+ strcpy(colname,"\"?column?\"");
+ append_select_def_info(analyze->select_ret,colname,typecast);
+ }
+ continue;
+ }
+ else if(obj && (IsA(obj, FuncCall)))
+ {
+ if(alias)
{
- delay_string_append_char(message,str, " max(args_$");
+ append_select_def_info(analyze->select_ret,alias,typecast);
}
- else if(strcmp(funcname,"min") == 0)
+ else
{
- delay_string_append_char(message,str, " min(args_$");
+ FuncCall *func = NULL;
+ char *funcname;
+ func = (FuncCall *) obj;
+ funcname = strVal(lfirst(list_head(func->funcname)));
+ append_select_def_info(analyze->select_ret,funcname,typecast);
}
- else if(strcmp(funcname,"sum") == 0)
+ continue;
+ }
+ else if (obj && (IsA(obj,SubLink)))
+ {
+ int i;
+ int max = message->analyze_num;
+ AnalyzeSelect *sublink = NULL;
+
+ for(i = select_num + 1; i < max;i++)
{
- delay_string_append_char(message,str, " sum(args_$");
+ sublink = message->analyze[i];
+
+ if(sublink->last_select == select_num
+ && sublink->part == 1 && c_sublink == n_sublink)
+ break;
}
- else if(strcmp(funcname,"count") == 0)
+
+ n_sublink++;
+ if(sublink && (sublink->select_ret->col_num == 1))
{
- delay_string_append_char(message,str, " sum(args_$");
+ if(alias)
+ {
+ append_select_def_info(analyze->select_ret,alias,sublink->select_ret->type_list[0]);
+ continue;
+ }
+ else
+ {
+ char *column;
+ column = (char *)palloc(sizeof(char) * strlen("\"?column?\"") + 1);
+ strcpy(column,"\"?column?\"");
+ append_select_def_info(analyze->select_ret,column,sublink->select_ret->type_list[0]);
+ continue;
+ }
}
- delay_string_append_char(message,str,number);
- delay_string_append_char(message,str, ")");
- args_num++;
}
- else if (target->val && (IsA(target->val, ColumnRef)))
+
+ if(colname && !strcmp(colname,"*"))
{
- ColumnRef *col = NULL;
- col = (ColumnRef *) target->val;
- delay_string_append_char(message,str,strVal(lfirst(list_head(col->fields))));
+ append_all_virtual(analyze,table_name);
+ continue;
}
- if(typename)
+ if(colname)
+ gettype = search_type_from_virtual(virtual,table_name,colname);
+ else
{
- delay_string_append_char(message,str, "::");
- _rewriteTypeName(NULL,message,NULL, str, typename);
+ char *column;
+ column = (char *)palloc(sizeof(char) * strlen("\"?column?\"") + 1);
+ strcpy(column,"\"?column?\"");
+ append_select_def_info(analyze->select_ret,column,NULL);
}
- if(target->name)
- {
- delay_string_append_char(message,str, " AS ");
- delay_string_append_char(message,str, target->name);
- delay_string_append_char(message,str, " ");
- }
- else if(funcname && !target->name && (strcmp(funcname,"count") == 0))
+ if(gettype)
{
- delay_string_append_char(message,str, " AS ");
- delay_string_append_char(message,str, "COUNT");
- delay_string_append_char(message,str, " ");
- }
- if(list_num != list_length(list))
- delay_string_append_char(message,str, ",");
- list_num++;
+ char *t_n;
+ char *t_t;
+
+ if(alias)
+ t_n = alias;
+ else
+ t_n = colname;
+
+ if(typecast)
+ t_t = typecast;
+ else
+ t_t = gettype;
+ append_select_def_info(analyze->select_ret,t_n,t_t);
+ }
}
- delay_string_append_char(message,str, " ");
}
-/*
- * check select stmt
- * this function check fromClause and where
- */
-int
-checkSelectStmtOneTable(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SelectStmt *node)
+static void
+append_virtual_table(RewriteQuery *message,VirtualTable *virtual,char **col_list,char **type_list,int col_num,char *table_name,char state,int next)
{
- RangeVar *var = NULL;
- int use_select_onetable = -1;
+ int base;
+ int i;
- if(node->larg || node->all || node->rarg || node->intoClause ||
- node->distinctClause || node->havingClause)
+ if(virtual->col_num == 0)
{
- return 0;
+ base = 0;
+ virtual->col_list = (char**) palloc(sizeof(char*) * col_num);
+ virtual->type_list = (char**) palloc(sizeof(char*) * col_num);
+ virtual->table_list = (char**) palloc(sizeof(char*) * col_num);
+ virtual->state_list = (char*) palloc(sizeof(char) * col_num);
+ virtual->column_no = (int*) palloc(sizeof(int) * col_num);
+ virtual->valid = (int*) palloc(sizeof(int) * col_num);
+ }
+ else
+ {
+ base = virtual->col_num;
+ virtual->col_list = (char**) repalloc(virtual->col_list,sizeof(char*) * (base + col_num));
+ virtual->type_list = (char**) repalloc(virtual->type_list,sizeof(char*) * (base + col_num));
+ virtual->table_list = (char**) repalloc(virtual->table_list,sizeof(char*)* (base + col_num));
+ virtual->state_list = (char*) repalloc(virtual->state_list,sizeof(char) * (base + col_num));
+ virtual->column_no = (int*) repalloc(virtual->column_no,sizeof(int) * (base + col_num));
+ virtual->valid = (int*) repalloc(virtual->valid,sizeof(int) * (base + col_num));
}
-
- if(message->r_code == SELECT_AEXPR)
+
+ for(i = 0; i< col_num; i++)
{
- KeepRewriteQueryCode(message, SELECT_AEXPR_FALSE);
- return 0;
- }
+ int j = base + i;
- KeepRewriteQueryCode(message, SELECT_ONETABLE);
+ virtual->col_list[j] = col_list[i];
+ virtual->type_list[j] = type_list[i];
+ virtual->table_list[j] = table_name;
+ virtual->state_list[j] = state;
- if(node->fromClause == NULL)
- return 0;
-
- if(!((list_length(node->fromClause) == 1)
- && list_head(node->fromClause)
- && lfirst(list_head(node->fromClause))
- && IsA(lfirst(list_head(node->fromClause)),RangeVar)))
- {
- return 0;
- }
+ /* (next > 0) means this function call from RangeSubselect */
+ if(next > 0)
+ {
+ AnalyzeSelect *analyze = message->analyze[next];
+ SelectDefInfo *select_ret;
+ select_ret = analyze->select_ret;
+ select_ret->return_list[i] = message->column;
+ pool_debug("append_virtual_table return_list[%d]=%d,next=%d",i,message->column,next);
+ }
+ virtual->column_no[j] = message->column;
+ message->column++;
+ virtual->valid[j] = -1;
+ pool_debug("append_virtual_table select=%d, no=%d,col=%s,type=%s,table=%s,state=%c,valid=%d",
+ message->current_select,
+ virtual->column_no[j],virtual->col_list[j],virtual->type_list[j]
+ ,virtual->table_list[j],virtual->state_list[j],virtual->valid[j]);
+ }
+ virtual->col_num = base + col_num;
+}
+
+
+static void append_join_using(AnalyzeSelect *analyze,char **col_list,char **type_list,int col_num,char *table_name)
+{
+ int num,i,j,k;
+ int same[analyze->join->col_num];
+ int lvalid[analyze->join->col_num];
+ int rvalid[col_num];
+ char **using;
+ int lc = 0;
+ int rc = 0;
+ int sc = 0;
+ int total;
+ JoinTable *join = analyze->join;
+ JoinTable *new = (JoinTable *) palloc(sizeof(JoinTable));
- var = (RangeVar *) lfirst(list_head(node->fromClause));
-
-
- message->table_relname = var->relname;
- if(var->schemaname)
- message->schemaname = var->schemaname;
- else
- message->schemaname = NULL;
+ using = join->using_list;
+ num = join->using_length;
- if(var->alias)
+ for(i = 0; i < join->col_num; i++)
{
- Alias *alias_p = (Alias *) var->alias;
- message->table_alias = alias_p->aliasname;
- } else
- message->table_alias = NULL;
-
- message->dbname = dblink->dbname;
+ char *colname = join->col_list[i];
+ int match = 0;
+ int array;
+ for(j = 0; j < num; j++)
+ {
+ if(!strcmp(colname, using[j]))
+ {
+ match++;
+ array = j;
+ }
+ }
+ if(match == 0)
+ {
+ lvalid[lc] = i;
+ lc++;
+ }
+ else if (match == 1)
+ {
+ same[sc] = array;
+ sc++;
+ }
+ /* XXX */
+ }
- if (node->whereClause)
+ for(i = 0; i < col_num; i++)
{
- int true_count = 0;
- KeepRewriteQueryCode(message, SELECT_AEXPR);
- true_count = _rewriteWhereClause(node->whereClause, message,dblink,str,true_count);
- if(true_count == 0)
- use_select_onetable = 1;
- } else {
- use_select_onetable = 1;
+ char *colname = col_list[i];
+ int match = 0;
+ for(j = 0; j < num; j++)
+ {
+ if(!strcmp(colname, using[j]))
+ {
+ match++;
+ }
+ }
+ if(match == 0)
+ {
+ rvalid[rc] = i;
+ rc++;
+ }
}
- return use_select_onetable;
-}
+ total= sc + lc + rc;
+ new->col_num = total;
+ new->col_list = (char**) palloc(sizeof(char*) * (total));
+ new->type_list = (char**) palloc(sizeof(char*) * (total));
+ new->table_list = (char**) palloc(sizeof(char*) * (total));
-/*
- * if message->r_code is SELECT_RELATION_ERROR
- * or SELECT_PGCATALOG, don't change message->r_code
- */
-static void KeepRewriteQueryCode(RewriteQuery *message, int r_code)
-{
- if((message->r_code != SELECT_RELATION_ERROR)
- && (message->r_code != SELECT_PGCATALOG))
+ for(k = 0; k < sc; k++)
{
- message->r_code = r_code;
+ new->col_list[k] = join->col_list[same[k]];
+ new->type_list[k] = join->type_list[same[k]];
+ if(table_name)
+ {
+ new->table_list[k] = table_name;
+ }
+ else
+ new->table_list[k] = join->table_list[same[k]];
+ }
+
+ for(k = sc; k < sc + lc; k++)
+ {
+ new->col_list[k] = join->col_list[lvalid[k - sc]];
+ new->type_list[k] = join->type_list[lvalid[k - sc]];
+ if(table_name)
+ {
+ new->table_list[k] = table_name;
+ }
+ else
+ new->table_list[k] = join->table_list[lvalid[k - sc]];
+ }
+ for(k = sc + lc; k < sc + lc + rc; k++)
+ {
+ new->col_list[k] = col_list[rvalid[k - lc - sc]];
+ new->type_list[k] = type_list[rvalid[k - lc - sc]];
+ if(table_name)
+ {
+ new->table_list[k] = table_name;
+ }
+ else
+ new->table_list[k] = NULL;
}
-}
-
-/*
- * Is this column member of table (message->table_relname) ?
- * return 1 -- member
- * return 0 or -1 -- not mmeber
- */
-static int
-_checkDistDefColumn(ColumnRef *col,RewriteQuery *message)
-{
- ListCell *c;
- List *list;
- char first = 0;
- char *tmp_table = NULL;
- char *colname = NULL;
- DistDefInfo *info = NULL;
- int i;
- int check_col = 0;
- list = col->fields;
+ analyze->join = new;
- if(list->length > 2 || list->length == 0)
+ for(i = 0; i< analyze->join->col_num; i++)
{
- /* send error message */
- return -1;
+ pool_debug("append_join_using no = %d ,col=%s,type=%s,table=%s",
+ i,new->col_list[i],new->type_list[i],new->table_list[i]);
}
+}
+
+static void append_join_natural(AnalyzeSelect *analyze,char **col_list,char **type_list,int base,int col_num,char *table_name)
+{
+ int same[base + 1];
+ int rvalid[base + 1];
+ int linvalid[col_num + 1];
+ int lvalid[col_num +1];
+ int i,j,k;
+ int sc = 0;
+ int vc = 0;
+ int ic = 0;
+ int total = 0;
+ int count;
+ JoinTable *join = analyze->join;
+ JoinTable *new = (JoinTable *) palloc(sizeof(JoinTable));
- foreach (c, col->fields)
+ for(i = 0; i < base; i++)
{
- Node *n = (Node *) lfirst(c);
-
- if (IsA(n, String))
+ char *colname = join->col_list[i];
+ int match = 0;
+ int array = -1;
+ for(j = 0; j < col_num; j++)
{
- Value *v = (Value *) lfirst(c);
- if(list->length == 2 && first == 0)
+ if(!strcmp(colname, col_list[j]))
{
- first = 1;
- tmp_table = v->val.str;
+ match++;
+ array = j;
}
- else
- colname = v->val.str;
}
+
+ if(match == 0)
+ {
+ rvalid[vc] = i;
+ vc++;
+ }
+ else if(match == 1)
+ {
+ same[sc] = i;
+ linvalid[ic] = array;
+ sc++;
+ ic++;
+ }
+ else
+ {
+ /* XXX */
+ }
}
- /* alias check */
- if(message->table_alias)
+ total=sc + vc + (col_num - ic);
+ new->col_num = total;
+ new->col_list = (char**) palloc(sizeof(char*) * (total));
+ new->type_list = (char**) palloc(sizeof(char*) * (total));
+ new->table_list = (char**) palloc(sizeof(char*) * (total));
+
+ for(k = 0; k < sc; k++)
{
- if(tmp_table && strcmp(message->table_alias,tmp_table) != 0)
+ new->col_list[k] = join->col_list[same[k]];
+ new->type_list[k] = join->type_list[same[k]];
+ if(table_name)
{
- return check_col;
+ new->table_list[k] = table_name;
}
- }
- else if (!message->table_alias && message->table_relname)
+ else
+ new->table_list[k] = join->table_list[same[k]];
+ }
+
+ for(k = sc; k < sc + vc; k++)
{
- if(tmp_table && strcmp(message->table_relname,tmp_table) != 0)
+ new->col_list[k] = join->col_list[rvalid[k - sc]];
+ new->type_list[k] = join->type_list[rvalid[k - sc]];
+ if(table_name)
{
- return check_col;
+ new->table_list[k] = table_name;
}
- }
- else
- {
- return check_col;
+ else
+ new->table_list[k] = join->table_list[rvalid[k - sc]];
}
-
- /* get dist_def_info */
- info = pool_get_dist_def_info(message->dbname, message->schemaname, message->table_relname);
- if(!info)
+ count = 0;
+ for(k = 0; k <col_num ; k++)
{
- /* send error message */
- return -1;
+ int l = 0;
+ bool test = true;
+ for(l = 0; l< ic; l++)
+ {
+ if(k == linvalid[l])
+ test = false;
+ }
+ if(test)
+ {
+ lvalid[count] = k;
+ count++;
+ }
}
- for(i = 0; i < info->col_num; i++)
+ for(k = sc + vc; k < sc + vc + count; k++)
{
- if(strcmp(info->col_list[i],colname) == 0)
+ new->col_list[k] = col_list[lvalid[k - sc - vc]];
+ new->type_list[k] = type_list[lvalid[k - sc - vc]];
+ if(table_name)
{
- check_col = 1;
+ new->table_list[k] = table_name;
}
-
+ else
+ new->table_list[k] = NULL;
}
+ analyze->join = new;
- return check_col;
+ for(i = 0; i< analyze->join->col_num; i++)
+ {
+ pool_debug("append_join_natural no = %d ,col=%s,type=%s,table=%s",
+ i,new->col_list[i],new->type_list[i],new->table_list[i]);
+ }
}
-/*
- * A_Expr check
- * if A_Expr-tree use subquery,function,or other table member
- * rewrite expression as "TRUE"
- *
- * return value is counter of rewriting query.
- */
-static int
-_writewhereClause(A_Expr *expr,RewriteQuery *message,ConInfoTodblink *dblink, String *str,int true_count)
+static void append_join_simple(JoinTable *join,char **col_list,char **type_list,int col_num,char *table_name)
{
- int message_r_code = message->r_code;
+ int base;
+ int i;
+ pool_debug("append_join_table start");
- switch (expr->kind)
+ if(join->col_num == 0)
{
- case AEXPR_OP:
- if (list_length(expr->name) == 1)
- {
- KeepRewriteQueryCode(message, SELECT_AEXPR);
- _rewriteNode(NULL, message, dblink, str, expr->lexpr);
- if(message->r_code == SELECT_AEXPR)
- {
- _rewriteNode(NULL, message, dblink, str, expr->rexpr);
- if(message->r_code == SELECT_AEXPR)
- {
- Value *op = (Value *) lfirst(list_head(expr->name));
- KeepRewriteQueryCode(message, message_r_code);
- _rewriteNode(NULL, message, dblink, str, expr->lexpr);
- delay_string_append_char(message, str, op->val.str);
- _rewriteNode(NULL, message, dblink, str, expr->rexpr);
- KeepRewriteQueryCode(message, message_r_code);
- } else {
- KeepRewriteQueryCode(message, message_r_code);
- delay_string_append_char(message, str,"TRUE");
- true_count++;
- break;
- }
- } else {
- KeepRewriteQueryCode(message, message_r_code);
- delay_string_append_char(message, str,"TRUE");
- true_count++;
- }
- break;
- }
- else
- {
- delay_string_append_char(message, str,"TRUE");
- true_count++;
- }
- break;
-
- case AEXPR_AND:
- delay_string_append_char(message, str, " (");
- true_count = true_count + _rewriteWhereClause(expr->lexpr,message,dblink,str,true_count);
- delay_string_append_char(message, str, " AND ");
- true_count = true_count + _rewriteWhereClause(expr->rexpr,message,dblink,str,true_count);
- delay_string_append_char(message, str, ")");
- break;
-
- case AEXPR_OR:
- delay_string_append_char(message, str, " (");
- true_count = true_count + _rewriteWhereClause(expr->lexpr,message,dblink,str,true_count);
- delay_string_append_char(message, str, " OR ");
- true_count = true_count + _rewriteWhereClause(expr->rexpr,message,dblink,str,true_count);
- delay_string_append_char(message, str, ")");
- break;
-
-
- case AEXPR_OP_ANY:
- /* not implemented yet */
- break;
-
- case AEXPR_OP_ALL:
- /* not implemented yet */
- break;
+ base = 0;
+ join->col_list = (char**) palloc(sizeof(char*) * col_num);
+ join->type_list = (char**) palloc(sizeof(char*) * col_num);
+ join->table_list = (char**) palloc(sizeof(char*) * col_num);
+ }
+ else
+ {
+ base = join->col_num;
+ join->col_list = (char**) repalloc(join->col_list,sizeof(char*) * (base + col_num));
+ join->type_list = (char**) repalloc(join->type_list,sizeof(char*) * (base + col_num));
+ join->table_list = (char**) repalloc(join->table_list,sizeof(char*) * (base + col_num));
+ }
+
+ for(i = 0; i< col_num; i++)
+ {
+ join->col_list[base + i] = col_list[i];
+ join->type_list[base + i] = type_list[i];
+ join->table_list[base + i] = table_name;
+ pool_debug("append_join_table no = %d ,col=%s,type=%s,table=%s",
+ base + i,join->col_list[base + i],join->type_list[base + i],join->table_list[base + i]);
+ }
+ join->col_num = base + col_num;
+}
+
+static void build_join_table(RewriteQuery *message,char *alias, int type)
+{
+ char *table_name = NULL;
+ char state;
+ char lstate;
+ int left_num,right_num,range_num,select_num;
+ DistDefInfo *distinfo = NULL;
+ RepliDefInfo *repliinfo = NULL;
+ SelectDefInfo *selectinfo = NULL;
+ JoinTable *join;
+ RangeInfo *range;
+ AnalyzeSelect *analyze;
+
+ select_num = message->current_select;
+ analyze=message->analyze[select_num];
+ join = analyze->join;
+ left_num = join->col_num;
+ range_num = analyze->rangeinfo_num;
+ range = analyze->range[range_num - 1];
+ state = range->state;
+ lstate = join->state;
+
+ distinfo = range->distinfo;
+ repliinfo = range->repliinfo;
+ selectinfo = range->selectinfo;
+
+ if(alias)
+ table_name = alias;
+ else if(!alias && range->alias)
+ table_name = range->alias;
+
+ if(distinfo && !repliinfo && !selectinfo)
+ {
+ right_num = distinfo->col_num;
+ if(!table_name)
+ table_name = distinfo->table_name;
+ pool_debug("inside build_join_info dist state=%c %s",range->state,table_name);
-#if 0
+ if(type == JNATURAL_INNER || type == JNATURAL_RIGHT
+ || type == JNATURAL_LEFT || type == JNATURAL_FULL)
+ append_join_natural(analyze,distinfo->col_list,distinfo->type_list,join->col_num,distinfo->col_num,table_name);
+ else
+ append_join_simple(join,distinfo->col_list,distinfo->type_list,distinfo->col_num,table_name);
+ }
+ else if (repliinfo && !distinfo && !selectinfo)
+ {
+ right_num = repliinfo->col_num;
+ if(!table_name)
+ table_name = repliinfo->table_name;
- case AEXPR_NOT:
- delay_string_append_char(message, str, " (NOT ");
- _rewriteWhereClause(expr->rexpr,str,tablename,dbname, schemaname,aliasname);
- delay_string_append_char(message, str, ")");
- break;
+ pool_debug("inside build_join_info dist state=%c %s",range->state,table_name);
- case AEXPR_DISTINCT:
- delay_string_append_char(message, str, " (");
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- delay_string_append_char(message, str, " IS DISTINCT FROM ");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
- break;
+ if(type == JNATURAL_INNER || type == JNATURAL_RIGHT
+ || type == JNATURAL_LEFT || type == JNATURAL_FULL)
+ append_join_natural(analyze,repliinfo->col_list,repliinfo->type_list,join->col_num,repliinfo->col_num,table_name);
+ else if(join->using_length != 0)
+ append_join_using(analyze,repliinfo->col_list,repliinfo->type_list,repliinfo->col_num,table_name);
+ else
+ append_join_simple(join,repliinfo->col_list,repliinfo->type_list,repliinfo->col_num,table_name);
+ }
+ else if (selectinfo && !repliinfo && !distinfo)
+ {
+ pool_debug("inside build_join_info dist state=%c %s",range->state,table_name);
+ right_num = selectinfo->col_num;
- case AEXPR_NULLIF:
- delay_string_append_char(message, str, " NULLIF(");
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- delay_string_append_char(message, str, ", ");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
- break;
+ if(type == JNATURAL_INNER || type == JNATURAL_RIGHT
+ || type == JNATURAL_LEFT || type == JNATURAL_FULL)
+ append_join_natural(analyze,selectinfo->col_list,selectinfo->type_list,join->col_num,selectinfo->col_num,table_name);
+ else if(join->using_length != 0)
+ append_join_using(analyze,selectinfo->col_list,selectinfo->type_list,selectinfo->col_num,table_name);
+ else
+ append_join_simple(join,selectinfo->col_list,selectinfo->type_list,selectinfo->col_num,table_name);
+ }
- case AEXPR_OF:
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- if (*(char *)lfirst(list_head(node->name)) == '!')
- delay_string_append_char(message, str, " IS NOT OF (");
- else
- delay_string_append_char(message, str, " IS OF (");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
- break;
-#endif
- default:
- delay_string_append_char(message, str,"TRUE");
- true_count++;
- break;
+ if(type == JDEFAULT)
+ {
+ join->state = state;
+ return;
}
- return true_count;
-}
-/*
- * Start whereClasue check
- * call _writewherelause();
- *
- * if messgae->r_code is SELECT DEFAULT,
- * write down whereClause
- * else
- * check whereClause
- *
- * return value is counter of rewriting query
- * if return value is zero, all whereclause can insert dblink function.
- */
-static int
-_rewriteWhereClause(Node *BaseSelect,RewriteQuery *message,ConInfoTodblink *dblink,String *str,int true_count)
-{
- A_Expr *expr = NULL;
+ join =analyze->join;
- if(!IsA(BaseSelect, A_Expr))
+ if(lstate =='E' || state == 'E')
{
- delay_string_append_char(message, str, "TRUE");
- true_count++;
- return true_count;
+ join->state = 'E';
+ return;
}
-
- expr = (A_Expr *) BaseSelect;
-
- if(expr->kind == AEXPR_NOT)
+ if(lstate =='L' && state =='L')
{
- delay_string_append_char(message, str, "TRUE");
- true_count++;
- return true_count;
+ join->state = 'L';
+ return;
}
- true_count = _writewhereClause(expr,message,dblink,str,true_count);
- return true_count;
+ if(lstate =='L' && state =='P')
+ {
+ int total = left_num + right_num;
+ if(type == JRIGHT)
+ join->state = 'P';
+ else if(type == JINNER)
+ join->state = 'P';
+ else if((join->col_num < total) && (type == JNATURAL_RIGHT || type == JNATURAL_INNER))
+ join->state = 'P';
+ else
+ join->state = 'S';
+ }
+ else if(lstate =='P' && state == 'L')
+ {
+ int total = left_num + right_num;
+ if(type == JLEFT)
+ join->state ='P';
+ else if(type == JINNER)
+ join->state ='P';
+ else if((join->col_num < total) && (type == JNATURAL_LEFT || type == JNATURAL_INNER))
+ join->state = 'P';
+ else
+ join->state ='S';
+ }
+ else
+ join->state = 'S';
}
-static void _rewriteIdList(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *node)
+static void change_analyze_state(AnalyzeSelect *analyze,char state)
{
- ListCell *lc;
- char first = 0;
+ if(state == 'E' || analyze->state == 'E')
+ return;
- foreach(lc, node)
+ if(!analyze->state)
+ analyze->state = state;
+ else if(analyze->state == 'L' && state =='L')
+ return;
+ else
+ analyze->state = 'S';
+}
+
+static void build_virtual_table(RewriteQuery *message,void *obj,int next)
+{
+ int select_num;
+ int range_num;
+ char *alias = NULL;
+ char *table_name = NULL;
+ char state;
+ AnalyzeSelect *analyze;
+ VirtualTable *virtual;
+ RangeInfo *range;
+ DistDefInfo *distinfo;
+ RepliDefInfo *repliinfo;
+ SelectDefInfo *selectinfo;
+ JoinTable *join;
+
+ select_num = message->current_select;
+ analyze=message->analyze[select_num];
+ virtual = analyze->virtual;
+
+ /* last range */
+ if(IsA(obj,RangeVar) || IsA(obj,RangeSubselect))
+ {
+ range_num = analyze->rangeinfo_num;
+ range = analyze->range[range_num - 1];
+ distinfo = range->distinfo;
+ repliinfo = range->repliinfo;
+ selectinfo = range->selectinfo;
+ }
+ else
{
- Value *v = lfirst(lc);
-
- if (first == 0)
- first = 1;
- else
- delay_string_append_char(message, str, ", ");
-
- delay_string_append_char(message, str, "\"");
- delay_string_append_char(message, str, v->val.str);
- delay_string_append_char(message, str, "\"");
+ /* join case */
+ int i;
+ join =analyze->join;
+ for(i =0; i < join->col_num;i++)
+ {
+ append_virtual_table(message,virtual,join->col_list++,join->type_list++,1,join->table_list[i],join->state,-1);
+ }
+ pool_debug("build_virtual_table join join->colnum=%d",join->col_num);
+ analyze->state = join->state;
+ return;
}
-}
-static void _rewriteList(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, List *node)
-{
- ListCell *lc;
- char first = 0;
+ if(range->alias)
+ alias = range->alias;
+
+ state = range->state;
- foreach(lc, node)
+ if(distinfo && !repliinfo && !selectinfo)
{
- if (first == 0)
- first = 1;
+ if(alias)
+ table_name =alias;
else
- {
- if(!IsA(lfirst(lc),A_Indices))
- delay_string_append_char(message, str, ",");
- }
- _rewriteNode(BaseSelect, message, dblink, str, lfirst(lc));
+ table_name = distinfo->table_name;
+
+ pool_debug("inside build_virtual_info dist state=%c %s",range->state,table_name);
+ append_virtual_table(message,virtual,distinfo->col_list,distinfo->type_list,distinfo->col_num,table_name,state,-2);
+ change_analyze_state(analyze,state);
+ return;
+ }
+ else if (repliinfo && !distinfo && !selectinfo)
+ {
+ if(alias)
+ table_name =alias;
+ else
+ table_name = repliinfo->table_name;
+ pool_debug("inside build_virtual_info dist state=%c %s",range->state,table_name);
+ append_virtual_table(message,virtual,repliinfo->col_list,repliinfo->type_list,repliinfo->col_num,table_name,state,-3);
+ change_analyze_state(analyze,state);
+ return;
+ }
+ else if (selectinfo && !repliinfo && !distinfo)
+ {
+ table_name = alias;
+ pool_debug("inside build_virtual_info select state=%c %s",range->state,table_name);
+ append_virtual_table(message,virtual,selectinfo->col_list,selectinfo->type_list,selectinfo->col_num,table_name,state,next);
+ change_analyze_state(analyze,state);
+ return;
+ }
+ else if (!selectinfo && !repliinfo && !distinfo)
+ {
+ pool_debug("inside build_virtual_info dist state=%c %s",range->state,alias);
+ change_analyze_state(analyze,'E');
}
}
-
-/*****************************************************************************
- *
- * Stuff from primnodes.h.
- *
- *****************************************************************************/
-
static void
-_rewriteAlias(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, Alias *node)
+build_range_info(RewriteQuery *message,DistDefInfo *info,RepliDefInfo *info2,SelectDefInfo *info3,char *alias, int select_num,int i_num)
{
- delay_string_append_char(message, str, " AS ");
- delay_string_append_char(message, str, node->aliasname);
- delay_string_append_char(message, str, " ");
+ int num;
+ AnalyzeSelect *analyze;
- if (node->colnames)
+ analyze=message->analyze[select_num];
+
+ if(analyze->range)
{
- delay_string_append_char(message, str, "(");
- _rewriteNode(BaseSelect, message, dblink, str, node->colnames);
- delay_string_append_char(message, str, ")");
+ ++analyze->rangeinfo_num;
+ num = analyze->rangeinfo_num;
+ analyze->range =
+ (RangeInfo **) repalloc(analyze->range,sizeof(RangeInfo *) * num);
+
+ analyze->range[num-1] = (RangeInfo *) palloc(sizeof(RangeInfo));
+ pool_debug("inside build_range_info num= %d current_select=%d",num,select_num);
+ }
+ else
+ {
+ num = 1;
+ analyze->range = (RangeInfo **) palloc(sizeof(RangeInfo *));
+ analyze->range[0] = (RangeInfo *) palloc(sizeof(RangeInfo));
+ analyze->rangeinfo_num = 1;
+ pool_debug("inside build_range_info num= %d current_select=%d",num,select_num);
+ }
+
+ analyze->range[num -1]->ret_num = num - 1;
+ analyze->range[num -1]->selectinfo = NULL;
+ /* set dist_def_info */
+ if(info && !info2 && !info3)
+ {
+ message->is_loadbalance = false;
+ analyze->range[num -1]->distinfo = info;
+ analyze->range[num -1]->repliinfo = NULL;
+ analyze->select_ret = NULL;
+ analyze->range[num -1]->alias = alias;
+ analyze->range[num -1]->state = 'P';
+ pool_debug("inside build_range_info dist %d",select_num);
+ }
+
+ /* set repli_def_info */
+ if(info2 && !info && !info3)
+ {
+ analyze->range[num -1]->distinfo = NULL;
+ analyze->range[num -1]->repliinfo = info2;
+ analyze->select_ret = NULL;
+ analyze->range[num -1]->alias = alias;
+ analyze->range[num -1]->state = 'L';
+ pool_debug("inside build_range_info repli %d",select_num);
+ return;
+ }
+
+ /* CALL FROM _rertiteRangeSubselect */
+ if(info3 && !info && !info2)
+ {
+ char state = (char)0;
+ analyze->range[num -1]->distinfo = NULL;
+ analyze->range[num -1]->repliinfo = NULL;
+ analyze->range[num -1]->selectinfo = info3;
+ analyze->range[num -1]->alias = alias;
+ state = message->analyze[i_num]->state;
+ analyze->range[num -1]->state = state;
+ pool_debug("inside build_range_info select %d, state = %c (%c)",select_num,state,analyze->state);
+ return;
+ }
+
+ if(info && info2)
+ {
+ /*TODO: error*/
+ analyze->select_ret = NULL;
+ pool_debug("inside build_range_info error %d",select_num);
+ return;
}
-}
+ if(!info && !info2 && !info3)
+ {
+ /*TODO: error*/
+ message->is_loadbalance = true;
+ analyze->range[num -1]->distinfo = NULL;
+ analyze->range[num -1]->repliinfo = NULL;
+ analyze->range[num -1]->state = 'E';
+ analyze->select_ret = NULL;
+ analyze->range[num -1]->alias = alias;
+ pool_debug("inside build_range_info const or func %d",select_num);
+ }
+}
static void
_rewriteRangeVar(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, RangeVar *node)
{
- if(message->r_code == SELECT_PGCATALOG)
- return ;
+ DistDefInfo *info;
+ RepliDefInfo *info2;
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ info = pool_get_dist_def_info(dblink->dbname, node->schemaname, node->relname);
+ info2 = pool_get_repli_def_info(dblink->dbname, node->schemaname, node->relname);
+ }
- if(message->r_code != SELECT_DEFAULT &&
- message->r_code != SELECT_ONETABLE)
+ if(!(message->r_code == SELECT_DEFAULT && message->rewritelock == -1 && message->ignore_rewrite == -1))
{
if (node->catalogname)
{
- delay_string_append_char(message, str, "\"");
delay_string_append_char(message, str, node->catalogname);
- delay_string_append_char(message, str, "\".");
+ delay_string_append_char(message, str, ".");
}
if (node->schemaname)
{
- delay_string_append_char(message, str, "\"");
delay_string_append_char(message, str, node->schemaname);
- delay_string_append_char(message, str, "\".");
+ delay_string_append_char(message, str, ".");
if(strcmp(node->schemaname,"pg_catalog") == 0)
{
- message->r_code = SELECT_PGCATALOG;
+ message->is_pg_catalog = true;
}
-
}
- delay_string_append_char(message, str, "\"");
delay_string_append_char(message, str, node->relname);
- delay_string_append_char(message, str, "\"");
if (node->alias)
+ {
+ Alias *alias = node->alias;
_rewriteNode(BaseSelect, message, dblink, str, node->alias);
+ if(message->r_code == SELECT_ANALYZE)
+ build_range_info(message,info,info2,NULL,alias->aliasname,message->current_select,-1);
+ }
+ else
+ {
+ if(message->r_code == SELECT_ANALYZE)
+ build_range_info(message,info,info2,NULL,node->relname,message->current_select,-1);
+ }
if (node->inhOpt == INH_YES)
{
delay_string_append_char(message, str, " * ");
}
- } else {
+ }
+ else
+ {
/* rewrite query using dblink connection */
- DistDefInfo *info;
- int i;
- char port[8];
+ char *alias_name = NULL;
SelectStmt *select = (SelectStmt *)BaseSelect;
+ if(node->alias)
+ {
+ alias_name = node->alias->aliasname;
+ }
/*
* iff schemaname is pg_catalog, send query to
* one node not system db.
@@ -1219,37 +1638,15 @@ _rewriteRangeVar(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
if(node->schemaname &&
(strcmp(node->schemaname,"pg_catalog") == 0))
{
- message->r_code = SELECT_PGCATALOG;
+ message->is_pg_catalog =true;
return;
}
- sprintf(port,"%d",dblink->port);
-
- if(node->alias)
- delay_string_append_char(message, str, "(SELECT * FROM dblink(");
- else
- delay_string_append_char(message, str, "dblink(");
-
- delay_string_append_char(message, str, "'");
- delay_string_append_char(message, str, "host=");
- delay_string_append_char(message, str, dblink->hostaddr);
- delay_string_append_char(message, str, " dbname=");
- delay_string_append_char(message, str, dblink->dbname);
- delay_string_append_char(message, str, " port=");
- delay_string_append_char(message, str, port);
- delay_string_append_char(message, str, " user=");
- delay_string_append_char(message, str, dblink->user);
-
- if(strlen(dblink->password))
- {
- delay_string_append_char(message, str, " password=");
- delay_string_append_char(message, str, dblink->password);
- }
- delay_string_append_char(message, str, "'");
- delay_string_append_char(message, str, ",");
- delay_string_append_char(message, str, "'");
- delay_string_append_char(message, str, "SELECT pool_parallel(\"select * from ");
+ info = pool_get_dist_def_info(dblink->dbname, node->schemaname, node->relname);
+ info2 = pool_get_repli_def_info(dblink->dbname, node->schemaname, node->relname);
+ writeRangeHeader(message,dblink,str,info,info2,alias_name);
+
if (node->catalogname)
{
delay_string_append_char(message, str, node->catalogname);
@@ -1264,7 +1661,7 @@ _rewriteRangeVar(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
if(strcmp(node->schemaname,"pg_catalog") == 0)
{
- message->r_code = SELECT_PGCATALOG;
+ message->is_pg_catalog = true;
}
}
@@ -1276,62 +1673,26 @@ _rewriteRangeVar(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
if(select->whereClause &&
!(message->r_code == SELECT_PGCATALOG))
{
- int message_code = message->r_code;
- if(node->alias)
- {
- Alias *alias_p = (Alias *) node->alias;
- _rewriteNode(BaseSelect, message, dblink, str, node->alias);
- delay_string_append_char(message, str, " WHERE ");
- message->table_alias = alias_p->aliasname;
- }
- else
- {
+ char * temp;
+ int message_code = message->r_code;
delay_string_append_char(message, str, " WHERE ");
- message->table_alias = NULL;
- }
- message->table_relname = node->relname;
- message->dbname = dblink->dbname;
- KeepRewriteQueryCode(message, SELECT_DEFAULT_INSIDE_DBLINK);
- _rewriteWhereClause(select->whereClause, message,dblink,str,0);
- KeepRewriteQueryCode(message, message_code);
- }
-
- delay_string_append_char(message, str, "\"");
- delay_string_append_char(message, str, ")");
- delay_string_append_char(message, str, "'");
- delay_string_append_char(message, str, ")");
-
-
- delay_string_append_char(message, str," AS ");
- delay_string_append_char(message, str, node->relname);
- delay_string_append_char(message, str, "(");
- /* get dist_def_info */
- info = pool_get_dist_def_info (dblink->dbname, node->schemaname, node->relname);
+ if(message->table_relname)
+ temp = message->table_relname;
- if(!info)
- {
- message->r_code = SELECT_RELATION_ERROR;
- return;
- }
+ if(alias_name)
+ message->table_relname = alias_name;
+ else
+ message->table_relname = node->relname;
- for(i = 0; i < info->col_num; i++)
- {
- delay_string_append_char(message, str, info->col_list[i]);
- delay_string_append_char(message, str, " ");
- delay_string_append_char(message, str, info->type_list[i]);
- if (i != info->col_num -1)
- delay_string_append_char(message, str, ",");
+ message->rewritelock = message->current_select;
+ CheckWhereCaluse(select->whereClause, message,dblink,str,0);
+ message->rewritelock = -1;
+ message->table_relname = temp;
+ KeepRewriteQueryReturnCode(message, message_code);
}
- delay_string_append_char(message, str, ")");
- if(node->alias)
- delay_string_append_char(message, str, ")");
-
- if(node->alias)
- {
- _rewriteNode(BaseSelect, message, dblink, str, node->alias);
- }
+ writeRangeFooter(message,dblink,str,info,info2,alias_name);
if (node->inhOpt == INH_YES)
{
@@ -1405,7 +1766,7 @@ _rewriteSubLink(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink
{
if(message->r_code == SELECT_AEXPR)
{
- KeepRewriteQueryCode(message, SELECT_AEXPR_FALSE);
+ KeepRewriteQueryReturnCode(message, SELECT_AEXPR_FALSE);
return;
}
_rewriteNode(BaseSelect, message, dblink, str, node->testexpr);
@@ -1453,9 +1814,12 @@ _rewriteSubLink(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink
if (node->subselect)
{
+ int count = message->current_select;
+ int part = message->part;
delay_string_append_char(message, str, "(");
_rewriteNode(BaseSelect, message, dblink, str, node->subselect);
delay_string_append_char(message, str, ")");
+ KeepMessages(message,count,part);
}
}
@@ -1655,34 +2019,111 @@ _rewriteRangeTblRef(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *db
static void
_rewriteJoinExpr(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, JoinExpr *node)
{
+ char l_state,r_state;
+ bool from;
+ int select_num;
+ bool natural = false;
+ bool inner = false;
+ bool cross = false;
+ bool full = false;
+ bool left = false;
+ bool right = false;
+ char *aliasname;
+ int using_length = 0;
+ char **using_list;
+
+ Alias *alias = node->alias;
+ if(alias)
+ aliasname = alias->aliasname;
+ else
+ aliasname = NULL;
+
+ from = message->fromClause;
+ select_num = message->current_select;
+
_rewriteNode(BaseSelect, message, dblink, str, node->larg);
+ /* reset message */
+ message->fromClause = from;
+ message->current_select = select_num;
+ l_state = message->table_state;
+
+ if(message->r_code == SELECT_ANALYZE &&
+ (IsA(node->larg, RangeVar) || IsA(node->larg,RangeSubselect)))
+ {
+ build_join_table(message,aliasname,JDEFAULT);
+ }
+
+
if (node->isNatural == TRUE)
+ {
+ natural = true;
delay_string_append_char(message, str, " NATURAL");
+ }
if (node->jointype == JOIN_INNER)
{
if (node->using == NIL && node->quals == NULL && !node->isNatural)
+ {
+ cross = true;
delay_string_append_char(message, str, " CROSS JOIN ");
+ }
else
+ {
delay_string_append_char(message, str, " JOIN ");
+ inner = true;
+ }
+
}
- else if (node->jointype == JOIN_INNER)
- delay_string_append_char(message, str, " JOIN ");
else if (node->jointype == JOIN_LEFT)
+ {
delay_string_append_char(message, str, " LEFT OUTER JOIN ");
+ left = true;
+ }
else if (node->jointype == JOIN_FULL)
+ {
delay_string_append_char(message, str, " FULL OUTER JOIN ");
+ full = true;
+ }
else if (node->jointype == JOIN_RIGHT)
+ {
delay_string_append_char(message, str, " RIGHT OUTER JOIN ");
+ right = true;
+ }
_rewriteNode(BaseSelect, message, dblink, str, node->rarg);
+ r_state = message->table_state;
+ message->fromClause = from;
+ message->current_select = select_num;
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ if(cross && (IsA(node->rarg, RangeVar) || IsA(node->rarg,RangeSubselect)))
+ build_join_table(message,aliasname,JNORMAL);
+
+ if(IsA(node->rarg, RangeVar) || IsA(node->rarg,RangeSubselect))
+ {
+ if(natural && inner)
+ build_join_table(message,aliasname,JNATURAL_INNER);
+ else if(natural && left)
+ build_join_table(message,aliasname,JNATURAL_LEFT);
+ else if(natural && right)
+ build_join_table(message,aliasname,JNATURAL_RIGHT);
+ else if(natural && full)
+ build_join_table(message,aliasname,JNORMAL);
+ }
+ }
+
if (node->using != NIL && IsA(node->using, List))
{
ListCell *lc;
char comma = 0;
+ int count = 0;
+ using_length= list_length(node->using);
+ using_list = (char **) palloc(sizeof(char *) * using_length);
+
delay_string_append_char(message, str, " USING(");
foreach (lc, node->using)
@@ -1698,22 +2139,42 @@ _rewriteJoinExpr(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
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++;
}
+
+ message->analyze[select_num]->join->using_length = using_length;
+ message->analyze[select_num]->join->using_list = using_list;
delay_string_append_char(message, str, ")");
}
+
if (node->quals)
{
delay_string_append_char(message, str, " ON ");
_rewriteNode(BaseSelect, message, dblink, str, node->quals);
}
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ if(IsA(node->rarg, RangeVar) || IsA(node->rarg,RangeSubselect))
+ {
+ if(!natural && inner)
+ build_join_table(message,aliasname,JINNER);
+ else if(!natural && left)
+ build_join_table(message,aliasname,JLEFT);
+ else if(!natural && right)
+ build_join_table(message,aliasname,JRIGHT);
+ else if(!natural && full)
+ build_join_table(message,aliasname,JNORMAL);
+ }
+ }
}
static void
_rewriteFromExpr(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, FromExpr *node)
{
-
}
/*****************************************************************************
@@ -1753,13 +2214,12 @@ _rewriteCreateStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
case ONCOMMIT_DELETE_ROWS:
delay_string_append_char(message, str, " ON COMMIT DELETE ROWS");
-
case ONCOMMIT_PRESERVE_ROWS:
delay_string_append_char(message, str, " ON COMMIT PRESERVE ROWS");
- break;
+ break;
default:
- break;
+ break;
}
if (node->tablespacename)
@@ -1778,10 +2238,10 @@ _rewriteIndexStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbli
if (node->unique == TRUE)
delay_string_append_char(message, str, "UNIQUE ");
- if (node->concurrent == true)
- delay_string_append_char(message, str, "INDEX CONCURRENTLY \"");
- else
- delay_string_append_char(message, str, "INDEX \"");
+ if (node->concurrent == true)
+ delay_string_append_char(message, str, "INDEX CONCURRENTLY \"");
+ else
+ delay_string_append_char(message, str, "INDEX \"");
delay_string_append_char(message, str, node->idxname);
delay_string_append_char(message, str, "\" ON ");
_rewriteNode(BaseSelect, message, dblink, str, node->relation);
@@ -1836,33 +2296,275 @@ _rewriteDeclareCursorStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodbli
if (node->options & CURSOR_OPT_HOLD)
delay_string_append_char(message, str, "WITH HOLD ");
- delay_string_append_char(message, str, "FOR");
- _rewriteNode(BaseSelect, message, dblink, str, node->query);
+ delay_string_append_char(message, str, "FOR");
+ _rewriteNode(BaseSelect, message, dblink, str, node->query);
+}
+
+static void
+initSelectStmt(RewriteQuery *message,SelectStmt *node)
+{
+ int count;
+ int last;
+ int i;
+ AnalyzeSelect *analyze;
+
+ if(message->r_code != SELECT_ANALYZE && message->r_code != SELECT_DEFAULT)
+ return;
+
+ count = message->analyze_num++;
+ last = message->current_select;
+ message->current_select = count;
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ if(count == 0)
+ {
+ message->analyze = (AnalyzeSelect **) palloc(sizeof(AnalyzeSelect *));
+ message->analyze[count]=(AnalyzeSelect *) palloc(sizeof(AnalyzeSelect));
+ message->part = SELECT_START;
+
+ analyze = message->analyze[count];
+ analyze->now_select = 0;
+ analyze->last_select = -1;
+ analyze->part = SELECT_START;
+ analyze->call_part = SELECT_START;
+ }
+ else
+ {
+ message->analyze = (AnalyzeSelect **) repalloc(message->analyze,sizeof(AnalyzeSelect *) * (count+1));
+ message->analyze[count]=(AnalyzeSelect *) palloc(sizeof(AnalyzeSelect));
+ analyze = message->analyze[count];
+ analyze->now_select = count;
+ analyze->last_select = last;
+ analyze->part = message->part;
+ analyze->call_part = message->analyze[last]->part;
+ }
+
+ analyze->range =NULL;
+ analyze->rangeinfo_num =-1;
+ analyze->virtual = NULL;
+ analyze->join = NULL;
+ analyze->state = (char) 0;
+ analyze->aggregate = false;
+ analyze->table_name = NULL;
+ analyze->select_range = false;
+ analyze->rarg_count = -1;
+ analyze->larg_count = -1;
+ analyze->ret_count = 0;
+
+ for(i = 0; i< 8; i++)
+ analyze->partstate[i] = (char)0;
+
+ if(node->larg && node->rarg)
+ analyze->select_union = true;
+ else
+ analyze->select_union = false;
+
+ pool_debug("initSelectStmt: ANALYZE now(%d)",message->current_select);
+
+ /* set default table_state */
+ /* S means that tuples was create by systemdb */
+ /* L means that tuples was create by one node */
+ /* P means that tuples was create by parallel node */
+ message->table_state = 'S';
+ }
+}
+
+static void writeRangeHeader(RewriteQuery *message,ConInfoTodblink *dblink, String *str,DistDefInfo *info, RepliDefInfo *info2,char *alias)
+{
+ char port[8];
+ char *schema = NULL;
+ char *table = NULL;
+
+
+ pool_debug("writeRangeHeader select_no=%d",message->current_select);
+
+ sprintf(port,"%d",dblink->port);
+
+ delay_string_append_char(message, str, "dblink(");
+ delay_string_append_char(message, str, "'");
+ delay_string_append_char(message, str, "host=");
+ delay_string_append_char(message, str, dblink->hostaddr);
+ delay_string_append_char(message, str, " dbname=");
+ delay_string_append_char(message, str, dblink->dbname);
+ delay_string_append_char(message, str, " port=");
+ delay_string_append_char(message, str, port);
+ delay_string_append_char(message, str, " user=");
+ delay_string_append_char(message, str, dblink->user);
+
+ if(strlen(dblink->password))
+ {
+ delay_string_append_char(message, str, " password=");
+ delay_string_append_char(message, str, dblink->password);
+ }
+
+ delay_string_append_char(message, str, "'");
+ delay_string_append_char(message, str, ",");
+ delay_string_append_char(message, str, "'");
+
+ if(info && !info2)
+ {
+ delay_string_append_char(message, str, "SELECT pool_parallel(\"");
+ schema = info->schema_name;
+ if(alias)
+ table = alias;
+ else
+ table = info->table_name;
+ }
+ else if (!info && info2)
+ {
+ delay_string_append_char(message, str, "SELECT pool_loadbalance(\"");
+ schema = info2->schema_name;
+ if(alias)
+ table = alias;
+ else
+ table = info2->table_name;
+ }
+
+ {
+ VirtualTable *virtual = NULL;
+ int no = message->current_select;
+ int v_colnum;
+ int i;
+ int first = 0;
+
+ AnalyzeSelect *analyze = message->analyze[no];
+ virtual = analyze->virtual;
+ v_colnum = virtual->col_num;
+
+ delay_string_append_char(message, str, "SELECT ");
+
+ for(i = 0; i < v_colnum; i++)
+ {
+ if(!strcmp(virtual->table_list[i],table) && virtual->valid[i] != -1)
+ {
+
+ if(first == 0)
+ {
+ delay_string_append_char(message, str, virtual->table_list[i]);
+ delay_string_append_char(message, str, ".");
+ delay_string_append_char(message, str, virtual->col_list[i]);
+ first = 1;
+ }
+ else
+ {
+ delay_string_append_char(message, str, ", ");
+ delay_string_append_char(message, str, virtual->table_list[i]);
+ delay_string_append_char(message, str, ".");
+ delay_string_append_char(message, str, virtual->col_list[i]);
+ }
+ }
+ }
+
+ if(first == 0)
+ delay_string_append_char(message, str, " * ");
+
+ delay_string_append_char(message, str, " FROM ");
+ }
+}
+
+static void writeRangeFooter(RewriteQuery *message,ConInfoTodblink *dblink, String *str,DistDefInfo *info, RepliDefInfo *info2,char *alias)
+{
+ int i,num;
+ char *schema = NULL;
+ char *table = NULL;
+
+ delay_string_append_char(message, str, "\"");
+ delay_string_append_char(message, str, ")");
+ delay_string_append_char(message, str, "'");
+ delay_string_append_char(message, str, ")");
+
+ delay_string_append_char(message, str," AS ");
+
+ if(alias)
+ table = alias;
+ else
+ {
+ if(info && !info2)
+ {
+ schema = info->schema_name;
+ table = info->table_name;
+ }
+ else if (!info && info2)
+ {
+ schema = info2->schema_name;
+ table = info2->table_name;
+ }
+ }
+
+ /* send one node */
+ if(!table)
+ {
+ message->r_code = SELECT_RELATION_ERROR;
+ }
+
+ delay_string_append_char(message, str, table);
+ delay_string_append_char(message, str, "(");
+
+ {
+ VirtualTable *virtual = NULL;
+ int first = 0;
+ int no = message->current_select;
+ AnalyzeSelect *analyze = message->analyze[no];
+ virtual = analyze->virtual;
+ num = virtual->col_num;
+
+ for(i = 0; i < num; i++)
+ {
+ char buf[16];
+ if(!strcmp(virtual->table_list[i],table) && virtual->valid[i] != -1)
+ {
+
+ if(first == 0)
+ first = 1;
+ else
+ delay_string_append_char(message, str, ",");
+
+ snprintf(buf, 16, "%d", analyze->virtual->column_no[i]);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf);
+ delay_string_append_char(message, str, " ");
+ delay_string_append_char(message, str,virtual->type_list[i]);
+ }
+ }
+
+ if(first == 0)
+ {
+ for(i = 0; i < num; i++)
+ {
+ char buf[16];
+ if(!strcmp(virtual->table_list[i],table))
+ {
+ if(first == 0)
+ first = 1;
+ else
+ delay_string_append_char(message, str, ",");
+
+ snprintf(buf, 16, "%d", analyze->virtual->column_no[i]);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf);
+ delay_string_append_char(message, str, " ");
+ delay_string_append_char(message, str,virtual->type_list[i]);
+ }
+ }
+ }
+ delay_string_append_char(message, str, ")");
+ }
}
-static void
-_rewriteSelectStmtIntensive(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SelectStmt *node)
+static void writeSelectHeader(RewriteQuery *message,ConInfoTodblink *dblink, String *str,int parallel,int state)
{
- int message_r_code = message->r_code;
- RangeVar *var = NULL;
char port[8];
- BaseSelect = (Node *) node;
-
- /* set table*/
- var = (RangeVar *) lfirst(list_head(node->fromClause));
-
- delay_string_append_char(message, str, " SELECT ");
- _rewriteIntensiveTargetList(message,node->targetList,str);
- delay_string_append_char(message, str, " FROM ");
-
sprintf(port,"%d",dblink->port);
+ pool_debug("writeSelectHeader select_no=%d state=%d",message->current_select,state);
- if(var->alias)
- delay_string_append_char(message, str, " (SELECT * FROM dblink(");
- else
- delay_string_append_char(message, str, " dblink(");
-
+ if(state == SELECT_START)
+ {
+ delay_string_append_char(message, str, "SELECT * FROM ");
+ }
+
+ delay_string_append_char(message, str, "dblink(");
delay_string_append_char(message, str, "'");
delay_string_append_char(message, str, "host=");
delay_string_append_char(message, str, dblink->hostaddr);
@@ -1881,115 +2583,334 @@ _rewriteSelectStmtIntensive(Node *BaseSelect, RewriteQuery *message, ConInfoTodb
delay_string_append_char(message, str, "'");
delay_string_append_char(message, str, ",");
delay_string_append_char(message, str, "'");
- delay_string_append_char(message, str, "SELECT pool_parallel(\"select ");
- KeepRewriteQueryCode(message, SELECT_DEFAULT_INSIDE_DBLINK);
- _rewriteIntensivedblink(BaseSelect,message,dblink, str,node->targetList);
- KeepRewriteQueryCode(message, message_r_code);
+ if(parallel)
+ delay_string_append_char(message, str, "SELECT pool_parallel(\"");
+ else
+ delay_string_append_char(message, str, "SELECT pool_loadbalance(\"");
- delay_string_append_char(message, str, " FROM ");
- if (var->catalogname)
+ if(state == SELECT_FROMCLAUSE)
{
- delay_string_append_char(message, str, var->catalogname);
- delay_string_append_char(message, str, ".");
- }
+ int no = message->current_select;
+ int v_colnum;
+ int i;
+ int first = 0;
- if (var->schemaname)
- {
- delay_string_append_char(message, str, var->schemaname);
- delay_string_append_char(message, str, ".");
- message->schemaname = var->schemaname;
- }
- else
- message->schemaname = NULL;
+ AnalyzeSelect *analyze = message->analyze[no];
- delay_string_append_char(message, str, var->relname);
-
- /* insert whereClause into dblink() */
- if(node->whereClause)
- {
- delay_string_append_char(message, str, " WHERE ");
- KeepRewriteQueryCode(message, SELECT_DEFAULT_INSIDE_DBLINK);
- _rewriteNode(BaseSelect, message, dblink, str, node->whereClause);
- KeepRewriteQueryCode(message, message_r_code);
- }
- if (node->groupClause)
- {
- delay_string_append_char(message, str, " GROUP BY ");
- _rewriteNode(BaseSelect, message, dblink, str, node->groupClause);
- KeepRewriteQueryCode(message, message_r_code);
+ v_colnum = analyze->virtual->col_num;
+
+ delay_string_append_char(message, str, "SELECT ");
+
+ for(i = 0; i < v_colnum; i++)
+ {
+ if(analyze->virtual->valid[i] != -1)
+ {
+ char *col_name = analyze->virtual->col_list[i];
+
+ if(first == 0)
+ {
+ delay_string_append_char(message, str, analyze->virtual->table_list[i]);
+ delay_string_append_char(message, str, ".");
+ if(strcmp(col_name,"\"?column?\""))
+ delay_string_append_char(message, str, col_name);
+ else
+ {
+ delay_string_append_char(message, str, "\"");
+ delay_string_append_char(message, str, col_name);
+ delay_string_append_char(message, str, "\"");
+ }
+ first = 1;
+ }
+ else
+ {
+ delay_string_append_char(message, str, ", ");
+ delay_string_append_char(message, str, analyze->virtual->table_list[i]);
+ delay_string_append_char(message, str, ".");
+ if(strcmp(col_name,"\"?column?\""))
+ delay_string_append_char(message, str, col_name);
+ else
+ {
+ delay_string_append_char(message, str, "\"");
+ delay_string_append_char(message, str, col_name);
+ delay_string_append_char(message, str, "\"");
+ }
+ }
+ }
+ }
+
+ if(first == 0)
+ delay_string_append_char(message, str, " * ");
+
+ delay_string_append_char(message, str, " FROM ");
}
+}
+static void writeSelectFooter(RewriteQuery *message,String *str,AnalyzeSelect *analyze,int state)
+{
+ int i,num;
+
delay_string_append_char(message, str, "\"");
delay_string_append_char(message, str, ")");
delay_string_append_char(message, str, "'");
delay_string_append_char(message, str, ")");
-
delay_string_append_char(message, str," AS ");
- delay_string_append_char(message, str, var->relname);
+ delay_string_append_char(message, str, analyze->table_name);
delay_string_append_char(message, str, "(");
- _rewriteReturnRecords(BaseSelect,message,node->targetList,str);
- delay_string_append_char(message, str, ")");
-
- if(var->alias)
- delay_string_append_char(message, str, ")");
-
- if(var->alias)
+ pool_debug("writeSelectFooter %s",analyze->table_name);
+
+ if(state != SELECT_FROMCLAUSE)
{
- _rewriteNode(BaseSelect, message, dblink, str, var->alias);
+ num = analyze->select_ret->col_num;
+ for(i = 0; i < num; i++)
+ {
+ delay_string_append_char(message, str, analyze->select_ret->col_list[i]);
+ delay_string_append_char(message, str, " ");
+ delay_string_append_char(message, str, analyze->select_ret->type_list[i]);
+ if (i != num -1)
+ delay_string_append_char(message, str, ",");
+ }
+ delay_string_append_char(message, str, ")");
}
-
- if (node->groupClause)
+ else
{
- delay_string_append_char(message, str, " GROUP BY ");
- _rewriteNode(BaseSelect, message, dblink, str, node->groupClause);
- KeepRewriteQueryCode(message, message_r_code);
+ int first = 0;
+ num = analyze->virtual->col_num;
+
+ for(i = 0; i < num; i++)
+ {
+ if(analyze->virtual->valid[i] != -1)
+ {
+ char buf[16];
+ if(first == 0)
+ {
+ snprintf(buf, 16, "%d", analyze->virtual->column_no[i]);
+ delay_string_append_char(message, str, "pool_c$");
+ delay_string_append_char(message, str, buf);
+ delay_string_append_char(message, str, " ");
+ delay_string_append_char(message, str, analyze->virtual->type_list[i]);
+ first = 1;
+ } else {
+ delay_string_append_char(message, str, ",");
+ snprintf(buf, 16, "%d", analyze->virtual->column_no[i]);
+ delay_string_append_char(message, str, "pool_c$");
+ delay_string_append_char(message, str, buf);
+ delay_string_append_char(message, str, " ");
+ delay_string_append_char(message, str, analyze->virtual->type_list[i]);
+ }
+ }
+ }
+
+ if(first == 0)
+ {
+ for(i = 0; i < num; i++)
+ {
+ if(first == 0)
+ {
+ delay_string_append_char(message, str, analyze->virtual->col_list[i]);
+ delay_string_append_char(message, str, " ");
+ delay_string_append_char(message, str, analyze->virtual->type_list[i]);
+ first = 1;
+ } else {
+ delay_string_append_char(message, str, ",");
+ delay_string_append_char(message, str, analyze->virtual->col_list[i]);
+ delay_string_append_char(message, str, " ");
+ delay_string_append_char(message, str, analyze->virtual->type_list[i]);
+ }
+ }
+ }
+
+ delay_string_append_char(message, str, ")");
}
+}
- if (node->sortClause)
+static void
+CopyFromLeftArg(RewriteQuery *message,int current_num)
+{
+ AnalyzeSelect *l_analyze = message->analyze[current_num + 1];
+ AnalyzeSelect *analyze = message->analyze[current_num];
+ VirtualTable *virtual = message->analyze[current_num]->virtual;
+ int col_num;
+ int i;
+ char **col_list = NULL;
+ char **type_list = NULL;
+ char state;
+ char *table_name;
+
+ table_name = l_analyze->table_name;
+ col_list = l_analyze->select_ret->col_list;
+ type_list = l_analyze->select_ret->type_list;
+ col_num = l_analyze->select_ret->col_num;
+
+ state = l_analyze->state;
+ append_virtual_table(message,virtual,col_list,type_list,col_num,table_name,state,current_num + 1);
+
+ analyze->select_ret = (SelectDefInfo *) palloc(sizeof(SelectDefInfo));
+ analyze->select_ret->valid = false;
+ analyze->select_ret->col_num = 0;
+ for(i =0; i<col_num; i++)
{
- delay_string_append_char(message, str, " ORDER BY ");
- _rewriteNode(BaseSelect, message, dblink, str, node->sortClause);
- KeepRewriteQueryCode(message, message_r_code);
+ append_select_def_info(analyze->select_ret,col_list[i],type_list[i]);
}
+}
- if (node->limitOffset)
+static void
+ChangeStateByCluase(RewriteQuery *message,void *obj,int before, int after)
+{
+ AnalyzeSelect *analyze;
+ int count = message->current_select;
+
+ if(message->r_code != SELECT_ANALYZE)
+ return;
+
+ analyze = message->analyze[count];
+
+ if (obj && analyze->partstate[before] == 'P')
+ analyze->partstate[after] = 'S';
+ else
+ analyze->partstate[after] = analyze->partstate[before];
+}
+
+static void
+ChangeStateRewriteFooter(RewriteQuery *message,String *str,int defore, int after)
+{
+ AnalyzeSelect *analyze;
+ int count = message->current_select;
+ char state;
+
+ if(message->r_code != SELECT_DEFAULT)
+ return;
+
+ analyze = message->analyze[count];
+
+ if(analyze->state == 'S' && message->rewritelock == count
+ && message->ignore_rewrite == -1)
{
- delay_string_append_char(message, str, " OFFSET ");
- _rewriteNode(BaseSelect, message, dblink, str, node->limitOffset);
- KeepRewriteQueryCode(message, message_r_code);
+ state = analyze->partstate[defore];
+
+ if(state == 'L' || state == 'P')
+ {
+ if(state != analyze->partstate[after])
+ {
+ writeSelectFooter(message,str,analyze,SELECT_FROMCLAUSE);
+ message->rewritelock = -1;
+ }
+ }
}
+}
- if (node->limitCount)
+static bool
+CheckUnionFromClause(RewriteQuery *message)
+{
+ int check;
+ int count = message->current_select;
+ AnalyzeSelect *analyze;
+ analyze = message->analyze[count];
+
+ check = analyze->last_select;
+ pool_debug("CheckUnion select=%d last_select=%d", count,check);
+
+ if(check == -1)
+ return false;
+ else
{
- delay_string_append_char(message, str, " LIMIT ");
- if (IsA(node->limitCount, A_Const) &&
- ((A_Const *)node->limitCount)->val.type == T_Null)
+ if(message->analyze[check]->select_union &&
+ message->analyze[check]->call_part == SELECT_FROMCLAUSE)
{
- delay_string_append_char(message, str, "ALL ");
+ pool_debug("CheckUnion true");
+ return true;
}
else
- {
- _rewriteNode(BaseSelect, message, dblink, str, node->limitCount);
- KeepRewriteQueryCode(message, message_r_code);
- }
+ return false;
}
}
static void
_rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SelectStmt *node)
{
- int message_r_code = message->r_code;
BaseSelect = (Node *) node;
+ AnalyzeSelect *analyze;
+ int count;
+ int analyze_num = 0;
+ int target_analyze = 0;
+ int from_analyze = 0;
+ bool lock = false;
+ bool direct = false;
+
+ count = message->analyze_num;
+ initSelectStmt(message,node);
+ if(message->r_code == SELECT_DEFAULT)
+ {
+ if(message->current_select == 0)
+ pool_debug("_rewriteSelectStmt:START QueryRewrite");
+ }
+
+ analyze_num = message->analyze_num;
+
+ analyze = message->analyze[count];
+ analyze->part = SELECT_START;
+
+ if(message->r_code == SELECT_DEFAULT && message->rewritelock == -1
+ && message->ignore_rewrite == -1)
+ {
+ if(analyze->state == 'P')
+ {
+ if(analyze->call_part != SELECT_FROMCLAUSE || !CheckUnionFromClause(message))
+ {
+ writeSelectHeader(message,dblink,str,PARALLEL,analyze->part);
+ message->rewritelock = count;
+ direct = true;
+ }
+ else
+ lock =true;
+ }
+ if(analyze->state == 'L')
+ {
+ if(analyze->call_part != SELECT_FROMCLAUSE && !CheckUnionFromClause(message))
+ {
+ writeSelectHeader(message,dblink,str,LOADBALANCE,analyze->part);
+ message->rewritelock = count;
+ direct = true;
+ }
+ else
+ lock =true;
+ }
+ }
+
if (node->larg) /* SETOP */
{
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ char buf[16];
+ char *temp = "pool_t$";
+ analyze->virtual = (VirtualTable *) palloc(sizeof(VirtualTable));
+ snprintf(buf, 16, "%d", message->virtual_num);
+ message->virtual_num++;
+ analyze->table_name = (char *) palloc(sizeof(char) * (strlen(temp) + strlen(buf) + 1));
+ strcpy(analyze->table_name,temp);
+ strcat(analyze->table_name,buf);
+ analyze->virtual->col_num = 0;
+ }
+
delay_string_append_char(message, str, "(");
+ message->part = SELECT_START;
_rewriteNode(BaseSelect, message, dblink, str, node->larg);
+ pool_debug("union larg select_no=%d(%d)",count,message->analyze_num);
+
delay_string_append_char(message, str, ")");
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_START);
+
+ /* COPY analyze of left arg */
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ //CopyFromLeftArg(message,count);
+ analyze->larg_count = count + 1;
+ analyze->rarg_count = message->analyze_num;
+ pool_debug("_rewriteSelectStmt: COUNT larg=%d, rarg=%d",analyze->larg_count, analyze->rarg_count);
+ }
switch (node->op)
{
@@ -2014,9 +2935,48 @@ _rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
if (node->rarg)
{
delay_string_append_char(message, str, "(");
+ message->part = SELECT_START;
_rewriteNode(BaseSelect, message, dblink, str, node->rarg);
delay_string_append_char(message, str, ")");
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_START);
+
+ if(message->r_code == SELECT_ANALYZE)
+ CopyFromLeftArg(message,count);
+ }
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ pool_debug("_rewriteSelectStmt: STATE larg=%c, rarg=%c",message->analyze[analyze->larg_count]->state,
+ message->analyze[analyze->rarg_count]->state);
+
+ }
+
+ if(message->r_code == SELECT_DEFAULT)
+ {
+ pool_debug("_rewriteSelectStmt: DEFAULT COUNT larg=%d, rarg=%d",analyze->larg_count, analyze->rarg_count);
+ }
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ int lcount = analyze->larg_count;
+ int rcount = analyze->rarg_count;
+ if(message->analyze[lcount]->state == 'L' &&
+ message->analyze[rcount]->state == 'L')
+ {
+ int j;
+ for(j = 0; j < SELECT_SORTCLAUSE; j++)
+ {
+ analyze->partstate[j]='L';
+ }
+ }
+ else
+ {
+ int j;
+ for(j = 0; j < SELECT_SORTCLAUSE; j++)
+ {
+ analyze->partstate[j]='S';
+ }
+ }
}
}
else
@@ -2031,14 +2991,14 @@ _rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
delay_string_append_char(message, str, "TEMP ");
delay_string_append_char(message, str, "TABLE ");
_rewriteNode(BaseSelect, message, dblink, str, into->rel);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepRewriteQueryReturnCode(message, SELECT_RELATION_ERROR);
if (into->colNames)
{
delay_string_append_char(message, str, " (");
_rewriteNode(BaseSelect, message, dblink, str, into->colNames);
delay_string_append_char(message, str, ") ");
- KeepRewriteQueryCode(message, message_r_code);
+ KeepRewriteQueryReturnCode(message, SELECT_RELATION_ERROR);
}
if (into->options)
@@ -2061,77 +3021,244 @@ _rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
default:
break;
}
-
delay_string_append_char(message, str, " AS");
}
delay_string_append_char(message, str, " SELECT ");
+ /*
+ * Check from-clause before Checking target-list
+ */
+ if(node->fromClause)
+ {
+ message->part = SELECT_FROMCLAUSE;
+ analyze->part = SELECT_FROMCLAUSE;
+ message->fromClause = true;
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ char buf[16];
+ char *temp = "pool_t$";
+ analyze->virtual = (VirtualTable *) palloc(sizeof(VirtualTable));
+ snprintf(buf, 16, "%d", message->virtual_num);
+ message->virtual_num++;
+ analyze->table_name = (char *) palloc(sizeof(char) * (strlen(temp) + strlen(buf) + 1));
+ strcpy(analyze->table_name,temp);
+ strcat(analyze->table_name,buf);
+ analyze->virtual->col_num = 0;
+ }
+
+ if(message->r_code == SELECT_DEFAULT && message->ignore_rewrite == -1)
+ message->ignore_rewrite = count;
+
+ /* remember analyze_num */
+ from_analyze = message->analyze_num;
+
+ _rewriteNode(BaseSelect, message, dblink, str, node->fromClause);
+
+ if(message->r_code == SELECT_DEFAULT && message->ignore_rewrite == count)
+ message->ignore_rewrite = -1;
+
+ message->fromClause = false;
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ message->table_state = analyze->state;
+ analyze->partstate[SELECT_FROMCLAUSE] = analyze->state;
+ }
+ }
+ else
+ {
+ /* this is const or function call*/
+ message->part = SELECT_TARGETLIST;
+ if(message->r_code == SELECT_ANALYZE)
+ build_range_info(message,NULL,NULL,NULL,NULL,message->current_select,-1);
+ }
+
+ message->part = SELECT_OTHER;
+ analyze->part = SELECT_OTHER;
+
if (node->distinctClause)
{
+ if(message->r_code == SELECT_ANALYZE)
+ analyze->aggregate = true;
+
delay_string_append_char(message, str, "DISTINCT ");
if (lfirst(list_head(node->distinctClause)) != NIL)
{
delay_string_append_char(message, str, "ON (");
_rewriteNode(BaseSelect, message, dblink,str, node->distinctClause);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
delay_string_append_char(message, str, ") ");
}
}
+ message->part = SELECT_TARGETLIST;
+ analyze->part = SELECT_TARGETLIST;
+
+ if(analyze->partstate[SELECT_FROMCLAUSE] == 'P' && analyze->aggregate)
+ analyze->partstate[SELECT_TARGETLIST] = 'S';
+
+ /* TARGETLIST START */
_rewriteNode(BaseSelect, message, dblink, str, node->targetList);
- KeepRewriteQueryCode(message, message_r_code);
- if (node->fromClause)
+ target_analyze = message->analyze_num;
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ if (analyze->aggregate && (analyze->partstate[SELECT_FROMCLAUSE] == 'P'
+ || analyze->partstate[SELECT_TARGETLIST] == 'P'))
+ analyze->partstate[SELECT_TARGETLIST] = 'S';
+ else if (!analyze->partstate[SELECT_TARGETLIST])
+ analyze->partstate[SELECT_TARGETLIST] = analyze->partstate[SELECT_FROMCLAUSE];
+ }
+
+ KeepMessages(message,count,SELECT_TARGETLIST);
+
+ if (node->fromClause && message->r_code != SELECT_ANALYZE)
{
+
+ message->analyze_num = from_analyze;
+
+ message->part = SELECT_FROMCLAUSE;
+ analyze->part = SELECT_FROMCLAUSE;
delay_string_append_char(message, str, " FROM ");
+
+ if(message->r_code == SELECT_DEFAULT &&
+ (analyze->state == 'S' || lock)
+ && message->rewritelock == -1 && message->ignore_rewrite ==-1)
+ {
+ if(analyze->partstate[SELECT_FROMCLAUSE] == 'L')
+ {
+ writeSelectHeader(message,dblink,str,LOADBALANCE, message->part);
+ message->rewritelock = count;
+ }
+ else if(analyze->partstate[SELECT_FROMCLAUSE] == 'P')
+ {
+ writeSelectHeader(message,dblink,str,PARALLEL, message->part);
+ message->rewritelock = count;
+ }
+ }
+
+ message->fromClause = true;
_rewriteNode(BaseSelect, message, dblink, str, node->fromClause);
- KeepRewriteQueryCode(message, message_r_code);
+ message->fromClause = false;
+ KeepMessages(message,count,SELECT_FROMCLAUSE);
+
+ if(message->r_code == SELECT_DEFAULT && analyze->state == 'S'
+ && message->rewritelock == count && message->ignore_rewrite == -1)
+ {
+ if(analyze->partstate[SELECT_FROMCLAUSE] == 'L')
+ {
+ if(analyze->partstate[SELECT_WHERECLAUSE] != 'L')
+ {
+ if(node->whereClause)
+ {
+ int message_code = message->r_code;
+ delay_string_append_char(message, str, " WHERE ");
+ CheckWhereCaluse(node->whereClause, message,dblink,str,0);
+ KeepRewriteQueryReturnCode(message, message_code);
+ }
+
+ writeSelectFooter(message,str,analyze,SELECT_FROMCLAUSE);
+ message->rewritelock = -1;
+ }
+ }
+ else if(analyze->partstate[SELECT_FROMCLAUSE] == 'P')
+ {
+ if(analyze->partstate[SELECT_WHERECLAUSE] != 'P')
+ {
+ if(node->whereClause)
+ {
+ int message_code = message->r_code;
+ delay_string_append_char(message, str, " WHERE ");
+ CheckWhereCaluse(node->whereClause, message,dblink,str,0);
+ KeepRewriteQueryReturnCode(message, message_code);
+ }
+ writeSelectFooter(message,str,analyze,SELECT_FROMCLAUSE);
+ message->rewritelock = -1;
+ }
+ }
+ }
+ message->analyze_num = target_analyze;
}
-
- /* reset where clause */
+
+ /* WHERE CLAUSE */
+
+ message->part = SELECT_OTHER;
if (node->whereClause)
BaseSelect = NULL;
- if (node->whereClause && message->r_code != SELECT_ONETABLE)
+ if (node->whereClause)
{
+ message->part = SELECT_WHERECLAUSE;
+ analyze->part = SELECT_WHERECLAUSE;
delay_string_append_char(message, str, " WHERE ");
_rewriteNode(BaseSelect, message, dblink, str, node->whereClause);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
}
+ if(!analyze->partstate[SELECT_WHERECLAUSE] && message->r_code == SELECT_ANALYZE)
+ analyze->partstate[SELECT_WHERECLAUSE] = analyze->partstate[SELECT_TARGETLIST];
+
+ ChangeStateRewriteFooter(message,str,SELECT_WHERECLAUSE, SELECT_GROUPBYCLAUSE);
+
+ /* GROUPBY CLAUSE */
if (node->groupClause)
{
+ analyze->part = SELECT_GROUPBYCLAUSE;
delay_string_append_char(message, str, " GROUP BY ");
_rewriteNode(BaseSelect, message, dblink, str, node->groupClause);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
}
+ ChangeStateByCluase(message,node->groupClause,SELECT_WHERECLAUSE,SELECT_GROUPBYCLAUSE);
+
+ /* HAVING CLAUSE */
if (node->havingClause)
{
+ analyze->part = SELECT_HAVINGCLAUSE;
delay_string_append_char(message, str, " HAVING ");
_rewriteNode(BaseSelect, message, dblink, str, node->havingClause);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
+
}
+
+ if(message->r_code == SELECT_ANALYZE)
+ analyze->partstate[SELECT_HAVINGCLAUSE] = analyze->partstate[SELECT_GROUPBYCLAUSE];
}
-
+
+ ChangeStateRewriteFooter(message,str,SELECT_HAVINGCLAUSE, SELECT_SORTCLAUSE);
+
if (node->sortClause)
{
+ analyze->part = SELECT_SORTCLAUSE;
delay_string_append_char(message, str, " ORDER BY ");
_rewriteNode(BaseSelect, message, dblink, str, node->sortClause);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
}
+
+ ChangeStateByCluase(message,node->sortClause,SELECT_HAVINGCLAUSE,SELECT_SORTCLAUSE);
+ ChangeStateRewriteFooter(message,str, SELECT_SORTCLAUSE, SELECT_OFFSETCLAUSE);
+
if (node->limitOffset)
{
+ analyze->part = SELECT_OFFSETCLAUSE;
delay_string_append_char(message, str, " OFFSET ");
_rewriteNode(BaseSelect, message, dblink, str, node->limitOffset);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
+
}
+ ChangeStateByCluase(message,node->limitOffset,SELECT_SORTCLAUSE,SELECT_OFFSETCLAUSE);
+
+ ChangeStateRewriteFooter(message,str,SELECT_OFFSETCLAUSE,SELECT_LIMITCLAUSE);
+
if (node->limitCount)
{
+ analyze->part = SELECT_LIMITCLAUSE;
delay_string_append_char(message, str, " LIMIT ");
if (IsA(node->limitCount, A_Const) &&
((A_Const *)node->limitCount)->val.type == T_Null)
@@ -2141,12 +3268,95 @@ _rewriteSelectStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
else
{
_rewriteNode(BaseSelect, message, dblink, str, node->limitCount);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
+ }
+ }
+
+ ChangeStateByCluase(message,node->limitCount,SELECT_OFFSETCLAUSE,SELECT_LIMITCLAUSE);
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ int i;
+ analyze->state = analyze->partstate[SELECT_LIMITCLAUSE];
+
+ for(i = 0; i< 8; i++)
+ {
+ char s = analyze->partstate[i];
+ if(s == 'S')
+ {
+ analyze->state = s;
+ break;
+ }
}
}
_rewriteNode(BaseSelect, message, dblink, str, node->lockingClause);
- KeepRewriteQueryCode(message, message_r_code);
+ KeepMessages(message,count,SELECT_OTHER);
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ if(node->targetList)
+ AnalyzeReturnRecord(BaseSelect,message,dblink,str,node->targetList);
+
+ pool_debug("_rewriteSelectStmt select_no=%d state=%s",message->current_select,analyze->partstate);
+
+ /* change state */
+ if(count != 0)
+ {
+ AnalyzeSelect *last = message->analyze[analyze->last_select];
+ if(last->part == SELECT_WHERECLAUSE)
+ {
+ char fromstate = last->partstate[SELECT_FROMCLAUSE];
+ char wherestate = (char) 0;
+
+ if(last->partstate[SELECT_WHERECLAUSE])
+ wherestate = last->partstate[SELECT_WHERECLAUSE];
+
+ if(fromstate == 'P' && analyze->state == 'L')
+ {
+ if(wherestate && wherestate == 'P')
+ last->partstate[SELECT_WHERECLAUSE] = 'P';
+ }
+ else if(fromstate == 'L' && analyze->state == 'L')
+ {
+ last->partstate[SELECT_WHERECLAUSE] = 'L';
+ }
+ else
+ last->partstate[SELECT_WHERECLAUSE] = 'S';
+ }
+
+ if(last->part == SELECT_TARGETLIST)
+ {
+ char fromstate = last->partstate[SELECT_FROMCLAUSE];
+ char targetstate = (char) 0;
+
+ if(last->partstate[SELECT_TARGETLIST])
+ targetstate = last->partstate[SELECT_TARGETLIST];
+
+ if(fromstate == 'P' && analyze->state == 'L')
+ {
+ last->partstate[SELECT_TARGETLIST] = 'P';
+ }
+ else if(fromstate == 'L' && analyze->state == 'L')
+ {
+ last->partstate[SELECT_TARGETLIST] = 'L';
+ }
+ else
+ last->partstate[SELECT_TARGETLIST] = 'S';
+ }
+ }
+ }
+
+ if(message->r_code == SELECT_DEFAULT && message->rewritelock == count
+ && message->ignore_rewrite ==-1 )
+ {
+ if(direct)
+ writeSelectFooter(message,str,analyze,analyze->call_part);
+ else
+ writeSelectFooter(message,str,analyze,SELECT_FROMCLAUSE);
+
+ message->rewritelock = -1;
+ }
}
static void
@@ -2155,13 +3365,34 @@ _rewriteFuncCall(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
char *funcname;
if(message->r_code == SELECT_AEXPR)
{
- KeepRewriteQueryCode(message, SELECT_AEXPR_FALSE);
+ KeepRewriteQueryReturnCode(message, SELECT_AEXPR_FALSE);
return;
}
_rewriteFuncName(BaseSelect, message, dblink, str, node->funcname);
funcname = strVal(lfirst(list_head(node->funcname)));
+ if(message->r_code == SELECT_ANALYZE && funcname)
+ {
+ /* aggregate functions */
+ if(!strcmp(funcname,"count") || !strcmp(funcname,"max") || !strcmp(funcname,"min")
+ || !strcmp(funcname,"sum") || !strcmp(funcname,"avg") || !strcmp(funcname,"bit_and")
+ || !strcmp(funcname,"bit_or") || !strcmp(funcname,"bool_and") || !strcmp(funcname,"bool_or")
+ || !strcmp(funcname,"every") || !strcmp(funcname,"corr") || !strcmp(funcname,"covar_pop")
+ || !strcmp(funcname,"covar_samp") || !strcmp(funcname,"regr_avgx") || !strcmp(funcname,"regr_avgy")
+ || !strcmp(funcname,"regr_count") || !strcmp(funcname,"regr_intercept") || !strcmp(funcname,"regr_r2")
+ || !strcmp(funcname,"regr_slope") || !strcmp(funcname,"regr_sxx") || !strcmp(funcname,"regr_sxy")
+ || !strcmp(funcname,"regr_syy") || !strcmp(funcname,"stddev") || !strcmp(funcname,"stddev_pop")
+ || !strcmp(funcname,"stddev_samp") || !strcmp(funcname,"variance") || !strcmp(funcname,"var_pop")
+ || !strcmp(funcname,"var_samp"))
+ {
+ AnalyzeSelect *analyze;
+ int no = message->current_select;
+ analyze = message->analyze[no];
+ analyze->aggregate = true;
+ }
+ }
+
if(strcmp(funcname,"user") == 0 ||
strcmp(funcname,"current_user") == 0 ||
strcmp(funcname,"session_user") == 0 ||
@@ -2231,7 +3462,7 @@ _rewriteTypeName(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblin
delay_string_append_char(message, str, ".");
if(node->typemod < 0)
{
- if(message->r_code == SELECT_DEFAULT_INSIDE_DBLINK)
+ if(message->rewritelock != -1)
{
delay_string_append_char(message, str, "\"\"");
delay_string_append_char(message, str, typename);
@@ -2326,111 +3557,519 @@ _rewriteAExpr(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink,
{
Value *op = (Value *) lfirst(list_head(node->name));
- delay_string_append_char(message, str, " (");
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- delay_string_append_char(message, str, op->val.str);
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, " )");
- }
- break;
+ delay_string_append_char(message, str, " (");
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ delay_string_append_char(message, str, op->val.str);
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, " )");
+ }
+ break;
+
+ case AEXPR_AND:
+ delay_string_append_char(message, str, " (");
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ delay_string_append_char(message, str, " AND ");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ case AEXPR_OR:
+ delay_string_append_char(message, str, " (");
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ delay_string_append_char(message, str, " OR ");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ case AEXPR_NOT:
+ delay_string_append_char(message, str, " (NOT ");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ case AEXPR_OP_ANY:
+ /* not implemented yet */
+ break;
+
+ case AEXPR_OP_ALL:
+ /* not implemented yet */
+ break;
+
+ case AEXPR_DISTINCT:
+ delay_string_append_char(message, str, " (");
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ delay_string_append_char(message, str, " IS DISTINCT FROM ");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ case AEXPR_NULLIF:
+ delay_string_append_char(message, str, " NULLIF(");
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ delay_string_append_char(message, str, ", ");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ case AEXPR_OF:
+ _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
+ if (*(char *)lfirst(list_head(node->name)) == '!')
+ delay_string_append_char(message, str, " IS NOT OF (");
+ else
+ delay_string_append_char(message, str, " IS OF (");
+ _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
+ delay_string_append_char(message, str, ")");
+ break;
+
+ default:
+ break;
+ }
+}
+
+static void
+_rewriteValue(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, Value *value)
+{
+ char buf[16];
+
+ switch (value->type)
+ {
+ case T_Integer:
+ sprintf(buf, "%ld", value->val.ival);
+ delay_string_append_char(message, str, buf);
+ break;
+
+ case T_Float:
+ delay_string_append_char(message, str, value->val.str);
+ break;
+
+ case T_String:
+ delay_string_append_char(message, str, "'");
+ delay_string_append_char(message, str, escape_string(value->val.str));
+ delay_string_append_char(message, str, "'");
+ break;
+
+ case T_Null:
+ delay_string_append_char(message, str, "NULL");
+ break;
+
+ default:
+ break;
+ }
+}
+
+static char
+GetInnerRef(RewriteQuery *message,int last,char *table_name,char *column,char state)
+{
+ AnalyzeSelect *analyze;
+ VirtualTable *virtual;
+ int num, i;
+ bool get = false;
+
+ analyze = message->analyze[last];
+ virtual = analyze->virtual;
+ num = virtual->col_num;
+
+ for(i=0; i < num; i++)
+ {
+ char *vcol = virtual->col_list[i];
+ char *vtable = virtual->table_list[i];
+ if(table_name
+ && !strcmp(table_name,vtable) && !strcmp(column,vcol))
+ {
+ pool_debug("GetInnerRef state = %c now select(%d), table_name=(%s), col_name=(%s) detect",state,last,vtable,vcol);
+ get = true;
+ break;
+ }
+ else if(!table_name && !strcmp(column,vcol))
+ {
+ pool_debug("GetInnerRef state = %c now select(%d), table_name=(%s), col_name=(%s) detect",state,last,vtable,vcol);
+ get = true;
+ break;
+ }
+ }
+
+ if(get)
+ return state;
+ else
+ {
+ /* there isn't the inner refarence at parallel part */
+ if(state == 'P' || analyze->partstate[SELECT_FROMCLAUSE] =='P')
+ return 'S';
+
+ if(last == 0)
+ return 'E';
+ else
+ return GetInnerRef(message,analyze->last_select,table_name,column,state);
+ }
+}
+
+static void
+ChangeStatebyColumnRef(RewriteQuery *message,ColumnRef *col)
+{
+ AnalyzeSelect *analyze;
+ VirtualTable *virtual;
+ ListCell *c;
+ List *list;
+ char first = 0;
+ char *table_name = NULL;
+ char *column;
+ int num,no;
+
+ list = col->fields;
+
+ no = message->current_select;
+ analyze = message->analyze[no];
+
+ if(list->length > 2 || list->length == 0)
+ {
+ /* Is this error ? */
+ message->analyze[no]->state = 'S';
+ return;
+ }
+
+ foreach (c, col->fields)
+ {
+ Node *n = (Node *) lfirst(c);
+
+ if (IsA(n, String))
+ {
+ Value *v = (Value *) lfirst(c);
+ if(list->length == 2 && first == 0)
+ {
+ first = 1;
+ table_name = v->val.str;
+ }
+ else
+ column = v->val.str;
+ }
+ }
+
+ if(!column)
+ {
+ message->table_state = 'S';
+ return;
+ }
+
+ pool_debug("ChangeStatebyColumnRef %s now(%d),last(%d) part(%d) call_part(%d)",
+ column,
+ analyze->now_select,
+ analyze->last_select,
+ analyze->part,
+ analyze->call_part
+ );
+
+ virtual = analyze->virtual;
+ num = virtual->col_num;
+
+ if(message->part==SELECT_WHERECLAUSE || message->part == SELECT_TARGETLIST)
+ {
+ if(analyze->partstate[message->part] != 'S')
+ {
+ analyze->partstate[message->part] =
+ GetInnerRef(message,no,table_name,column,analyze->partstate[SELECT_FROMCLAUSE]);
+ pool_debug("return state is %c",analyze->partstate[message->part]);
+ }
+ }
+}
+
+static bool DetectValidColumn(RewriteQuery *message,char *table_name,char *column_name,int no, int call)
+{
+ AnalyzeSelect *analyze;
+ VirtualTable *virtual;
+ int v_num,i;
+ int get = 0;
+ int call_num;
+ bool star = false;
+
+ call_num = message->current_select;
+
+ if(call != -1 && no != call_num && call == SELECT_FROMCLAUSE)
+ {
+ int last = message->analyze[no]->last_select;
+ int call_part = message->analyze[no]->call_part;
+
+ if(last == -1)
+ {
+ return false;
+ }
+ else
+ {
+ return DetectValidColumn(message,table_name,column_name,last,call_part);
+ }
+ }
+
+ analyze = message->analyze[no];
+ virtual = analyze->virtual;
+ v_num = virtual->col_num;
+
+ for(i = 0; i< v_num; i++)
+ {
+ char *vcol = virtual->col_list[i];
+ char *vtable = virtual->table_list[i];
+
+ if(table_name && !strcmp(table_name,vtable) && !strcmp(column_name,"*"))
+ {
+ virtual->valid[i] = message->current_select;
+ star = true;
+ }
+ else if (!table_name && !strcmp(column_name,"*"))
+ {
+ virtual->valid[i] = message->current_select;
+ star = true;
+ }
+
+ if(table_name
+ && !strcmp(table_name,vtable) && !strcmp(column_name,vcol))
+ {
+ pool_debug("DetectValidColumn no = %d, table_name=(%s), col_name=(%s) detect",no, vtable, vcol);
+ get++;
+
+ if(virtual->valid[i] == -1)
+ {
+ virtual->valid[i] = message->current_select;
+ }
+ else if (virtual->valid[i] > no)
+ {
+ virtual->valid[i] = message->current_select;
+ }
+ }
+ else if(!table_name && !strcmp(column_name,vcol))
+ {
+ pool_debug("DetectValidColumn no = %d, col_name=(%s) detect",no, vcol);
+ get++;
+
+ if(virtual->valid[i] == -1)
+ {
+ virtual->valid[i] = message->current_select;
+ }
+ else if (virtual->valid[i] > no)
+ {
+ virtual->valid[i] = message->current_select;
+ }
+ }
+ }
+
+ if(star)
+ return true;
+
+ if(get == 1)
+ {
+ return true;
+ }
+ else if(get == 0)
+ {
+ int last = analyze->last_select;
+ if(last != -1)
+ {
+ return DetectValidColumn(message,table_name,column_name,analyze->last_select,analyze->call_part);
+ }
+ }
+ else if(get>= 2)
+ pool_debug("DetectValidColumn select_no=(%d) col_name=(%s) ambiguous",message->current_select,column_name);
+
+ return false;
+}
+
+static bool GetPoolColumn(RewriteQuery *message,String *str,char *table_name,char *column_name,int no, int call,bool state)
+{
+ AnalyzeSelect *analyze,*analyze_now;
+ VirtualTable *virtual;
+ int v_num,i;
+ int get = 0;
+ int call_num;
+ bool star = false;
+
+ call_num = message->current_select;
+
+ if(call != -1 && no != call_num && call == SELECT_FROMCLAUSE)
+ {
+ int last = message->analyze[no]->last_select;
+ int call_part = message->analyze[no]->call_part;
+
+ if(last == -1)
+ {
+ return false;
+ }
+ else
+ {
+ return GetPoolColumn(message,str,table_name,column_name,last,call_part,state);
+ }
+ }
+
+ analyze = message->analyze[no];
+ analyze_now = message->analyze[call_num];
+ virtual = analyze->virtual;
+ v_num = virtual->col_num;
- case AEXPR_AND:
- delay_string_append_char(message, str, " (");
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- delay_string_append_char(message, str, " AND ");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
+ for(i = 0; i< v_num; i++)
+ {
+ char *vcol = virtual->col_list[i];
+ char *vtable = virtual->table_list[i];
+
+ if(table_name && !strcmp(table_name,vtable) && !strcmp(column_name,"*"))
+ {
+ star = true;
break;
-
- case AEXPR_OR:
- delay_string_append_char(message, str, " (");
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- delay_string_append_char(message, str, " OR ");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
+ }
+ else if (!table_name && !strcmp(column_name,"*"))
+ {
+ star = true;
break;
-
- case AEXPR_NOT:
- delay_string_append_char(message, str, " (NOT ");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
+ }
+
+ if(table_name
+ && !strcmp(table_name,vtable) && !strcmp(column_name,vcol))
+ {
+ pool_debug("GetPoolColumn no = %d, table_name=(%s), col_name=(%s) detect",no, vtable, vcol);
+ get++;
break;
-
- case AEXPR_OP_ANY:
- /* not implemented yet */
+ }
+ else if(!table_name && !strcmp(column_name,vcol))
+ {
+ pool_debug("GetPoolColumn no = %d, col_name=(%s) detect",no, vcol);
+ get++;
break;
+ }
+ }
- case AEXPR_OP_ALL:
- /* not implemented yet */
- break;
+ if(star)
+ {
+ if(table_name)
+ {
+ int first = 0;
+ for(i = 0; i < v_num; i++)
+ {
+ char buf[16];
+ if(!strcmp(virtual->table_list[i],table_name) && virtual->valid[i] != -1)
+ {
+ if(first == 0)
+ first = 1;
+ else
+ delay_string_append_char(message, str, ",");
- case AEXPR_DISTINCT:
- delay_string_append_char(message, str, " (");
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- delay_string_append_char(message, str, " IS DISTINCT FROM ");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
- break;
+ if(message->rewritelock == -1)
+ {
+ snprintf(buf, 16, "%d", analyze->virtual->column_no[i]);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf);
- case AEXPR_NULLIF:
- delay_string_append_char(message, str, " NULLIF(");
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- delay_string_append_char(message, str, ", ");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
- break;
+ if(message->ignore_rewrite == -1 && call_num != 0 && analyze->call_part == SELECT_FROMCLAUSE)
+ {
+ char buf2[16];
+ int col_no = analyze->select_ret->return_list[analyze->ret_count];
+ delay_string_append_char(message, str, " AS ");
+ snprintf(buf2, 16, "%d", col_no);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf2);
+ delay_string_append_char(message, str," ");
+ analyze->ret_count++;
+ continue;
+ }
+ } else {
+ delay_string_append_char(message, str,analyze->virtual->col_list[i]);
+ }
- case AEXPR_OF:
- _rewriteNode(BaseSelect, message, dblink, str, node->lexpr);
- if (*(char *)lfirst(list_head(node->name)) == '!')
- delay_string_append_char(message, str, " IS NOT OF (");
- else
- delay_string_append_char(message, str, " IS OF (");
- _rewriteNode(BaseSelect, message, dblink, str, node->rexpr);
- delay_string_append_char(message, str, ")");
- break;
+ delay_string_append_char(message, str, " ");
- default:
- break;
- }
-}
+ if(state && message->rewritelock == -1)
+ {
+ delay_string_append_char(message, str, " AS ");
+ delay_string_append_char(message, str, "\"");
+ delay_string_append_char(message, str, analyze->select_ret->col_list[message->ret_num]);
+ delay_string_append_char(message, str, "\" ");
+ message->ret_num++;
+ }
+ }
+ }
+ }
+ else
+ {
+ int first = 0;
+ for(i = 0; i < v_num; i++)
+ {
+ char buf[16];
+ if(virtual->valid[i] != -1)
+ {
+ if(first == 0)
+ first = 1;
+ else
+ delay_string_append_char(message, str, ",");
-static void
-_rewriteValue(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, Value *value)
-{
- char buf[16];
+ if(message->rewritelock == -1)
+ {
+ snprintf(buf, 16, "%d", analyze->virtual->column_no[i]);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf);
- switch (value->type)
- {
- case T_Integer:
- sprintf(buf, "%ld", value->val.ival);
- delay_string_append_char(message, str, buf);
- break;
+ if(message->ignore_rewrite == -1 && call_num != 0 && analyze->call_part == SELECT_FROMCLAUSE)
+ {
+ char buf2[16];
+ int col_no = analyze->select_ret->return_list[analyze->ret_count];
+ delay_string_append_char(message, str, " AS ");
+ snprintf(buf2, 16, "%d", col_no);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf2);
+ delay_string_append_char(message, str," ");
+ analyze->ret_count++;
+ continue;
+ }
- case T_Float:
- delay_string_append_char(message, str, value->val.str);
- break;
+ } else {
+ delay_string_append_char(message, str,analyze->virtual->col_list[i]);
+ }
+ delay_string_append_char(message, str, " ");
- case T_String:
- delay_string_append_char(message, str, "'");
- delay_string_append_char(message, str, escape_string(value->val.str));
- delay_string_append_char(message, str, "'");
- break;
+ if(state && message->rewritelock == -1)
+ {
+ delay_string_append_char(message, str, " AS ");
+ delay_string_append_char(message, str, analyze->select_ret->col_list[message->ret_num]);
+ message->ret_num++;
+ }
+ }
+ }
+ }
+ return true;
+ }
- case T_Null:
- delay_string_append_char(message, str, "NULL");
- break;
+ if(get == 1)
+ {
+ char buf[16];
+ snprintf(buf, 16, "%d", virtual->column_no[i]);
- default:
- break;
+ if(analyze_now->partstate[SELECT_FROMCLAUSE] != 'S')
+ delay_string_append_char(message, str, analyze->table_name);
+ 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, buf);
+
+ return true;
}
+ else if(get == 0)
+ {
+ int last = analyze->last_select;
+ if(last != -1)
+ {
+ return GetPoolColumn(message,str,table_name,column_name,analyze->last_select,analyze->call_part,state);
+ }
+ else
+ return false;
+ }
+
+ return false;
}
static void
_rewriteColumnRef(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, ColumnRef *node)
{
ListCell *c;
+ List *list;
char first = 0;
+ char *table_name = NULL;
+ char *column_name = NULL;
+
+ list = node->fields;
foreach (c, node->fields)
{
@@ -2438,24 +4077,59 @@ _rewriteColumnRef(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbli
if (IsA(n, String))
{
- Value *v = (Value *) lfirst(c);
if(message->r_code == SELECT_AEXPR &&
- (_checkDistDefColumn(node, message) != 1))
+ (_checkVirtualColumn(node, message) != 1))
{
- KeepRewriteQueryCode(message, SELECT_AEXPR_FALSE);
+ KeepRewriteQueryReturnCode(message, SELECT_AEXPR_FALSE);
return;
}
- if (first == 0)
+ Value *v = (Value *) lfirst(c);
+ if(list->length == 2 && first == 0)
+ {
first = 1;
+ table_name = v->val.str;
+ }
else
- delay_string_append_char(message, str, ".");
+ column_name = v->val.str;
+ }
+ }
- /* delay_string_append_char(message, str, "\""); */
- delay_string_append_char(message, str, v->val.str);
- /* delay_string_append_char(message, str, "\""); */
- delay_string_append_char(message, str, " ");
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ if(!DetectValidColumn(message,table_name,column_name,message->current_select,-1))
+ pool_debug("_rewriteColumnRef: wrong column select_no=%d",message->current_select);
+
+ if(strcmp(column_name,"*"))
+ ChangeStatebyColumnRef(message,node);
+ }
+ else if(message->r_code == SELECT_DEFAULT)
+ {
+
+ if(message->rewritelock == -1)
+ {
+ GetPoolColumn(message,str,table_name,column_name,message->current_select,-1,false);
+ }
+ else
+ {
+ if(table_name)
+ {
+ if(message->rewritelock == -1)
+ {
+ AnalyzeSelect *analyze = message->analyze[message->current_select];
+ char s = analyze->partstate[SELECT_FROMCLAUSE];
+
+ if(s == 'L' || s =='P')
+ {
+ delay_string_append_char(message, str, analyze->table_name);
+ }
+ }
+ else
+ delay_string_append_char(message, str, table_name);
+ delay_string_append_char(message, str, ".");
+ }
+ delay_string_append_char(message, str, column_name);
}
}
}
@@ -2487,7 +4161,7 @@ _rewriteAConst(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink,
break;
case T_String:
- if(message->r_code == SELECT_DEFAULT_INSIDE_DBLINK)
+ if(message->rewritelock != -1)
{
delay_string_append_char(message, str, "\'\'");
delay_string_append_char(message, str, escape_string(node->val.val.str));
@@ -2529,9 +4203,65 @@ _rewriteA_Indirection(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *
}
+static bool
+AliasToResTargetCondition(RewriteQuery *message,String *str)
+{
+ int select_no = message->current_select;
+ AnalyzeSelect *n_analyze;
+ AnalyzeSelect *u_analyze;
+ char buf[16];
+ int ret;
+ int col_no;
+
+ if(select_no == 0 || message->r_code != SELECT_DEFAULT
+ || message->rewritelock != -1 || message->ignore_rewrite != -1)
+ return false;
+
+ n_analyze=message->analyze[select_no];
+ u_analyze=message->analyze[select_no - 1];
+
+ if(n_analyze->call_part == SELECT_FROMCLAUSE)
+ {
+ ret = n_analyze->ret_count;
+ col_no = n_analyze->select_ret->return_list[n_analyze->ret_count];
+ pool_debug("_rewirteResTarget select no =%d,ret_no = %d,col_no =%d,colname=%s", select_no,ret,col_no,n_analyze->select_ret->col_list[ret]);
+ delay_string_append_char(message, str, " AS ");
+ snprintf(buf, 16, "%d", col_no);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf);
+ delay_string_append_char(message, str," ");
+ return true;
+ }
+ else if(u_analyze->select_union && u_analyze->call_part == SELECT_FROMCLAUSE)
+ {
+ ret = u_analyze->ret_count;
+ col_no = u_analyze->select_ret->return_list[u_analyze->ret_count];
+ pool_debug("_rewirteResTarget(union) select now=%d up=%d,ret_no = %d,col_no =%d,colname=%s", select_no,select_no-1,ret,col_no,u_analyze->select_ret->col_list[ret]);
+ delay_string_append_char(message, str, " AS ");
+ snprintf(buf, 16, "%d", col_no);
+ delay_string_append_char(message, str,"pool_c$");
+ delay_string_append_char(message, str,buf);
+ delay_string_append_char(message, str," ");
+ u_analyze->ret_count++;
+ return true;
+ }
+ else
+ return false;
+}
+
static void
_rewriteResTarget(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, ResTarget *node)
{
+ int select_no = message->current_select;
+ AnalyzeSelect *analyze = NULL;
+ SelectDefInfo *select = NULL;
+
+ if(message->r_code == SELECT_DEFAULT)
+ {
+ analyze= message->analyze[select_no];
+ select = analyze->select_ret;
+ }
+
if (node->indirection != NIL)
{
delay_string_append_char(message, str, "\"");
@@ -2541,14 +4271,75 @@ _rewriteResTarget(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbli
}
else
{
- _rewriteNode(BaseSelect, message, dblink, str, node->val);
+ char *star = NULL;
+ char *table_name = NULL;
+
+ if(message->r_code == SELECT_DEFAULT && !node->name)
+ {
+ if (node->val && (IsA(node->val, ColumnRef)))
+ {
+ int first = 0;
+ ListCell *c;
+ ColumnRef *col;
+
+ col = (ColumnRef *) node->val;
+ foreach (c, col->fields)
+ {
+ Node *n = (Node *) lfirst(c);
+
+ if (IsA(n, String))
+ {
+ Value *v = (Value *) lfirst(c);
+ if(col->fields->length == 2 && first == 0)
+ {
+ first = 1;
+ table_name = v->val.str;
+ }
+ else
+ star = v->val.str;
+ }
+ }
+ }
+
+ if(star && strcmp(star,"*"))
+ {
+ star = NULL;
+ }
+ }
+
+ if(select_no == 0 && star)
+ {
+ GetPoolColumn(message,str,table_name,star,message->current_select,-1,true);
+ return;
+ }
+ else
+ {
+ _rewriteNode(BaseSelect, message, dblink, str, node->val);
+ if(star)
+ return;
+ }
+
+ if(AliasToResTargetCondition(message,str))
+ return;
if (node->name)
{
delay_string_append_char(message, str, " AS ");
- delay_string_append_char(message, str, "\"");
delay_string_append_char(message, str, node->name);
- delay_string_append_char(message, str, "\" ");
+ if(message->r_code == SELECT_DEFAULT && select_no == 0)
+ message->ret_num++;
+ return;
+ }
+ else if(message->r_code == SELECT_DEFAULT && select_no == 0
+ && !node->name)
+ {
+ char *col_name = select->col_list[message->ret_num];
+ pool_debug("_rewriteResTarget: check(%d) ret_num=%d",message->current_select,message->ret_num);
+ pool_debug("_rewriteResTarget: col ret_num=%d",message->current_select,col_name);
+ delay_string_append_char(message, str, " AS ");
+ delay_string_append_char(message, str, col_name);
+
+ message->ret_num++;
}
}
}
@@ -2719,7 +4510,7 @@ static void
_rewriteSortBy(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, SortBy *node)
{
_rewriteNode(BaseSelect, message, dblink, str, node->node);
-
+
if (node->sortby_dir == SORTBY_USING)
{
delay_string_append_char(message, str, " USING ");
@@ -2797,7 +4588,7 @@ static void _rewriteUpdateStmt(Node *BaseSelect, RewriteQuery *message, ConInfoT
delay_string_append_char(message, str, "\"");
delay_string_append_char(message, str, node->name);
delay_string_append_char(message, str, "\" =");
- _rewriteNode(BaseSelect, message, dblink, str, node->val);
+ _rewriteNode(BaseSelect, message, dblink, str, node->val);
}
if (node->fromClause)
@@ -2915,7 +4706,7 @@ static void _rewriteVacuumStmt(Node *BaseSelect, RewriteQuery *message, ConInfoT
if (node->full == TRUE)
delay_string_append_char(message, str, "FULL ");
-
+
if (node->freeze_min_age == 0)
delay_string_append_char(message, str, "FREEZE ");
@@ -3035,7 +4826,6 @@ static void _rewriteCopyStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTod
delay_string_append_char(message, str, ")");
}
-
if (binary == TRUE)
delay_string_append_char(message, str, "BINARY ");
@@ -3403,7 +5193,6 @@ _rewriteAlterRoleStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *
delay_string_append_char(message, str, "ALTER ROLE \"");
delay_string_append_char(message, str, node->role);
delay_string_append_char(message, str, "\"");
-
if (node->options)
_rewriteOptRoleList(BaseSelect, message, dblink, str, node->options);
}
@@ -3563,7 +5352,7 @@ _rewriteVariableSetStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink
if (node->is_local)
delay_string_append_char(message, str, "LOCAL ");
-
+
_rewriteSetRest(BaseSelect, message, dblink, str, node);
}
@@ -3701,7 +5490,6 @@ _rewriteAlterTableCmd(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *
/* not implemented */
break;
-
case AT_DropCluster:
delay_string_append_char(message, str, "SET WITHOUT CLUSTER");
break;
@@ -3833,7 +5621,7 @@ _rewriteOptSeqList(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
delay_string_append_char(message, str, " START ");
else if (strcmp(e->defname, "restart") == 0)
delay_string_append_char(message, str, " RESTART ");
-
+
if (IsA(e->arg, String))
delay_string_append_char(message, str, v->val.str);
else
@@ -4121,11 +5909,11 @@ _rewriteDefineStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
case OBJECT_OPERATOR:
delay_string_append_char(message, str, "CREATE OPERATOR ");
-
+
foreach (lc, node->defnames)
{
Value *v = lfirst(lc);
-
+
if (dot == 0)
dot = 1;
else
@@ -4133,7 +5921,7 @@ _rewriteDefineStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
delay_string_append_char(message, str, v->val.str);
}
-
+
delay_string_append_char(message, str, " ");
_rewriteDefinition(BaseSelect, message, dblink, str, node->definition);
break;
@@ -4168,7 +5956,7 @@ _rewriteDefineStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dbl
_rewriteIdList(BaseSelect, message, dblink, str, node->defnames);
_rewriteDefinition(BaseSelect, message, dblink, str, node->definition);
break;
-
+
default:
break;
}
@@ -5456,18 +7244,32 @@ _rewriteCommentStmt(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *db
static void
_rewriteRangeSubselect(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, RangeSubselect *node)
{
-
- if(message->r_code == SELECT_AEXPR)
+ int last = message->current_select;
+ Alias *alias = node->alias;
+ char *table_name = alias->aliasname;
+ int sub_no = message->analyze_num;
+
+ if(message->r_code == SELECT_AEXPR)
{
- KeepRewriteQueryCode(message, SELECT_AEXPR_FALSE);
+ KeepRewriteQueryReturnCode(message, SELECT_AEXPR_FALSE);
return;
}
delay_string_append_char(message, str, "(");
_rewriteNode(BaseSelect, message, dblink, str, node->subquery);
delay_string_append_char(message, str, ")");
-
_rewriteNode(BaseSelect, message, dblink, str, node->alias);
+
+ if(message->r_code == SELECT_ANALYZE)
+ {
+ AnalyzeSelect *analyze;
+ analyze=message->analyze[sub_no];
+
+ message->analyze[last]->select_range = true;
+
+ pool_debug("_rewriteTRangeSubSelect: select range ture %d",sub_no);
+ build_range_info(message,NULL,NULL,analyze->select_ret,table_name,last,sub_no);
+ }
}
static void
@@ -5496,7 +7298,7 @@ _rewriteWithDefinition(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink
{
DefElem *elem;
Value *v;
-
+
elem = linitial(def_list);
v = (Value *)elem->arg;
if (strcmp(elem->defname, "oids") == 0)
@@ -5538,7 +7340,7 @@ _rewriteCurrentOfExpr(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *
static void
_rewriteNode(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, String *str, void *obj)
{
- if (obj == NULL)
+ if(!obj)
return;
else if (IsA(obj, List) ||IsA(obj, IntList) || IsA(obj, OidList))
_rewriteList(BaseSelect, message, dblink, str, obj);
@@ -5673,30 +7475,7 @@ _rewriteNode(Node *BaseSelect, RewriteQuery *message, ConInfoTodblink *dblink, S
_rewriteDeclareCursorStmt(BaseSelect, message, dblink, str, obj);
break;
case T_SelectStmt:
- {
- int message_r_code = message->r_code;
-
- if(message_r_code != SELECT_ONETABLE &&
- checkSelectStmtOneTable(BaseSelect, message, dblink, str,obj) == 1)
- {
- /* Special Query rewriteing when fromClause has one-table */
- KeepRewriteQueryCode(message, SELECT_ONETABLE);
-
- /* Intensive Query rewriting starts */
- if(checkIntensiveOneTable(BaseSelect, message, dblink, str, obj) == 1)
- {
- pool_debug("OneTable Intensive optimization");
- _rewriteSelectStmtIntensive(BaseSelect, message, dblink, str, obj);
- KeepRewriteQueryCode(message, message_r_code);
- break;
- }
- } else {
- KeepRewriteQueryCode(message, SELECT_DEFAULT);
- }
- /* default dblink action */
_rewriteSelectStmt(BaseSelect, message, dblink, str, obj);
- KeepRewriteQueryCode(message, message_r_code);
- }
break;
case T_ColumnDef:
_rewriteColumnDef(BaseSelect, message, dblink, str, obj);
@@ -6075,6 +7854,7 @@ nodeToRewriteString(RewriteQuery *message, ConInfoTodblink *dblink, void *obj)
String *str;
str = init_string("");
+ message->analyze_num = 0;
_rewriteNode(NULL, message, dblink, str, obj);
message->rewrite_query = str->data;
}
diff --git a/pool_rewrite_query.c b/pool_rewrite_query.c
index 87363a4b4..b3afdb2ac 100644
--- a/pool_rewrite_query.c
+++ b/pool_rewrite_query.c
@@ -33,8 +33,11 @@ static int getInsertRule(ListCell *lc,List *list_t ,DistDefInfo *info, int div_k
static void examInsertStmt(Node *node,POOL_CONNECTION_POOL *backend,RewriteQuery *message);
static void examSelectStmt(Node *node,POOL_CONNECTION_POOL *backend,RewriteQuery *message);
static char *delimistr(char *str);
-static int direct_parallel_query(SelectStmt *stmt,POOL_CONNECTION_POOL *backend);
+static int direct_parallel_query(RewriteQuery *message);
static int check_whereClause(Node *where);
+static void initMessage(RewriteQuery *message,bool analyze);
+static void initdblink(ConInfoTodblink *dblink, POOL_CONNECTION_POOL *backend);
+static void analyze_debug(RewriteQuery *message);
char *pool_error_message(char *message)
{
@@ -161,9 +164,7 @@ static void examInsertStmt(Node *node,POOL_CONNECTION_POOL *backend, RewriteQuer
if (!info)
{
/* send error message to frontend */
- message->r_code = INSERT_SQL_RESTRICTION;
- message->r_node = -1;
- message->rewrite_query = pool_error_message("cannot find dist_def table info");
+ message->r_code = INSERT_DIST_NO_RULE;
return;
}
@@ -245,22 +246,48 @@ static void examSelectStmt(Node *node,POOL_CONNECTION_POOL *backend,RewriteQuery
static ConInfoTodblink dblink;
/* initialize dblink info */
- dblink.dbname = MASTER_CONNECTION(backend)->sp->database;
- dblink.hostaddr = pool_config->pgpool2_hostname;
- dblink.user = MASTER_CONNECTION(backend)->sp->user;
- dblink.port = pool_config->port;
- dblink.password = MASTER_CONNECTION(backend)->con->password;
+ initdblink(&dblink,backend);
/* initialize message */
+ initMessage(message,true);
message->type = node->type;
message->r_code = SELECT_DEFAULT;
- message->dbname = NULL;
+
+ nodeToRewriteString(message,&dblink,node);
+}
+
+
+static void initMessage(RewriteQuery *message, bool analyze)
+{
+ message->r_code = 0;
+ message->r_node = 0;
+ message->column = 0;
+ message->virtual_num = 0;
+ message->is_pg_catalog = false;
+ message->is_loadbalance = false;
+ message->is_parallel = false;
message->table_relname = NULL;
message->table_alias = NULL;
- message->schemaname = NULL;
message->dbname = NULL;
+ message->schemaname = NULL;
+ message->rewrite_query = NULL;
+ message->rewritelock = -1;
+ message->ignore_rewrite = -1;
+ message->ret_num = 0;
- nodeToRewriteString(message,&dblink,node);
+ if(analyze)
+ {
+ message->analyze_num = 0;
+ }
+}
+
+static void initdblink(ConInfoTodblink *dblink,POOL_CONNECTION_POOL *backend)
+{
+ dblink->dbname = MASTER_CONNECTION(backend)->sp->database;
+ dblink->hostaddr = pool_config->pgpool2_hostname;
+ dblink->user = MASTER_CONNECTION(backend)->sp->user;
+ dblink->port = pool_config->port;
+ dblink->password = MASTER_CONNECTION(backend)->con->password;
}
int IsSelectpgcatalog(Node *node,POOL_CONNECTION_POOL *backend)
@@ -269,33 +296,28 @@ int IsSelectpgcatalog(Node *node,POOL_CONNECTION_POOL *backend)
static RewriteQuery message;
/* initialize dblink info */
- dblink.dbname = MASTER_CONNECTION(backend)->sp->database;
- dblink.hostaddr = pool_config->pgpool2_hostname;
- dblink.user = MASTER_CONNECTION(backend)->sp->user;
- dblink.port = pool_config->port;
- dblink.password = MASTER_CONNECTION(backend)->con->password;
+ initdblink(&dblink,backend);
/* initialize message */
+ initMessage(&message,false);
+
message.type = node->type;
- message.r_code = SELECT_CHECK_PGCATALOG;
- message.dbname = NULL;
- message.table_relname = NULL;
- message.table_alias = NULL;
- message.schemaname = NULL;
- message.dbname = NULL;
- nodeToRewriteString(&message,&dblink,node);
+ initdblink(&dblink,backend);
- if(message.r_code == SELECT_PGCATALOG)
+ if(message.is_pg_catalog)
+ {
+ pool_debug("Isselectpgcatalog %d",message.is_pg_catalog);
return 1;
+ }
else
+ {
return 0;
+ }
}
-RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNECTION_POOL *backend)
+RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNECTION_POOL *backend,RewriteQuery *message)
{
- static RewriteQuery message;
-
switch(node->type)
{
case T_SelectStmt:
@@ -310,70 +332,70 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
pool_send_readyforquery(frontend);
- message.status=POOL_CONTINUE;
+ message->status=POOL_CONTINUE;
break;
}
- examSelectStmt(node,backend,&message);
+ examSelectStmt(node,backend,message);
- if (message.r_code != SELECT_PGCATALOG &&
- message.r_code != SELECT_RELATION_ERROR)
+ if (message->r_code != SELECT_PGCATALOG &&
+ message->r_code != SELECT_RELATION_ERROR)
{
POOL_CONNECTION_POOL_SLOT *system_db = pool_system_db_connection();
- message.status = OneNode_do_command(frontend,
+ message->status = OneNode_do_command(frontend,
system_db->con,
- message.rewrite_query,
+ message->rewrite_query,
backend->info->database);
}
else
{
if(TSTATE(backend) == 'T' &&
- message.r_code == SELECT_RELATION_ERROR)
+ message->r_code == SELECT_RELATION_ERROR)
{
pool_debug("pool_rewrite_stmt(select): inside transaction");
- message.rewrite_query = nodeToString(node);
- message.status = pool_parallel_exec(frontend,backend,message.rewrite_query,node,true);
+ message->rewrite_query = nodeToString(node);
+ message->status = pool_parallel_exec(frontend,backend,message->rewrite_query,node,true);
}
else
{
pool_debug("pool_rewrite_stmt: executed by Master");
- message.rewrite_query = nodeToString(node);
- message.status = OneNode_do_command(frontend,
+ message->rewrite_query = nodeToString(node);
+ message->status = OneNode_do_command(frontend,
MASTER(backend),
- message.rewrite_query,
+ message->rewrite_query,
backend->info->database);
}
}
- pool_debug("pool_rewrite_stmt: XXX message_code %d",message.r_code);
+ pool_debug("pool_rewrite_stmt: XXX message_code %d",message->r_code);
}
break;
case T_InsertStmt:
- examInsertStmt(node,backend,&message);
+ examInsertStmt(node,backend,message);
- if(message.r_code == 0 )
+ if(message->r_code == 0 )
{
- message.status = OneNode_do_command(frontend,
- CONNECTION(backend,message.r_node),
- message.rewrite_query,
+ message->status = OneNode_do_command(frontend,
+ CONNECTION(backend,message->r_node),
+ message->rewrite_query,
backend->info->database);
}
- else if (message.r_code == INSERT_SQL_RESTRICTION)
+ else if (message->r_code == INSERT_SQL_RESTRICTION)
{
pool_send_error_message(frontend, MAJOR(backend), "XX000",
"pgpool2 sql restriction",
- message.rewrite_query, "", __FILE__,
+ message->rewrite_query, "", __FILE__,
__LINE__);
if(TSTATE(backend) == 'T')
{
pool_debug("rewrite_query_stmt(insert): inside transaction");
- message.status = pool_parallel_exec(frontend,backend, "POOL_RESET_TSTATE",node,false);
+ message->status = pool_parallel_exec(frontend,backend, "POOL_RESET_TSTATE",node,false);
}
else
{
pool_send_readyforquery(frontend);
- message.status=POOL_CONTINUE;
+ message->status=POOL_CONTINUE;
}
}
break;
@@ -383,17 +405,18 @@ RewriteQuery *rewrite_query_stmt(Node *node,POOL_CONNECTION *frontend,POOL_CONNE
break;
#endif
default:
- message.type = node->type;
- message.status = POOL_CONTINUE;
+ message->type = node->type;
+ message->status = POOL_CONTINUE;
break;
}
pool_debug("pool_rewrite_stmt: XXX rule %d",node->type);
- return &message;
+ return message;
}
#define POOL_PARALLEL "pool_parallel"
+#define POOL_LOADBALANCE "pool_loadbalance"
static int check_whereClause(Node *where)
{
@@ -424,64 +447,12 @@ static int check_whereClause(Node *where)
}
}
-static int direct_parallel_query(SelectStmt *stmt,POOL_CONNECTION_POOL *backend)
+static int direct_parallel_query(RewriteQuery *message)
{
-
- if(stmt->lockingClause)
- {
- pool_debug("lockingClasue is exist");
+ if(message && message->analyze[0] && message->analyze[0]->state == 'P')
return 1;
- }
-
- if (!stmt->distinctClause && !stmt->intoClause &&
- !stmt->groupClause && !stmt->havingClause && !stmt->sortClause &&
- !stmt->limitOffset && !stmt->limitCount &&!stmt->larg && !stmt->rarg)
- {
- if(stmt->fromClause && (list_length(stmt->fromClause) == 1)
- && list_head(stmt->fromClause) && lfirst(list_head(stmt->fromClause))
- && IsA(lfirst(list_head(stmt->fromClause)),RangeVar))
- {
- ListCell *lc;
- DistDefInfo *info = NULL;
- RangeVar *var = NULL;
-
- var = (RangeVar *) lfirst(list_head(stmt->fromClause));
- info = pool_get_dist_def_info (MASTER_CONNECTION(backend)->sp->database, var->schemaname, var->relname);
-
- if(!info)
- return 0;
-
- if(stmt->whereClause &&
- (check_whereClause(stmt->whereClause)))
- {
- return 0;
- }
-
- foreach (lc, stmt->targetList)
- {
- Node *n = lfirst(lc);
- if (IsA(n, ResTarget))
- {
- ResTarget *target = (ResTarget *) n;
- if (target->val && (IsA(target->val, FuncCall) ||
- IsA(target->val,CaseExpr) || IsA(target->val,SubLink)))
- {
- return 0;
- }
- else if(target->val && (IsA(target->val, TypeCast)))
- {
- TypeCast *type = (TypeCast *)target->val;
- if(type->arg && (IsA(type->arg,FuncCall) ||
- IsA(type->arg,CaseExpr) || IsA(type->arg,SubLink)))
- return 0;
- }
- }
- }
-
- return 1;
- }
- }
- return 0;
+ else
+ return 0;
}
static char *delimistr(char *str)
@@ -494,10 +465,12 @@ static char *delimistr(char *str)
for(i = 0; i < len; i++)
{
- unsigned char ch = (unsigned char) str[i];
+ char c = (unsigned char) str[i];
if((i != 0) && (i != len -1))
{
- result[j] = (char) ch;
+ if(c=='\'' && (char) str[i+1]=='\'')
+ i++;
+ result[j] = c;
j++;
}
}
@@ -507,9 +480,24 @@ static char *delimistr(char *str)
return result;
}
-char *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
+void analyze_debug(RewriteQuery *message)
+{
+ int analyze_num,i;
+ analyze_num = message->analyze_num;
+
+ for(i = 0; i< analyze_num; i++)
+ {
+ AnalyzeSelect *analyze = message->analyze[i];
+ pool_debug("analyze_debug :select no(%d), last select(%d), last_part(%d), state(%c)",
+ analyze->now_select,analyze->last_select,analyze->call_part,analyze->state);
+ }
+}
+
+RewriteQuery *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
{
- char *parallel = NULL;
+ static RewriteQuery message;
+ static ConInfoTodblink dblink;
+
if (IsA(node, SelectStmt))
{
@@ -519,48 +507,103 @@ char *is_parallel_query(Node *node, POOL_CONNECTION_POOL *backend)
stmt = (SelectStmt *) node;
- direct_ok = direct_parallel_query(stmt,backend);
- if(direct_ok == 1)
- {
- parallel = nodeToString(node);
- pool_debug("can pool_parallel_exec %s",parallel);
- return parallel;
- }
-
- if (stmt->distinctClause || stmt->intoClause ||
+ /* LOADBALANCE OR PARALLEL */
+ if (!(stmt->distinctClause || stmt->intoClause ||
stmt->fromClause || stmt->groupClause || stmt->havingClause ||
stmt->sortClause || stmt->limitOffset || stmt->limitCount ||
- stmt->lockingClause || stmt->larg || stmt->rarg)
- return NULL;
-
- n = lfirst(list_head(stmt->targetList));
- if (IsA(n, ResTarget))
+ stmt->lockingClause || stmt->larg || stmt->rarg) &&
+ (n = lfirst(list_head(stmt->targetList))) && IsA(n, ResTarget))
{
ResTarget *target = (ResTarget *) n;
if (target->val && IsA(target->val, FuncCall))
{
FuncCall *func = (FuncCall *) target->val;
- if (list_length(func->funcname) == 1 &&
- strcmp(strVal(lfirst(list_head(func->funcname))),
- POOL_PARALLEL) == 0)
+ if (list_length(func->funcname) == 1 && func->args)
{
Node *func_args = (Node *) lfirst(list_head(func->args));
- parallel = delimistr(nodeToString(func_args));
- return parallel;
- }
+ message.rewrite_query = delimistr(nodeToString(func_args));
+
+ /* PARALLEL */
+ if(strcmp(strVal(lfirst(list_head(func->funcname))),
+ POOL_PARALLEL) == 0)
+ {
+ message.r_code = SEND_PARALLEL_ENGINE;
+ message.is_parallel = true;
+ message.is_loadbalance = false;
+ pool_debug("can pool_parallel_exec %s",message.rewrite_query);
+ return &message;
+ }
+ else /* LOADBALANCE */
+ if(strcmp(strVal(lfirst(list_head(func->funcname))),
+ POOL_LOADBALANCE) == 0)
+ {
+ message.r_code = SEND_LOADBALANCE_ENGINE;
+ message.is_loadbalance = true;
+ message.is_parallel = false;
+ pool_debug("can loadbalance_mode %s",message.rewrite_query);
+ return &message;
+ }
+ }
}
}
+
+ /* ANALYZE QUERY */
+ initMessage(&message,true);
+ message.r_code = SELECT_ANALYZE;
+ message.is_loadbalance = true;
+
+ initdblink(&dblink,backend);
+ nodeToRewriteString(&message,&dblink,node);
+
+ if(message.is_pg_catalog)
+ {
+ message.is_loadbalance = false;
+ message.is_parallel = false;
+ pool_debug("is_parallel_query: query is done by loadbalance(pgcatalog)");
+ return &message;
+ }
+
+ if(message.is_loadbalance)
+ {
+ message.is_parallel = false;
+ pool_debug("is_parallel_query: query is done by loadbalance");
+ return &message;
+ }
+
+ analyze_debug(&message);
+
+ /* PARALLEL*/
+ direct_ok = direct_parallel_query(&message);
+ if(direct_ok == 1)
+ {
+ message.rewrite_query = nodeToString(node);
+ message.is_parallel = true;
+ message.is_loadbalance = false;
+ pool_debug("can pool_parallel_exec %s",message.rewrite_query);
+ return &message;
+ }
}
else if (IsA(node, CopyStmt))
{
CopyStmt *stmt = (CopyStmt *)node;
+
+ initMessage(&message,false);
+
if (stmt->is_from == FALSE && stmt->filename == NULL)
{
- parallel = nodeToString(stmt);
- return parallel;
+ RangeVar *relation = (RangeVar *)stmt->relation;
+
+ /* check on distribution table or replicate table */
+
+ if(pool_get_dist_def_info (MASTER_CONNECTION(backend)->sp->database, relation->schemaname, relation->relname))
+ {
+ message.rewrite_query = nodeToString(stmt);
+ message.is_parallel = true;
+ message.is_loadbalance = false;
+ message.r_code = SEND_PARALLEL_ENGINE;
+ }
}
}
-
- return NULL;
+ return &message;
}
diff --git a/pool_rewrite_query.h b/pool_rewrite_query.h
index b4f86beb1..952a1c6ce 100644
--- a/pool_rewrite_query.h
+++ b/pool_rewrite_query.h
@@ -35,7 +35,7 @@
#define SELECT_BACKEND_CONNECT 3
#define SELECT_NOT_BACKEND_CONNECT 4
#define SELECT_PGCATALOG 5
-#define SELECT_CHECK_PGCATALOG 6
+#define SELECT_CHECK_PGCATALOG_REPLICATION 6
#define SELECT_DEFAULT 7
#define SELECT_DEFAULT_INSIDE_DBLINK 8
#define SELECT_AEXPR 9
@@ -43,17 +43,98 @@
#define SELECT_ONETABLE 11
#define SELECT_ONETABLE_FALSE 12
#define SELECT_RELATION_ERROR 13
+#define INSERT_DIST_NO_RULE 14
+#define SEND_PARALLEL_ENGINE 15
+#define SEND_LOADBALANCE_ENGINE 16
+#define SELECT_NOT_REPLICATION 17
+#define SELECT_REWRITE 18
+#define SELECT_ANALYZE 19
+#define SELECT_DEFAULT_PREP 20
+
+typedef struct {
+ char **col_list; /* column list */
+ char **type_list; /* type list */
+ int *return_list;
+ int col_num;
+ bool valid;
+}SelectDefInfo;
+
+typedef struct {
+ DistDefInfo *distinfo;
+ RepliDefInfo *repliinfo;
+ SelectDefInfo *selectinfo;
+ char *alias;
+ char state;
+ int ret_num;
+} RangeInfo;
+
+typedef struct {
+ char **col_list; /* column list */
+ char **type_list; /* type list */
+ char **table_list; /* table list */
+ char *state_list;
+ int *column_no;
+ int *valid;
+ int col_num;
+} VirtualTable;
+
+typedef struct {
+ char **col_list; /* column list */
+ char **type_list; /* type list */
+ char **table_list; /* table list */
+ char state;
+ int *valid;
+ int col_num;
+ char **using_list;
+ int using_length;
+} JoinTable;
+
+typedef struct {
+ int now_select;
+ int part;
+ int last_select;
+ int call_part;
+ int from_num;
+ int larg_count;
+ int rarg_count;
+ int ret_count;
+ char state;
+ char *table_name;
+ char partstate[8];
+ bool select_union;
+ bool select_range;
+ bool aggregate;
+ RangeInfo **range;
+ int rangeinfo_num;
+ VirtualTable *virtual;
+ JoinTable *join;
+ SelectDefInfo *select_ret;
+} AnalyzeSelect;
typedef struct {
int r_code;
int r_node;
+ int part;
+ int rewritelock;
+ int analyze_num;
+ int current_select;
+ int ignore_rewrite;
+ int column;
+ int virtual_num;
+ int ret_num;
+ bool is_pg_catalog;
+ bool is_loadbalance;
+ bool is_parallel;
+ bool fromClause;
char *table_relname;
char *table_alias;
char *schemaname;
char *dbname;
char *rewrite_query;
+ char table_state;
POOL_STATUS status;
NodeTag type;
+ AnalyzeSelect **analyze;
} RewriteQuery;
typedef struct {
@@ -64,11 +145,11 @@ typedef struct {
char *password;
} ConInfoTodblink;
-extern RewriteQuery *rewrite_query_stmt(Node *node, POOL_CONNECTION *frontend,POOL_CONNECTION_POOL *backend);
+extern RewriteQuery *rewrite_query_stmt(Node *node, POOL_CONNECTION *frontend,POOL_CONNECTION_POOL *backend,RewriteQuery *message);
extern void nodeToRewriteString(RewriteQuery *message, ConInfoTodblink *dblink,void *obj);
char *pool_error_message(char *message);
extern int IsSelectpgcatalog(Node *node,POOL_CONNECTION_POOL *backend);
-extern char *is_parallel_query(Node *node,POOL_CONNECTION_POOL *backend);
+extern RewriteQuery *is_parallel_query(Node *node,POOL_CONNECTION_POOL *backend);
extern POOL_STATUS pool_parallel_exec(POOL_CONNECTION *frontend,POOL_CONNECTION_POOL *backend, char *string,Node *node,bool send_to_frontend);
diff --git a/pool_system.c b/pool_system.c
index 73cbd2d25..624a00402 100644
--- a/pool_system.c
+++ b/pool_system.c
@@ -29,6 +29,7 @@
static int create_prepared_statement(DistDefInfo *dist_info);
static int get_col_list(DistDefInfo *info);
+static int get_col_list2(RepliDefInfo *info);
static
int get_col_list(DistDefInfo *info)
@@ -84,6 +85,57 @@ int get_col_list(DistDefInfo *info)
return 0;
}
+static
+int get_col_list2(RepliDefInfo *info)
+{
+ int i;
+ static char sql[1024];
+ PGresult *result;
+
+ if (!system_db_info->pgconn ||
+ (PQstatus(system_db_info->pgconn) != CONNECTION_OK))
+ {
+ if (system_db_connect())
+ return -1;
+ }
+
+ for (i = 0; i < info->col_num; i++)
+ {
+ snprintf(sql,
+ sizeof(sql),
+ "SELECT col_list[%d],type_list[%d] FROM %s.replicate_def where dbname = '%s' and schema_name = '%s' and table_name = '%s'",
+ i + 1,
+ i + 1,
+ pool_config->system_db_schema,
+ info->dbname,info->schema_name,
+ info->table_name);
+
+ result = PQexec(system_db_info->pgconn, sql);
+
+ if (!result || PQresultStatus(result) != PGRES_TUPLES_OK)
+ {
+ pool_error("get_col_list2: PQexec failed: %s",
+ PQerrorMessage(system_db_info->pgconn));
+ return -1;
+ }
+ else
+ {
+ info->col_list[i] = malloc(strlen(PQgetvalue(result,0,0)) + 1);
+ info->type_list[i] = malloc(strlen(PQgetvalue(result,0,1)) + 1);
+
+ if (info->col_list[i] == NULL || info->type_list[i] == NULL)
+ {
+ pool_error("get_col_list2: malloc failed: %s", strerror(errno));
+ PQclear(result);
+ return -1;
+ }
+ strcpy(info->col_list[i],PQgetvalue(result,0,0));
+ strcpy(info->type_list[i],PQgetvalue(result,0,1));
+ PQclear(result);
+ }
+ }
+ return 0;
+}
/*
* system_db_connect:
@@ -134,9 +186,10 @@ int system_db_connect (void)
int pool_memset_system_db_info (SystemDBInfo *info)
{
int i;
- static char sql[1024];
+ static char sql[1024],sql2[1024];
PGresult *result;
DistDefInfo *dist_info = NULL;
+ RepliDefInfo *repli_info = NULL;
if (!system_db_info->pgconn ||
(PQstatus(system_db_info->pgconn) != CONNECTION_OK))
@@ -145,6 +198,7 @@ int pool_memset_system_db_info (SystemDBInfo *info)
return -1;
}
+ /* get distribution rules */
snprintf(sql,
sizeof(sql),
"SELECT dbname, schema_name, table_name,col_name,array_upper(col_list,1),col_list,type_list, dist_def_func FROM %s.dist_def",
@@ -286,6 +340,130 @@ int pool_memset_system_db_info (SystemDBInfo *info)
}
PQclear(result);
+
+ /* get replication rules */
+ snprintf(sql2,
+ sizeof(sql2),
+ "SELECT dbname, schema_name, table_name, array_upper(col_list,1),col_list,type_list FROM %s.replicate_def",
+ pool_config->system_db_schema);
+
+ result = PQexec(system_db_info->pgconn, sql2);
+
+ if (!result)
+ {
+ pool_error("PQexec failed: %s", PQerrorMessage(system_db_info->pgconn));
+ return -1;
+ }
+ else if (PQresultStatus(result) != PGRES_TUPLES_OK)
+ {
+ info->repli_def_num = 0;
+ info->repli_def_slot = NULL;
+ }
+ else
+ {
+ info->repli_def_num = PQntuples(result);
+ if (info->repli_def_num != 0)
+ {
+ repli_info = malloc(sizeof(RepliDefInfo) * info->repli_def_num);
+ }
+
+ if (repli_info == NULL && info->repli_def_num != 0)
+ {
+ pool_error("pool_memset_system_db_info: malloc failed: %s",
+ strerror(errno));
+ PQclear(result);
+ pool_close_libpq_connection();
+ return -1;
+ }
+
+ info->repli_def_slot = repli_info;
+
+ for (i = 0; i < PQntuples(result); ++i)
+ {
+ char *t_dbname;
+ char *t_schema_name;
+ char *t_table_name;
+ int num;
+ int len;
+
+ num = atol(PQgetvalue(result, i ,3));
+ t_dbname = malloc(strlen(PQgetvalue(result,i,0)) + 1);
+ if (t_dbname == NULL)
+ {
+ pool_error("pool_memset_system_db_info: malloc failed: %s",
+ strerror(errno));
+ PQclear(result);
+ pool_close_libpq_connection();
+ return -1;
+ }
+ strcpy(t_dbname, PQgetvalue(result,i,0));
+ repli_info[i].dbname = t_dbname;
+
+ t_schema_name = malloc(strlen(PQgetvalue(result,i,1)) + 1);
+ if (t_schema_name == NULL)
+ {
+ pool_error("pool_memset_system_db_info: malloc failed: %s",
+ strerror(errno));
+ PQclear(result);
+ pool_close_libpq_connection();
+ return -1;
+ }
+ strcpy(t_schema_name, PQgetvalue(result,i,1));
+ repli_info[i].schema_name = t_schema_name;
+
+ t_table_name = malloc(strlen(PQgetvalue(result,i,2)) + 1);
+ if (t_table_name == NULL)
+ {
+ pool_error("pool_memset_system_db_info: malloc failed: %s",
+ strerror(errno));
+ PQclear(result);
+ pool_close_libpq_connection();
+ return -1;
+ }
+ strcpy(t_table_name, PQgetvalue(result,i,2));
+ repli_info[i].table_name = t_table_name;
+
+ repli_info[i].col_num = num;
+
+ repli_info[i].col_list = calloc(num, sizeof(char *));
+ repli_info[i].type_list = calloc(num, sizeof(char *));
+ if (repli_info[i].col_list == NULL || repli_info[i].type_list == NULL)
+ {
+ pool_error("pool_memset_system_db_info: calloc failed: %s",
+ strerror(errno));
+ PQclear(result);
+ pool_close_libpq_connection();
+ return -1;
+ }
+
+ if (get_col_list2(&repli_info[i]) < 0)
+ {
+ pool_error("get_col_list() failed");
+ PQclear(result);
+ pool_close_libpq_connection();
+ return -1;
+ }
+
+ /* create PREPARE statement */
+ len = strlen(t_dbname) + strlen(t_schema_name) +
+ strlen(t_table_name) + strlen("pgpool_");
+
+ repli_info[i].prepare_name = malloc(len + 1);
+ if (repli_info[i].prepare_name == NULL)
+ {
+ pool_error("pool_memset_system_db_info: malloc failed: %s",
+ strerror(errno));
+ return -1;
+ }
+
+ snprintf(repli_info[i].prepare_name, len+1, "pgpool_%s%s%s",
+ t_dbname, t_schema_name, t_table_name);
+ repli_info[i].prepare_name[len] = '\0';
+ }
+ }
+
+ PQclear(result);
+
pool_close_libpq_connection();
return i;
}
@@ -330,6 +508,46 @@ DistDefInfo *pool_get_dist_def_info (char *dbname, char *schema_name, char *tabl
return NULL;
}
+/*
+ * pool_get_repli_def_info:
+ * Looks up replication rule with dbname, schema_name and table_name.
+ */
+RepliDefInfo *pool_get_repli_def_info (char *dbname, char *schema_name, char *table_name)
+{
+ int i;
+ int repli_def_num = system_db_info->info->repli_def_num;
+ char *public ="public";
+
+ if (!dbname || !table_name)
+ {
+ return NULL;
+ }
+
+ if (!schema_name)
+ {
+ schema_name = public;
+ }
+
+ for (i = 0; i < repli_def_num; i++)
+ {
+ char *mem_dbname;
+ char *mem_schema_name;
+ char *mem_table_name;
+
+ mem_dbname = system_db_info->info->repli_def_slot[i].dbname;
+ mem_schema_name = system_db_info->info->repli_def_slot[i].schema_name;
+ mem_table_name = system_db_info->info->repli_def_slot[i].table_name;
+
+ if ((strcmp(mem_dbname, dbname) == 0) &&
+ (strcmp(mem_schema_name, schema_name) == 0) &&
+ (strcmp(mem_table_name, table_name) ==0))
+ {
+ return &system_db_info->info->repli_def_slot[i];
+ }
+ }
+ return NULL;
+}
+
/*
* pool_get_id:
* Returns the backend node id from value.
diff --git a/pool_type.h b/pool_type.h
index f8f3cd01a..773ab9067 100644
--- a/pool_type.h
+++ b/pool_type.h
@@ -142,19 +142,32 @@ typedef struct {
int is_created_prepare; /* is prepare statement created? */
} DistDefInfo;
+typedef struct {
+ char *dbname; /* database name */
+ char *schema_name; /* schema name */
+ char *table_name; /* table name */
+ int col_num; /* number of clumn*/
+ char **col_list; /* column list */
+ char **type_list; /* type list */
+ char *prepare_name; /* prepared statement name */
+ int is_created_prepare; /* is prepare statement created? */
+} RepliDefInfo;
+
typedef struct {
int has_prepared_statement; /* true if the current session has prepared statement created */
char *register_prepared_statement; /* prepared statement name for cache register */
} QueryCacheTableInfo;
typedef struct {
- char *hostname; /* host name */
- int port; /* port number */
- char *user; /* login user name */
- char *password; /* login password */
- char *schema_name; /* schema name */
- char *database_name; /* database name */
- int dist_def_num; /* number of distribution table */
+ char *hostname; /* host name */
+ int port; /* port number */
+ char *user; /* login user name */
+ char *password; /* login password */
+ char *schema_name; /* schema name */
+ char *database_name; /* database name */
+ int repli_def_num; /* number of replication table */
+ int dist_def_num; /* number of distribution table */
+ RepliDefInfo *repli_def_slot; /* replication rule list */
DistDefInfo *dist_def_slot; /* distribution rule list */
QueryCacheTableInfo query_cache_table_info; /* query cache db session info */
BACKEND_STATUS system_db_status;
[parent: 049be8dd53a8]