Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/tsdb_refact

include/os/osSysinfo.h

@@ -68,6 +68,7 @@ typedef struct {
 } SysNameInfo;
 
 SysNameInfo taosGetSysNameInfo();
+bool taosCheckCurrentInDll();
 
 #ifdef __cplusplus
 }

source/client/src/clientHb.c

@@ -592,7 +592,9 @@ void hbThreadFuncUnexpectedStopped(void) {
 static void *hbThreadFunc(void *param) {
   setThreadName("hb");
 #ifdef WINDOWS
-  atexit(hbThreadFuncUnexpectedStopped);
+  if (taosCheckCurrentInDll()) {
+    atexit(hbThreadFuncUnexpectedStopped);
+  }
 #endif
   while (1) {
     int8_t threadStop = atomic_val_compare_exchange_8(&clientHbMgr.threadStop, 1, 2);

source/libs/catalog/src/ctgCache.c

@@ -1938,7 +1938,9 @@ void ctgCleanupCacheQueue(void) {
 void* ctgUpdateThreadFunc(void* param) {
   setThreadName("catalog");
 #ifdef WINDOWS
-  atexit(ctgUpdateThreadUnexpectedStopped);
+  if (taosCheckCurrentInDll()) {
+    atexit(ctgUpdateThreadUnexpectedStopped);
+  }
 #endif
   qInfo("catalog update thread started");
 

source/libs/executor/src/timewindowoperator.c

@@ -1257,6 +1257,10 @@ static int32_t getAllIntervalWindow(SHashObj* pHashMap, SArray* resWins) {
   return TSDB_CODE_SUCCESS;
 }
 
+bool isCloseWindow(STimeWindow *pWin, STimeWindowAggSupp* pSup) {
+  return pWin->ekey < pSup->maxTs - pSup->waterMark;
+}
+
 static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup, SInterval* pInterval,
                                    SArray* closeWins) {
   void*  pIte = NULL;
@@ -1269,7 +1273,7 @@ static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup,
     SResultRowInfo dumyInfo;
     dumyInfo.cur.pageId = -1;
     STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, pInterval->precision, NULL);
-    if (win.ekey < pSup->maxTs - pSup->waterMark) {
+    if (isCloseWindow(&win, pSup)) {
       char keyBuf[GET_RES_WINDOW_KEY_LEN(sizeof(TSKEY))];
       SET_RES_WINDOW_KEY(keyBuf, &ts, sizeof(TSKEY), groupId);
       taosHashRemove(pHashMap, keyBuf, keyLen);
@@ -2036,7 +2040,55 @@ _error:
   return NULL;
 }
 
-static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBlock, int32_t tableGroupId,
+bool isFinalInterval(SStreamFinalIntervalOperatorInfo* pInfo) { return pInfo->pChildren != NULL; }
+
+void compactFunctions(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int32_t numOfOutput,
+                      SExecTaskInfo* pTaskInfo) {
+  for (int32_t k = 0; k < numOfOutput; ++k) {
+    if (fmIsWindowPseudoColumnFunc(pDestCtx[k].functionId)) {
+      continue;
+    }
+    int32_t code = TSDB_CODE_SUCCESS;
+    if (functionNeedToExecute(&pDestCtx[k]) && pDestCtx[k].fpSet.combine != NULL) {
+      code = pDestCtx[k].fpSet.combine(&pDestCtx[k], &pSourceCtx[k]);
+      if (code != TSDB_CODE_SUCCESS) {
+        qError("%s apply functions error, code: %s", GET_TASKID(pTaskInfo), tstrerror(code));
+        pTaskInfo->code = code;
+        longjmp(pTaskInfo->env, code);
+      }
+    }
+  }
+}
+
+static void rebuildIntervalWindow(SStreamFinalIntervalOperatorInfo* pInfo, SArray* pWinArray, int32_t groupId,
+                                  int32_t numOfOutput, SExecTaskInfo* pTaskInfo) {
+  int32_t size = taosArrayGetSize(pWinArray);
+  ASSERT(pInfo->pChildren);
+  for (int32_t i = 0; i < size; i++) {
+    STimeWindow* pParentWin = taosArrayGet(pWinArray, i);
+    SResultRow*  pCurResult = NULL;
+    setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, pParentWin, true, &pCurResult, 0, pInfo->binfo.pCtx,
+                           numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup, pTaskInfo);
+    int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
+    for (int32_t j = 0; j < numOfChildren; j++) {
+      SOperatorInfo*            pChildOp = taosArrayGetP(pInfo->pChildren, j);
+      SIntervalAggOperatorInfo* pChInfo = pChildOp->info;
+      SResultRow*               pChResult = NULL;
+      setTimeWindowOutputBuf(&pChInfo->binfo.resultRowInfo, pParentWin, true, &pChResult, 0, pChInfo->binfo.pCtx,
+                             pChildOp->numOfExprs, pChInfo->binfo.rowCellInfoOffset, &pChInfo->aggSup, pTaskInfo);
+      compactFunctions(pInfo->binfo.pCtx, pChInfo->binfo.pCtx, numOfOutput, pTaskInfo);
+    }
+  }
+}
+
+bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup) {
+  SET_RES_WINDOW_KEY(pSup->keyBuf, &pWin->skey, sizeof(int64_t), groupId);
+  SResultRowPosition* p1 = (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable,
+      pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(sizeof(int64_t)));
+  return p1 == NULL;
+}
+
+static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBlock, uint64_t tableGroupId,
                            SArray* pUpdated) {
   SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)pOperatorInfo->info;
   SResultRowInfo*                   pResultRowInfo = &(pInfo->binfo.resultRowInfo);
@@ -2060,6 +2112,14 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
   STimeWindow nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval,
                                             pInfo->interval.precision, NULL);
   while (1) {
+    if (isFinalInterval(pInfo) && isCloseWindow(&nextWin, &pInfo->twAggSup) &&
+        isDeletedWindow(&nextWin, tableGroupId, &pInfo->aggSup)) {
+      SArray* pUpWins = taosArrayInit(8, sizeof(STimeWindow));
+      taosArrayPush(pUpWins, &nextWin);
+      rebuildIntervalWindow(pInfo, pUpWins, pInfo->binfo.pRes->info.groupId,
+          pOperatorInfo->numOfExprs, pOperatorInfo->pTaskInfo);
+      taosArrayDestroy(pUpWins);
+    }
     int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, true, &pResult, tableGroupId, pInfo->binfo.pCtx,
                                           numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup, pTaskInfo);
     if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
@@ -2089,47 +2149,6 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
   }
 }
 
