Changeset: 1f773ff776bb for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=1f773ff776bb
Modified Files:
monetdb5/optimizer/opt_centipede.mx
Branch: default
Log Message:
Switch to OID-range fragmentation.
diffs (truncated from 747 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
@@ -37,8 +37,8 @@ address OPTcentipedeMaterialize
comment "Implement the centipede partition operation. Throw an exception if
the partition was empty,
because then the subquery should produce a NIL ";
-pattern centipede.vector(b:bat[:oid,:any_1]) :any_1...
-address OPTvector
+pattern centipede.vector(b:bat[:oid,:any_1]) (:oid,:oid...)
+address OPTvectorOid
comment "Derive a series of slices values based on sampling";
@h
@@ -51,10 +51,11 @@ comment "Derive a series of slices value
@:exportOptimizer(centipede)@
opt_export str OPTvector(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
+opt_export str OPTvectorOid(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
InstrPtr pci);
opt_export str OPTcentipedeMaterialize(int *result, int *bid, ptr low, ptr
high);
opt_export str OPTpeers(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
-#define REMOTE_EXECUTION /* or LOCAL_EXECUTION */
+#define LOCAL_EXECUTION /* or REMOTE_EXECUTION */
#define OPTDEBUGcentipede if ( optDebug & ((lng)1 <<DEBUG_OPT_CENTIPEDE) )
#endif
@@ -67,13 +68,14 @@ opt_export str OPTpeers(Client cntxt, Ma
#include "mal_interpreter.h"
/* #define DEBUG_DETAIL*/
-/*#define _DEBUG_OPT_CENTIPEDE_ */
+#define _DEBUG_OPT_CENTIPEDE_
typedef struct{
InstrPtr target;
str schema, table, column;
int type, slice;
- int lslices, hslices;
+ int lslices, hslices; /* variables holding the range bound */
+ lng rowcnt;
ValRecord bounds[MAXSITES];
} Slices;
@@ -83,14 +85,11 @@ typedef struct{
* TODO pack is expensive, move to mat.new
*/
static MalBlkPtr
-OPTplanCntrl(Client cntxt, MalBlkPtr mb, MalBlkPtr pmb, Slices *slices, oid
plantag)
+OPTexecController(Client cntxt, MalBlkPtr mb, MalBlkPtr pmb, Slices *slices,
oid plantag)
{
MalBlkPtr cmb;
Symbol s;
char nme[BUFSIZ], *plan, *stub;
-#ifndef REMOTE_EXECUTION
- int base;
-#endif
int barrier, x, i, j, k, *alias, nrpack;
InstrPtr ret, p, q, *pack;
@@ -119,7 +118,7 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
getFunctionId( getInstrPtr(cmb,0)) = putName(nme,strlen(nme));
insertSymbol(cntxt->nspace,s);
- /* initialize binds and histograms */
+ /* initialize binds */
q= newFcnCall(cmb, sqlRef, mvcRef);
x= getArg(q,0);
alias = (int*) GDKzalloc(nrservers * sizeof(int));
@@ -142,7 +141,7 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
getFunctionId(q) = vectorRef;
q = pushArgument(cmb, q, j);
for ( j = 0; j < nrservers +1; j++) {
- k = alias[j] = newTmpVariable(cmb, slices->type);
+ k = alias[j] = newTmpVariable(cmb, TYPE_oid);
q= pushReturn(cmb,q, k);
}
pushInstruction(cmb,q);
@@ -169,12 +168,15 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
*/
#ifdef REMOTE_EXECUTION
- /* PHASE 1: run in distributed mode */
q= newFcnCall(cmb,schedulerRef,srvpoolRef);
+#else
+ q= newFcnCall(cmb,languageRef,dataflowRef);
+#endif
q->barrier= BARRIERsymbol;
barrier = getArg(q,0);
setVarType(cmb,x,TYPE_int);
+#ifdef REMOTE_EXECUTION
/* get servers to execute the query */
q= newStmt(cmb,srvpoolRef,putName("query",5));
q->retc= q->argc = 0;
@@ -182,6 +184,7 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
q= pushReturn(cmb,q, newTmpVariable(cmb,TYPE_str));
q= pushStr(cmb,q,plan);
p= q;
+#endif
for ( i = 0; i < nrservers ; i++) {
q= copyInstruction(getInstrPtr(pmb,0));
@@ -198,9 +201,13 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
q= pushArgument(cmb,q,alias[i]);
q= pushArgument(cmb,q,alias[i+1]);
}
+#ifdef REMOTE_EXECUTION
/* for distributed execution we use the stub */
getFunctionId(q) = stub;
q= setArgument(cmb,q,q->retc, getArg(p,i));
+#else
+ getFunctionId(q) = plan;
+#endif
pushInstruction(cmb,q);
}
@@ -218,26 +225,6 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
q= newAssignment(cmb);
q->barrier = EXITsymbol;
getArg(q,0) = barrier;
-#else
- /* PHASE 2 The second block runs the query in local mode */
- /* re-use instructions, injecting missing server selections */
- base = cmb->stop;
- for ( i = 0; i < nrservers ; i++) {
- q= copyInstruction(getInstrPtr(cmb,base + i));
- getFunctionId(q) = plan;
- delArgument(q,q->retc);
- pushInstruction(cmb,q);
- }
- /* put all mat.pack instructions into the program
- make sure that they have contiguous void headed columns
- */
- p = getInstrPtr(pmb,0);
- if ( slices->column)
- for ( k=0 ; k < nrpack; k++) {
- pushInstruction(cmb, copyInstruction(pack[k]));
- getArg(pack[k],0)= getArg(p,k);
- }
-#endif
/* consolidate the result */
ret = copyInstruction(getInstrPtr(cmb,0));
@@ -356,58 +343,34 @@ OPTplanStub(Client cntxt, MalBlkPtr mb,
return smb;
}
-/* prepare access to partitions by injection of the materialize instructions */
-static int
-OPTpreparePartition(MalBlkPtr nmb, InstrPtr p, Slices *slices, int pc)
+
+/* derive an OID partition from a bind column */
+/* it leads to horizontal fragmentation by OID ranges */
+static void
+OPTmaterializePartition(MalBlkPtr mb, InstrPtr p, oid low, oid hgh)
{
- InstrPtr q;
- int i,fnd,parallel = 0;
+ int v,oldvar;
- if ( slices->slice == 0){
- slices->slice = newTmpVariable(nmb, getVarType(nmb,
getArg(slices->target,0)));
- setVarUDFtype(nmb, slices->slice);
- setVarUsed(nmb, slices->slice);
- nmb->stmt[0] = pushArgument(nmb, nmb->stmt[0], slices->lslices);
- nmb->stmt[0] = pushArgument(nmb, nmb->stmt[0], slices->hslices);
- parallel=2; /* return number of arguments added for later
disposal */
- }
- /* check if the partition column instruction is already added */
- fnd =0;
- for ( i = pc -1; i > 0 && fnd == 0; i--){
- q= getInstrPtr(nmb,i);
- fnd = getFunctionId(q) && getModuleId(q) == sqlRef &&
getFunctionId(q) == bindRef &&
- strcmp(slices->column, getVarConstant(nmb,
getArg(p,2)).val.sval) == 0 &&
- strcmp(slices->column, getVarConstant(nmb,
getArg(p,3)).val.sval) == 0;
- }
- if ( fnd == 0) {
- i = newTmpVariable(nmb, getVarType(nmb,
getArg(slices->target,0)));
- pushInstruction(nmb, p = copyInstruction(slices->target));
- copyProperties(nmb, getArg(p,0),i); /* retain the properties */
- getArg(p,0) = i;
- }
- q= newInstruction(nmb,ASSIGNsymbol);
- /* add the slices->to the call */
- q= pushArgument(nmb,q, getArg(p,0));
- q= pushArgument(nmb,q, slices->lslices);
- q= pushArgument(nmb,q, slices->hslices);
- setModuleId(q,algebraRef);
- setFunctionId(q,selectRef);
- getArg(q,0)= slices->slice;
- getArg(q,1) = getArg(p,0);
- q= pushBit(nmb,q,TRUE);
- q= pushBit(nmb,q,FALSE);
- pushInstruction(nmb,q);
- return parallel;
+ oldvar = getArg(p,0);
+ getArg(p,0) = v = newTmpVariable(mb, getVarType(mb,oldvar));
+ setVarUDFtype(mb, v);
+ setVarFixed(mb,v);
+
+ p = newStmt(mb, algebraRef, sliceRef);
+ p = pushArgument(mb, p, v);
+ p = pushArgument(mb, p, low);
+ p = pushArgument(mb, p, hgh);
+ getArg(p,0)= oldvar;
}
+
/*
- * For bind instructions we have to inject materialize and semijoin
instructions
+ * For bind instructions we have to inject the horizontal slicing action
*/
static int
-OPTsliceColumn(Client cntxt, MalBlkPtr nmb, MalBlkPtr mb, InstrPtr p, Slices
*slices, int pc)
+OPTsliceColumn(Client cntxt, MalBlkPtr nmb, MalBlkPtr mb, InstrPtr p, Slices
*slices)
{
- int k, parallel = 0;
- InstrPtr q;
+ int parallel = 0;
(void) cntxt;
if ( ! (getModuleId(p) == sqlRef && getFunctionId(p) == bindRef ) &&
@@ -421,35 +384,17 @@ OPTsliceColumn(Client cntxt, MalBlkPtr n
return 0;
}
- if ( slices->slice == 0)
- parallel =OPTpreparePartition(nmb, p, slices, pc);
-
- if (strcmp(slices->column, getVarConstant(nmb, getArg(p,4)).val.sval) )
{
- /* sideways projection the range selection */
- pushInstruction(nmb,p);
- /* not needed. single column partitioning is enough
- q= newInstruction(nmb,ASSIGNsymbol);
- setModuleId(q,algebraRef);
- setFunctionId(q,semijoinRef);
- k = newTmpVariable(nmb, getVarType(mb, getArg(p,0)));
- setVarUDFtype(nmb,k);
- setVarUsed(nmb,k);
- getArg(q,0)= getArg(p,0);
- getArg(p,0)= k;
- q= pushArgument(nmb,q, k);
- q= pushArgument(nmb,q, slices->slice);
- pushInstruction(nmb,q);
- */
- } else {
- q= newInstruction(nmb,ASSIGNsymbol);
- k = newTmpVariable(nmb, getVarType(mb, getArg(p,0)));
- setVarUDFtype(nmb,k);
- setVarUsed(nmb,k);
- getArg(q,0)= getArg(p,0);
- getArg(p,0)= k;
- q= pushArgument(nmb,q, slices->slice);
- pushInstruction(nmb,q);
- }
+ if ( slices->slice == 0){
+ slices->slice = newTmpVariable(nmb, getVarType(nmb,
getArg(slices->target,0)));
+ setVarUDFtype(nmb, slices->slice);
+ setVarUsed(nmb, slices->slice);
+ nmb->stmt[0] = pushArgument(nmb, nmb->stmt[0], slices->lslices);
+ nmb->stmt[0] = pushArgument(nmb, nmb->stmt[0], slices->hslices);
+ parallel=2; /* return number of arguments added for later
disposal */
+ }
+ pushInstruction(nmb,p);
+ /* prepare access to partitions by injection of the materialize
instructions */
+ OPTmaterializePartition(nmb, p, slices->lslices, slices->hslices);
return parallel;
}
@@ -470,44 +415,44 @@ OPTsliceColumn(Client cntxt, MalBlkPtr n
#define KEEPLOCAL 5
#ifdef _DEBUG_OPT_CENTIPEDE_
-static char *status[6]= {"", "blocked ", "required ", "support ", "exported
", "keeplocal "};
+static char *statusname[6]= {"", "blocked ", "required ", "support ",
"exported ", "keeplocal "};
#endif
-static int
-OPTplanFragment(Client cntxt, MalBlkPtr mb, Slices *slices)
+static void
+OPTbakePlans(Client cntxt, MalBlkPtr mb, Slices *slices)
{
- int *plan,*vars;
+ int *status,*vars;
int i, j, k, limit, last;
InstrPtr ret, call, p, *old;
Symbol s;
- MalBlkPtr nmb, cmb, smb;
+ MalBlkPtr plan, cntrl, stub;
str msg= MAL_SUCCEED;
char nme[BUFSIZ];
char *head, *tail; /* oid reference to target table*/
oid plantag= OIDnew(1);
(void) msg; /* only used when _DEBUG_OPT_CENTIPEDE_ is defined */
- plan = GDKzalloc(mb->ssize * sizeof(int));
- if( plan == 0)
- return 0;
+ status = GDKzalloc(mb->ssize * sizeof(int));
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
