xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(stream): interval offset

This commit is contained in:
54liuyao 2022-07-14 10:15:20 +08:00
parent f02d8c2596
commit 13519f8fd9
4 changed files with 9 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/libs/executor/src/executil.c
View File

@ -853,6 +853,9 @@ static STimeWindow doCalculateTimeWindow(int64_t ts, SInterval* pInterval) {
w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1; w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
} else { } else {
int64_t st = w.skey; int64_t st = w.skey;
if (pInterval->offset > 0) {
st = taosTimeAdd(st, pInterval->offset, pInterval->offsetUnit, pInterval->precision);
}
if (st > ts) { if (st > ts) {
st -= ((st - ts + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding; st -= ((st - ts + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;

2
source/libs/stream/src/streamUpdate.c
View File

@ -17,7 +17,7 @@
#include "ttime.h" #include "ttime.h"
#define DEFAULT_FALSE_POSITIVE 0.01 #define DEFAULT_FALSE_POSITIVE 0.01
#define DEFAULT_BUCKET_SIZE 1024 #define DEFAULT_BUCKET_SIZE 131072
#define ROWS_PER_MILLISECOND 1 #define ROWS_PER_MILLISECOND 1
#define MAX_NUM_SCALABLE_BF 100000 #define MAX_NUM_SCALABLE_BF 100000
#define MIN_NUM_SCALABLE_BF 10 #define MIN_NUM_SCALABLE_BF 10

1
tests/script/jenkins/basic.txt
View File

@ -114,6 +114,7 @@
# ./test.sh -f tsim/stream/schedSnode.sim # ./test.sh -f tsim/stream/schedSnode.sim
./test.sh -f tsim/stream/windowClose.sim ./test.sh -f tsim/stream/windowClose.sim
./test.sh -f tsim/stream/ignoreExpiredData.sim ./test.sh -f tsim/stream/ignoreExpiredData.sim
./test.sh -f tsim/stream/sliding.sim
# ---- transaction # ---- transaction
./test.sh -f tsim/trans/lossdata1.sim ./test.sh -f tsim/trans/lossdata1.sim

8
tests/script/tsim/stream/sliding.sim
View File

@ -17,10 +17,10 @@ sql use test
sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int); sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int);
sql create table t1 using st tags(1,1,1); sql create table t1 using st tags(1,1,1);
sql create table t2 using st tags(2,2,2); sql create table t2 using st tags(2,2,2);
sql create stream streams1 trigger at_once into streamt as select _wstartts, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s) sliding (5s); sql create stream streams1 trigger at_once into streamt as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s) sliding (5s);
sql create stream streams2 trigger at_once watermark 1d into streamt2 as select _wstartts, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s) sliding (5s); sql create stream streams2 trigger at_once watermark 1d into streamt2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s) sliding (5s);
sql create stream stream_t1 trigger at_once into streamtST as select _wstartts, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s) sliding (5s); sql create stream stream_t1 trigger at_once into streamtST as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s) sliding (5s);
sql create stream stream_t2 trigger at_once watermark 1d into streamtST2 as select _wstartts, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s) sliding (5s); sql create stream stream_t2 trigger at_once watermark 1d into streamtST2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s) sliding (5s);
sql insert into t1 values(1648791210000,1,2,3,1.0); sql insert into t1 values(1648791210000,1,2,3,1.0);
sql insert into t1 values(1648791216000,2,2,3,1.1); sql insert into t1 values(1648791216000,2,2,3,1.1);