Changeset: f31305f645f8 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=f31305f645f8
Modified Files:
monetdb5/optimizer/opt_centipede.mx
Branch: default
Log Message:
Latest tryout of centipede.
diffs (truncated from 302 to 300 lines):
diff --git a/monetdb5/optimizer/opt_centipede.mx
b/monetdb5/optimizer/opt_centipede.mx
--- a/monetdb5/optimizer/opt_centipede.mx
+++ b/monetdb5/optimizer/opt_centipede.mx
@@ -67,7 +67,7 @@ opt_export str OPTpeers(Client cntxt, Ma
#include "mal_recycle.h"
#include "mal_interpreter.h"
-/* #define DEBUG_DETAIL*/
+/* #define DEBUG_OPT_DETAIL*/
#define _DEBUG_OPT_CENTIPEDE_
typedef struct{
@@ -79,6 +79,25 @@ typedef struct{
ValRecord bounds[MAXSITES];
} Slices;
+/* Glue the groups together, propagating the extension oids */
+/* assume you trigger on the 'grp' */
+static void
+OPTglueGroup(MalBlkPtr mb, InstrPtr pci)
+{
+ int ext, grp, i,j;
+ InstrPtr p;
+
+ for (i = 0; i < mb->stop; i++) {
+ p= getInstrPtr(mb,i);
+ if ( sscanf(getVarName(mb, getArg(p,i)),"ext%d", &ext) == 1
&&
+ sscanf(getVarName(mb, getArg(pci,i)),"grp%d",&grp) == 1
&&
+ ext == grp){
+ /* correct the components */
+ for ( j = pci->retc; j < pci->argc; j++){
+ }
+ }
+ }
+}
/*
* The query will be controlled from the coordinator with a plan
* geared at parallel execution
@@ -93,10 +112,6 @@ OPTexecController(Client cntxt, MalBlkPt
int barrier, x, i, j, k, *alias, nrpack;
InstrPtr ret, p, q, *pack;
-#ifdef _DEBUG_OPT_CENTIPEDE_
- mnstr_printf(cntxt->fdout,"#original plan \n");
- printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT);
-#endif
/* define the query controller */
snprintf(nme, BUFSIZ, "%s_plan"OIDFMT, getFunctionId(
getInstrPtr(mb,0)), plantag);
plan = putName(nme, strlen(nme));
@@ -150,7 +165,8 @@ OPTexecController(Client cntxt, MalBlkPt
/* pack[i] := mat.pack(x1,...xn) */
if ( slices->column) {
p = getInstrPtr(pmb,0);
- for ( k=0 ;k < nrpack ; k++){
+ for ( k=0 ;k < nrpack ; k++)
+ {
pack[k] = newInstruction(cmb,ASSIGNsymbol);
getModuleId(pack[k]) = matRef;
getFunctionId(pack[k]) = packRef;
@@ -217,7 +233,14 @@ OPTexecController(Client cntxt, MalBlkPt
*/
p = getInstrPtr(pmb,0);
if ( slices->column)
- for ( k=0 ; k < nrpack; k++) {
+ for ( k=0 ; k < nrpack; k++)
+ if ( strncmp(getVarName(cmb, getArg(pack[k],0)),"grp",3) == 0){
+ OPTglueGroup(cmb, pack[k]);
+ /* take care of grp targets */
+ pushInstruction(cmb, pack[k]);
+ getArg(pack[k],0)= getArg(p,k);
+ } else
+ {
pushInstruction(cmb, pack[k]);
getArg(pack[k],0)= getArg(p,k);
}
@@ -484,7 +507,7 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
strcmp(slices->table, getVarConstant(mb,
getArg(p,3)).val.sval) == 0 ) {
status[i] = PARTITION;
head[getArg(p,0)] = 1;
- tail[getArg(p,0)] = getFunctionId(p) != bindidxRef;
+ tail[getArg(p,0)] = 0;
}
/* blocking instructions are those that require data exchange,
aggregation or total view */
@@ -495,24 +518,63 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
status[i] = PARTITION;
head[getArg(p,0)] = 1;
} else
- if ( vars[getArg(p,1)] == PARTITION &&
vars[getArg(p,2)] == PARTITION )
- status[i] = BLOCKED;
+ if ( vars[getArg(p,1)] == PARTITION &&
vars[getArg(p,2)] == PARTITION ) {
+ if (tail[getArg(p,1)] == 1 ) {
+ status[i] = PARTITION;
+ tail[getArg(p,0)] = tail[getArg(p,2)];
+ } else status[i] = BLOCKED;
+ }
} else
if ( getModuleId(p) == algebraRef && getFunctionId(p) ==
leftjoinRef ) {
- head[getArg(p,0)] = head[getArg(p,1)];
- tail[getArg(p,0)] = tail[getArg(p,1)];
- if ( head[getArg(p,0)] + tail[getArg(p,0)])
- status[i] = PARTITION;
+ head[getArg(p,0)] = head[getArg(p,1)];
+ tail[getArg(p,0)] = tail[getArg(p,1)];
+ if ( head[getArg(p,0)] + tail[getArg(p,0)])
+ status[i] = PARTITION;
} else
- if ( getModuleId(p) == algebraRef &&
(getFunctionId(p)==markTRef || getFunctionId(p)== thetauselectRef) ) {
+ if ( getModuleId(p) == algebraRef && (getFunctionId(p)==
thetauselectRef || getFunctionId(p) == uselectRef || getFunctionId(p) ==
selectRef ) ) {
if (vars[getArg(p,p->retc)] == PARTITION ) {
head[getArg(p,0)] = 1;
status[i] = PARTITION;
}
} else
+ if ( getModuleId(p) == algebraRef && getFunctionId(p) ==
kdifferenceRef ) {
+ /* kdifference can be pushed if the second argument is
a partition variable */
+ if (vars[getArg(p,2)] == PARTITION ){
+ head[getArg(p,0)] = head[getArg(p,1)];
+ tail[getArg(p,0)] = tail[getArg(p,1)];
+ status[i] = PARTITION;
+ }
+ } else
+ if ( getModuleId(p) == algebraRef && getFunctionId(p) ==
kunionRef ) {
+ /* kunion can always be pushed when either argument is
partition */
+ head[getArg(p,0)] = head[getArg(p,1)];
+ tail[getArg(p,0)] = tail[getArg(p,1)];
+ if ( head[getArg(p,0)] + tail[getArg(p,0)])
+ status[i] = PARTITION;
+ } else
+ if ( getModuleId(p) == algebraRef &&
getFunctionId(p)==markTRef ) {
+ if (vars[getArg(p,p->retc)] == PARTITION ){
+ head[getArg(p,0)] = 1;
+ status[i] = PARTITION;
+ }
+ } else
if ( getModuleId(p) == algebraRef &&
getFunctionId(p)==markHRef ) {
- if (vars[getArg(p,p->retc)] == PARTITION )
+ if (vars[getArg(p,p->retc)] == PARTITION ){
tail[getArg(p,0)] = 1;
+ status[i] = PARTITION;
+ }
+ } else
+ if ( getModuleId(p) == algebraRef && getFunctionId(p) ==
sortTailRef ) {
+ if (tail[getArg(p,1)]){
+ tail[getArg(p,0)] = 1;
+ status[i] = PARTITION;
+ }
+ } else
+ if ( getModuleId(p) == batRef && (getFunctionId(p) ==
mirrorRef || getFunctionId(p) == reverseRef) ) {
+ if (vars[getArg(p,p->retc)] == PARTITION ){
+ tail[getArg(p,0)] = 1;
+ status[i] = PARTITION;
+ }
} else
if ( getModuleId(p) == batRef &&
getFunctionId(p)==reverseRef ) {
if (vars[getArg(p,p->retc)] == PARTITION ){
@@ -521,16 +583,25 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
status[i] = PARTITION;
}
} else
- if ( getModuleId(p) == algebraRef &&
(getFunctionId(p)==sortRef || getFunctionId(p)==sortTailRef)) {
+ if ( getModuleId(p) == algebraRef && (getFunctionId(p)==sortRef
|| getFunctionId(p)==sortTailRef)) {
/* this is always blocking */
if (vars[getArg(p,p->retc)] == PARTITION )
status[i]= BLOCKED;
} else
- if ( (getModuleId(p) == groupRef && (getFunctionId(p) ==
doneRef || getFunctionId(p) == newRef ||getFunctionId(p) == deriveRef) ) ||
- getModuleId(p) == pqueueRef || getModuleId(p)
== aggrRef ){
- /* this is always blocking */
- status[i]= BLOCKED;
- } else
+ if ( getModuleId(p) == groupRef && ( getFunctionId(p) == newRef
||getFunctionId(p) == deriveRef || getFunctionId(p) == doneRef) ){
+ if ( vars[getArg(p, p->retc)] == PARTITION )
+ /* groups against the partition column is
allowed.
+ It calls for a proper group reconstruction
at the receiver
+ */
+ head[getArg(p,0)] = 1;
+ tail[getArg(p,0)] = 0;
+ head[getArg(p,1)] = 1;
+ tail[getArg(p,1)] = 1;
+ status[i] = PARTITION;
+ }else
+ if ( getModuleId(p) == groupRef && (getFunctionId(p) ==
pqueueRef || getFunctionId(p) == aggrRef ))
+ status[i] = BLOCKED;
+ else
if ( (getModuleId(p) == sqlRef && (getFunctionId(p) ==
resultSetRef || getFunctionId(p) == putName("exportValue",11) ) ) ||
getModuleId(p) == ioRef )
status[i] = BLOCKED;
@@ -631,10 +702,11 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
p = old[i];
for( j = p->retc; j < p->argc; j++)
if ( (vars[getArg(p,j)] == PARTITION || vars[getArg(p,j)] ==
SUPPORTIVE) && isaBatType(getArgType(plan,p,j)) ){
- /* limit the number of returned BATs to those that are
expensive */
+ /* limit the number of returned BATs to those that are
expensive
if ( (getModuleId(p) == algebraRef && (getFunctionId(p)
== markHRef || getFunctionId(p) == markTRef)) ||
(getModuleId(p) == batRef && (getFunctionId(p)
== reverseRef || getFunctionId(p) == mirrorRef )) )
continue;
+ */
for ( k = 0; k < ret->retc; k++)
if (getArg(ret,k) == getArg(p,j))
break;
@@ -656,6 +728,16 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
for ( i = 1; i < limit ; i++)
if( status[i] == PARTITION || status[i] == SUPPORTIVE ) {
p = copyInstruction(getInstrPtr(mb, i));
+ if ( old[i]->barrier == RETURNsymbol) {
+ /* convert back the grouped variables */
+ int ext,grp;
+ for ( k = 0; k < p->retc ; k++)
+ if (sscanf(getVarName(plan, getArg(p,k)),"ext%d",&ext)
== 1){
+ } else
+ if (sscanf(getVarName(plan, getArg(p,k)),"grp%d",&grp)
== 1){
+ }
+ pushInstruction(plan,p);
+ } else
if ( old[i]->token == ENDsymbol) {
getFunctionId(plan->stmt[0]) = putName(nme,strlen(nme));
/* fix the return statement to become a complete
assignment */
@@ -687,21 +769,12 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
insertSymbol(cntxt->nspace,s);
#ifdef _DEBUG_OPT_CENTIPEDE_
+ mnstr_printf(cntxt->fdout,"#original plan \n");
+ printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT);
mnstr_printf(cntxt->fdout,"#rough plan errors %d \n", plan->errors);
//printFunction(cntxt->fdout, plan, 0, LIST_MAL_STMT);
#endif
- /* construct the remote stub plan */
- stub = OPTplanStub(cntxt, mb, plan, plantag);
- if ( stub) {
- msg= optimizeMALBlock(cntxt, stub);
- chkProgram(cntxt->fdout, cntxt->nspace, stub);
- }
-#ifdef _DEBUG_OPT_CENTIPEDE_
- mnstr_printf(cntxt->fdout,"#stub plan errors %d %s
\n",stub->errors,msg?msg:"");
- printFunction(cntxt->fdout, stub, 0, LIST_MAL_STMT);
-#endif
-
/* construct the control plan for local/remote execution */
cntrl = OPTexecController(cntxt,mb,plan, slices, plantag);
if ( cntrl) {
@@ -720,14 +793,27 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
getFunctionId(call) = getFunctionId(getInstrPtr(cntrl,0));
insertInstruction(mb,call, 1);
-#ifdef _DEBUG_OPT_CENTIPEDE_
+#ifdef _DEBUG_OPT_DEBUG
mnstr_printf(cntxt->fdout,"#non-optimized remote plan error %d %s\n",
plan->errors, msg?msg:"");
printFunction(cntxt->fdout, plan, 0, LIST_MAL_STMT);
+#endif
+#ifdef _DEBUG_OPT_CENTIPEDE_
msg= optimizeMALBlock(cntxt, plan);
chkProgram(cntxt->fdout, cntxt->nspace, plan);
mnstr_printf(cntxt->fdout,"#optimized remote plan error %d %s\n",
plan->errors, msg?msg:"");
printFunction(cntxt->fdout, plan, 0, LIST_MAL_STMT);
#endif
+
+ /* construct the remote stub plan */
+ stub = OPTplanStub(cntxt, mb, plan, plantag);
+ if ( stub) {
+ msg= optimizeMALBlock(cntxt, stub);
+ chkProgram(cntxt->fdout, cntxt->nspace, stub);
+ }
+#ifdef _DEBUG_OPT_CENTIPEDE_
+ mnstr_printf(cntxt->fdout,"#stub plan errors %d %s
\n",stub->errors,msg?msg:"");
+ printFunction(cntxt->fdout, stub, 0, LIST_MAL_STMT);
+#endif
GDKfree(old);
GDKfree(vars);
}
@@ -802,14 +888,14 @@ OPTcentipedeImplementation(Client cntxt,
/* derive a local plan based on forward flow reasoning */
OPTbakePlans(cntxt, mb, &slices);
-#ifdef DEBUG_DETAIL
+#ifdef DEBUG_OPT_DETAIL
mnstr_printf(cntxt->fdout,"#orig BEFORE optimizer test\n");
printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT);
#endif
msg= optimizeMALBlock(cntxt, mb);
chkProgram(cntxt->fdout, cntxt->nspace, mb);
#ifdef _DEBUG_OPT_CENTIPEDE_
- mnstr_printf(cntxt->fdout,"#partition %s.%s.%s type %d %s\n",
+ mnstr_printf(cntxt->fdout,"#final plan %s.%s.%s type %d %s\n",
slices.schema,
slices.table,
(slices.column ? slices.column: ""),
@@ -920,11 +1006,11 @@ OPTvectorOid(Client cntxt, MalBlkPtr mb,
o= (oid*) getArgReference(stk,pci,0);
*o = 0;
if ( pci->retc > 2 )
- for ( i= 1; i < pci->retc-2; i++){
+ for ( i= 1; i < pci->retc-1; i++){
o= (oid*) getArgReference(stk, pci, i);
*o = (rows * i ) / (pci->retc - 1) + 1; /* last one excluded */
}
- o= (oid*) getArgReference(stk,pci,pci->retc-1);
+ o= (oid*) getArgReference(stk,pci,i);
*o = oid_nil;
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