-bool isFinalInterval(SStreamFinalIntervalOperatorInfo* pInfo) { return pInfo->pChildren != NULL; }
-
-void compactFunctions(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int32_t numOfOutput,
-                      SExecTaskInfo* pTaskInfo) {
-  for (int32_t k = 0; k < numOfOutput; ++k) {
-    if (fmIsWindowPseudoColumnFunc(pDestCtx[k].functionId)) {
-      continue;
-    }
-    int32_t code = TSDB_CODE_SUCCESS;
-    if (functionNeedToExecute(&pDestCtx[k]) && pDestCtx[k].fpSet.combine != NULL) {
-      code = pDestCtx[k].fpSet.combine(&pDestCtx[k], &pSourceCtx[k]);
-      if (code != TSDB_CODE_SUCCESS) {
-        qError("%s apply functions error, code: %s", GET_TASKID(pTaskInfo), tstrerror(code));
-        pTaskInfo->code = code;
-        longjmp(pTaskInfo->env, code);
-      }
-    }
-  }
-}
-
-static void rebuildIntervalWindow(SStreamFinalIntervalOperatorInfo* pInfo, SArray* pWinArray, int32_t groupId,
-                                  int32_t numOfOutput, SExecTaskInfo* pTaskInfo) {
-  int32_t size = taosArrayGetSize(pWinArray);
-  ASSERT(pInfo->pChildren);
-  for (int32_t i = 0; i < size; i++) {
-    STimeWindow* pParentWin = taosArrayGet(pWinArray, i);
-    SResultRow*  pCurResult = NULL;
-    setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, pParentWin, true, &pCurResult, 0, pInfo->binfo.pCtx,
-                           numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup, pTaskInfo);
-    int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
-    for (int32_t j = 0; j < numOfChildren; j++) {
-      SOperatorInfo*            pChildOp = taosArrayGetP(pInfo->pChildren, j);
-      SIntervalAggOperatorInfo* pChInfo = pChildOp->info;
-      SResultRow*               pChResult = NULL;
-      setTimeWindowOutputBuf(&pChInfo->binfo.resultRowInfo, pParentWin, true, &pChResult, 0, pChInfo->binfo.pCtx,
-                             pChildOp->numOfExprs, pChInfo->binfo.rowCellInfoOffset, &pChInfo->aggSup, pTaskInfo);
-      compactFunctions(pInfo->binfo.pCtx, pChInfo->binfo.pCtx, numOfOutput, pTaskInfo);
-    }
-  }
-}
-
 static void clearStreamIntervalOperator(SStreamFinalIntervalOperatorInfo* pInfo) {
   taosHashClear(pInfo->aggSup.pResultRowHashTable);
   clearDiskbasedBuf(pInfo->aggSup.pResultBuf);
@@ -2169,6 +2188,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
   SStreamFinalIntervalOperatorInfo* pInfo = pOperator->info;
   SOperatorInfo*                    downstream = pOperator->pDownstream[0];
   SArray*                           pUpdated = taosArrayInit(4, POINTER_BYTES);
+  TSKEY                             maxTs = INT64_MIN;
 
   if (pOperator->status == OP_EXEC_DONE) {
     return NULL;
@@ -2222,6 +2242,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
     }
 
     setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
+    doHashInterval(pOperator, pBlock, pBlock->info.groupId, pUpdated);
     if (isFinalInterval(pInfo)) {
       int32_t chIndex = getChildIndex(pBlock);
       int32_t size = taosArrayGetSize(pInfo->pChildren);
@@ -2238,10 +2259,10 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
       setInputDataBlock(pChildOp, pChInfo->binfo.pCtx, pBlock, pChInfo->order, MAIN_SCAN, true);
       doHashInterval(pChildOp, pBlock, pBlock->info.groupId, NULL);
     }
-    doHashInterval(pOperator, pBlock, pBlock->info.groupId, pUpdated);
-    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
+    maxTs = TMAX(maxTs, pBlock->info.window.ekey);
   }
 
