Changeset: 2d127f2cc152 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=2d127f2cc152
Modified Files:
monetdb5/optimizer/opt_centipede.mx
Branch: default
Log Message:
Simplify the centipede optimizer
The code produces a horizontal partitioning based on oid ranges.
The TPCH queries run correctly on non-remote version (except
for double column expression Q1)).
diffs (truncated from 355 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
@@ -152,7 +152,7 @@ OPTexecController(Client cntxt, MalBlkPt
for ( k=0 ;k < nrpack ; k++){
pack[k] = newInstruction(cmb,ASSIGNsymbol);
getModuleId(pack[k]) = matRef;
- getFunctionId(pack[k]) = putName("pack3",5);
+ getFunctionId(pack[k]) = packRef;
getArg(pack[k],0) = newTmpVariable(cmb,
newBatType(TYPE_oid, getTailType(getArgType(cmb,p,k))) );
}
}
@@ -409,13 +409,13 @@ OPTsliceColumn(Client cntxt, MalBlkPtr n
* when a connection is re-used by different client sessions.
*/
#define BLOCKED 1
-#define REQUIRED 2
+#define PARTITION 2
#define SUPPORTIVE 3
#define EXPORTED 4
#define KEEPLOCAL 5
#ifdef _DEBUG_OPT_CENTIPEDE_
-static char *statusname[6]= {"", "blocked ", "required ", "support ",
"exported ", "keeplocal "};
+static char *statusname[6]= {"", "blocked ", "partition ", "support ",
"exported ", "keeplocal "};
#endif
static void
@@ -469,200 +469,181 @@ OPTbakePlans(Client cntxt, MalBlkPtr mb,
/* Phase 1: determine all variables/instructions indirectly dependent
on a fragmented column */
last = limit;
+ status[0]= PARTITION;
+ for ( j = old[0]->retc; j < old[0]->argc; j++)
+ vars[getArg(old[0],j)]= SUPPORTIVE;
for ( i = 1; i < limit ; i++) {
p = old[i];
if ( p->token == ENDsymbol || i > last) {
- status[i] = REQUIRED;
+ status[i] = PARTITION;
last = i;
} else
if ( getModuleId(p) == sqlRef && (getFunctionId(p) == bindRef
|| getFunctionId(p) == bindidxRef) &&
strcmp(slices->schema, getVarConstant(mb,
getArg(p,2)).val.sval) == 0 &&
strcmp(slices->table, getVarConstant(mb,
getArg(p,3)).val.sval) == 0 ) {
- status[i] = REQUIRED;
+ status[i] = PARTITION;
head[getArg(p,0)] = 1;
tail[getArg(p,0)] = getFunctionId(p) != bindidxRef;
}
- /* blocking instructions are those that require data exchange
or total view */
+ /* blocking instructions are those that require data exchange,
aggregation or total view */
if ( getModuleId(p) == algebraRef && getFunctionId(p) ==
joinRef ) {
/* be aware that supportive subqueries may produce
pivot sets */
/* this means we have to enforce the following */
- if ( vars[getArg(p,1)] == REQUIRED && vars[getArg(p,2)]
!= REQUIRED ) {
- status[i] = SUPPORTIVE;
+ if ( vars[getArg(p,1)] == PARTITION &&
vars[getArg(p,2)] != PARTITION ) {
+ status[i] = PARTITION;
head[getArg(p,0)] = 1;
- }
+ } else
+ if ( vars[getArg(p,1)] == PARTITION &&
vars[getArg(p,2)] == PARTITION )
+ 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] = REQUIRED;
+ status[i] = PARTITION;
} else
if ( getModuleId(p) == algebraRef &&
(getFunctionId(p)==markTRef || getFunctionId(p)== thetauselectRef) ) {
- if (vars[getArg(p,p->retc)] == REQUIRED ) {
+ if (vars[getArg(p,p->retc)] == PARTITION ) {
head[getArg(p,0)] = 1;
- status[i] = REQUIRED;
+ status[i] = PARTITION;
}
} else
if ( getModuleId(p) == algebraRef &&
getFunctionId(p)==markHRef ) {
- if (vars[getArg(p,p->retc)] == REQUIRED )
+ if (vars[getArg(p,p->retc)] == PARTITION )
tail[getArg(p,0)] = 1;
} else
if ( getModuleId(p) == batRef &&
getFunctionId(p)==reverseRef ) {
- if (vars[getArg(p,p->retc)] == REQUIRED ){
+ if (vars[getArg(p,p->retc)] == PARTITION ){
head[getArg(p,0)] = tail[getArg(p,p->retc)];
tail[getArg(p,0)] = head[getArg(p,p->retc)];
- status[i] = REQUIRED;
+ status[i] = PARTITION;
}
} else
+ 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 blocking for partitioned columns, this can
be achieved by propagating the REQUIRED property over variables */
- for( j = p->retc; j < p->argc; j++)
- if (vars[getArg(p,j)] == REQUIRED )
- break;
+ /* this is always blocking */
+ status[i]= BLOCKED;
+ } else
+ if ( (getModuleId(p) == sqlRef && (getFunctionId(p) ==
resultSetRef || getFunctionId(p) == putName("exportValue",11) ) ) ||
+ getModuleId(p) == ioRef )
+ status[i] = BLOCKED;
+ else
+ if ( getModuleId(p) == batcalcRef ){
+ if ( p->argc == 2 /* coercions and unaries */ &&
vars[getArg(p,1)] == PARTITION ) {
+ status[i]= PARTITION;
+ head[getArg(p,0)] = head[getArg(p,1)];
+ tail[getArg(p,0)] = tail[getArg(p,1)];
+ }
+ if ( p->argc == 3 /* binaries */ && (vars[getArg(p,1)]
== PARTITION || vars[getArg(p,2)] == PARTITION)) {
+ status[i]= PARTITION;
+ head[getArg(p,0)] = head[getArg(p,1)];
+ tail[getArg(p,0)] = tail[getArg(p,1)];
+ }
}
+ for( j = p->retc; j < p->argc; j++)
+ if (vars[getArg(p,j)] == BLOCKED )
+ break;
+ if ( j != p->argc && p->argc - p->retc > 0 )
+ status[i]= BLOCKED;
- if( status[i] == BLOCKED){
- for ( j= 0; j< p->retc; j++)
- vars[getArg(p,j)] = BLOCKED;
- } else {
- /* one blocking argument blocks the instruction */
- for( j = p->retc; j < p->argc; j++)
- if (vars[getArg(p,j)] == BLOCKED )
- break;
- if ( j != p->argc )
- status[i]= BLOCKED;
+ for ( j= 0; j< p->retc; j++)
+ if (vars[getArg(p,j)] == 0)
+ vars[getArg(p,j)] = status[i];
- /* one required then instruction is needed */
- k = 0;
- for( j = p->retc; j < p->argc; j++)
- if (vars[getArg(p,j)] == REQUIRED )
- k++;
- if (k && k == p->argc-p->retc && status[i] != BLOCKED )
- status[i]= REQUIRED;
- else
- if ( k && status[i] != BLOCKED )
- status[i] = REQUIRED;
+ if ( status[i] == PARTITION)
+ for( j = p->retc; j < p->argc; j++)
+ if (vars[getArg(p,j)] == 0)
+ vars[getArg(p,j)] = SUPPORTIVE;
+ }
+#ifdef _DEBUG_OPT_CENTIPEDE_
+ mnstr_printf(cntxt->fdout,"\n#phase 1\n");
+ for( i= 0; i< limit; i++)
+ if (status[i] ) {
+ int x = old[i]->argv[0];
+ mnstr_printf(cntxt->fdout,"%s %d %s %c%c
",statusname[status[i]], x, statusname[vars[x]], head[x]+'0', tail[x]+'0');
+ if( old[i])
+ printInstruction(cntxt->fdout,
mb,0,old[i],LIST_MAL_STMT);
+ }
+#endif
- for ( j= 0; j< p->retc; j++)
- if ( vars[getArg(p,j)] == 0)
- vars[getArg(p,j)] = status[i];
- }
- }
-
- /* Phase 2: extend the set of supportive instructions.
+ /* Phase 2: extend the set with supportive instructions.
we have to avoid common ancestor dependency on partitioned variables
*/
for ( i = limit -1; i >= 0 ; i--)
- if ( status[i] != BLOCKED ){
+ if ( status[i] == 0 ){
p = old[i];
for( j = 0; j < p->argc; j++)
- if ( vars[getArg(p,j)] == SUPPORTIVE || vars[getArg(p,j)] ==
REQUIRED)
+ if ( vars[getArg(p,j)] == BLOCKED)
break;
- if( j< p->argc) {
- for ( j= p->retc; j< p->argc; j++)
- if ( vars[getArg(p,j)] != REQUIRED)
+ if ( j == p->argc ) {
+ /* does it produce partitioned or support variables */
+ for( j = 0; j < p->retc; j++)
+ if ( vars[getArg(p,j)] == SUPPORTIVE ||
vars[getArg(p,j)] == PARTITION)
+ break;
+ } else {
+ status[i] = BLOCKED;
+ for( j = 0; j < p->retc; j++)
+ if ( vars[getArg(p,j)] == 0)
+ vars[getArg(p,j)] = BLOCKED;
+ }
+
+ if( j< p->retc && status[i] != BLOCKED ){
+ for ( j= 0; j< p->argc; j++)
+ if ( vars[getArg(p,j)] == 0)
vars[getArg(p,j)] = SUPPORTIVE;
- if ( status[i] != REQUIRED)
- status[i] = SUPPORTIVE;
- }
- /* be careful in adding operations that require exchange or
full view */
- if ( (getModuleId(p) == sqlRef && (getFunctionId(p) ==
resultSetRef || getFunctionId(p) == putName("exportValue",11) ) ) ||
- getModuleId(p) == ioRef )
+ status[i] = SUPPORTIVE;
+ } else {
+ for ( j= 0; j< p->retc; j++)
+ if ( vars[getArg(p,j)] == 0)
+ vars[getArg(p,j)] = BLOCKED;
status[i] = BLOCKED;
- if (getModuleId(p) == algebraRef && getFunctionId(p) == joinRef
) {
- /* if we produce an oid/oid based on the target, then
we need an exchange */
- if ( head[getArg(p,1)] && tail[getArg(p,2)] )
- status[i] = BLOCKED;
- }
- if ( (getModuleId(p) == groupRef && (getFunctionId(p) ==
doneRef || getFunctionId(p) == newRef ||getFunctionId(p) == deriveRef) ) ||
- getModuleId(p) == pqueueRef || getModuleId(p)
== aggrRef || getModuleId(p) == ioRef ) {
- /* this is blocking for partitioned columns, this can
be achieved by propagating the REQUIRED property over variables */
- for( j = p->retc; j < p->argc; j++)
- if (vars[getArg(p,j)] == REQUIRED )
- break;
- if ( j != p->argc)
- status[i] = BLOCKED;
}
}
- /* Phase 3: check all instructions for blocked arguments.
- The variable may not be produced by a blocked instruction
- and all arguments must be required/supportive/constant
- (not blocked)
- */
- for ( i = 0; i < limit; i++)
- if( status[i] != 0){
- p = old[i];
- for( j = 0; j < p->argc; j++)
- if (vars[getArg(p,j)] == BLOCKED)
- break;
- if( j != p->argc || status[i] == BLOCKED) {
- status[i] = BLOCKED;
- for( j = 0; j < p->retc; j++)
- vars[getArg(p,j)] = BLOCKED;
-
- }
- for( j = 0; j < p->argc; j++)
- if (vars[getArg(p,j)] == REQUIRED)
- break;
- if( j != p->argc && status[i] != BLOCKED )
- for( j = 0; j < p->retc; j++)
- vars[getArg(p,j)] = REQUIRED;
-
- }
#ifdef _DEBUG_OPT_CENTIPEDE_
- mnstr_printf(cntxt->fdout,"\n#phase 3\n");
+ mnstr_printf(cntxt->fdout,"\n#phase 2\n");
for( i= 0; i< limit; i++)
if (status[i] ) {
int x = old[i]->argv[0];
- mnstr_printf(cntxt->fdout,"%s %c%c ",statusname[status[i]],
head[x]+'0', tail[x]+'0');
+ mnstr_printf(cntxt->fdout,"%s %s %c%c ",statusname[status[i]],
statusname[vars[x]], head[x]+'0', tail[x]+'0');
if( old[i])
printInstruction(cntxt->fdout,
mb,0,old[i],LIST_MAL_STMT);
}
#endif
/* Phase 4: determine all variables to be exported
- this is limited to all variables produced but not consumed
- and only those that have an oid of the target
+ this is limited to all variables produced and consumed by a blocked
instruction
*/
ret= newInstruction(plan,ASSIGNsymbol);
ret->barrier = RETURNsymbol;
ret->argc= ret->retc = 0;
- for ( i = limit-1; i >= 0 ; i--)
- if ( status[i] == REQUIRED || status[i] == SUPPORTIVE ){
+ for ( i = 0; i< limit; i++)
+ if ( status[i] == BLOCKED )
+ {
p = old[i];
- for ( j=0; j < p->retc; j++)
- if ( vars[getArg(p,j)] == REQUIRED &&
isaBatType(getArgType(plan,p,j)) )
- vars[getArg(p,j)] = EXPORTED;
- else
- if ( vars[getArg(p,j)] != 0 && vars[getArg(p,j)] != KEEPLOCAL
&& vars[getArg(p,j)] != SUPPORTIVE && isaBatType(getArgType(plan,p,j)) ) {
+ 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 */
if ( (getModuleId(p) == algebraRef && (getFunctionId(p)
== markHRef || getFunctionId(p) == markTRef)) ||
(getModuleId(p) == batRef && (getFunctionId(p)
== reverseRef || getFunctionId(p) == mirrorRef )) )
continue;
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
