Changeset: 40496207a997 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=40496207a997
Modified Files:
MonetDB5/src/optimizer/opt_mapreduce.mx
Branch: default
Log Message:
major changes to solve 'kip en ei' problem, due to some long standing MAL bugs
'ben ik nu m'n ei kwijt'... (according to Jennie). Plan generation for
individual min/max/avg queries is ok, combination causes an order problem
diffs (truncated from 421 to 300 lines):
diff -r 1bbd1bdb15e3 -r 40496207a997 MonetDB5/src/optimizer/opt_mapreduce.mx
--- a/MonetDB5/src/optimizer/opt_mapreduce.mx Fri Jun 11 12:14:28 2010 +0200
+++ b/MonetDB5/src/optimizer/opt_mapreduce.mx Fri Jun 11 14:53:12 2010 +0200
@@ -144,15 +144,19 @@
}
+enum poper { pBAT = 1, SUM, MAX, MIN, SORT, SORTDESC, LIMIT };
typedef struct _mapcol {
- int mapid; /* original column var in map program we
+ int mapid; /* var in map plan that is in its signature
+ and return */
+ int reduceid; /* original column var in reduce program we
eventually need to replace */
- int reduceid; /* var in reduce plan that is in its signature
- and return */
int type; /* type of the map plan var */
int mapbat; /* the var that is a BAT containing all values
- returned from map nodes (function) */
+ returned from map nodes (function), can only be
+ used *after* MRdistributework */
+ enum poper postop;/* the operation that needs to be performed on
+ mapbat to turn it into reduceid */
struct _mapcol *next;
} mapcol;
@@ -264,7 +268,7 @@
/* delayed bat.inserts for easily creating a deterministic flow */
for (lcol = col, j = 0; lcol != NULL; lcol = lcol->next, j++) {
- q = getArg(packs[j], 0);
+ q = lcol->mapbat;
/* p := bat.insert(b, y1) */
for (i = 0; i < n; i++) {
p = newStmt(reduce, batRef, insertRef);
@@ -280,14 +284,55 @@
* confused by possible duplicate ids */
p = newFcnCall(reduce, algebraRef, markHRef);
p = pushArgument(reduce, p, q);
- getArg(p, 0) = lcol->reduceid;
- } else {
- lcol->mapbat = getArg(p, 0);
- lcol->type = newBatType(TYPE_void, lcol->type);
- p = newAssignment(reduce);
- p = pushArgument(reduce, p, lcol->mapbat);
- getArg(p, 0) = lcol->reduceid;
- setArgType(reduce, p, 0, lcol->type);
+ }
+ lcol->mapbat = getArg(p, 0);
+
+ /* We must deliver here the variables (reduceid) that the rest
+ * of the reduce plan uses in such a way that they can deal with
+ * it. Since this code is ran at latest (after the possible
+ * optimisations are known) the optimisation code cannot know
+ * what vars come out of here (in particular mapbat), so it must
+ * be able to rely on what it knows (reduceid). */
+ switch (lcol->postop) {
+ case pBAT:
+ getArg(p, 0) = lcol->reduceid;
+ break;
+ case SUM:
+ lcol->type = newBatType(TYPE_void, lcol->type);
+ p = newFcnCall(reduce, aggrRef, sumRef);
+ p = pushArgument(reduce, p, lcol->mapbat);
+ getArg(p, 0) = lcol->reduceid;
+ break;
+ case MAX:
+ lcol->type = newBatType(TYPE_void, lcol->type);
+ p = newFcnCall(reduce, aggrRef, maxRef);
+ p = pushArgument(reduce, p, lcol->mapbat);
+ getArg(p, 0) = lcol->reduceid;
+ break;
+ case MIN:
+ lcol->type = newBatType(TYPE_void, lcol->type);
+ p = newFcnCall(reduce, aggrRef, minRef);
+ p = pushArgument(reduce, p, lcol->mapbat);
+ getArg(p, 0) = lcol->reduceid;
+ break;
+ case SORT:
+ lcol->type = newBatType(TYPE_void, lcol->type);
+ p = newFcnCall(reduce, algebraRef, sortTailRef);
+ p = pushArgument(reduce, p, lcol->mapbat);
+ getArg(p, 0) = lcol->reduceid;
+ break;
+ case SORTDESC:
+ lcol->type = newBatType(TYPE_void, lcol->type);
+ p = newFcnCall(reduce, algebraRef,
sortReverseTailRef);
+ p = pushArgument(reduce, p, lcol->mapbat);
+ getArg(p, 0) = lcol->reduceid;
+ break;
+ case LIMIT:
+ lcol->type = newBatType(TYPE_void, lcol->type);
+ p = newFcnCall(reduce, algebraRef, sliceRef);
+ p = pushArgument(reduce, p, lcol->mapbat);
+ getArg(p, 0) = lcol->reduceid;
+ break;
}
}
@@ -304,7 +349,8 @@
struct stack {
int *stack; /* array of ints */
size_t len; /* max capacity of alloced array stack */
- size_t cur; /* current pointer */
+ size_t cur; /* current tail pointer */
+ size_t pos; /* current head pointer */
};
static inline void
@@ -313,6 +359,7 @@
if (stk->stack == NULL) {
stk->len = 10;
stk->cur = 0;
+ stk->pos = 0;
stk->stack = GDKmalloc(sizeof(int) * stk->len);
} else if (stk->cur == stk->len) {
stk->len *= 2;
@@ -327,7 +374,7 @@
static inline char
trackstack_isempty(struct stack *stk)
{
- return(stk->stack == NULL || stk->cur == 0);
+ return(stk->stack == NULL || stk->cur == stk->pos);
}
static inline char
@@ -338,18 +385,29 @@
if (trackstack_isempty(stk))
return(0);
- for (tsci = stk->cur; tsci > 0; tsci--) {
+ for (tsci = stk->cur; tsci > stk->pos; tsci--) {
if (stk->stack[tsci - 1] == val)
return(1);
}
return(0);
}
+static inline int
+trackstack_head(struct stack *stk)
+{
+ if (trackstack_isempty(stk))
+ return(0);
+
+ return(stk->stack[stk->pos++]);
+}
+
static inline void
trackstack_clear(struct stack *stk)
{
- if (!trackstack_isempty(stk))
+ if (!trackstack_isempty(stk)) {
stk->cur = 0;
+ stk->pos = 0;
+ }
}
static inline void
@@ -401,10 +459,12 @@
lastcol->mapid = getArg(p, 1);
lastcol->reduceid =
getArg(oreduce[i], 1);
lastcol->type = getArgType(map,
p, 1);
+ lastcol->postop = SUM;
lastcol = lastcol->next =
alloca(sizeof(mapcol));
lastcol->mapid = getArg(p, 2);
lastcol->reduceid =
getArg(oreduce[i], 2);
lastcol->type = getArgType(map,
p, 2);
+ lastcol->postop = SUM;
lastcol->next = NULL;
/* got it, time to copy
instructions */
j = i;
@@ -446,8 +506,8 @@
char mrcluster[BUFSIZ];
Symbol new;
mapcol *col, *lastcol;
- struct stack tracker = { NULL, 0, 0 };
- struct stack avgtrack = { NULL, 0, 0 };
+ struct stack tracker = { NULL, 0, 0, 0 };
+ struct stack avgtrack = { NULL, 0, 0, 0 };
char hadBinds = 0;
str mapreduceRef = putName("mapreduce", 9);
@@ -573,6 +633,7 @@
lastcol->reduceid =
getArg(oreduce[i], 0);
lastcol->type =
getArgType(map, p, 0);
lastcol->mapbat = -1;
+ lastcol->postop = pBAT;
lastcol->next = NULL;
newComment(map, "= sql
column bat");
@@ -621,7 +682,7 @@
break;
}
}
- trackstack_destroy(&avgtrack);
+ trackstack_clear(&avgtrack);
if (hadBinds == 0) {
GDKfree(reduce->stmt);
@@ -645,41 +706,85 @@
}
/* skip all NOOPs */
- if (p->token == NOOPsymbol)
+ if (p->token == NOOPsymbol) {
+ trackstack_push(&avgtrack, i);
continue;
+ }
- if (getModuleId(p) == algebraRef && (
- getFunctionId(p) == sortTailRef ||
- getFunctionId(p) == sortReverseTailRef
||
- getFunctionId(p) == sliceRef))
- {
- /* simple ORDER BY [DESC] or LIMIT/OFFSET */
- for (lastcol = col; lastcol != NULL; lastcol =
lastcol->next) {
- if (getArg(p, 1) == lastcol->reduceid) {
- copy = SINGLE_DUP;
- newComment(map, "ORDER BY [DESC] or
LIMIT/OFFSET");
- /* fix return */
- lastcol->mapid = getArg(omap[i], 0);
- lastcol->type = getArgType(map, p, 0);
- break;
+ if (getModuleId(p) == algebraRef) {
+ if (getFunctionId(p) == sortTailRef) {
+ /* simple ORDER BY */
+ for (lastcol = col; lastcol != NULL; lastcol =
lastcol->next) {
+ if (getArg(p, 1) == lastcol->reduceid) {
+ newComment(map, "= ORDER BY");
+ /* fix return */
+ lastcol->mapid =
getArg(omap[i], 0);
+ lastcol->type = getArgType(map,
p, 0);
+ lastcol->reduceid = getArg(p,
0);
+ lastcol->postop = SORT;
+ copy = cNONE;
+ break;
+ }
+ }
+ } else if (getFunctionId(p) == sortReverseTailRef) {
+ /* simple ORDER BY DESC*/
+ for (lastcol = col; lastcol != NULL; lastcol =
lastcol->next) {
+ if (getArg(p, 1) == lastcol->reduceid) {
+ newComment(map, "= ORDER BY
DESC");
+ /* fix return */
+ lastcol->mapid =
getArg(omap[i], 0);
+ lastcol->type = getArgType(map,
p, 0);
+ lastcol->reduceid = getArg(p,
0);
+ lastcol->postop = SORTDESC;
+ copy = cNONE;
+ break;
+ }
+ }
+ } else if (getFunctionId(p) == sliceRef) {
+ /* simple LIMIT/OFFSET */
+ for (lastcol = col; lastcol != NULL; lastcol =
lastcol->next) {
+ if (getArg(p, 1) == lastcol->reduceid) {
+ newComment(map, "=
LIMIT/OFFSET");
+ /* fix return */
+ lastcol->mapid =
getArg(omap[i], 0);
+ lastcol->type = getArgType(map,
p, 0);
+ lastcol->reduceid = getArg(p,
0);
+ lastcol->postop = LIMIT;
+ copy = cNONE;
+ break;
+ }
}
}
- } else if (getModuleId(p) == aggrRef && (
- getFunctionId(p) == maxRef ||
- getFunctionId(p) == minRef))
- {
- /* MAX/MIN aggregation */
- for (lastcol = col; lastcol != NULL; lastcol =
lastcol->next) {
- if (getArg(p, 1) == lastcol->reduceid) {
- newComment(map, "MAX/MIN");
- /* basically perform a MAX over all
MAXes */
- pushInstruction(map, omap[i]);
- lastcol->mapid = getArg(omap[i], 0);
- lastcol->type = getArgType(map, p, 0);
- /* we already DUP_SINGLE'd */
- copy = STICK;
- p = oreduce[i];
- break;
+ } else if (getModuleId(p) == aggrRef) {
+ if (getFunctionId(p) == maxRef) {
+ /* MAX aggregation */
+ for (lastcol = col; lastcol != NULL; lastcol =
lastcol->next) {
+ if (getArg(p, 1) == lastcol->reduceid) {
+ newComment(map, "= MAX");
+ /* basically perform a MAX over
all MAXes */
+ pushInstruction(map, omap[i]);
+ lastcol->mapid =
getArg(omap[i], 0);
+ lastcol->type = getArgType(map,
p, 0);
+ lastcol->reduceid = getArg(p,
0);
+ lastcol->postop = MAX;
+ copy = cNONE;
+ break;
+ }
+ }
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