+  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
   if (isFinalInterval(pInfo)) {
     closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pUpdated);
   }
@@ -2564,7 +2585,7 @@ int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pTs, int32_t
 }
 
 static int32_t setWindowOutputBuf(SResultWindowInfo* pWinInfo, SResultRow** pResult, SqlFunctionCtx* pCtx,
-                                  int32_t groupId, int32_t numOfOutput, int32_t* rowCellInfoOffset,
+                                  uint64_t groupId, int32_t numOfOutput, int32_t* rowCellInfoOffset,
                                   SStreamAggSupporter* pAggSup, SExecTaskInfo* pTaskInfo) {
   assert(pWinInfo->win.skey <= pWinInfo->win.ekey);
   // too many time window in query
@@ -2642,7 +2663,7 @@ int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap)
   return size - startIndex - 1;
 }
 
-void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, int32_t groupId,
+void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, uint64_t groupId,
                        int32_t numOfOutput, SExecTaskInfo* pTaskInfo, SHashObj* pStUpdated, SHashObj* pStDeleted) {
   SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pCurWins, startIndex);
   SResultRow*        pCurResult = NULL;
@@ -2667,13 +2688,18 @@ void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex,
   }
 }
 
+typedef struct SWinRes {
+  TSKEY    ts;
+  uint64_t groupId;
+} SWinRes;
+
 static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SHashObj* pStUpdated,
                                    SHashObj* pStDeleted) {
   SExecTaskInfo*                 pTaskInfo = pOperator->pTaskInfo;
   SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
   bool                           masterScan = true;
   int32_t                        numOfOutput = pOperator->numOfExprs;
-  int64_t                        groupId = pSDataBlock->info.groupId;
+  uint64_t                       groupId = pSDataBlock->info.groupId;
   int64_t                        gap = pInfo->gap;
   int64_t                        code = TSDB_CODE_SUCCESS;
 
@@ -2693,7 +2719,7 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
   SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
   for (int32_t i = 0; i < pSDataBlock->info.rows;) {
     int32_t            winIndex = 0;
-    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], pSDataBlock->info.groupId, gap, &winIndex);
+    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, tsCols[i], groupId, gap, &winIndex);
     winRows = updateSessionWindowInfo(pCurWin, tsCols, pSDataBlock->info.rows, i, pInfo->gap, pStDeleted);
     code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pTaskInfo);
     if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
@@ -2709,7 +2735,8 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
     }
     pCurWin->isClosed = false;
     if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
