Changeset: c3532ceb7fb6 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=c3532ceb7fb6
Modified Files:
sql/backends/monet5/datacell/Tests/emili.sql
sql/backends/monet5/datacell/basket.c
sql/backends/monet5/datacell/emitter.c
sql/backends/monet5/datacell/opt_datacell.c
sql/backends/monet5/datacell/receptor.c
sql/backends/monet5/datacell/sensor.c
Branch: default
Log Message:
Enable transactions on base table.
diffs (297 lines):
diff --git a/sql/backends/monet5/datacell/Tests/emili.sql
b/sql/backends/monet5/datacell/Tests/emili.sql
--- a/sql/backends/monet5/datacell/Tests/emili.sql
+++ b/sql/backends/monet5/datacell/Tests/emili.sql
@@ -1,7 +1,7 @@
-- the raw input stream received at receptor
CREATE TABLE datacell.istream(
ip INET,
- location VARCHAR(20),
+ location VARCHAR(5),
kind VARCHAR(50),
value DOUBLE
);
@@ -10,7 +10,8 @@ CALL datacell.receptor('datacell.istream
-- collect the sensors in certain areas
CREATE TABLE datacell.area(
ip INET,
- location varchar(20)
+ location varchar(5),
+ primary key(ip)
);
-- tag the events with their arrival time
@@ -22,23 +23,36 @@ CALL datacell.basket('datacell.sensors')
-- administer the fire state in locations
CREATE TABLE datacell.states(
- location varchar(20),
+ location varchar(5),
time timestamp,
status varchar(20) default 'normal'
);
--- warden mailbox
+-- warden emitter mailbox
CREATE TABLE datacell.warden(
- location varchar(20),
+ location varchar(5),
message varchar(20)
);
CALL datacell.emitter('datacell.warden','localhost',50600);
+-- observations made by the warden
+CREATE TABLE datacell.observations(
+ location varchar(5),
+ message varchar(20)
+);
+CALL datacell.receptor('datacell.observations','localhost',50501);
+
-- enrich at the arrival time of each stream event
CREATE PROCEDURE datacell.enrich()
BEGIN
- INSERT INTO datacell.sensors(ip, location,kind,value) SELECT * FROM
datacell.istream;
- INSERT INTO datacell.area SELECT ip, location FROM datacell.istream;
+ DECLARE cnt INTEGER;
+ SET cnt = (SELECT count(*) FROM datacell.area ) ;
+ INSERT INTO datacell.sensors(ip, location, kind,value)
+ SELECT ip, substring(location,0,3), kind, value FROM
datacell.istream;
+ IF cnt = 0
+ THEN
+ INSERT INTO datacell.area SELECT ip, substring(location,0,3)
FROM datacell.istream;
+ END IF;
END;
CALL datacell.query('datacell.enrich');
@@ -56,7 +70,7 @@ BEGIN
INSERT INTO datacell.hotsensors
SELECT ip, time, value
FROM datacell.sensors
- WHERE kind LIKE 'temperature' AND value > 27;
+ WHERE kind LIKE 'temperature' AND value > 25;
END;
CALL datacell.query('datacell.hot');
@@ -91,11 +105,18 @@ BEGIN
INSERT into datacell.states
SELECT S.location, H.time, 'confirmed'
FROM datacell.area A, datacell.states S, datacell.area B,
datacell.hotsensors2 H
- WHERE S.status ='unconfirmed' AND A.ip <> H.ip AND B.ip = H.ip AND A.ip
<> B.ip AND S.location = A.location
- AND (H.time - S.time) > '3' as minutes;
+ WHERE S.status ='unconfirmed' AND A.ip <> H.ip AND B.ip = H.ip AND A.ip
<> B.ip AND S.location = A.location;
END;
CALL datacell.query('datacell.firespotted');
+-- Warden confirms location status
+CREATE PROCEDURE datacell.observation()
+BEGIN
+ DELETE FROM datacell.states WHERE location IN (SELECT location FROM
datacell.observations);
+ INSERT INTO datacell.states SELECT O.location, now(), O.message FROM
datacell.observations O;
+END;
+
CALL datacell.resume();
-SELECT * FROM datacell.receptors(); SELECT * FROM datacell.emitters(); SELECT
* FROM datacell.queries(); SELECT * FROM datacell.baskets();
+SELECT * FROM datacell.receptors(); SELECT * FROM datacell.emitters(); SELECT
* FROM datacell.queries(); SELECT * FROM datacell.baskets(); SELECT * FROM
datacell.area; SELECT * FROM datacell.states;
+
diff --git a/sql/backends/monet5/datacell/basket.c
b/sql/backends/monet5/datacell/basket.c
--- a/sql/backends/monet5/datacell/basket.c
+++ b/sql/backends/monet5/datacell/basket.c
@@ -119,6 +119,7 @@ BSKTlocate(str tbl)
return i;
/* try prefixing it with datacell */
snprintf(buf,BUFSIZ,"datacell.%s",tbl);
+ BSKTtolower(buf);
for (i = 1; i < bsktTop; i++)
if (baskets[i].name && strcmp(buf, baskets[i].name) == 0)
return i;
diff --git a/sql/backends/monet5/datacell/emitter.c
b/sql/backends/monet5/datacell/emitter.c
--- a/sql/backends/monet5/datacell/emitter.c
+++ b/sql/backends/monet5/datacell/emitter.c
@@ -66,6 +66,7 @@ EMfind(str nme)
if (strcmp(nme, r->name) == 0)
return r;
snprintf(buf,BUFSIZ,"datacell.%s",nme);
+ BSKTtolower(buf);
for (r = emAnchor; r; r = r->nxt)
if (strcmp(buf, r->name) == 0)
return r;
@@ -493,7 +494,8 @@ EMtable(int *nameId, int *hostId, int *p
goto wrapup;
BATseqbase(status, 0);
- for (; em; em = em->nxt){
+ for (; em; em = em->nxt)
+ if ( em->table.format[1].c[0]){
BUNappend(name, em->name, FALSE);
BUNappend(host, em->host, FALSE);
BUNappend(port, &em->port, FALSE);
diff --git a/sql/backends/monet5/datacell/opt_datacell.c
b/sql/backends/monet5/datacell/opt_datacell.c
--- a/sql/backends/monet5/datacell/opt_datacell.c
+++ b/sql/backends/monet5/datacell/opt_datacell.c
@@ -72,7 +72,7 @@ OPTdatacellImplementation(Client cntxt,
removeDataflow(old, limit);
pushInstruction(mb, old[0]);
- //newFcnCall(mb, sqlRef, putName("transaction", 11));
+ newFcnCall(mb, sqlRef, putName("transaction", 11));
for (i = 1; i < limit; i++)
if (old[i]) {
p = old[i];
@@ -116,7 +116,7 @@ OPTdatacellImplementation(Client cntxt,
getArg(r, 0) = j;
r->barrier = EXITsymbol;
- //(void) newFcnCall(mb, sqlRef, commitRef);
+ (void) newFcnCall(mb, sqlRef, commitRef);
break;
}
if (getModuleId(p) == sqlRef && getFunctionId(p) ==
putName("affectedRows", 12)) {
diff --git a/sql/backends/monet5/datacell/receptor.c
b/sql/backends/monet5/datacell/receptor.c
--- a/sql/backends/monet5/datacell/receptor.c
+++ b/sql/backends/monet5/datacell/receptor.c
@@ -82,6 +82,7 @@ RCfind(str nme)
if (strcmp(nme, r->name) == 0)
return r;
snprintf(buf,BUFSIZ,"datacell.%s",nme);
+ BSKTtolower(buf);
for (r = rcAnchor; r; r = r->nxt)
if (strcmp(buf, r->name) == 0)
return r;
@@ -103,6 +104,9 @@ RCreceptorStartInternal(int *ret, str *t
if (RCfind(*tbl))
throw(MAL, "receptor.new", "Duplicate receptor");
+ idx = BSKTlocate(*tbl);
+ if (idx == 0) /* should not happen */
+ throw(MAL, "receptor.new", "Basket '%s' not found",*tbl);
for (rc = rcAnchor; rc; rc = rc->nxt)
if (rc->port == *port)
throw(MAL, "receptor.new", "Port already in use");
@@ -124,9 +128,7 @@ RCreceptorStartInternal(int *ret, str *t
rc->protocol = protocol;
rc->lastseen = *timestamp_nil;
- rc->bskt = idx = BSKTlocate(*tbl);
- if (idx == 0) /* should not happen */
- throw(MAL, "receptor.new", "Basket not found");
+ rc->bskt = idx;
len = BSKTmemberCount(*tbl);
fmt = rc->table.format = GDKzalloc(sizeof(Column) * len);
@@ -744,7 +746,8 @@ RCtable(int *nameId, int *hostId, int *p
goto wrapup;
BATseqbase(status, 0);
- for (; rc; rc = rc->nxt){
+ for (; rc; rc = rc->nxt)
+ if ( rc->table.format[1].c[0]){
BUNappend(name, rc->name, FALSE);
BUNappend(host, rc->host, FALSE);
BUNappend(port, &rc->port, FALSE);
diff --git a/sql/backends/monet5/datacell/sensor.c
b/sql/backends/monet5/datacell/sensor.c
--- a/sql/backends/monet5/datacell/sensor.c
+++ b/sql/backends/monet5/datacell/sensor.c
@@ -126,8 +126,9 @@ static int timecolumn = -1; /* use firs
static char *host = "localhost";
static int port = 50500;
static int trace = 0;
+static int replay = 0;
static char *sensor = "X";
-static char *datafile;
+static char *datafile= 0;
static int server = 0;
@@ -146,7 +147,8 @@ usage(void)
mnstr_printf(SEout, "--timestamp, default=on\n");
mnstr_printf(SEout, "--columns=<number>, default=1\n");
mnstr_printf(SEout, "--events=<events length>, (-1=forever,>0),
default=1\n");
- mnstr_printf(SEout, "--file=<data file>\n");
+ mnstr_printf(SEout, "--file=<data file> \n");
+ mnstr_printf(SEout, "--replay use file or standard input\n");
mnstr_printf(SEout, "--time=<column> where to find the exact time\n");
mnstr_printf(SEout, "--batch=<batchsize> , default=1\n");
mnstr_printf(SEout, "--delay=<ticks> interbatch delay in ms,
default=1\n");
@@ -198,7 +200,7 @@ int main(int argc, char **argv)
char hostname[1024];
Sensor se = NULL;
static SOCKET sockfd;
- static struct option long_options[17] = {
+ static struct option long_options[18] = {
{ "increment", 0, 0, 'i' },
{ "batch", 1, 0, 'b' },
{ "columns", 1, 0, 'c' },
@@ -209,6 +211,7 @@ int main(int argc, char **argv)
{ "events", 1, 0, 'e' },
{ "sensor", 1, 0, 's' },
{ "server", 0, 0, 's' },
+ { "replay", 0, 0, 'r' },
{ "delay", 1, 0, 'd' },
{ "file", 1, 0, 'f' },
{ "host", 1, 0, 'h' },
@@ -279,7 +282,7 @@ int main(int argc, char **argv)
}
break;
case 'f':
- datafile = optarg;
+ datafile = optarg && *optarg? optarg:0;
break;
case 'e':
if (strcmp(long_options[option_index].name, "events")
== 0) {
@@ -297,6 +300,9 @@ int main(int argc, char **argv)
exit(0);
}
break;
+ case 'r':
+ replay= 1;
+ break;
case 's':
if (strcmp(long_options[option_index].name, "sensor")
== 0) {
sensor = optarg;
@@ -372,6 +378,7 @@ int main(int argc, char **argv)
mnstr_printf(SEout, "--time=%d\n", timecolumn);
mnstr_printf(SEout, "--events=%d\n", events);
mnstr_printf(SEout, "--batch=%d\n", batchsize);
+ mnstr_printf(SEout, "--replay=%d\n", replay);
mnstr_printf(SEout, "--delay=%d\n", delay);
mnstr_printf(SEout, "--protocol %s\n", protocolname[protocol]);
mnstr_printf(SEout, "--trace=%d\n", trace);
@@ -612,18 +619,22 @@ produceDataStream(Sensor se)
int i, snr;
time_t lasttime;
- /* read a events of messages from a file.
+ /* read a events of messages from a file or standard input.
It is processed multiple times.
The header is the delay imposed */
snr = 0;
do {
- fd = fopen(datafile, "r");
- if (fd == NULL) {
- mnstr_printf(SEout, "Could not open file '%s'\n",
datafile);
- close_stream(se->toServer);
- se->toServer = NULL;
- return;
+ if ( datafile == 0){
+ fd = stdin;
+ } else {
+ fd = fopen(datafile, "r");
+ if (fd == NULL) {
+ mnstr_printf(SEout, "Could not open file
'%s'\n", datafile);
+ close_stream(se->toServer);
+ se->toServer = NULL;
+ return;
+ }
}
/* read the event requests and sent when the becomes */
@@ -680,7 +691,7 @@ produceDataStream(Sensor se)
static void
produceServerStream(Sensor se)
{
- if (datafile)
+ if (replay)
produceDataStream(se);
else
produceStream(se);
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
