Merge pull request #14881 from taosdata/feature/TD-17357

feat(stream): interval offset

source/libs/executor/src/executil.c

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

source/libs/stream/src/streamUpdate.c

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

tests/script/jenkins/basic.txt

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

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  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from t1 interval(10s) sliding (5s);
-sql create stream streams2 trigger at_once  watermark 1d into streamt2 as select  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from t1 interval(10s) sliding (5s);
-sql create stream stream_t1 trigger at_once  into streamtST as select  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from st interval(10s) sliding (5s);
-sql create stream stream_t2 trigger at_once watermark 1d into streamtST2 as select  _wstartts, count(*) c1, sum(a) c3 , max(b)  c4, min(c) c5 from st interval(10s) sliding (5s);
+sql create stream 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 insert into t1 values(1648791210000,1,2,3,1.0);
 sql insert into t1 values(1648791216000,2,2,3,1.1);