Merge pull request #13996 from taosdata/feature/TD-16616

feat(stream): add ci

source/libs/executor/src/timewindowoperator.c

@@ -2892,7 +2892,10 @@ static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWin
       SArray*                        pChWins = getWinInfos(&pChInfo->streamAggSup, groupId);
       int32_t                        chWinSize = taosArrayGetSize(pChWins);
       int32_t index = binarySearch(pChWins, chWinSize, pParentWin->win.skey, TSDB_ORDER_DESC, getSessionWindowEndkey);
-      for (int32_t k = index; k > 0 && k < chWinSize; k++) {
+      if (index < 0) {
+        index  = 0;
+      }
+      for (int32_t k = index; k < chWinSize; k++) {
         SResultWindowInfo* pcw = taosArrayGet(pChWins, k);
         if (pParentWin->win.skey <= pcw->win.skey && pcw->win.ekey <= pParentWin->win.ekey) {
           SResultRow* pChResult = NULL;

source/libs/function/src/builtinsimpl.c

@@ -1481,11 +1481,13 @@ int32_t minMaxCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int3
     if (pSBuf->assign && ((((*(double*)&pDBuf->v) < (*(double*)&pSBuf->v)) ^ isMinFunc) || !pDBuf->assign)) {
       *(double*)&pDBuf->v = *(double*)&pSBuf->v;
       replaceTupleData(&pDBuf->tuplePos, &pSBuf->tuplePos);
+      pDBuf->assign = true;
     }
   } else {
     if (pSBuf->assign && (((pDBuf->v < pSBuf->v) ^ isMinFunc) || !pDBuf->assign)) {
       pDBuf->v = pSBuf->v;
       replaceTupleData(&pDBuf->tuplePos, &pSBuf->tuplePos);
+      pDBuf->assign = true;
     }
   }
   pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes);

tests/script/jenkins/basic.txt

@@ -73,6 +73,7 @@
 ./test.sh -f tsim/stream/basic1.sim
 ./test.sh -f tsim/stream/basic2.sim
 # ./test.sh -f tsim/stream/distributeInterval0.sim
+# ./test.sh -f tsim/stream/distributesession0.sim
 # ./test.sh -f tsim/stream/session0.sim
 # ./test.sh -f tsim/stream/session1.sim
 # ./test.sh -f tsim/stream/state0.sim

tests/script/tsim/stream/distributesession0.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/deploy.sh -n dnode2 -i 2
+
+system sh/exec.sh -n dnode1 -s start 
+sleep 50
+sql connect
+
+sql create dnode $hostname2 port 7200
+
+system sh/exec.sh -n dnode2 -s start
+
+sql create database test  vgroups 4;
+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  _wstartts, count(*) c1, sum(a) c2 ,  max(b)  c3 from st session(ts, 10s) ;
+
+sleep 1000
+
+sql insert into ts1 values(1648791211000,1,1,1) (1648791211005,1,1,1);
+sql insert into ts2 values(1648791221004,1,2,3) (1648791221008,2,2,3);
+sql insert into ts1 values(1648791211005,1,1,1);
+sql insert into ts2 values(1648791221006,5,5,5) (1648791221007,5,5,5);
+sql insert into ts2 values(1648791221008,5,5,5) (1648791221008,5,5,5)(1648791221006,5,5,5);
+sql insert into ts1 values(1648791231000,1,1,1) (1648791231002,1,1,1) (1648791231006,1,1,1);
+sql insert into ts1 values(1648791211000,6,6,6) (1648791231002,2,2,2);
+sql insert into ts1 values(1648791211002,7,7,7);
+sql insert into ts1 values(1648791211002,7,7,7) ts2 values(1648791221008,5,5,5) ;
+
+$loop_count = 0
+loop1:
+sql select * from streamtST;
+
+sleep 300
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+# row 0
+if $data01 != 10 then
+  print =====data01=$data01
+  goto loop1
+endi
+
+if $data02 != 34 then
+  print =====data02=$data02
+  goto loop1
+endi
+
+if $data03 != 7 then
+  print ======$data03
+  return -1
+endi
+
+system sh/stop_dnodes.sh
\ No newline at end of file