Changeset: 61a7a8be67c7 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=61a7a8be67c7
Modified Files:
monetdb5/optimizer/Makefile.ag
monetdb5/optimizer/opt_partition.mx
monetdb5/optimizer/opt_partitions.mx
monetdb5/optimizer/opt_prelude.mx
monetdb5/optimizer/opt_slicing.mx
monetdb5/optimizer/opt_support.mx
Branch: default
Log Message:
Commit of intermediate stage of partitions
The current code base is experimental and mostly geared at
creation of MAL blocks walking their way through a partitioned
database. Consolidation of the partial results is not yet
working properly.
diffs (truncated from 1486 to 300 lines):
diff --git a/monetdb5/optimizer/Makefile.ag b/monetdb5/optimizer/Makefile.ag
--- a/monetdb5/optimizer/Makefile.ag
+++ b/monetdb5/optimizer/Makefile.ag
@@ -74,7 +74,7 @@
opt_statistics.mx \
opt_strengthReduction.mx \
opt_support.mx \
- opt_slicing.mx \
+ opt_partition.mx \
opt_trace.mx
}
@@ -94,7 +94,7 @@
opt_accessmode.mx opt_joinpath.mx opt_heuristics.mx
opt_remap.mx \
opt_statistics.mx opt_trace.mx opt_recycler.mx opt_dataflow.mx
\
opt_cluster.mx opt_dictionary.mx opt_mapreduce.mx opt_origin.mx
\
- opt_mitosis.mx opt_octopus.mx opt_slicing.mx opt_history.mx
+ opt_mitosis.mx opt_octopus.mx opt_partition.mx opt_history.mx
}
EXTRA_DIST = optimizerWrapper.mx
diff --git a/monetdb5/optimizer/opt_slicing.mx
b/monetdb5/optimizer/opt_partition.mx
rename from monetdb5/optimizer/opt_slicing.mx
rename to monetdb5/optimizer/opt_partition.mx
--- a/monetdb5/optimizer/opt_slicing.mx
+++ b/monetdb5/optimizer/opt_partition.mx
@@ -16,31 +16,32 @@
Copyright August 2008-2011 MonetDB B.V.
All Rights Reserved.
@
-Use a single gauge structure to split the largest table.
-@f opt_slicing
+@f opt_partition
@a M. Kersten
-@- Slicing
-todo: alias required for all object returned from subqueries
+@- Partitioning
+Documentation in accompanying paper.
@{
@mal
-pattern optimizer.slicing():str
-address OPTslicing;
-pattern optimizer.slicing(mod:str, fcn:str):str
-address OPTslicing
+module partition;
+pattern optimizer.partition():str
+address OPTpartition;
+pattern optimizer.partition(mod:str, fcn:str):str
+address OPTpartition
comment "Modify the plan to exploit parallel processing on multiple cores";
-command sql.slice(b:bat[:any_1,:any_2], low:any_2, high:any_2)
:bat[:any_1,:any_2]
-address OPTslice
-comment "Implement the slice operation. Throw an exception if the slice was
empty,
+command partition.materialize(b:bat[:any_1,:any_2], low:any_2, high:any_2)
:bat[:any_1,:any_2]
+address OPTpartitionMaterialize
+comment "Implement the partition operation. Throw an exception if the
partition was empty,
because then the subquery should produce a NIL ";
-command sql.markH( b:bat[:any_1,:any_2] ) :bat[:oid,:any_2]
+command partition.markH( b:bat[:any_1,:any_2] ) :bat[:oid,:any_2]
address OPTmarkHead
comment "Ignore a NIL bat";
-pattern sql.gauges(b:bat[:oid,:any_1]) :any_1...
-address OPTgauges
-comment "Derive a series of gauge values based on sampling";
+pattern partition.vector(b:bat[:oid,:any_1]) :any_1...
+address OPTvector
+comment "Derive a series of slices values based on sampling";
+
@h
#ifndef _OPT_SLICING_
#define _OPT_SLICING_
@@ -48,19 +49,19 @@
#include "opt_support.h"
#define MAXSITES 4 /* to be refined */
-@:exportOptimizer(slicing)@
+@:exportOptimizer(partition)@
-opt_export str OPTgauges(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
-opt_export str OPTslice(int *result, int *bid, ptr low, ptr high);
+opt_export str OPTvector(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
+opt_export str OPTpartitionMaterialize(int *result, int *bid, ptr low, ptr
high);
opt_export str OPTmarkHead(int *result, int *bid);
/* #define DEBUG_DETAIL*/
-#define _DEBUG_OPT_SLICE_
-#define OPTDEBUGslicing if ( optDebug & ((lng)1 <<DEBUG_OPT_SLICING) )
+#define _DEBUG_OPT_PARTITION_
+#define OPTDEBUGpartition if ( optDebug & ((lng)1 <<DEBUG_OPT_PARTITION) )
#endif
@c
#include "monetdb_config.h"
-#include "opt_slicing.h"
+#include "opt_partition.h"
#include "opt_deadcode.h"
#include "mal_builder.h"
#include "mal_recycle.h"
@@ -73,9 +74,9 @@
InstrPtr target;
str schema, table, column;
int type, slice;
- int lgauge, hgauge;
+ int lslices, hslices;
ValRecord bounds[MAXSITES];
-} Gauge;
+} Slices;
static int
OPTinitcode(Client cntxt, MalBlkPtr mb){
@@ -112,19 +113,24 @@
for ( i=0; i< p->retc; i++){
q = newAssignment(nmb);
getArg(q,0) = getArg(p,i);
- q= pushNil(nmb,q, getArgType(nmb,p,i));
+ q= pushNil(nmb,q, TYPE_void);
}
+ /* clear the exception */
+ q= newAssignment(nmb);
+ getArg(q,0) = findVariable(nmb,"ANYexception");
+ pushNil(nmb,q,TYPE_str);
q = newExitStmt(nmb,"ANYexception");
}
+
static MalBlkPtr
-OPTslicingStub(Client cntxt, MalBlkPtr mb, int pc, MalBlkPtr pmb, Gauge gauge)
+OPTpartitionStub(Client cntxt, MalBlkPtr mb, int pc, MalBlkPtr pmb, Slices
slices)
{
MalBlkPtr smb = 0;
Symbol s;
InstrPtr sig, q, ret;
int j,k,conn, *arg;
char nme[BUFSIZ];
- (void) gauge;
+ (void) slices;
/* define the sub query stub for remote processing */
snprintf(nme,BUFSIZ,"%s_stub%d",getFunctionId( getInstrPtr(mb,0)), pc);
@@ -154,7 +160,7 @@
}
/* (k1,...kn):= remote.exec(conn,slicing,qry,version....) */
- snprintf(nme,BUFSIZ,"%s_qry%d",getFunctionId( getInstrPtr(mb,0)),pc);
+ snprintf(nme,BUFSIZ,"%s_qry_%d",getFunctionId( getInstrPtr(mb,0)),pc);
q = newInstruction(smb,ASSIGNsymbol);
getModuleId(q) = remoteRef;
getFunctionId(q) = execRef;
@@ -204,17 +210,17 @@
}
static MalBlkPtr
-OPTslicingCntrl(Client cntxt, MalBlkPtr mb, int pc, MalBlkPtr pmb, Gauge gauge)
+OPTpartitionCntrl(Client cntxt, MalBlkPtr mb, int pc, MalBlkPtr pmb, Slices
slices)
{
MalBlkPtr cmb;
Symbol s;
char nme[BUFSIZ];
- int /*tpe,*/ x, i, j, k, *alias, nrpack;
+ int x, i, j, k, *alias, nrpack;
InstrPtr call, ret, p, q, *pack;
- (void) gauge;
+ (void) slices;
/* define the query controller */
- snprintf(nme,BUFSIZ,"%s_cntrl%d",getFunctionId( getInstrPtr(mb,0)),pc);
+ snprintf(nme,BUFSIZ,"%s_cntrl_%d",getFunctionId( getInstrPtr(mb,0)),pc);
s = newFunction(userRef, putName(nme, strlen(nme)),FUNCTIONsymbol);
if ( s == NULL)
return 0;
@@ -234,8 +240,7 @@
nrpack = q->retc;
pack = (InstrPtr *) GDKzalloc(sizeof(InstrPtr) * nrpack);
pushInstruction(cmb, copyInstruction(pmb->stmt[0]));
- call = copyInstruction(getInstrPtr(pmb,0));
- cmb->stmt[0]->argc -= 2; /* no bounds */
+ call = copyInstruction(getInstrPtr(mb,0));
getFunctionId( getInstrPtr(cmb,0)) = putName(nme,strlen(nme));
insertSymbol(cntxt->nspace,s);
@@ -243,32 +248,32 @@
q= newFcnCall(cmb, sqlRef, mvcRef);
x= getArg(q,0);
alias = (int*) GDKzalloc(nrpeers * sizeof(int));
- if( gauge.column) {
+ if( slices.column) {
q= newInstruction(cmb, ASSIGNsymbol);
getModuleId(q) = sqlRef;
getFunctionId(q) = bindRef;
q = pushArgument(cmb,q,x);
- j = getArg(q,0) = newTmpVariable(cmb,newBatType(TYPE_oid,
gauge.type));
+ j = getArg(q,0) = newTmpVariable(cmb,newBatType(TYPE_oid,
slices.type));
setVarUDFtype(cmb,j);
setVarFixed(cmb,j);
- q= pushStr(cmb,q, gauge.schema);
- q= pushStr(cmb,q, gauge.table);
- q= pushStr(cmb,q, gauge.column);
+ q= pushStr(cmb,q, slices.schema);
+ q= pushStr(cmb,q, slices.table);
+ q= pushStr(cmb,q, slices.column);
q= pushInt(cmb,q, 0);
pushInstruction(cmb,q);
q= newInstruction(cmb, ASSIGNsymbol);
- getModuleId(q) = sqlRef;
- getFunctionId(q) = putName("gauges",6);
+ getModuleId(q) = partitionRef;
+ getFunctionId(q) = vectorRef;
q = pushArgument(cmb, q, j);
for ( j = 0; j < nrpeers +1; j++) {
- k = alias[j] = newTmpVariable(cmb, gauge.type);
+ k = alias[j] = newTmpVariable(cmb, slices.type);
q= pushReturn(cmb,q, k);
}
pushInstruction(cmb,q);
}
/* dataflow */
- if ( gauge.column) {
+ if ( slices.column) {
p = getInstrPtr(pmb,0);
for ( k=0 ;k < nrpack ; k++){
pack[k] = newInstruction(cmb,ASSIGNsymbol);
@@ -279,6 +284,7 @@
}
if ( slicingLocal == 0){
+ /* under dataflow control, initialize the variables */
for ( k=0 ; k < nrpack ; k++){
q = newInstruction(cmb,ASSIGNsymbol);
getArg(q,0) = getArg(pack[k],0);
@@ -286,16 +292,16 @@
pushInstruction(cmb,q);
}
+ /* mark start of dataflow block */
q= newFcnCall(cmb,languageRef,dataflowRef);
q->barrier= BARRIERsymbol;
x = getArg(q,0);
setVarType(cmb,x,TYPE_int);
}
- /* tpe = getArgType( pmb, getInstrPtr(pmb,0),0); */
-
for ( i = 0; i < nrpeers ; i++) {
- q= copyInstruction(call);
+ q= copyInstruction(getInstrPtr(pmb,0));
+ q->argc = call->argc- call->retc + pmb->stmt[0]->retc;;
q->token = ASSIGNsymbol;
q->barrier = 0;
for ( k=0 ; k < q->retc; k++) {
@@ -303,37 +309,40 @@
pack[k] = pushArgument(cmb,pack[k], getArg(q,k));
}
- if ( gauge.column){
- getArg(q, q->argc-2 ) = alias[i];
- getArg(q, q->argc-1) = alias[i+1];
+ if ( slices.column){
+ /* add the splitter arguments */
+ q= pushArgument(cmb,q,alias[i]);
+ q= pushArgument(cmb,q,alias[i+1]);
}
pushInstruction(cmb,q);
}
- /* put all mat.pak instructions into the program
- and make sure that they have correct void headed columns
+ /* put all mat.pack instructions into the program
+ and make sure that they have contiguous void headed columns
*/
- if ( gauge.column)
+ if ( slices.column)
for ( k=0 ; k < nrpack; k++) {
pushInstruction(cmb, pack[k]);
j = newTmpVariable(cmb,getArgType(cmb,pack[k],0));
- q= newFcnCall(cmb,sqlRef,markHRef);
+ q= newFcnCall(cmb,partitionRef,markHRef);
getArg(q,0) = getArg(pack[k],0);
q= pushArgument(cmb,q, j);
getArg(pack[k],0) = j;
pack[k] = q;
}
-
+ /* finalize the dataflow block */
if ( slicingLocal == 0){
q= newAssignment(cmb);
q->barrier = EXITsymbol;
getArg(q,0) = x;
}
+ /* consolidate the result */
q = copyInstruction(getInstrPtr(mb,pc));
if ( getFunctionId(q) == countRef || getFunctionId(q) == sumRef ){
getFunctionId(q) = sumRef;
getArg(q,1) = getArg(pack[0],0);
+
setVarType(cmb,getArg(getInstrPtr(cmb,0),0),getArgType(cmb,q,0));
} else
if ( getFunctionId(q) == minRef || getFunctionId(q) == maxRef) {
getArg(q,1) = getArg(pack[0],0);
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
