Changeset: abf669bad837 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=abf669bad837
Added Files:
sql/backends/monet5/iot/Tests/iot04.sql
sql/backends/monet5/iot/Tests/iot14.sql
sql/backends/monet5/iot/Tests/logger.sql
Modified Files:
sql/backends/monet5/iot/50_iot.sql
sql/backends/monet5/iot/Tests/All
sql/backends/monet5/iot/Tests/iot12.sql
sql/backends/monet5/iot/Tests/iot13.sql
sql/backends/monet5/iot/basket.c
sql/backends/monet5/iot/basket.h
sql/backends/monet5/iot/basket.mal
sql/backends/monet5/iot/iot.mal
sql/backends/monet5/iot/petrinet.c
sql/backends/monet5/iot/petrinet.h
sql/backends/monet5/iot/petrinet.mal
Branch: iot
Log Message:
Add heartbeat and cycles to Petrinet
Queries can have their own heartbeat, which is sufficient to run the query.
The cycles places an upper limit on the number of successful
invocations of the query.
diffs (truncated from 559 to 300 lines):
diff --git a/sql/backends/monet5/iot/50_iot.sql
b/sql/backends/monet5/iot/50_iot.sql
--- a/sql/backends/monet5/iot/50_iot.sql
+++ b/sql/backends/monet5/iot/50_iot.sql
@@ -80,6 +80,9 @@ create procedure iot.tumble("schema" str
create procedure iot.window("schema" string, "table" string, elem integer)
external name iot.window;
+create procedure iot.cycles("schema" string, "query" string, elem integer)
+ external name iot.cycles;
+
-- Inspection tables
create procedure iot."explain"("schema" string, "table" string)
external name iot."explain";
diff --git a/sql/backends/monet5/iot/Tests/All
b/sql/backends/monet5/iot/Tests/All
--- a/sql/backends/monet5/iot/Tests/All
+++ b/sql/backends/monet5/iot/Tests/All
@@ -1,10 +1,12 @@
iot00
iot02
iot03
+iot04
iot06
iot10
iot12
iot13
+iot14
#iot15
receptor00
receptor01
@@ -12,3 +14,4 @@ receptor01
webtest
inputoutput
export00
+logger
diff --git a/sql/backends/monet5/iot/Tests/iot04.sql
b/sql/backends/monet5/iot/Tests/iot04.sql
new file mode 100644
--- /dev/null
+++ b/sql/backends/monet5/iot/Tests/iot04.sql
@@ -0,0 +1,41 @@
+-- A simple continuous query using the cycles bound.
+set schema iot;
+set optimizer='iot_pipe';
+
+create stream table stmp (t timestamp, sensor integer, val decimal(8,2)) ;
+create table result(like stmp);
+
+create procedure cq00()
+begin
+ insert into result select min(t), count(*), avg(val) from stmp;
+end;
+
+call iot.query('iot','cq00');
+--call iot.query('insert into iot.result select min(t), count(*), avg(val)
from iot.stmp;');
+
+--select * from iot.baskets();
+--select * from iot.queries();
+--select * from iot.inputs();
+--select * from iot.outputs();
+
+-- stop all continuous queries and wait for it
+call iot.pause();
+
+insert into stmp values('2005-09-23 12:34:26.736',1,12.34);
+select * from stmp;
+
+-- let the cq run only wanse
+call iot.cycles('iot','cq00',2);
+call iot.resume();
+-- wait for 1 cycle in the scheduler
+
+select 'RESULT';
+select * from result;
+
+--select * from iot.baskets();
+--select * from iot.queries();
+select * from iot.errors();
+call iot.stop();
+drop procedure cq00;
+drop table stmp;
+drop table result;
diff --git a/sql/backends/monet5/iot/Tests/iot12.sql
b/sql/backends/monet5/iot/Tests/iot12.sql
--- a/sql/backends/monet5/iot/Tests/iot12.sql
+++ b/sql/backends/monet5/iot/Tests/iot12.sql
@@ -22,8 +22,8 @@ call iot.pause();
--select * from iot.queries();
select * from clocks;
+call iot.cycles('iot','clk1',5);
call iot.resume();
-call iot.wait(5);
select * from clocks;
call iot.stop();
diff --git a/sql/backends/monet5/iot/Tests/iot13.sql
b/sql/backends/monet5/iot/Tests/iot13.sql
--- a/sql/backends/monet5/iot/Tests/iot13.sql
+++ b/sql/backends/monet5/iot/Tests/iot13.sql
@@ -3,28 +3,20 @@ set schema iot;
set optimizer='iot_pipe';
create stream table tmp13 (t timestamp, sensor integer, val decimal(8,2)) ;
-create table agenda(like tmp13);
+create table agenda13(t timestamp, cnt integer, msg string);
--- Queries can fire based on the actual state of the streams and heartbeat
+-- Queries can fire based on bohth the actual state of the streams and
heartbeat
call iot.window('iot','tmp13',4);
--- at least one tuple in basket
-call iot.window('iot','cq13b','iot','tmp13',1);
-- every 5 seconds inspect the basket regardless filling
-call iot.heartbeat('iot','cq13a',5000);
+call iot.heartbeat('iot','tmp13',5000);
-create procedure cq13a()
+create procedure cq13()
begin
- insert into agenda select count(*), 'full batch' from tmp13;
+ insert into agenda13 select count(*), 'full batch' from tmp13;
end;
-create procedure cq13b()
-begin
- insert into agenda select count(*), 'partial batch' from tmp13;
-end;
-
-call iot.query('iot','cq13a');
-call iot.query('iot','cq13b');
+call iot.query('iot','cq13');
call iot.pause();
insert into tmp13 values('2005-09-23 12:34:26.736',1,12.34);
@@ -36,11 +28,10 @@ call iot.resume();
call iot.wait(1);
select 'RESULT';
-select * from agenda;
+select * from agenda13;
select * from iot.errors();
call iot.stop();
-drop procedure cq13a;
-drop procedure cq13b;
+drop procedure cq13;
drop table tmp13;
-drop table agenda;
+drop table agenda13;
diff --git a/sql/backends/monet5/iot/Tests/iot14.sql
b/sql/backends/monet5/iot/Tests/iot14.sql
new file mode 100644
--- /dev/null
+++ b/sql/backends/monet5/iot/Tests/iot14.sql
@@ -0,0 +1,38 @@
+-- removal from a stream
+set schema iot;
+set optimizer='iot_pipe';
+
+create stream table sdel (t timestamp, sensor integer, val decimal(8,2)) ;
+
+insert into sdel values('2005-09-23 12:34:26.736',1,12.34);
+insert into sdel values('2005-09-23 12:34:26.736',2,12.34);
+insert into sdel values('2005-09-23 12:34:26.736',3,12.34);
+insert into sdel values('2005-09-23 12:34:26.736',4,12.34);
+select * from sdel;
+
+delete from sdel where sensor = 2;
+select * from sdel;
+
+-- don't remove tuples automatically
+call iot.tumble('iot','sdel',0);
+
+create procedure sdel00()
+begin
+ delete from sdel where sensor = 3;
+end;
+
+call iot.query('iot','sdel00');
+call iot.show('iot','sdel00');
+call iot.pause();
+insert into sdel values('2005-09-23 12:34:26.736',1,12.34);
+insert into sdel values('2005-09-23 12:34:26.736',3,12.34);
+insert into sdel values('2005-09-23 12:34:26.736',4,12.34);
+insert into sdel values('2005-09-23 12:34:26.736',3,12.34);
+select * from sdel;
+
+call iot.resume();
+call iot.wait(2);
+select * from sdel;
+
+drop procedure sdel00;
+drop table sdel;
diff --git a/sql/backends/monet5/iot/Tests/logger.sql
b/sql/backends/monet5/iot/Tests/logger.sql
new file mode 100644
--- /dev/null
+++ b/sql/backends/monet5/iot/Tests/logger.sql
@@ -0,0 +1,24 @@
+-- A simple heartbeat continuous query.
+set schema iot;
+set optimizer='iot_pipe';
+
+create table log(t timestamp,b integer);
+
+create procedure cqlogger()
+begin
+ insert into log values(now(), iot.getheartbeat('iot','cqlogger'));
+end;
+
+call iot.query('iot','cqlogger');
+call iot.heartbeat('iot','cqlogger',1000);
+
+-- wait for 1 cycle in the scheduler
+call iot.wait(2);
+
+select 'RESULT';
+select * from log;
+
+select * from iot.errors();
+call iot.stop();
+drop procedure cqlogger;
+drop table log;
diff --git a/sql/backends/monet5/iot/basket.c b/sql/backends/monet5/iot/basket.c
--- a/sql/backends/monet5/iot/basket.c
+++ b/sql/backends/monet5/iot/basket.c
@@ -30,6 +30,7 @@
#include "gdk.h"
#include "iot.h"
#include "basket.h"
+#include "petrinet.h"
#include "mal_exception.h"
#include "mal_builder.h"
#include "opt_prelude.h"
@@ -228,12 +229,8 @@ BSKTheartbeat(Client cntxt, MalBlkPtr mb
if( ticks < 0)
throw(SQL,"basket.heartbeat","Positive heartbeat expected]n");
idx = BSKTlocate(sch, tbl);
- if( idx == 0){
- BSKTregisterInternal(cntxt, mb, sch, tbl);
- idx = BSKTlocate(sch, tbl);
- if( idx ==0)
- throw(SQL,"basket.heartbeat","Stream table %s.%s not
accessible to deactivate\n",sch,tbl);
- }
+ if( idx == 0)
+ return PNheartbeat(sch,tbl,ticks);
baskets[idx].heartbeat = ticks;
return MAL_SUCCEED;
}
@@ -250,8 +247,12 @@ BSKTgetheartbeat(Client cntxt, MalBlkPtr
(void) mb;
idx = BSKTlocate(sch, tbl);
- if( idx == 0)
- throw(SQL,"basket.heartbeat","Stream table %s.%s not accessible
to deactivate\n",sch,tbl);
+ if( idx == 0){
+ BSKTregisterInternal(cntxt, mb, sch, tbl);
+ idx = BSKTlocate(sch, tbl);
+ if( idx == 0)
+ throw(SQL,"basket.heartbeat","Stream table %s.%s not
accessible to deactivate\n",sch,tbl);
+ }
*ret = baskets[idx].heartbeat;
return MAL_SUCCEED;
}
@@ -598,7 +599,7 @@ BSKTexportInternal(Client cntxt, int bsk
case TYPE_hge:
#endif
/* append the binary partition */
- fsize = BATcount(b) * ATOMsize(b->ttype);
+ fsize = (long) BATcount(b) * ATOMsize(b->ttype);
if( fwrite(Tloc(b, BUNlast(b)),1,fsize, f) != (size_t)
fsize){
(void) fclose(f);
msg= createException(MAL,"iot.export","Could
not write complete basket file %s\n",baskets[bskt].cols[i]);
@@ -907,6 +908,41 @@ BSKTupdate(Client cntxt, MalBlkPtr mb, M
}
str
+BSKTdelete(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
+{
+ int *res = getArgReference_int(stk, pci, 0);
+ str sname = *getArgReference_str(stk, pci, 2);
+ str tname = *getArgReference_str(stk, pci, 3);
+ bat rows = *getArgReference_bat(stk, pci, 4);
+ BAT *b=0, *rid=0;
+ int i,idx;
+
+ (void) cntxt;
+ (void) mb;
+ *res = 0;
+
+ rid = BATdescriptor(rows);
+ if( rid == NULL)
+ throw(SQL, "basket.delete", "Cannot access source oid descriptor");
+
+ idx = BSKTlocate(sname,tname);
+ if( idx == 0)
+ throw(SQL, "basket.delete", "Cannot access basket descriptor
%s.%s",sname,tname);
+ for( i=0; baskets[idx].cols[i]; i++){
+ b = baskets[idx].bats[i];
+ if(b){
+ (void) BATdel(b, rid);
+ BATderiveProps(b, FALSE);
+ }
+ }
+
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
