Changeset: 16a1c6090fa0 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=16a1c6090fa0
Modified Files:
sql/backends/monet5/datacell/50_datacell.sql
sql/backends/monet5/datacell/Tests/scenario06.sql
sql/backends/monet5/datacell/datacell.mx
sql/backends/monet5/datacell/opt_datacell.mx
sql/backends/monet5/datacell/petrinet.mx
Branch: default
Log Message:
Add procedural continuous queries
Any SQL procedure can be compiled for continous execution.
Scenario06 shows how this can be used to both pass events
to an emitter and to keep a statistics log in an ordinary table,
i.e. mixing baskets and tables in a multi-query setting.
Overall performance is as predicted, i.e. high.
diffs (276 lines):
diff --git a/sql/backends/monet5/datacell/50_datacell.sql
b/sql/backends/monet5/datacell/50_datacell.sql
--- a/sql/backends/monet5/datacell/50_datacell.sql
+++ b/sql/backends/monet5/datacell/50_datacell.sql
@@ -41,6 +41,9 @@
create procedure datacell.query(proc string, def string)
external name datacell.query;
+create procedure datacell.query(proc string)
+ external name datacell.query;
+
create procedure datacell.remove (obj string)
external name datacell.remove;
diff --git a/sql/backends/monet5/datacell/Tests/scenario06.sql
b/sql/backends/monet5/datacell/Tests/scenario06.sql
new file mode 100644
--- /dev/null
+++ b/sql/backends/monet5/datacell/Tests/scenario06.sql
@@ -0,0 +1,41 @@
+-- Scenario to exercise the datacell implementation
+-- using a single receptor and emitter
+-- use a compound query to deliver the events to both emitter
+-- and store aggregated information in a log
+
+create schema datacell;
+set optimizer='datacell_pipe';
+
+create table datacell.barrelin(
+ id integer,
+ tag timestamp,
+ payload integer
+);
+create table datacell.barrelout (like datacell.barrelin);
+create table datacell.barrellog( ts timestamp, cnt integer);
+
+call datacell.receptor('datacell.barrelin','localhost',50500);
+
+call datacell.emitter('datacell.barrelout','localhost',50600);
+
+create procedure datacell.splitter()
+begin
+ insert into datacell.barrelout select * from datacell.barrelin;
+ insert into datacell.barrellog values (now(), (select count(*) from
datacell.barrelin));
+end;
+call datacell.query('datacell.splitter');
+
+call datacell.resume();
+call datacell.dump();
+
+-- externally, activate the sensor
+--sensor --host=localhost --port=50500 --events=100 --columns=3 --delay=1
+-- externally, activate the actuator server to listen
+-- actuator
+
+
+-- wrapup
+call datacell.postlude();
+drop table datacell.barrelin;
+drop table datacell.barrelout;
+
diff --git a/sql/backends/monet5/datacell/datacell.mx
b/sql/backends/monet5/datacell/datacell.mx
--- a/sql/backends/monet5/datacell/datacell.mx
+++ b/sql/backends/monet5/datacell/datacell.mx
@@ -60,6 +60,10 @@
address DCquery
comment "Add a new continuous query.";
+pattern query(name:str):void
+address DCquery
+comment "Add a new continuous query procedure.";
+
pattern pause()
address DCpauseScheduler
comment "(Re)start the petrinet scheduler.";
@@ -148,6 +152,7 @@
#include "datacell.h"
#include "opt_datacell.h"
#include "sql_optimizer.h"
+#include "sql_gencode.h"
#ifdef WIN32
#include "winsock2.h"
@@ -168,6 +173,33 @@
* grab all tables in the datacell schema and turn them into baskets.
* The same for all procedures, turn them into continuous queries.
*/
+static str
+DCprocedureStmt(Client cntxt, MalBlkPtr mb, str schema, str nme)
+{
+ mvc *m = NULL;
+ str msg = getContext(cntxt, mb, &m, NULL);
+ sql_schema *s;
+ backend *be;
+ node *o;
+ sql_func *f;
+ sql_trans *tr;
+
+ if ( msg)
+ return msg;
+ s = mvc_bind_schema(m, schema);
+ if (s == NULL)
+ throw(SQL, "datacell.query", "Schema missing");
+ tr = m->session->tr;
+ for (o = s->funcs.set->h; o; o = o->next) {
+ f = o->data;
+ if ( strcmp(f->base.name, nme) == 0 ){
+ be = (void *) backend_create(m, cntxt);
+ backend_create_func(be, f);
+ break;
+ }
+ }
+ return MAL_SUCCEED;
+}
str
DCprelude(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
@@ -334,34 +366,62 @@
DCquery(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
str nme = *(str*) getArgReference(stk,pci,1);
- str *def = (str*) getArgReference(stk,pci,2);
- Symbol s;
+ str def;
+ Symbol s = NULL;
+ MalBlkPtr qry;
str msg;
InstrPtr p;
Module scope;
lng clk = GDKusec();
- char buf[BUFSIZ], *lsch, *ltbl;
+ char buf[BUFSIZ], *lsch, *lnme;
- BSKTelements( nme, buf, &lsch, <bl);
+ BSKTelements( nme, buf, &lsch, &lnme);
BSKTtolower(lsch);
- BSKTtolower(ltbl);
+ BSKTtolower(lnme);
(void) mb;
/* check if the argument denotes a procedure name */
/* if so, get its definition to be compiled */
- msg = SQLstatementIntern(cntxt, def, ltbl, 0, 0);
- if ( msg )
- return msg;
+ /* check existing of the pre-compiled function */
scope = findModule(cntxt->nspace,putName(lsch,strlen(lsch)));
- s = newFunction(putName(lsch,strlen(lsch)), putName(ltbl,
strlen(ltbl)),FUNCTIONsymbol);
+ if ( scope)
+ s = findSymbolInModule(scope,putName(lnme,strlen(lnme)));
+ /* is it defined in module user */
+ if ( s == NULL)
+ s =
findSymbolInModule(cntxt->nspace,putName(lnme,strlen(lnme)));
+
+ if (s == NULL){
+ if ( pci->argc == 3 ) {
+ def = *(str*) getArgReference(stk,pci,2);
+ msg = SQLstatementIntern(cntxt, &def, lnme, 0, 0);
+ if ( msg )
+ return msg;
+ qry =cntxt->curprg->def;
+ } else {
+ /* get definition from catalog */
+ msg =DCprocedureStmt(cntxt, mb, lsch, lnme);
+ if ( msg)
+ return msg;
+ s =
findSymbolInModule(cntxt->nspace,putName(lnme,strlen(lnme)));
+ if ( s == NULL)
+ throw(SQL,"datacell.query","Definition
missing");
+ qry= s->def;
+ }
+ } else
+ if ( pci->argc == 3)
+ throw(SQL,"datacell.query","Query already defined");
+ else qry = s->def;
+
+ scope = findModule(cntxt->nspace,putName(lsch,strlen(lsch)));
+ s = newFunction(putName(lsch,strlen(lsch)), putName(lnme,
strlen(lnme)),FUNCTIONsymbol);
if ( s == NULL)
throw(SQL,"datacell.query","Procedure code does not exist");
freeMalBlk(s->def);
- s->def = copyMalBlk(cntxt->curprg->def);
+ s->def = copyMalBlk(qry);
p= getInstrPtr(s->def,0);
setModuleId(p, putName(lsch,strlen(lsch)));
- setFunctionId(p, putName(ltbl,strlen(ltbl)));
+ setFunctionId(p, putName(lnme,strlen(lnme)));
insertSymbol(scope,s);
/* printFunction(cntxt->fdout, s->def, 0, LIST_MAL_STMT);*/
/* optimize the code and register at scheduler */
diff --git a/sql/backends/monet5/datacell/opt_datacell.mx
b/sql/backends/monet5/datacell/opt_datacell.mx
--- a/sql/backends/monet5/datacell/opt_datacell.mx
+++ b/sql/backends/monet5/datacell/opt_datacell.mx
@@ -58,7 +58,7 @@
int
OPTdatacellImplementation(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci)
{
- int actions = 0, fnd;
+ int actions = 0, fnd, mvc= 0;
int bskt, i, j, k, limit, vlimit, slimit;
InstrPtr r, p, *old;
str col;
@@ -141,6 +141,9 @@
freeInstruction(p);
continue;
}
+
+ if ( getModuleId(p) == sqlRef && getFunctionId(p) == mvcRef)
+ mvc= getArg(p,0);
if ( getModuleId(p) == sqlRef && (getFunctionId(p) == bindRef
||getFunctionId(p) == binddbatRef)){
snprintf(buf,BUFSIZ,"%s.%s", getVarConstant(mb,
getArg(p,2)).val.sval, getVarConstant(mb, getArg(p,3)).val.sval);
col = getVarConstant(mb, getArg(p,4)).val.sval;
@@ -156,6 +159,9 @@
tables[m++]= GDKstrdup(buf);
actions = 1;
}
+ } else {
+ pushInstruction(mb,p);
+ continue;
}
fnd = 0;
@@ -228,6 +234,9 @@
getArg(p, 1) = getArg(p, 5);
p->argc = 2;
}
+ } else {
+ getArg(p,1) = mvc;
+ mvc = getArg(p,0);
}
}
pushInstruction(mb, p);
@@ -235,6 +244,7 @@
(void)stk;
(void)pci;
+ printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT | LIST_MAPI);
if (actions)
{
addPipeDefinition("datacell_pipe",
diff --git a/sql/backends/monet5/datacell/petrinet.mx
b/sql/backends/monet5/datacell/petrinet.mx
--- a/sql/backends/monet5/datacell/petrinet.mx
+++ b/sql/backends/monet5/datacell/petrinet.mx
@@ -29,6 +29,11 @@
@end example
@mal
module petrinet;
+pattern register(nme:str):void
+address PNregister
+comment "Add a continous query to the Petri net. It will analyse
+the MAL block to determine the input/output dependencies. ";
+
pattern register(nme:str, def:str):void
address PNregister
comment "Add a continous query to the Petri net. It will analyse
@@ -192,7 +197,6 @@
Symbol s = 0;
int *ret = (int *)getArgReference(stk, pci, 0);
str nme = *(str *)getArgReference(stk, pci, 1);
- str def = *(str*) getArgReference(stk,pci,2);
int i;
char buf[BUFSIZ], *modnme, *fcnnme;
@@ -220,8 +224,11 @@
if ( i != pnettop)
throw(MAL, "petrinet.register", "Duplicate definition
of transition");
pnet[pnettop].name = GDKstrdup(nme);
- pnet[pnettop].def = GDKstrdup(def);;
- pnet[pnettop].mb = s->def;
+ if ( pci->argc == 3)
+ pnet[pnettop].def = GDKstrdup(*(str*)
getArgReference(stk,pci,2));
+ else
+ pnet[pnettop].def = GDKstrdup("");
+
pnet[pnettop].status = PNwaiting;
pnet[pnettop].cycles = 0;
/* all the rest is zero */
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
