From fe4123e814db9f76460df070d84186361b937a6a Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao@163.com> Date: Tue, 28 Feb 2023 14:19:21 +0800 Subject: [PATCH] feat:window close && ignore expired --- include/common/tmsg.h | 2 +- source/libs/parser/src/parAstCreater.c | 2 +- source/libs/parser/test/parInitialCTest.cpp | 10 ++----- tests/script/tsim/stream/basic1.sim | 16 +++++----- tests/script/tsim/stream/deleteInterval.sim | 6 ++-- tests/script/tsim/stream/deleteSession.sim | 10 +++---- tests/script/tsim/stream/deleteState.sim | 6 ++-- .../stream/distributeIntervalRetrive0.sim | 4 +-- .../script/tsim/stream/distributeSession0.sim | 2 +- .../script/tsim/stream/fillHistoryBasic1.sim | 4 +-- .../script/tsim/stream/fillHistoryBasic2.sim | 6 ++-- .../script/tsim/stream/fillHistoryBasic3.sim | 2 +- .../tsim/stream/fillIntervalDelete0.sim | 20 ++++++------- .../tsim/stream/fillIntervalDelete1.sim | 30 +++++++++---------- .../script/tsim/stream/fillIntervalLinear.sim | 6 ++-- .../tsim/stream/fillIntervalPartitionBy.sim | 10 +++---- .../tsim/stream/fillIntervalPrevNext.sim | 8 ++--- .../tsim/stream/fillIntervalPrevNext1.sim | 4 +-- .../script/tsim/stream/fillIntervalRange.sim | 6 ++-- .../script/tsim/stream/fillIntervalValue.sim | 12 ++++---- .../script/tsim/stream/ignoreCheckUpdate.sim | 4 +-- tests/script/tsim/stream/partitionby.sim | 6 ++-- tests/script/tsim/stream/partitionby1.sim | 6 ++-- .../tsim/stream/partitionbyColumnInterval.sim | 10 +++---- .../tsim/stream/partitionbyColumnSession.sim | 8 ++--- .../tsim/stream/partitionbyColumnState.sim | 4 +-- tests/script/tsim/stream/schedSnode.sim | 2 +- tests/script/tsim/stream/session0.sim | 18 +++++------ tests/script/tsim/stream/session1.sim | 4 +-- tests/script/tsim/stream/sliding.sim | 20 ++++++------- tests/script/tsim/stream/state0.sim | 20 ++++++------- tests/script/tsim/stream/triggerInterval0.sim | 2 +- tests/script/tsim/stream/triggerSession0.sim | 2 +- tests/script/tsim/stream/udTableAndTag0.sim | 12 ++++---- tests/script/tsim/stream/udTableAndTag1.sim | 10 +++---- tests/script/tsim/stream/udTableAndTag2.sim | 14 ++++----- .../system-test/1-insert/database_pre_suf.py | 2 +- tests/system-test/1-insert/drop.py | 8 ++--- utils/test/c/tmq_taosx_ci.c | 2 +- 39 files changed, 158 insertions(+), 162 deletions(-) diff --git a/include/common/tmsg.h b/include/common/tmsg.h index cd63f7d278..d6a301c38b 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1812,7 +1812,7 @@ typedef struct { #define STREAM_TRIGGER_AT_ONCE 1 #define STREAM_TRIGGER_WINDOW_CLOSE 2 #define STREAM_TRIGGER_MAX_DELAY 3 -#define STREAM_DEFAULT_IGNORE_EXPIRED 0 +#define STREAM_DEFAULT_IGNORE_EXPIRED 1 #define STREAM_FILL_HISTORY_ON 1 #define STREAM_FILL_HISTORY_OFF 0 #define STREAM_DEFAULT_FILL_HISTORY STREAM_FILL_HISTORY_OFF diff --git a/source/libs/parser/src/parAstCreater.c b/source/libs/parser/src/parAstCreater.c index f613b1bd3d..634a239399 100644 --- a/source/libs/parser/src/parAstCreater.c +++ b/source/libs/parser/src/parAstCreater.c @@ -1810,7 +1810,7 @@ SNode* createStreamOptions(SAstCreateContext* pCxt) { CHECK_PARSER_STATUS(pCxt); SStreamOptions* pOptions = (SStreamOptions*)nodesMakeNode(QUERY_NODE_STREAM_OPTIONS); CHECK_OUT_OF_MEM(pOptions); - pOptions->triggerType = STREAM_TRIGGER_AT_ONCE; + pOptions->triggerType = STREAM_TRIGGER_WINDOW_CLOSE; pOptions->fillHistory = STREAM_DEFAULT_FILL_HISTORY; pOptions->ignoreExpired = STREAM_DEFAULT_IGNORE_EXPIRED; pOptions->ignoreUpdate = STREAM_DEFAULT_IGNORE_UPDATE; diff --git a/source/libs/parser/test/parInitialCTest.cpp b/source/libs/parser/test/parInitialCTest.cpp index 8ba5802ad6..2dac35590e 100644 --- a/source/libs/parser/test/parInitialCTest.cpp +++ b/source/libs/parser/test/parInitialCTest.cpp @@ -781,16 +781,10 @@ TEST_F(ParserInitialCTest, createStream) { snprintf(expect.targetStbFullName, sizeof(expect.targetStbFullName), "0.test.%s", pDstStb); expect.igExists = igExists; expect.sql = taosStrdup(pSql); - expect.createStb = STREAM_CREATE_STABLE_TRUE; - expect.triggerType = STREAM_TRIGGER_AT_ONCE; - expect.maxDelay = 0; - expect.watermark = 0; - expect.fillHistory = STREAM_DEFAULT_FILL_HISTORY; - expect.igExpired = STREAM_DEFAULT_IGNORE_EXPIRED; }; auto setStreamOptions = - [&](int8_t createStb = STREAM_CREATE_STABLE_TRUE, int8_t triggerType = STREAM_TRIGGER_AT_ONCE, + [&](int8_t createStb = STREAM_CREATE_STABLE_TRUE, int8_t triggerType = STREAM_TRIGGER_WINDOW_CLOSE, int64_t maxDelay = 0, int64_t watermark = 0, int8_t igExpired = STREAM_DEFAULT_IGNORE_EXPIRED, int8_t fillHistory = STREAM_DEFAULT_FILL_HISTORY, int8_t igUpdate = STREAM_DEFAULT_IGNORE_UPDATE) { expect.createStb = createStb; @@ -852,6 +846,7 @@ TEST_F(ParserInitialCTest, createStream) { }); setCreateStreamReq("s1", "test", "create stream s1 into st3 as select count(*) from t1 interval(10s)", "st3"); + setStreamOptions(); run("CREATE STREAM s1 INTO st3 AS SELECT COUNT(*) FROM t1 INTERVAL(10S)"); clearCreateStreamReq(); @@ -872,6 +867,7 @@ TEST_F(ParserInitialCTest, createStream) { "st3"); addTag("tname", TSDB_DATA_TYPE_VARCHAR, 10 + VARSTR_HEADER_SIZE); addTag("id", TSDB_DATA_TYPE_INT); + setStreamOptions(); run("CREATE STREAM s1 INTO st3 TAGS(tname VARCHAR(10), id INT) SUBTABLE(CONCAT('new-', tname)) " "AS SELECT _WSTART wstart, COUNT(*) cnt FROM st1 PARTITION BY TBNAME tname, tag1 id INTERVAL(10S)"); clearCreateStreamReq(); diff --git a/tests/script/tsim/stream/basic1.sim b/tests/script/tsim/stream/basic1.sim index a8f04aac64..e69875d69f 100644 --- a/tests/script/tsim/stream/basic1.sim +++ b/tests/script/tsim/stream/basic1.sim @@ -17,7 +17,7 @@ sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams1 trigger at_once into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); sql insert into t1 values(1648791213000,1,2,3,1.0); sql insert into t1 values(1648791223001,2,2,3,1.1); sql insert into t1 values(1648791233002,3,2,3,2.1); @@ -545,8 +545,8 @@ sql create table t2 using st tags(2,2,2); sql create table t3 using st tags(2,2,2); sql create table t4 using st tags(2,2,2); sql create table t5 using st tags(2,2,2); -sql create stream streams2 trigger at_once into streamt as select _wstart, count(*) c1, sum(a) c3,max(b) c4 from st partition by tbname interval(10s); -sql create stream streams3 trigger at_once into streamt3 as select _wstart, count(*) c1, sum(a) c3,max(b) c4, now c5 from st partition by tbname interval(10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, sum(a) c3,max(b) c4 from st partition by tbname interval(10s); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart, count(*) c1, sum(a) c3,max(b) c4, now c5 from st partition by tbname interval(10s); sql insert into t1 values(1648791213000,1,1,1,1.0) t2 values(1648791213000,2,2,2,2.0) t3 values(1648791213000,3,3,3,3.0) t4 values(1648791213000,4,4,4,4.0); @@ -667,7 +667,7 @@ sql create database test3 vgroups 1; sql use test3; sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int); sql create table ts1 using st tags(1,1,1); -sql create stream stream_t3 trigger at_once into streamtST3 as select ts, min(a) c6, a, b, c, ta, tb, tc from ts1 interval(10s) ; +sql create stream stream_t3 trigger at_once IGNORE EXPIRED 0 into streamtST3 as select ts, min(a) c6, a, b, c, ta, tb, tc from ts1 interval(10s) ; sql insert into ts1 values(1648791211000,1,2,3); sleep 50 @@ -701,7 +701,7 @@ endi sql create database test4 vgroups 1; sql use test4; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams4 trigger at_once into streamt4 as select _wstart, count(*) c1 from t1 where a > 5 interval(10s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart, count(*) c1 from t1 where a > 5 interval(10s); sql insert into t1 values(1648791213000,1,2,3,1.0); $loop_count = 0 @@ -797,8 +797,8 @@ sql create database test5 vgroups 1; sql use test5; sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int); sql create table ts1 using st tags(1,1,1); -sql create stream streams5 trigger at_once into streamt5 as select count(*), _wstart, _wend, max(a) from ts1 interval(10s) ; -sql create stream streams6 trigger at_once into streamt6 as select count(*), _wstart, _wend, max(a), _wstart as ts from ts1 interval(10s) ; +sql create stream streams5 trigger at_once IGNORE EXPIRED 0 into streamt5 as select count(*), _wstart, _wend, max(a) from ts1 interval(10s) ; +sql create stream streams6 trigger at_once IGNORE EXPIRED 0 into streamt6 as select count(*), _wstart, _wend, max(a), _wstart as ts from ts1 interval(10s) ; sql_error create stream streams7 trigger at_once into streamt7 as select _wstart, count(*), _wstart, _wend, max(a) from ts1 interval(10s) ; sql_error create stream streams8 trigger at_once into streamt8 as select count(*), _wstart, _wstart, _wend, max(a) from ts1 interval(10s) ; @@ -840,7 +840,7 @@ sql create database test7 vgroups 1; sql use test7; sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int); sql create table ts1 using st tags(1,1,1); -sql create stream streams7 trigger at_once into streamt7 as select _wstart, count(*) from ts1 interval(10s) ; +sql create stream streams7 trigger at_once IGNORE EXPIRED 0 into streamt7 as select _wstart, count(*) from ts1 interval(10s) ; sql insert into ts1 values(1648791211000,1,2,3); sql_error insert into ts1 values(-1648791211000,1,2,3); diff --git a/tests/script/tsim/stream/deleteInterval.sim b/tests/script/tsim/stream/deleteInterval.sim index 7532b2d5de..9540e448d4 100644 --- a/tests/script/tsim/stream/deleteInterval.sim +++ b/tests/script/tsim/stream/deleteInterval.sim @@ -16,7 +16,7 @@ sql drop database if exists test; sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3 from t1 interval(10s); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from t1 interval(10s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sleep 200 @@ -193,7 +193,7 @@ sql use test2; 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 t2 using st tags(2,2,2); -sql create stream streams2 trigger at_once into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st interval(10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st interval(10s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL); @@ -419,7 +419,7 @@ sql use test3; 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 t2 using st tags(2,2,2); -sql create stream streams3 trigger at_once into test.streamt3 as select _wstart c1, count(*) c2, max(a) c3 from st interval(10s); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into test.streamt3 as select _wstart c1, count(*) c2, max(a) c3 from st interval(10s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL); diff --git a/tests/script/tsim/stream/deleteSession.sim b/tests/script/tsim/stream/deleteSession.sim index c3c64a5977..3645939178 100644 --- a/tests/script/tsim/stream/deleteSession.sim +++ b/tests/script/tsim/stream/deleteSession.sim @@ -16,7 +16,7 @@ sql drop database if exists test; sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3 from t1 session(ts, 5s); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from t1 session(ts, 5s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sleep 200 @@ -191,7 +191,7 @@ sql use test2; 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 t2 using st tags(2,2,2); -sql create stream streams2 trigger at_once into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st session(ts,5s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st session(ts,5s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL); @@ -422,7 +422,7 @@ sql use test3; 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 t2 using st tags(2,2,2); -sql create stream streams3 trigger at_once into test.streamt3 as select _wstart c1, count(*) c2, max(a) c3 from st session(ts,5s); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into test.streamt3 as select _wstart c1, count(*) c2, max(a) c3 from st session(ts,5s); sql insert into t1 values(1648791210000,1,1,1,NULL); sql insert into t1 values(1648791210001,2,2,2,NULL); @@ -532,8 +532,8 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -print create stream streams4 trigger at_once into streamt4 as select _wstart, count(*) c1 from st partition by tbname session(ts, 2s); -sql create stream streams4 trigger at_once into streamt4 as select _wstart, count(*) c1 from st partition by tbname session(ts, 2s); +print create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart, count(*) c1 from st partition by tbname session(ts, 2s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart, count(*) c1 from st partition by tbname session(ts, 2s); sql insert into t1 values(1648791210000,1,2,3); sql insert into t1 values(1648791220000,2,2,3); diff --git a/tests/script/tsim/stream/deleteState.sim b/tests/script/tsim/stream/deleteState.sim index 45d9bc3e39..dd74b73dce 100644 --- a/tests/script/tsim/stream/deleteState.sim +++ b/tests/script/tsim/stream/deleteState.sim @@ -16,7 +16,7 @@ sql drop database if exists test; sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(b) c3 from t1 state_window(a); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(b) c3 from t1 state_window(a); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sleep 200 @@ -197,8 +197,8 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -print create stream streams4 trigger at_once into streamt4 as select _wstart, count(*) c1 from st partition by tbname state_window(c); -sql create stream streams4 trigger at_once into streamt4 as select _wstart, count(*) c1 from st partition by tbname state_window(c); +print create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart, count(*) c1 from st partition by tbname state_window(c); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart, count(*) c1 from st partition by tbname state_window(c); sql insert into t1 values(1648791210000,1,2,1); sql insert into t1 values(1648791220000,2,2,2); diff --git a/tests/script/tsim/stream/distributeIntervalRetrive0.sim b/tests/script/tsim/stream/distributeIntervalRetrive0.sim index ae2f9afdb5..529a2a1b30 100644 --- a/tests/script/tsim/stream/distributeIntervalRetrive0.sim +++ b/tests/script/tsim/stream/distributeIntervalRetrive0.sim @@ -44,7 +44,7 @@ sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); sql create table ts3 using st tags(3,2,2); sql create table ts4 using st tags(4,2,2); -sql create stream stream_t1 trigger at_once into streamtST1 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s); +sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 into streamtST1 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s); sleep 1000 @@ -243,7 +243,7 @@ sql use test1; sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams1 trigger at_once into streamt1 as select _wstart as c0, count(*) c1, count(a) c2 from st interval(10s) ; +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart as c0, count(*) c1, count(a) c2 from st interval(10s) ; sql insert into t1 values(1648791211000,1,2,3); diff --git a/tests/script/tsim/stream/distributeSession0.sim b/tests/script/tsim/stream/distributeSession0.sim index d752f5c29c..25ac479a38 100644 --- a/tests/script/tsim/stream/distributeSession0.sim +++ b/tests/script/tsim/stream/distributeSession0.sim @@ -39,7 +39,7 @@ sql use test; sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); -sql create stream stream_t1 trigger at_once into streamtST as select _wstart, count(*) c1, sum(a) c2 , max(b) c3 from st session(ts, 10s) ; +sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 into streamtST as select _wstart, count(*) c1, sum(a) c2 , max(b) c3 from st session(ts, 10s) ; sleep 1000 diff --git a/tests/script/tsim/stream/fillHistoryBasic1.sim b/tests/script/tsim/stream/fillHistoryBasic1.sim index 772a09c017..e7a8da90e2 100644 --- a/tests/script/tsim/stream/fillHistoryBasic1.sim +++ b/tests/script/tsim/stream/fillHistoryBasic1.sim @@ -17,7 +17,7 @@ sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream stream1 trigger at_once fill_history 1 into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); +sql create stream stream1 trigger at_once fill_history 1 IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); sql insert into t1 values(1648791213000,1,2,3,1.0); sql insert into t1 values(1648791223001,2,2,3,1.1); @@ -479,7 +479,7 @@ sql insert into t1 values(1648791233002,3,2,3,2.1); sql insert into t1 values(1648791243003,4,2,3,3.1); sql insert into t1 values(1648791213004,4,2,3,4.1); -sql create stream stream2 trigger at_once fill_history 1 into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); +sql create stream stream2 trigger at_once fill_history 1 IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); sleep 5000 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt; diff --git a/tests/script/tsim/stream/fillHistoryBasic2.sim b/tests/script/tsim/stream/fillHistoryBasic2.sim index 3af198259d..2f6c3ea92d 100644 --- a/tests/script/tsim/stream/fillHistoryBasic2.sim +++ b/tests/script/tsim/stream/fillHistoryBasic2.sim @@ -79,7 +79,7 @@ sql insert into ts3 values(1648791243005,4,42,3,3.1) (1648791243003,4,2,33,3.1) sql insert into ts4 values(1648791243005,4,42,3,3.1) (1648791243003,4,2,33,3.1) (1648791243006,4,2,3,3.1) (1648791213001,1,52,13,1.0) (1648791223001,22,22,83,1.1) (1648791233004,13,12,13,2.1) ; sql insert into ts3 values(1648791243006,4,2,3,3.1) (1648791213001,1,52,13,1.0) (1648791223001,22,22,83,1.1) ; -sql create stream stream_t1 trigger at_once fill_history 1 watermark 1d into streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s); +sql create stream stream_t1 trigger at_once fill_history 1 watermark 1d IGNORE EXPIRED 0 into streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s); sleep 1000 @@ -211,7 +211,7 @@ sql insert into ts1 values(1648791222001,2,2,3); sql insert into ts2 values(1648791211000,1,2,3); sql insert into ts2 values(1648791222001,2,2,3); -sql create stream stream_t2 trigger at_once fill_history 1 watermark 20s into streamtST1 as select _wstart, count(*) c1, count(a) c2 , sum(a) c3 , max(b) c5, min(c) c6 from st interval(10s) ; +sql create stream stream_t2 trigger at_once fill_history 1 watermark 20s IGNORE EXPIRED 0 into streamtST1 as select _wstart, count(*) c1, count(a) c2 , sum(a) c3 , max(b) c5, min(c) c6 from st interval(10s) ; $loop_count = 0 loop2: @@ -241,7 +241,7 @@ sql use test3; sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int); sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); -sql create stream stream_t3 trigger at_once into streamtST3 as select ts, min(a) c6, a, b, c, ta, tb, tc from st interval(10s) ; +sql create stream stream_t3 trigger at_once IGNORE EXPIRED 0 into streamtST3 as select ts, min(a) c6, a, b, c, ta, tb, tc from st interval(10s) ; sql insert into ts1 values(1648791211000,1,2,3); sleep 50 diff --git a/tests/script/tsim/stream/fillHistoryBasic3.sim b/tests/script/tsim/stream/fillHistoryBasic3.sim index db8d6bc2d0..44d7ee9d9e 100644 --- a/tests/script/tsim/stream/fillHistoryBasic3.sim +++ b/tests/script/tsim/stream/fillHistoryBasic3.sim @@ -17,7 +17,7 @@ sql create table t2 using st tags(2,2,2); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL); -sql create stream streams2 trigger at_once fill_history 1 into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); +sql create stream streams2 trigger at_once fill_history 1 IGNORE EXPIRED 0 into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); sleep 3000 diff --git a/tests/script/tsim/stream/fillIntervalDelete0.sim b/tests/script/tsim/stream/fillIntervalDelete0.sim index 1c0647d57b..41b018a862 100644 --- a/tests/script/tsim/stream/fillIntervalDelete0.sim +++ b/tests/script/tsim/stream/fillIntervalDelete0.sim @@ -16,11 +16,11 @@ sql drop database if exists test1; sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); -sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); -sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); -sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); -sql create stream streams5 trigger at_once into streamt5 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); +sql create stream streams5 trigger at_once IGNORE EXPIRED 0 into streamt5 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); sql insert into t1 values(1648791213000,1,1,1,1.0,'aaa'); sleep 200 @@ -256,11 +256,11 @@ sql use test6; sql create stable st(ts timestamp, a int, b int , c int, d double, s varchar(20)) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(1,1,1); -sql create stream streams6 trigger at_once into streamt6 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); -sql create stream streams7 trigger at_once into streamt7 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); -sql create stream streams8 trigger at_once into streamt8 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); -sql create stream streams9 trigger at_once into streamt9 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); -sql create stream streams10 trigger at_once into streamt10 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); +sql create stream streams6 trigger at_once IGNORE EXPIRED 0 into streamt6 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); +sql create stream streams7 trigger at_once IGNORE EXPIRED 0 into streamt7 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); +sql create stream streams8 trigger at_once IGNORE EXPIRED 0 into streamt8 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); +sql create stream streams9 trigger at_once IGNORE EXPIRED 0 into streamt9 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); +sql create stream streams10 trigger at_once IGNORE EXPIRED 0 into streamt10 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); sql insert into t1 values(1648791210000,1,1,1,1.0,'aaa'); sql insert into t1 values(1648791217000,1,1,1,1.0,'aaa'); diff --git a/tests/script/tsim/stream/fillIntervalDelete1.sim b/tests/script/tsim/stream/fillIntervalDelete1.sim index 0206b88fdc..108f5f862d 100644 --- a/tests/script/tsim/stream/fillIntervalDelete1.sim +++ b/tests/script/tsim/stream/fillIntervalDelete1.sim @@ -18,11 +18,11 @@ sql drop database if exists test1; sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); -sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); -sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); -sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); -sql create stream streams5 trigger at_once into streamt5 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); +sql create stream streams5 trigger at_once IGNORE EXPIRED 0 into streamt5 as select _wstart as ts, max(a), sum(b), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); sql insert into t1 values(1648791210000,0,0,0,0.0,'aaa'); sql insert into t1 values(1648791213000,1,1,1,1.0,'bbb'); @@ -221,11 +221,11 @@ sql use test6; sql create stable st(ts timestamp, a int, b int , c int, d double, s varchar(20)) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(1,1,1); -sql create stream streams6 trigger at_once into streamt6 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); -sql create stream streams7 trigger at_once into streamt7 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); -sql create stream streams8 trigger at_once into streamt8 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); -sql create stream streams9 trigger at_once into streamt9 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); -sql create stream streams10 trigger at_once into streamt10 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); +sql create stream streams6 trigger at_once IGNORE EXPIRED 0 into streamt6 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(NULL); +sql create stream streams7 trigger at_once IGNORE EXPIRED 0 into streamt7 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value,100,200,300); +sql create stream streams8 trigger at_once IGNORE EXPIRED 0 into streamt8 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); +sql create stream streams9 trigger at_once IGNORE EXPIRED 0 into streamt9 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); +sql create stream streams10 trigger at_once IGNORE EXPIRED 0 into streamt10 as select _wstart as ts, max(a), sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); sql insert into t1 values(1648791210000,1,1,1,1.0,'aaa'); sql insert into t1 values(1648791215000,6,8,8,8.0,'bbb'); @@ -353,11 +353,11 @@ sql drop database if exists test7; sql create database test7 vgroups 1; sql use test7; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams11 trigger at_once into streamt11 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(NULL); -sql create stream streams12 trigger at_once into streamt12 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(value,100.0,200); -sql create stream streams13 trigger at_once into streamt13 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(next); -sql create stream streams14 trigger at_once into streamt14 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(prev); -sql create stream streams15 trigger at_once into streamt15 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(linear); +sql create stream streams11 trigger at_once IGNORE EXPIRED 0 into streamt11 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(NULL); +sql create stream streams12 trigger at_once IGNORE EXPIRED 0 into streamt12 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(value,100.0,200); +sql create stream streams13 trigger at_once IGNORE EXPIRED 0 into streamt13 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(next); +sql create stream streams14 trigger at_once IGNORE EXPIRED 0 into streamt14 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(prev); +sql create stream streams15 trigger at_once IGNORE EXPIRED 0 into streamt15 as select _wstart as ts, avg(a), count(*), timezone(), to_iso8601(1) from t1 where ts >= 1648791210000 and ts < 1648791240000 interval(1s) fill(linear); sql insert into t1 values(1648791210000,1,1,1,1.0,'aaa'); diff --git a/tests/script/tsim/stream/fillIntervalLinear.sim b/tests/script/tsim/stream/fillIntervalLinear.sim index b9f301d81b..3fa369d8d5 100644 --- a/tests/script/tsim/stream/fillIntervalLinear.sim +++ b/tests/script/tsim/stream/fillIntervalLinear.sim @@ -16,7 +16,7 @@ sql drop database if exists test1; sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a)+sum(c), avg(b), first(s), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart as ts, max(a)+sum(c), avg(b), first(s), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791216000,5,5,5,5.0,'bbb'); sql insert into t1 values(1648791210000,1,1,1,1.0,'ccc') (1648791219000,2,2,2,2.0,'ddd') (1648791222000,3,3,3,3.0,'eee'); @@ -205,7 +205,7 @@ sql drop database if exists test2; sql create database test2 vgroups 1; sql use test2; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a)+sum(c), avg(b), first(s), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart as ts, max(a)+sum(c), avg(b), first(s), count(*) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(linear); sql insert into t1 values(1648791210000,1,1,1,1.0,'ccc') (1648791219000,2,2,2,2.0,'ddd') (1648791222000,3,3,3,3.0,'eee'); sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791216000,5,5,5,5.0,'bbb'); @@ -393,7 +393,7 @@ sql drop database if exists test3; sql create database test3 vgroups 1; sql use test3; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a), b+c, s, b+1, 1 from t1 where ts >= 1648791150000 and ts < 1648791261000 interval(1s) fill(linear); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart as ts, max(a), b+c, s, b+1, 1 from t1 where ts >= 1648791150000 and ts < 1648791261000 interval(1s) fill(linear); sql insert into t1 values(1648791215000,1,1,1,1.0,'aaa'); sql insert into t1 values(1648791217000,2,2,2,2.0,'bbb'); sql insert into t1 values(1648791211000,3,3,3,3.0,'ccc'); diff --git a/tests/script/tsim/stream/fillIntervalPartitionBy.sim b/tests/script/tsim/stream/fillIntervalPartitionBy.sim index 168452fdc8..6a11b9952c 100644 --- a/tests/script/tsim/stream/fillIntervalPartitionBy.sim +++ b/tests/script/tsim/stream/fillIntervalPartitionBy.sim @@ -18,11 +18,11 @@ sql use test1; sql create stable st(ts timestamp, a int, b int , c int, d double, s varchar(20)) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(NULL); -sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(value,100,200,300); -sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(next); -sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(prev); -sql create stream streams5 trigger at_once into streamt5 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(linear); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(NULL); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(value,100,200,300); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(next); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(prev); +sql create stream streams5 trigger at_once IGNORE EXPIRED 0 into streamt5 as select _wstart as ts, max(a) c1, sum(b), count(*) from st where ts >= 1648791210000 and ts < 1648791261000 partition by ta interval(1s) fill(linear); sql insert into t1 values(1648791210000,0,0,0,0.0,'aaa'); sql insert into t1 values(1648791213000,1,1,1,1.0,'bbb'); diff --git a/tests/script/tsim/stream/fillIntervalPrevNext.sim b/tests/script/tsim/stream/fillIntervalPrevNext.sim index 99f7dcb5cb..ec963e1d4a 100644 --- a/tests/script/tsim/stream/fillIntervalPrevNext.sim +++ b/tests/script/tsim/stream/fillIntervalPrevNext.sim @@ -15,8 +15,8 @@ sql drop database if exists test1; sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams1 trigger at_once into streamt1 as select _wstart as ts, count(*) c1, max(b)+sum(a) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); -sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, count(*) c1, max(a)+min(c), avg(b) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart as ts, count(*) c1, max(b)+sum(a) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart as ts, count(*) c1, max(a)+min(c), avg(b) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791215000,5,5,5,5.0,'aaa'); sql insert into t1 values(1648791211000,1,1,1,1.0,'aaa') (1648791217000,2,2,2,2.0,'aaa') (1648791220000,3,3,3,3.0,'aaa'); @@ -263,8 +263,8 @@ sql drop database if exists test5; sql create database test5 vgroups 1; sql use test5; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams5 trigger at_once into streamt5 as select _wstart as ts, count(*) c1, max(b)+sum(a) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); -sql create stream streams6 trigger at_once into streamt6 as select _wstart as ts, count(*) c1, max(a)+min(c), avg(b) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); +sql create stream streams5 trigger at_once IGNORE EXPIRED 0 into streamt5 as select _wstart as ts, count(*) c1, max(b)+sum(a) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(prev); +sql create stream streams6 trigger at_once IGNORE EXPIRED 0 into streamt6 as select _wstart as ts, count(*) c1, max(a)+min(c), avg(b) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(next); sql insert into t1 values(1648791211000,1,1,1,1.0,'aaa') (1648791217000,2,2,2,2.0,'aaa') (1648791220000,3,3,3,3.0,'aaa'); sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791215000,5,5,5,5.0,'aaa'); diff --git a/tests/script/tsim/stream/fillIntervalPrevNext1.sim b/tests/script/tsim/stream/fillIntervalPrevNext1.sim index 8058065bcf..40ef895c5a 100644 --- a/tests/script/tsim/stream/fillIntervalPrevNext1.sim +++ b/tests/script/tsim/stream/fillIntervalPrevNext1.sim @@ -16,8 +16,8 @@ sql drop database if exists test7; sql create database test7 vgroups 1; sql use test7; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams7 trigger at_once into streamt7 as select _wstart as ts, max(a), b+c, s from t1 where ts >= 1648791150000 and ts < 1648791261000 interval(1s) fill(prev); -sql create stream streams8 trigger at_once into streamt8 as select _wstart as ts, max(a), 1, b+1 from t1 where ts >= 1648791150000 and ts < 1648791261000 interval(1s) fill(next); +sql create stream streams7 trigger at_once IGNORE EXPIRED 0 into streamt7 as select _wstart as ts, max(a), b+c, s from t1 where ts >= 1648791150000 and ts < 1648791261000 interval(1s) fill(prev); +sql create stream streams8 trigger at_once IGNORE EXPIRED 0 into streamt8 as select _wstart as ts, max(a), 1, b+1 from t1 where ts >= 1648791150000 and ts < 1648791261000 interval(1s) fill(next); sql insert into t1 values(1648791215000,1,1,1,1.0,'aaa'); sql insert into t1 values(1648791217000,2,2,2,2.0,'bbb'); sql insert into t1 values(1648791211000,3,3,3,3.0,'ccc'); diff --git a/tests/script/tsim/stream/fillIntervalRange.sim b/tests/script/tsim/stream/fillIntervalRange.sim index a0905141f2..0e0dfb46d8 100644 --- a/tests/script/tsim/stream/fillIntervalRange.sim +++ b/tests/script/tsim/stream/fillIntervalRange.sim @@ -13,7 +13,7 @@ sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));; -sql create stream streams1 trigger at_once into streamt as select _wstart ts, count(*) c1 from t1 interval(1s) fill(NULL); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart ts, count(*) c1 from t1 interval(1s) fill(NULL); sql insert into t1 values(1648791211000,1,2,3,1.0,'aaa'); sleep 100 sql insert into t1 values(1648795308000,1,2,3,1.0,'aaa'); @@ -126,10 +126,10 @@ sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); print create stream streams1 trigger at_once into streamt as select _wstart ts, max(a) c1 from t1 interval(1s) fill(linear); -sql create stream streams1 trigger at_once into streamt as select _wstart ts, max(a) c1 from t1 interval(1s) fill(linear); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart ts, max(a) c1 from t1 interval(1s) fill(linear); print create stream streams2 trigger at_once into streamt2 as select _wstart ts, max(a) c1 from t1 interval(1s) fill(prev); -sql create stream streams2 trigger at_once into streamt2 as select _wstart ts, max(a) c1 from t1 interval(1s) fill(prev); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart ts, max(a) c1 from t1 interval(1s) fill(prev); sql insert into t1 values(1648791211000,1,2,3,1.0,'aaa'); sleep 100 diff --git a/tests/script/tsim/stream/fillIntervalValue.sim b/tests/script/tsim/stream/fillIntervalValue.sim index 2cd419397f..b447e9a559 100644 --- a/tests/script/tsim/stream/fillIntervalValue.sim +++ b/tests/script/tsim/stream/fillIntervalValue.sim @@ -13,8 +13,8 @@ sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20));; -sql create stream streams1 trigger at_once into streamt as select _wstart ts, count(*) c1 from t1 where ts > 1648791210000 and ts < 1648791413000 interval(10s) fill(value, 100); -sql create stream streams1a trigger at_once into streamta as select _wstart ts, count(*) c1 from t1 where ts > 1648791210000 and ts < 1648791413000 interval(10s) fill(value_f, 100); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart ts, count(*) c1 from t1 where ts > 1648791210000 and ts < 1648791413000 interval(10s) fill(value, 100); +sql create stream streams1a trigger at_once IGNORE EXPIRED 0 into streamta as select _wstart ts, count(*) c1 from t1 where ts > 1648791210000 and ts < 1648791413000 interval(10s) fill(value_f, 100); sql insert into t1 values(1648791213000,1,2,3,1.0,'aaa'); sleep 100 sql insert into t1 values(1648791233000,1,2,3,1.0,'aaa'); @@ -146,7 +146,7 @@ sql drop database if exists test2; sql create database test2 vgroups 1; sql use test2; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams2 trigger at_once into streamt2 as select _wstart as ts, count(*) c1, max(b)+sum(a) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value, 100,200); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart as ts, count(*) c1, max(b)+sum(a) from t1 where ts >= 1648791210000 and ts < 1648791261000 interval(1s) fill(value, 100,200); sql insert into t1 values(1648791211000,1,1,1,1.0,'aaa') (1648791217000,2,2,2,2.0,'aaa') (1648791220000,3,3,3,3.0,'aaa'); sql insert into t1 values(1648791213000,4,4,4,4.0,'aaa') (1648791215000,5,5,5,5.0,'aaa'); @@ -280,7 +280,7 @@ sql drop database if exists test3; sql create database test3 vgroups 1; sql use test3; sql create table t1(ts timestamp, a int, b int , c int, d double, s varchar(20)); -sql create stream streams3 trigger at_once into streamt3 as select _wstart as ts, max(b), a+b, c from t1 where ts >= 1648791200000 and ts < 1648791261000 interval(10s) sliding(3s) fill(value, 100,200,300); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart as ts, max(b), a+b, c from t1 where ts >= 1648791200000 and ts < 1648791261000 interval(10s) sliding(3s) fill(value, 100,200,300); sql insert into t1 values(1648791220000,1,1,1,1.0,'aaa'); sleep 100 @@ -471,8 +471,8 @@ sql create stable st(ts timestamp,a int,b int,c int, d double, s varchar(20) ) t sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams4 trigger at_once into streamt4 as select _wstart ts, count(*) c1, concat(tbname, 'aaa') as pname, timezone() from st where ts > 1648791000000 and ts < 1648793000000 partition by tbname interval(10s) fill(NULL); -sql create stream streams4a trigger at_once into streamt4a as select _wstart ts, count(*) c1, concat(tbname, 'aaa') as pname, timezone() from st where ts > 1648791000000 and ts < 1648793000000 partition by tbname interval(10s) fill(NULL_F); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart ts, count(*) c1, concat(tbname, 'aaa') as pname, timezone() from st where ts > 1648791000000 and ts < 1648793000000 partition by tbname interval(10s) fill(NULL); +sql create stream streams4a trigger at_once IGNORE EXPIRED 0 into streamt4a as select _wstart ts, count(*) c1, concat(tbname, 'aaa') as pname, timezone() from st where ts > 1648791000000 and ts < 1648793000000 partition by tbname interval(10s) fill(NULL_F); sql insert into t1 values(1648791213000,1,2,3,1.0,'aaa'); sql insert into t1 values(1648791233000,1,2,3,1.0,'aaa'); sql insert into t1 values(1648791273000,1,2,3,1.0,'aaa'); diff --git a/tests/script/tsim/stream/ignoreCheckUpdate.sim b/tests/script/tsim/stream/ignoreCheckUpdate.sim index 7f99c534c8..2cd0117feb 100644 --- a/tests/script/tsim/stream/ignoreCheckUpdate.sim +++ b/tests/script/tsim/stream/ignoreCheckUpdate.sim @@ -12,9 +12,9 @@ sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int); -print create stream streams0 trigger at_once ignore update 1 into streamt as select _wstart c1, count(*) c2, max(b) c3 from t1 interval(10s); +print create stream streams0 trigger at_once IGNORE EXPIRED 0 ignore update 1 into streamt as select _wstart c1, count(*) c2, max(b) c3 from t1 interval(10s); -sql create stream streams0 trigger at_once ignore update 1 into streamt as select _wstart c1, count(*) c2, max(b) c3 from t1 interval(10s); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 ignore update 1 into streamt as select _wstart c1, count(*) c2, max(b) c3 from t1 interval(10s); sql insert into t1 values(1648791213000,1,1,1); sql insert into t1 values(1648791213000,2,2,2); diff --git a/tests/script/tsim/stream/partitionby.sim b/tests/script/tsim/stream/partitionby.sim index bc2c07b951..e63459e97d 100644 --- a/tests/script/tsim/stream/partitionby.sim +++ b/tests/script/tsim/stream/partitionby.sim @@ -12,7 +12,7 @@ sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); sql create table ts3 using st tags(3,2,2); sql create table ts4 using st tags(4,2,2); -sql create stream stream_t1 trigger at_once into test0.streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st partition by ta,tb,tc interval(10s); +sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 into test0.streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st partition by ta,tb,tc interval(10s); sql insert into ts1 values(1648791213001,1,12,3,1.0); sql insert into ts2 values(1648791213001,1,12,3,1.0); @@ -67,7 +67,7 @@ sql create table ts1 using st tags(1,2,3); sql create table ts2 using st tags(1,3,4); sql create table ts3 using st tags(1,4,5); -sql create stream streams1 trigger at_once into streamt as select _wstart, count(*) c1, count(a) c2 from st partition by ta,tb,tc interval(10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, count(a) c2 from st partition by ta,tb,tc interval(10s); sql insert into ts1 values(1648791211000,1,2,3); @@ -98,7 +98,7 @@ sql create stable st(ts timestamp,a int,b int,c int,id int) tags(ta int,tb int,t sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); -sql create stream stream_t2 trigger at_once watermark 20s into streamtST as select _wstart, count(*) c1, count(a) c2 , sum(a) c3 , max(b) c5, min(c) c6, max(id) c7 from st partition by ta interval(10s) ; +sql create stream stream_t2 trigger at_once watermark 20s IGNORE EXPIRED 0 into streamtST as select _wstart, count(*) c1, count(a) c2 , sum(a) c3 , max(b) c5, min(c) c6, max(id) c7 from st partition by ta interval(10s) ; sql insert into ts1 values(1648791211000,1,2,3,1); sql insert into ts1 values(1648791222001,2,2,3,2); sql insert into ts2 values(1648791211000,1,2,3,3); diff --git a/tests/script/tsim/stream/partitionby1.sim b/tests/script/tsim/stream/partitionby1.sim index b29666cad7..c8bb25e0dd 100644 --- a/tests/script/tsim/stream/partitionby1.sim +++ b/tests/script/tsim/stream/partitionby1.sim @@ -11,7 +11,7 @@ sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); sql create table ts3 using st tags(3,2,2); sql create table ts4 using st tags(4,2,2); -sql create stream stream_t1 trigger at_once into streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st partition by tbname interval(10s); +sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 into streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st partition by tbname interval(10s); sql insert into ts1 values(1648791213001,1,12,3,1.0); sql insert into ts2 values(1648791213001,1,12,3,1.0); @@ -43,7 +43,7 @@ sql create table ts1 using st tags(1,2,3); sql create table ts2 using st tags(1,3,4); sql create table ts3 using st tags(1,4,5); -sql create stream streams1 trigger at_once into streamt as select _wstart, count(*) c1, count(a) c2 from st partition by tbname interval(10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, count(a) c2 from st partition by tbname interval(10s); sql insert into ts1 values(1648791211000,1,2,3); @@ -74,7 +74,7 @@ sql create stable st(ts timestamp,a int,b int,c int,id int) tags(ta int,tb int,t sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); -sql create stream stream_t2 trigger at_once into streamtST as select _wstart, count(*) c1, count(a) c2 , sum(a) c3 , max(b) c5, min(c) c6, max(id) c7 from st partition by tbname interval(10s) ; +sql create stream stream_t2 trigger at_once IGNORE EXPIRED 0 into streamtST as select _wstart, count(*) c1, count(a) c2 , sum(a) c3 , max(b) c5, min(c) c6, max(id) c7 from st partition by tbname interval(10s) ; sql insert into ts1 values(1648791211000,1,2,3,1); sql insert into ts1 values(1648791222001,2,2,3,2); sql insert into ts2 values(1648791211000,1,2,3,3); diff --git a/tests/script/tsim/stream/partitionbyColumnInterval.sim b/tests/script/tsim/stream/partitionbyColumnInterval.sim index 2e57e8d699..94053990e4 100644 --- a/tests/script/tsim/stream/partitionbyColumnInterval.sim +++ b/tests/script/tsim/stream/partitionbyColumnInterval.sim @@ -16,7 +16,7 @@ sql drop database if exists test; sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a interval(10s); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a interval(10s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); @@ -197,7 +197,7 @@ sql drop database if exists test1; sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams1 trigger at_once into streamt1 as select _wstart c1, count(*) c2, max(c) c3, _group_key(a+b) c4 from t1 partition by a+b interval(10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart c1, count(*) c2, max(c) c3, _group_key(a+b) c4 from t1 partition by a+b interval(10s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); @@ -284,7 +284,7 @@ sql use test2; 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 t2 using st tags(2,2,2); -sql create stream streams2 trigger at_once into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); @@ -481,7 +481,7 @@ sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); sql create table t3 using st tags(2,2,2); sql create table t4 using st tags(2,2,2); -sql create stream streams4 trigger at_once into test.streamt4 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into test.streamt4 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); sql insert into t1 values(1648791213000,2,2,3,1.0); sql insert into t2 values(1648791213000,2,2,3,1.0); @@ -571,7 +571,7 @@ sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); sql create table t3 using st tags(2,2,2); sql create table t4 using st tags(2,2,2); -sql create stream streams5 trigger at_once into test.streamt5 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); +sql create stream streams5 trigger at_once IGNORE EXPIRED 0 into test.streamt5 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a interval(10s); sql insert into t1 values(1648791213000,1,2,3,1.0); sql insert into t2 values(1648791213000,2,2,3,1.0); diff --git a/tests/script/tsim/stream/partitionbyColumnSession.sim b/tests/script/tsim/stream/partitionbyColumnSession.sim index 1742d52cf0..bb3f6015c7 100644 --- a/tests/script/tsim/stream/partitionbyColumnSession.sim +++ b/tests/script/tsim/stream/partitionbyColumnSession.sim @@ -16,7 +16,7 @@ sql drop database if exists test; sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a session(ts, 5s); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a session(ts, 5s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); @@ -196,7 +196,7 @@ sql drop database if exists test1; sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams1 trigger at_once into streamt1 as select _wstart c1, count(*) c2, max(c) c3, _group_key(a+b) c4 from t1 partition by a+b session(ts, 5s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart c1, count(*) c2, max(c) c3, _group_key(a+b) c4 from t1 partition by a+b session(ts, 5s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); @@ -282,7 +282,7 @@ sql use test2; 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 t2 using st tags(2,2,2); -sql create stream streams2 trigger at_once into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a session(ts, 5s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into test.streamt2 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a session(ts, 5s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); @@ -478,7 +478,7 @@ sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); sql create table t3 using st tags(2,2,2); sql create table t4 using st tags(2,2,2); -sql create stream streams4 trigger at_once into test.streamt4 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a session(ts, 5s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into test.streamt4 as select _wstart c1, count(*) c2, max(a) c3 from st partition by a session(ts, 5s); sql insert into t1 values(1648791213000,2,2,3,1.0); sql insert into t2 values(1648791213000,2,2,3,1.0); diff --git a/tests/script/tsim/stream/partitionbyColumnState.sim b/tests/script/tsim/stream/partitionbyColumnState.sim index 75d01b17ec..62262a490c 100644 --- a/tests/script/tsim/stream/partitionbyColumnState.sim +++ b/tests/script/tsim/stream/partitionbyColumnState.sim @@ -11,7 +11,7 @@ sql drop database if exists test; sql create database test vgroups 1; sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a state_window(b); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(a) c3, _group_key(a) c4 from t1 partition by a state_window(b); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); @@ -191,7 +191,7 @@ sql drop database if exists test1; sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d int); -sql create stream streams1 trigger at_once into streamt1 as select _wstart c1, count(*) c2, max(d) c3, _group_key(a+b) c4 from t1 partition by a+b state_window(c); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart c1, count(*) c2, max(d) c3, _group_key(a+b) c4 from t1 partition by a+b state_window(c); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL); diff --git a/tests/script/tsim/stream/schedSnode.sim b/tests/script/tsim/stream/schedSnode.sim index 2caecf50a2..6a4d6f79bb 100644 --- a/tests/script/tsim/stream/schedSnode.sim +++ b/tests/script/tsim/stream/schedSnode.sim @@ -20,7 +20,7 @@ sql create table ts1 using st tags(1,1,1); sql create table ts2 using st tags(2,2,2); sql create table ts3 using st tags(3,2,2); sql create table ts4 using st tags(4,2,2); -sql create stream stream_t1 trigger at_once into target.streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s); +sql create stream stream_t1 trigger at_once IGNORE EXPIRED 0 into target.streamtST1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s); sleep 1000 diff --git a/tests/script/tsim/stream/session0.sim b/tests/script/tsim/stream/session0.sim index 5e95428e0a..622c5f7c6d 100644 --- a/tests/script/tsim/stream/session0.sim +++ b/tests/script/tsim/stream/session0.sim @@ -17,7 +17,7 @@ sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double,id int); -sql create stream streams1 trigger at_once into streamt as select _wstart, count(*) c1, sum(a), max(a), min(d), stddev(a), last(a), first(d), max(id) s from t1 session(ts,10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, sum(a), max(a), min(d), stddev(a), last(a), first(d), max(id) s from t1 session(ts,10s); sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL,1); sql insert into t1 values(1648791223001,10,2,3,1.1,2); sql insert into t1 values(1648791233002,3,2,3,2.1,3); @@ -179,7 +179,7 @@ endi sql create database test2 vgroups 1; sql use test2; sql create table t2(ts timestamp, a int, b int , c int, d double, id int); -sql create stream streams2 trigger at_once watermark 1d into streamt2 as select _wstart,apercentile(a,30) c1, apercentile(a,70), apercentile(a,20,"t-digest") c2, apercentile(a,60,"t-digest") c3, max(id) c4 from t2 session(ts,10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 watermark 1d into streamt2 as select _wstart,apercentile(a,30) c1, apercentile(a,70), apercentile(a,20,"t-digest") c2, apercentile(a,60,"t-digest") c3, max(id) c4 from t2 session(ts,10s); sql insert into t2 values(1648791213001,1,1,3,1.0,1); sql insert into t2 values(1648791213002,2,2,6,3.4,2); sql insert into t2 values(1648791213003,4,9,3,4.8,3); @@ -229,13 +229,13 @@ endi sql create database test3 vgroups 1; sql use test3; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams3 trigger at_once watermark 1d into streamt3 as select _wstart, min(b), a,c from t1 session(ts,10s); -sql create stream streams4 trigger at_once watermark 1d into streamt4 as select _wstart, max(b), a,c from t1 session(ts,10s); -# sql create stream streams5 trigger at_once watermark 1d into streamt5 as select _wstart, top(b,3), a,c from t1 session(ts,10s); -# sql create stream streams6 trigger at_once watermark 1d into streamt6 as select _wstart, bottom(b,3), a,c from t1 session(ts,10s); -# sql create stream streams7 trigger at_once watermark 1d into streamt7 as select _wstart, spread(a), elapsed(ts), hyperloglog(a) from t1 session(ts,10s); -sql create stream streams7 trigger at_once watermark 1d into streamt7 as select _wstart, spread(a), hyperloglog(a) from t1 session(ts,10s); -# sql create stream streams8 trigger at_once watermark 1d into streamt8 as select _wstart, histogram(a,"user_input", "[1,3,5,7]", 1), histogram(a,"user_input", "[1,3,5,7]", 0) from t1 session(ts,10s); +sql create stream streams3 trigger at_once watermark 1d IGNORE EXPIRED 0 into streamt3 as select _wstart, min(b), a,c from t1 session(ts,10s); +sql create stream streams4 trigger at_once watermark 1d IGNORE EXPIRED 0 into streamt4 as select _wstart, max(b), a,c from t1 session(ts,10s); +# sql create stream streams5 trigger at_once watermark 1d IGNORE EXPIRED 0 into streamt5 as select _wstart, top(b,3), a,c from t1 session(ts,10s); +# sql create stream streams6 trigger at_once watermark 1d IGNORE EXPIRED 0 into streamt6 as select _wstart, bottom(b,3), a,c from t1 session(ts,10s); +# sql create stream streams7 trigger at_once watermark 1d IGNORE EXPIRED 0 into streamt7 as select _wstart, spread(a), elapsed(ts), hyperloglog(a) from t1 session(ts,10s); +sql create stream streams7 trigger at_once watermark 1d IGNORE EXPIRED 0 into streamt7 as select _wstart, spread(a), hyperloglog(a) from t1 session(ts,10s); +# sql create stream streams8 trigger at_once watermark 1d IGNORE EXPIRED 0 into streamt8 as select _wstart, histogram(a,"user_input", "[1,3,5,7]", 1), histogram(a,"user_input", "[1,3,5,7]", 0) from t1 session(ts,10s); sql insert into t1 values(1648791213001,1,1,1,1.0); sql insert into t1 values(1648791213002,2,3,2,3.4); sql insert into t1 values(1648791213003,4,9,3,4.8); diff --git a/tests/script/tsim/stream/session1.sim b/tests/script/tsim/stream/session1.sim index f535fd619f..3ad7c6f04e 100644 --- a/tests/script/tsim/stream/session1.sim +++ b/tests/script/tsim/stream/session1.sim @@ -17,7 +17,7 @@ sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double,id int); -sql create stream streams2 trigger at_once into streamt as select _wstart, count(*) c1, sum(a), min(b), max(id) s from t1 session(ts,10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, sum(a), min(b), max(id) s from t1 session(ts,10s); sql insert into t1 values(1648791210000,1,1,1,1.1,1); sql insert into t1 values(1648791220000,2,2,2,2.1,2); sql insert into t1 values(1648791230000,3,3,3,3.1,3); @@ -200,7 +200,7 @@ endi sql create database test1 vgroups 1; sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams3 trigger at_once into streamt3 as select _wstart, count(*) c1 from t1 where a > 5 session(ts, 5s); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart, count(*) c1 from t1 where a > 5 session(ts, 5s); sql insert into t1 values(1648791213000,1,2,3,1.0); $loop_count = 0 diff --git a/tests/script/tsim/stream/sliding.sim b/tests/script/tsim/stream/sliding.sim index 8287274cd2..e12e3c6686 100644 --- a/tests/script/tsim/stream/sliding.sim +++ b/tests/script/tsim/stream/sliding.sim @@ -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 table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -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 _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 _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 _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s) sliding (5s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 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 IGNORE EXPIRED 0 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 IGNORE EXPIRED 0 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 IGNORE EXPIRED 0 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(1648791216000,2,2,3,1.1); @@ -309,8 +309,8 @@ sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams11 trigger at_once into streamt as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s, 5s); -sql create stream streams12 trigger at_once into streamt2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s, 5s); +sql create stream streams11 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s, 5s); +sql create stream streams12 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s, 5s); sql insert into t1 values(1648791213000,1,2,3,1.0); sql insert into t1 values(1648791223001,2,2,3,1.1); @@ -442,9 +442,9 @@ sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams21 trigger at_once into streamt as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s, 5s); -sql create stream streams22 trigger at_once into streamt2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s, 5s); -sql create stream streams23 trigger at_once into streamt3 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(20s) sliding(10s); +sql create stream streams21 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s, 5s); +sql create stream streams22 trigger at_once IGNORE EXPIRED 0 into streamt2 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(10s, 5s); +sql create stream streams23 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart, count(*) c1, sum(a) c3 , max(b) c4, min(c) c5 from st interval(20s) sliding(10s); sql insert into t1 values(1648791213000,1,1,1,1.0); sql insert into t1 values(1648791223001,2,2,2,1.1); @@ -683,7 +683,7 @@ sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, count(*),min(a) c1 from st interval(10s) sliding(5s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into streamt4 as select _wstart as ts, count(*),min(a) c1 from st interval(10s) sliding(5s); sql insert into t1 values(1648791213000,1,1,1,1.0); sql insert into t1 values(1648791243000,2,1,1,1.0); diff --git a/tests/script/tsim/stream/state0.sim b/tests/script/tsim/stream/state0.sim index 07abdc0040..d009f742b7 100644 --- a/tests/script/tsim/stream/state0.sim +++ b/tests/script/tsim/stream/state0.sim @@ -17,9 +17,9 @@ sql use test; sql create table t1(ts timestamp, a int, b int , c int, d double, id int); -print create stream streams1 trigger at_once into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); +print create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); -sql create stream streams1 trigger at_once into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); sql insert into t1 values(1648791213000,1,2,3,1.0,1); sql insert into t1 values(1648791213000,1,2,3,1.0,2); @@ -453,9 +453,9 @@ sql use test1; sql create table t1(ts timestamp, a int, b int , c int, d double, id int); -print create stream streams2 trigger at_once into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); +print create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); -sql create stream streams2 trigger at_once into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into streamt1 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(a) c4, min(c) c5, max(id) c from t1 state_window(a); sql insert into t1 values(1648791212000,2,2,3,1.0,1); sql insert into t1 values(1648791213000,1,2,3,1.0,1); @@ -501,9 +501,9 @@ sql use test3; sql create table t1(ts timestamp, a int, b int , c int, d double, id int); -print create stream streams3 trigger at_once into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a); +print create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a); -sql create stream streams3 trigger at_once into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into streamt3 as select _wstart, count(*) c1, sum(b) c3 from t1 state_window(a); sql insert into t1 values(1648791212000,1,2,3,1.0,1); sql insert into t1 values(1648791213000,2,2,3,1.0,1); sql insert into t1 values(1648791214000,3,2,4,1.0,2); @@ -553,9 +553,9 @@ sql create table st (ts timestamp, c1 tinyint, c2 smallint) tags (t1 tinyint) ; sql create table t1 using st tags (-81) ; sql create table t2 using st tags (-81) ; -print create stream if not exists streams4 trigger window_close into streamt4 as select _wstart AS startts, min(c1),count(c1) from t1 state_window(c1); +print create stream if not exists streams4 trigger window_close IGNORE EXPIRED 0 into streamt4 as select _wstart AS startts, min(c1),count(c1) from t1 state_window(c1); -sql create stream if not exists streams4 trigger window_close into streamt4 as select _wstart AS startts, min(c1),count(c1) from t1 state_window(c1); +sql create stream if not exists streams4 trigger window_close IGNORE EXPIRED 0 into streamt4 as select _wstart AS startts, min(c1),count(c1) from t1 state_window(c1); sql insert into t1 (ts, c1) values (1668073288209, 11); sql insert into t1 (ts, c1) values (1668073288210, 11); @@ -742,9 +742,9 @@ sql create table tb (ts timestamp, a int); sql insert into tb values (now + 1m , 1 ); sql create table b (c timestamp, d int, e int , f int, g double); -print create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a); +print create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a); -sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a); +sql create stream streams0 trigger at_once IGNORE EXPIRED 0 into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a); sql insert into b values(1648791213000,NULL,NULL,NULL,NULL); sql select * from streamt order by c1, c2, c3; diff --git a/tests/script/tsim/stream/triggerInterval0.sim b/tests/script/tsim/stream/triggerInterval0.sim index b522dcf035..1c62f689ac 100644 --- a/tests/script/tsim/stream/triggerInterval0.sim +++ b/tests/script/tsim/stream/triggerInterval0.sim @@ -15,7 +15,7 @@ print $data00 $data01 $data02 sql use test sql create table t1(ts timestamp, a int, b int , c int, d double); -sql create stream streams1 trigger window_close into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); +sql create stream streams1 trigger window_close IGNORE EXPIRED 0 into streamt as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t1 interval(10s); sql insert into t1 values(1648791213001,1,2,3,1.0); sleep 300 diff --git a/tests/script/tsim/stream/triggerSession0.sim b/tests/script/tsim/stream/triggerSession0.sim index 4c664cf7c7..81a016be2b 100644 --- a/tests/script/tsim/stream/triggerSession0.sim +++ b/tests/script/tsim/stream/triggerSession0.sim @@ -15,7 +15,7 @@ print $data00 $data01 $data02 sql use test; sql create table t2(ts timestamp, a int, b int , c int, d double); -sql create stream streams2 trigger window_close into streamt2 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t2 session(ts, 10s); +sql create stream streams2 trigger window_close IGNORE EXPIRED 0 into streamt2 as select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5 from t2 session(ts, 10s); sql insert into t2 values(1648791213000,1,2,3,1.0); sql insert into t2 values(1648791222999,1,2,3,1.0); diff --git a/tests/script/tsim/stream/udTableAndTag0.sim b/tests/script/tsim/stream/udTableAndTag0.sim index 8bf34dc54c..e3ab344bbe 100644 --- a/tests/script/tsim/stream/udTableAndTag0.sim +++ b/tests/script/tsim/stream/udTableAndTag0.sim @@ -20,8 +20,8 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -#sql_error create stream streams1 trigger at_once into result.streamt SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); -sql create stream streams1 trigger at_once into result.streamt SUBTABLE(concat("aaa-", tbname)) as select _wstart, count(*) c1 from st partition by tbname interval(10s); +#sql_error create stream streams1 trigger at_once IGNORE EXPIRED 0 into result.streamt SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into result.streamt SUBTABLE(concat("aaa-", tbname)) as select _wstart, count(*) c1 from st partition by tbname interval(10s); sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,1,2,3); @@ -88,7 +88,7 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams2 trigger at_once into result2.streamt2 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st partition by concat("tag-", tbname) as cc interval(10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into result2.streamt2 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st partition by concat("tag-", tbname) as cc interval(10s); sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,1,2,3); @@ -173,7 +173,7 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams3 trigger at_once into result3.streamt3 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", tbname)) as select _wstart, count(*) c1 from st partition by concat("tag-", tbname) as dd, tbname interval(10s); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into result3.streamt3 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", tbname)) as select _wstart, count(*) c1 from st partition by concat("tag-", tbname) as dd, tbname interval(10s); sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,1,2,3); @@ -285,7 +285,7 @@ sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); sql create table t3 using st tags(3,3,3); -sql create stream streams4 trigger at_once into result4.streamt4 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", tbname)) as select _wstart, count(*) c1 from st partition by concat("tag-", tbname) as dd, tbname interval(10s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into result4.streamt4 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", tbname)) as select _wstart, count(*) c1 from st partition by concat("tag-", tbname) as dd, tbname interval(10s); sql insert into t1 values(1648791213000,1,1,1) t2 values(1648791213000,2,2,2) t3 values(1648791213000,3,3,3); @@ -404,7 +404,7 @@ sql create table t1 using st tags("1",1,1); sql create table t2 using st tags("2",2,2); sql create table t3 using st tags("3",3,3); -sql create stream streams6 trigger at_once into result6.streamt6 TAGS(dd int) as select _wstart, count(*) c1 from st partition by concat(ta, "0") as dd, tbname interval(10s); +sql create stream streams6 trigger at_once IGNORE EXPIRED 0 into result6.streamt6 TAGS(dd int) as select _wstart, count(*) c1 from st partition by concat(ta, "0") as dd, tbname interval(10s); sql insert into t1 values(1648791213000,1,1,1) t2 values(1648791213000,2,2,2) t3 values(1648791213000,3,3,3); diff --git a/tests/script/tsim/stream/udTableAndTag1.sim b/tests/script/tsim/stream/udTableAndTag1.sim index 4229de2cf0..fbedaa6e4e 100644 --- a/tests/script/tsim/stream/udTableAndTag1.sim +++ b/tests/script/tsim/stream/udTableAndTag1.sim @@ -20,8 +20,8 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -#sql_error create stream streams1 trigger at_once into result.streamt SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); -sql create stream streams1 trigger at_once into result.streamt SUBTABLE( concat("aaa-", cast(a as varchar(10) ) ) ) as select _wstart, count(*) c1 from st partition by a interval(10s); +#sql_error create stream streams1 trigger at_once IGNORE EXPIRED 0 into result.streamt SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into result.streamt SUBTABLE( concat("aaa-", cast(a as varchar(10) ) ) ) as select _wstart, count(*) c1 from st partition by a interval(10s); print ===== insert into 1 sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,2,2,3); @@ -87,7 +87,7 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams2 trigger at_once into result2.streamt2 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st partition by concat("col-", cast(a as varchar(10) ) ) as cc interval(10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into result2.streamt2 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st partition by concat("col-", cast(a as varchar(10) ) ) as cc interval(10s); print ===== insert into 2 sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,2,2,3); @@ -171,7 +171,7 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams3 trigger at_once into result3.streamt3 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", cast(a as varchar(10) ) ) ) as select _wstart, count(*) c1 from st partition by concat("col-", cast(a as varchar(10) ) ) as dd, a interval(10s); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into result3.streamt3 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", cast(a as varchar(10) ) ) ) as select _wstart, count(*) c1 from st partition by concat("col-", cast(a as varchar(10) ) ) as dd, a interval(10s); print ===== insert into 3 sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,2,2,3); @@ -283,7 +283,7 @@ sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); sql create table t3 using st tags(3,3,3); -sql create stream streams4 trigger at_once into result4.streamt4 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", dd)) as select _wstart, count(*) c1 from st partition by concat("t", cast(a as varchar(10) ) ) as dd interval(10s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into result4.streamt4 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", dd)) as select _wstart, count(*) c1 from st partition by concat("t", cast(a as varchar(10) ) ) as dd interval(10s); sql insert into t1 values(1648791213000,1,1,1) t2 values(1648791213000,2,2,2) t3 values(1648791213000,3,3,3); diff --git a/tests/script/tsim/stream/udTableAndTag2.sim b/tests/script/tsim/stream/udTableAndTag2.sim index bacc301ad0..c0e72712df 100644 --- a/tests/script/tsim/stream/udTableAndTag2.sim +++ b/tests/script/tsim/stream/udTableAndTag2.sim @@ -20,7 +20,7 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams1 trigger at_once into result.streamt SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams1 trigger at_once IGNORE EXPIRED 0 into result.streamt SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); print ===== insert into 1 sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,2,2,3); @@ -93,7 +93,7 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams2 trigger at_once into result2.streamt2 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams2 trigger at_once IGNORE EXPIRED 0 into result2.streamt2 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st interval(10s); print ===== insert into 2 sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,2,2,3); @@ -195,7 +195,7 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams3 trigger at_once into result3.streamt3 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", "1") ) as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams3 trigger at_once IGNORE EXPIRED 0 into result3.streamt3 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", "1") ) as select _wstart, count(*) c1 from st interval(10s); print ===== insert into 3 sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,2,2,3); @@ -305,7 +305,7 @@ sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); sql create table t3 using st tags(3,3,3); -sql create stream streams4 trigger at_once into result4.streamt4 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", "1")) as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams4 trigger at_once IGNORE EXPIRED 0 into result4.streamt4 TAGS(dd varchar(100)) SUBTABLE(concat("tbn-", "1")) as select _wstart, count(*) c1 from st interval(10s); sql insert into t1 values(1648791213000,1,1,1) t2 values(1648791213000,2,2,2) t3 values(1648791213000,3,3,3); @@ -375,9 +375,9 @@ sql create stable st(ts timestamp,a int,b int,c int) tags(ta int,tb int,tc int); sql create table t1 using st tags(1,1,1); sql create table t2 using st tags(2,2,2); -sql create stream streams51 trigger at_once into result5.streamt51 SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); -sql create stream streams52 trigger at_once into result5.streamt52 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st interval(10s); -sql create stream streams53 trigger at_once into result5.streamt53 TAGS(dd varchar(100)) SUBTABLE(concat("aaa-", "1") ) as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams51 trigger at_once IGNORE EXPIRED 0 into result5.streamt51 SUBTABLE("aaa") as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams52 trigger at_once IGNORE EXPIRED 0 into result5.streamt52 TAGS(cc varchar(100)) as select _wstart, count(*) c1 from st interval(10s); +sql create stream streams53 trigger at_once IGNORE EXPIRED 0 into result5.streamt53 TAGS(dd varchar(100)) SUBTABLE(concat("aaa-", "1") ) as select _wstart, count(*) c1 from st interval(10s); sql insert into t1 values(1648791213000,1,2,3); sql insert into t2 values(1648791213000,2,2,3); diff --git a/tests/system-test/1-insert/database_pre_suf.py b/tests/system-test/1-insert/database_pre_suf.py index 488dfebff5..a6ff95ab3f 100755 --- a/tests/system-test/1-insert/database_pre_suf.py +++ b/tests/system-test/1-insert/database_pre_suf.py @@ -108,7 +108,7 @@ class TDTestCase: # create stream - tdSql.execute('''create stream current_stream into stream_max_stable_1 as select _wstart as startts, _wend as wend, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 where ts is not null interval (5s);''') + tdSql.execute('''create stream current_stream trigger at_once IGNORE EXPIRED 0 into stream_max_stable_1 as select _wstart as startts, _wend as wend, max(q_int) as max_int, min(q_bigint) as min_int from stable_1 where ts is not null interval (5s);''') # insert data for i in range(num_random*n): diff --git a/tests/system-test/1-insert/drop.py b/tests/system-test/1-insert/drop.py index c15a9bbc35..a8bfea2741 100644 --- a/tests/system-test/1-insert/drop.py +++ b/tests/system-test/1-insert/drop.py @@ -138,14 +138,14 @@ class TDTestCase: stream_name = tdCom.getLongName(5,"letters") tdSql.execute(f'create table {stbname} (ts timestamp,c0 int) tags(t0 int)') tdSql.execute(f'create table tb using {stbname} tags(1)') - tdSql.execute(f'create stream {stream_name} into stb as select * from {self.dbname}.{stbname} partition by tbname') + tdSql.execute(f'create stream {stream_name} trigger at_once ignore expired 0 into stb as select * from {self.dbname}.{stbname} partition by tbname') tdSql.query(f'select * from information_schema.ins_streams where stream_name = "{stream_name}"') print(tdSql.queryResult) - tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} into stb as select * from {self.dbname}.{stbname} partition by tbname') + tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} trigger at_once ignore expired 0 into stb as select * from {self.dbname}.{stbname} partition by tbname') tdSql.execute(f'drop stream {stream_name}') - tdSql.execute(f'create stream {stream_name} into stb1 as select * from tb') + tdSql.execute(f'create stream {stream_name} trigger at_once ignore expired 0 into stb1 as select * from tb') tdSql.query(f'select * from information_schema.ins_streams where stream_name = "{stream_name}"') - tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} into stb1 as select * from tb') + tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} trigger at_once ignore expired 0 into stb1 as select * from tb') tdSql.execute(f'drop database {self.dbname}') def run(self): self.drop_ntb_check() diff --git a/utils/test/c/tmq_taosx_ci.c b/utils/test/c/tmq_taosx_ci.c index 3d458d90c1..1f25eae366 100644 --- a/utils/test/c/tmq_taosx_ci.c +++ b/utils/test/c/tmq_taosx_ci.c @@ -409,7 +409,7 @@ int buildStable(TAOS* pConn, TAOS_RES* pRes) { taos_free_result(pRes); pRes = taos_query(pConn, - "create stream meters_summary_s into meters_summary as select _wstart, max(current) as current, " + "create stream meters_summary_s trigger at_once IGNORE EXPIRED 0 into meters_summary as select _wstart, max(current) as current, " "groupid, location from meters partition by groupid, location interval(10m)"); if (taos_errno(pRes) != 0) { printf("failed to create super table meters_summary, reason:%s\n", taos_errstr(pRes));