-      code = taosHashPut(pStUpdated, &pCurWin->pos, sizeof(SResultRowPosition), &(pCurWin->win.skey), sizeof(TSKEY));
+      SWinRes value = {.ts = pCurWin->win.skey, .groupId = groupId};
+      code = taosHashPut(pStUpdated, &pCurWin->pos, sizeof(SResultRowPosition), &value, sizeof(SWinRes));
       if (code != TSDB_CODE_SUCCESS) {
         longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
@@ -2736,7 +2763,7 @@ static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SOptrBasicInfo*
   }
 }
 
-static int32_t copyUpdateResult(SHashObj* pStUpdated, SArray* pUpdated, int32_t groupId) {
+static int32_t copyUpdateResult(SHashObj* pStUpdated, SArray* pUpdated) {
   void*  pData = NULL;
   size_t keyLen = 0;
   while ((pData = taosHashIterate(pStUpdated, pData)) != NULL) {
@@ -2746,9 +2773,9 @@ static int32_t copyUpdateResult(SHashObj* pStUpdated, SArray* pUpdated, int32_t
     if (pos == NULL) {
       return TSDB_CODE_QRY_OUT_OF_MEMORY;
     }
-    pos->groupId = groupId;
+    pos->groupId = ((SWinRes*)pData)->groupId;
     pos->pos = *(SResultRowPosition*)key;
-    *(int64_t*)pos->key = *(uint64_t*)pData;
+    *(int64_t*)pos->key = ((SWinRes*)pData)->ts;
     taosArrayPush(pUpdated, &pos);
   }
   return TSDB_CODE_SUCCESS;
@@ -2815,7 +2842,9 @@ int32_t closeSessionWindow(SHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SArra
     __get_win_info_ fn) {
   // Todo(liuyao) save window to tdb
   void **pIte = NULL;
+  size_t keyLen = 0;
   while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
+    uint64_t* pGroupId = taosHashGetKey(pIte, &keyLen);
     SArray *pWins = (SArray *) (*pIte);
     int32_t size = taosArrayGetSize(pWins);
     for (int32_t i = 0; i < size; i++) {
@@ -2825,7 +2854,7 @@ int32_t closeSessionWindow(SHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SArra
         if (!pSeWin->isClosed) {
           pSeWin->isClosed = true;
           if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
-            int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
+            int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, *pGroupId, pClosed);
             pSeWin->isOutput = true;
           }
         }
@@ -2892,7 +2921,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
         SStreamSessionAggOperatorInfo* pChildInfo = pChildOp->info;
         doClearSessionWindows(&pChildInfo->streamAggSup, &pChildInfo->binfo, pBlock, 0, pChildOp->numOfExprs,
                               pChildInfo->gap, NULL);
-        rebuildTimeWindow(pInfo, pWins, pInfo->binfo.pRes->info.groupId, pOperator->numOfExprs, pOperator->pTaskInfo);
+        rebuildTimeWindow(pInfo, pWins, pBlock->info.groupId, pOperator->numOfExprs, pOperator->pTaskInfo);
       }
       taosArrayDestroy(pWins);
       continue;
@@ -2916,7 +2945,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
 
   closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated,
                      getSessionWinInfo);
-  copyUpdateResult(pStUpdated, pUpdated, pBInfo->pRes->info.groupId);
+  copyUpdateResult(pStUpdated, pUpdated);
   taosHashCleanup(pStUpdated);
 
   finalizeUpdatedResult(pOperator->numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated,
@@ -3216,8 +3245,9 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
     }
     pCurWin->winInfo.isClosed = false;
     if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
-      code = taosHashPut(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition), &(pCurWin->winInfo.win.skey),
-                         sizeof(TSKEY));
+      SWinRes value = {.ts = pCurWin->winInfo.win.skey, .groupId = groupId};
+      code = taosHashPut(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition),
+          &value, sizeof(SWinRes));
       if (code != TSDB_CODE_SUCCESS) {
         longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
@@ -3274,7 +3304,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
 
   closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated,
                      getStateWinInfo);
