Changeset: 06e9b35bb178 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=06e9b35bb178
Modified Files:
MonetDB5/src/optimizer/opt_mapreduce.mx
Branch: default
Log Message:
Rename some variables for clarity.
mc -> map
mb -> reduce
old -> omap
orig -> oreduce
diffs (truncated from 555 to 300 lines):
diff -r 608619e3fb60 -r 06e9b35bb178 MonetDB5/src/optimizer/opt_mapreduce.mx
--- a/MonetDB5/src/optimizer/opt_mapreduce.mx Tue May 25 10:34:09 2010 +0200
+++ b/MonetDB5/src/optimizer/opt_mapreduce.mx Tue May 25 10:47:11 2010 +0200
@@ -163,7 +163,7 @@
static void
MRdistributework(
Client cntxt,
- MalBlkPtr mb,
+ MalBlkPtr reduce,
mapcol *col,
InstrPtr sig,
str mrcluster)
@@ -190,66 +190,66 @@
w = (int *)alloca(retc * sizeof(int));
for (lcol = col, j = 0; lcol != NULL; lcol = lcol->next, j++) {
- packs[j] = p = newFcnCall(mb, batRef, newRef);
- p = pushType(mb, p, getHeadType(lcol->val1type));
- p = pushType(mb, p, getTailType(lcol->val1type));
+ packs[j] = p = newFcnCall(reduce, batRef, newRef);
+ p = pushType(reduce, p, getHeadType(lcol->val1type));
+ p = pushType(reduce, p, getTailType(lcol->val1type));
getArg(p, 0) = lcol->val1;
/* same for all sub results that we push into the mat.pack as
* arguments at the same time */
for (i = 0; i < n; i++) {
- p = newFcnCall(mb, batRef, newRef);
- p = pushType(mb, p, getHeadType(lcol->val1type));
- p = pushType(mb, p, getTailType(lcol->val1type));
- setArgType(mb, p, 0, lcol->val1type);
+ p = newFcnCall(reduce, batRef, newRef);
+ p = pushType(reduce, p, getHeadType(lcol->val1type));
+ p = pushType(reduce, p, getTailType(lcol->val1type));
+ setArgType(reduce, p, 0, lcol->val1type);
gets[(i * retc) + j] = getArg(p, 0);
}
}
for (i = 0; i < n; i++) {
/* q := remote.connect("uri", "user", "pass"); */
- p = newStmt(mb, remoteRef, connectRef);
- p = pushStr(mb, p, mapnodes[i].uri);
- p = pushStr(mb, p, mapnodes[i].user);
- p = pushStr(mb, p, mapnodes[i].pass);
- p = pushStr(mb, p, "msql");
+ p = newStmt(reduce, remoteRef, connectRef);
+ p = pushStr(reduce, p, mapnodes[i].uri);
+ p = pushStr(reduce, p, mapnodes[i].user);
+ p = pushStr(reduce, p, mapnodes[i].pass);
+ p = pushStr(reduce, p, "msql");
q = getArg(p, 0);
/* remote.register(q, "mod", "fcn"); */
- p = newStmt(mb, remoteRef, putName("register", 8));
- p = pushArgument(mb, p, q);
- p = pushStr(mb, p, getModuleId(sig));
- p = pushStr(mb, p, getFunctionId(sig));
+ p = newStmt(reduce, remoteRef, putName("register", 8));
+ p = pushArgument(reduce, p, q);
+ p = pushStr(reduce, p, getModuleId(sig));
+ p = pushStr(reduce, p, getFunctionId(sig));
/* (x1, x2, ..., xn) := remote.exec(q, "mod", "fcn"); */
- p = newInstruction(mb, ASSIGNsymbol);
+ p = newInstruction(reduce, ASSIGNsymbol);
setModuleId(p, remoteRef);
setFunctionId(p, execRef);
- p = pushArgument(mb, p, q);
- p = pushStr(mb, p, getModuleId(sig));
- p = pushStr(mb, p, getFunctionId(sig));
+ p = pushArgument(reduce, p, q);
+ p = pushStr(reduce, p, getModuleId(sig));
+ p = pushStr(reduce, p, getFunctionId(sig));
for (j = 0; j < retc; j++) {
/* x1 := remote.put(q, :type) */
- o = newFcnCall(mb, remoteRef, putRef);
- o = pushArgument(mb, o, q);
- o = pushArgument(mb, o, gets[(i * retc) + j]);
+ o = newFcnCall(reduce, remoteRef, putRef);
+ o = pushArgument(reduce, o, q);
+ o = pushArgument(reduce, o, gets[(i * retc) + j]);
v = getArg(o, 0);
- p = pushReturn(mb, p, v);
+ p = pushReturn(reduce, p, v);
w[j] = v;
}
- pushInstruction(mb, p);
+ pushInstruction(reduce, p);
/* y1 := remote.get(q, x1); */
for (j = 0; j < retc; j++) {
- p = newFcnCall(mb, remoteRef, getRef);
- p = pushArgument(mb, p, q);
- p = pushArgument(mb, p, w[j]);
+ p = newFcnCall(reduce, remoteRef, getRef);
+ p = pushArgument(reduce, p, q);
+ p = pushArgument(reduce, p, w[j]);
getArg(p, 0) = gets[(i * retc) + j];
}
/* remote.disconnect(q); */
- p = newStmt(mb, remoteRef, disconnectRef);
- p = pushArgument(mb, p, q);
+ p = newStmt(reduce, remoteRef, disconnectRef);
+ p = pushArgument(reduce, p, q);
}
/* delayed bat.inserts for easily creating an deterministic flow */
@@ -257,23 +257,23 @@
q = getArg(packs[j], 0);
/* p := bat.insert(b, y1) */
for (i = 0; i < n; i++) {
- p = newStmt(mb, batRef, insertRef);
- p = pushArgument(mb, p, q);
+ p = newStmt(reduce, batRef, insertRef);
+ p = pushArgument(reduce, p, q);
if (!isaBatType(lcol->val1type))
- p = pushNil(mb, p, TYPE_void);
- p = pushArgument(mb, p, gets[(i * retc) + j]);
+ p = pushNil(reduce, p, TYPE_void);
+ p = pushArgument(reduce, p, gets[(i * retc) + j]);
q = getArg(p, 0);
}
/* markH all result bats such that further operations don't get
* confused by possible duplicate ids */
- p = newFcnCall(mb, algebraRef, markHRef);
- p = pushArgument(mb, p, q);
+ p = newFcnCall(reduce, algebraRef, markHRef);
+ p = pushArgument(reduce, p, q);
getArg(p, 0) = lcol->val1;
}
#if defined(_DEBUG_OPT_MAPREDUCE) && _DEBUG_OPT_MAPREDUCE == 0
- chkProgram(cntxt->nspace, mb);
- printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT);
+ chkProgram(cntxt->nspace, reduce);
+ printFunction(cntxt->fdout, reduce, 0, LIST_MAL_STMT);
#endif
MRcleanCloud();
@@ -337,15 +337,15 @@
static int
OPTmapreduceImplementation(
Client cntxt,
- MalBlkPtr mb,
+ MalBlkPtr reduce,
MalStkPtr stk,
InstrPtr pc)
{
int i, j, limit;
int lastUnion = -1;
enum copymode copy;
- InstrPtr p, *old, *orig, ret, sig;
- MalBlkPtr mc;
+ InstrPtr p, *omap, *oreduce, ret, sig;
+ MalBlkPtr map;
char nme[IDLENGTH];
char mrcluster[BUFSIZ];
Symbol new;
@@ -357,7 +357,7 @@
(void)pc;
#if defined(_DEBUG_OPT_MAPREDUCE) && _DEBUG_OPT_MAPREDUCE == 0
- printFunction(cntxt->fdout, mb, 0, LIST_MAL_ALL);
+ printFunction(cntxt->fdout, reduce, 0, LIST_MAL_ALL);
#endif
/* For now we assume that the default ritual of the SQL compiler is
@@ -378,28 +378,28 @@
col = lastcol = NULL;
- mc = copyMalBlk(mb); /* for the map-program */
- old = mc->stmt;
- orig = mb->stmt;
- limit = mc->stop;
+ map = copyMalBlk(reduce); /* for the map-program */
+ omap = map->stmt;
+ oreduce = reduce->stmt;
+ limit = map->stop;
- snprintf(nme, IDLENGTH, "%smap", getFunctionId(getInstrPtr(mb, 0)));
+ snprintf(nme, IDLENGTH, "%smap", getFunctionId(getInstrPtr(reduce, 0)));
/* zap */
- if (newMalBlkStmt(mc, mc->ssize) < 0) {
+ if (newMalBlkStmt(map, map->ssize) < 0) {
return 0;
}
- if (newMalBlkStmt(mb, mb->ssize) < 0) {
- freeMalBlk(mc);
+ if (newMalBlkStmt(reduce, reduce->ssize) < 0) {
+ freeMalBlk(map);
return 0;
}
new = newFunction(userRef, putName(nme, strlen(nme)), FUNCTIONsymbol);
sig = copyInstruction(getInstrPtr(new->def, 0));
freeMalBlk(new->def);
- new->def = mc;
- mc->keephistory = mb->keephistory;
- pushInstruction(mc, sig);
+ new->def = map;
+ map->keephistory = reduce->keephistory;
+ pushInstruction(map, sig);
/* We do a two-phase scan over the original plan to get a MAP and
* REDUCE program. We cannot do it in a single scan, because
@@ -409,15 +409,15 @@
copy = FREE; /* free original copied signature */
mrcluster[0] = '\0';
for (i = 0; i < limit; i++) { /* phase 1 */
- p = old[i];
+ p = omap[i];
if (getModuleId(p) == sqlRef) {
if (getFunctionId(p) == mvcRef) {
/* sql.mvc(): we need this statement everywhere
*/
copy = SINGLE_DUP;
} else if (getFunctionId(p) == bindRef) {
- if (*(int *)VALget(&getVar(mc, getArg(p,
5))->value) == 0) {
- str schema = VALget(&getVar(mc,
getArg(p, 2))->value);
+ if (*(int *)VALget(&getVar(map, getArg(p,
5))->value) == 0) {
+ str schema = VALget(&getVar(map,
getArg(p, 2))->value);
/* check if this is a column from a
mapreduce schema (mr_*) */
if (strncmp(schema, "mr_", 3) != 0)
break;
@@ -462,10 +462,10 @@
if (lastcol == NULL) {
col = lastcol =
alloca(sizeof(mapcol));
/* this is the
first one, leave a marker */
-
freeInstruction(orig[i]);
+
freeInstruction(oreduce[i]);
/* FIXME:
should leave some symbol */
- orig[i] =
(InstrPtr)1;
-
pushInstruction(mc, p);
+ oreduce[i] =
(InstrPtr)1;
+
pushInstruction(map, p);
copy = LEAVE;
} else {
lastcol =
lastcol->next =
@@ -473,9 +473,9 @@
copy = SINGLE;
}
lastcol->val1 =
getArg(p, 0);
- lastcol->val1type =
getArgType(mc, p, 0);
+ lastcol->val1type =
getArgType(map, p, 0);
lastcol->next = NULL;
- newComment(mc, "= sql
column bat");
+ newComment(map, "= sql
column bat");
/* break to avoid
tracking the return */
break;
/* TODO: we should do a
forward search for
@@ -504,19 +504,19 @@
switch (copy) {
case FREE:
freeInstruction(p);
- old[i] = NULL;
+ omap[i] = NULL;
break;
case LEAVE:
/* make GCC happy */
break;
case SINGLE_DUP:
copy = FREE;
- pushInstruction(mc, p);
+ pushInstruction(map, p);
break;
case SINGLE:
copy = LEAVE;
- pushInstruction(mc, p);
- orig[i]->token = NOOPsymbol;
+ pushInstruction(map, p);
+ oreduce[i]->token = NOOPsymbol;
break;
case STICK:
case DUP:
@@ -528,34 +528,34 @@
trackstack_destroy(&tracker);
if (hadBinds == 0) {
- GDKfree(mb->stmt);
- mb->stmt = orig;
- mb->stop = limit;
- GDKfree(old);
+ GDKfree(reduce->stmt);
+ reduce->stmt = oreduce;
+ reduce->stop = limit;
+ GDKfree(omap);
freeSymbol(new);
return 0;
}
copy = STICK;
for (i = 0; i < limit; i++) { /* phase 2 */
- p = orig[i];
+ p = oreduce[i];
if (p == (InstrPtr)1) {
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
