feat:window close && ignore expired

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

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;

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();

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);

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);

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);

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);

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);
 

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
 

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;

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

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
 

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');

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');

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');

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');

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');
 

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');

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

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');

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);

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);

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);

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);

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);

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);

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
 

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);

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

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);

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;
 

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

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);

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);
 
 

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);
 
 

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);

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):        

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()

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));