-  copyUpdateResult(pSeUpdated, pUpdated, pBInfo->pRes->info.groupId);
+  copyUpdateResult(pSeUpdated, pUpdated);
   taosHashCleanup(pSeUpdated);
 
   finalizeUpdatedResult(pOperator->numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated,

source/os/src/osSysinfo.c

@@ -945,3 +945,19 @@ SysNameInfo taosGetSysNameInfo() {
   return info;
 #endif
 }
+
+
+bool taosCheckCurrentInDll() {
+#ifdef WINDOWS
+  MEMORY_BASIC_INFORMATION mbi;
+  char path[PATH_MAX] = {0};
+  GetModuleFileName(((VirtualQuery(taosCheckCurrentInDll,&mbi,sizeof(mbi)) != 0) ? (HMODULE)mbi.AllocationBase : NULL), path, PATH_MAX);
+  int strLastIndex = strlen(path);
+  if ((path[strLastIndex-3] == 'd' || path[strLastIndex-3] == 'D') && (path[strLastIndex-2] == 'l' || path[strLastIndex-2] == 'L') && (path[strLastIndex-1] == 'l' || path[strLastIndex-1] == 'L')) {
+    return true;
+  }
+  return false;
+#else
+  return false;
+#endif
+}
\ No newline at end of file

source/util/src/tcache.c

@@ -829,7 +829,11 @@ void *taosCacheTimedRefresh(void *handle) {
 
   const int32_t SLEEP_DURATION = 500;  // 500 ms
   int64_t       count = 0;
-  atexit(taosCacheRefreshWorkerUnexpectedStopped);
+#ifdef WINDOWS
+  if (taosCheckCurrentInDll()) {
+    atexit(taosCacheRefreshWorkerUnexpectedStopped);
+  }
+#endif
 
   while (1) {
     taosMsleep(SLEEP_DURATION);

tests/script/tsim/stream/distributeInterval0.sim

@@ -173,4 +173,39 @@ endi
 
 sql select  _wstartts, count(*) c1, count(d) c2 , sum(a) c3 , max(b)  c4, min(c) c5, avg(d) from st interval(10s);
 
-system sh/exec.sh -n dnode1 -s stop -x SIGINT
+
\ No newline at end of file
+sql create database test1  vgroups 1;
+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 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 streamtST1 as select  _wstartts, count(*) c1, count(a) c2 , sum(a) c3 ,  max(b)  c5, min(c) c6 from st interval(10s) ;
+
+sql insert into ts1 values(1648791211000,1,2,3);
+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);
+
+$loop_count = 0
+loop2:
+sql select * from streamtST1;
+
+sleep 300
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+# row 0
+if $data01 != 2 then
+  print =====data01=$data01
+  goto loop2
+endi
+
+#rows 1
+if $data11 != 2 then
+  print =====data11=$data11
+  goto loop2
+endi
+
+system sh/stop_dnodes.sh
\ No newline at end of file

tests/script/tsim/stream/partitionby.sim

@@ -34,6 +34,7 @@ print =====rows=$rows
 goto loop0
 endi
 
+print =====loop0
 
 sql create database test1  vgroups 1;
 sql use test1;
@@ -51,7 +52,7 @@ sql insert into ts2 values(1648791211000,1,2,3);
 
 $loop_count = 0
 
-loop0:
+loop1:
 sleep 300
 sql select * from streamt;
 
@@ -62,7 +63,62 @@ endi
 
 if $rows != 2 then
 print =====rows=$rows
-goto loop0
+goto loop1
 endi
 
-system sh/exec.sh -n dnode1 -s stop -x SIGINT
+print =====loop1
\ No newline at end of file
+
+sql create database test2  vgroups 1;
+sql use test2;
+sql create stable st(ts timestamp,a int,b int,c int,id 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_t2 trigger at_once  into streamtST as select  _wstartts, 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);
+sql insert into ts2 values(1648791222001,2,2,3,4);
+
+sql insert into ts2 values(1648791222002,2,2,3,5);
+sql insert into ts2 values(1648791222002,2,2,3,6);
+
+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);
+sql insert into ts2 values(1648791222001,2,2,3,4);
+
+$loop_count = 0
+
+loop2:
+sleep 300
+sql select * from streamtST;
+
+$loop_count = $loop_count + 1
+if $loop_count == 10 then
+  return -1
+endi
+
+if $data01 != 1 then
+print =====data01=$data01
+goto loop2
+endi
+
+if $data02 != 1 then
+print =====data02=$data02
+goto loop2
+endi
+
+if $data03 != 1 then
+print =====data03=$data03
+goto loop2
+endi
+
+if $data04 != 2 then
+print =====data04=$data04
+goto loop2
+endi
+
+print =====loop2
+
+system sh/stop_dnodes.sh
\ No newline at end of file