Merge pull request #14243 from taosdata/feature/TD-16728

feat(stream): auto pull data

include/common/tcommon.h

@@ -42,12 +42,13 @@ enum {
 typedef enum EStreamType {
   STREAM_NORMAL = 1,
   STREAM_INVERT,
-  STREAM_REPROCESS,
+  STREAM_CLEAR,
   STREAM_INVALID,
   STREAM_GET_ALL,
   STREAM_DELETE,
   STREAM_RETRIEVE,
   STREAM_PUSH_DATA,
+  STREAM_PUSH_EMPTY,
 } EStreamType;
 
 typedef struct {

include/common/tdatablock.h

@@ -224,6 +224,7 @@ size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
 int32_t blockDataTrimFirstNRows(SSDataBlock* pBlock, size_t n);
 
 int32_t      assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src);
+int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src);
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData);
 SSDataBlock* createDataBlock();
 int32_t      blockDataAppendColInfo(SSDataBlock* pBlock, SColumnInfoData* pColInfoData);
@@ -236,6 +237,8 @@ void        blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t*
 const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t numOfRows, const char* pData);
 
 void blockDebugShowData(const SArray* dataBlocks, const char* flag);
+// for debug
+char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
 
 int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks, STSchema* pTSchema, int32_t vgId,
                                     tb_uid_t suid);

include/libs/stream/tstream.h

@@ -319,7 +319,7 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem
       return -1;
     }
     taosWriteQitem(pTask->inputQueue->queue, pSubmitClone);
-  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK) {
+  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
     taosWriteQitem(pTask->inputQueue->queue, pItem);
   } else if (pItem->type == STREAM_INPUT__CHECKPOINT) {
     taosWriteQitem(pTask->inputQueue->queue, pItem);

include/libs/stream/tstreamUpdate.h

@@ -32,12 +32,15 @@ typedef struct SUpdateInfo {
   int64_t interval;
   int64_t watermark;
   TSKEY minTS;
+  SScalableBf* pCloseWinSBF;
 } SUpdateInfo;
 
 SUpdateInfo *updateInfoInitP(SInterval* pInterval, int64_t watermark);
 SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t watermark);
 bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts);
 void updateInfoDestroy(SUpdateInfo *pInfo);
+void updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo);
+void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo);
 
 #ifdef __cplusplus
 }

source/common/src/tdatablock.c

@@ -1164,7 +1164,7 @@ int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows)
 
 int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
   int32_t code = 0;
-  ASSERT(numOfRows > 0);
+  //ASSERT(numOfRows > 0);
 
   if (numOfRows == 0) {
     return TSDB_CODE_SUCCESS;
@@ -1230,6 +1230,32 @@ int32_t assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
   return 0;
 }
 
+int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
+  ASSERT(src != NULL && dst != NULL);
+
+  blockDataCleanup(dst);
+  int32_t code = blockDataEnsureCapacity(dst, src->info.rows);
+  if (code != TSDB_CODE_SUCCESS) {
+    terrno = code;
+    return code;
+  }
+
+  size_t numOfCols = taosArrayGetSize(src->pDataBlock);
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pDst = taosArrayGet(dst->pDataBlock, i);
+    SColumnInfoData* pSrc = taosArrayGet(src->pDataBlock, i);
+    if (pSrc->pData == NULL) {
+      continue;
+    }
+
+    colDataAssign(pDst, pSrc, src->info.rows, &src->info);
+  }
+
+  dst->info.rows = src->info.rows;
+  dst->info.window = src->info.window;
+  return TSDB_CODE_SUCCESS;
+}
+
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) {
   if (pDataBlock == NULL) {
     return NULL;
@@ -1627,6 +1653,56 @@ void blockDebugShowData(const SArray* dataBlocks, const char* flag) {
   }
 }
 
+// for debug
+char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf) {
+  int32_t size = 2048;
+  *pDataBuf = taosMemoryCalloc(size, 1);
+  char* dumpBuf = *pDataBuf;
+  char pBuf[128] = {0};
+  int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
+  int32_t rows = pDataBlock->info.rows;
+  int32_t len = 0;
+  len += snprintf(dumpBuf + len, size - len, "\n%s |block type %d |child id %d|\n", flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId);
+  for (int32_t j = 0; j < rows; j++) {
+    len += snprintf(dumpBuf + len, size - len, "%s |", flag);
+    for (int32_t k = 0; k < colNum; k++) {
+      SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
+      void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
+      if (colDataIsNull(pColInfoData, rows, j, NULL)) {
+        len += snprintf(dumpBuf + len, size - len, " %15s |", "NULL");
+        continue;
+      }
+      switch (pColInfoData->info.type) {
+        case TSDB_DATA_TYPE_TIMESTAMP:
+          formatTimestamp(pBuf, *(uint64_t*)var, TSDB_TIME_PRECISION_MILLI);
+          len += snprintf(dumpBuf + len, size - len, " %25s |", pBuf);
+          break;
+        case TSDB_DATA_TYPE_INT:
+          len += snprintf(dumpBuf + len, size - len, " %15d |", *(int32_t*)var);
+          break;
+        case TSDB_DATA_TYPE_UINT:
+          len += snprintf(dumpBuf + len, size - len, " %15u |", *(uint32_t*)var);
+          break;
+        case TSDB_DATA_TYPE_BIGINT:
+          len += snprintf(dumpBuf + len, size - len, " %15ld |", *(int64_t*)var);
+          break;
+        case TSDB_DATA_TYPE_UBIGINT:
+          len += snprintf(dumpBuf + len, size - len, " %15lu |", *(uint64_t*)var);
+          break;
+        case TSDB_DATA_TYPE_FLOAT:
+          len += snprintf(dumpBuf + len, size - len, " %15f |", *(float*)var);
+          break;
+        case TSDB_DATA_TYPE_DOUBLE:
+          len += snprintf(dumpBuf + len, size - len, " %15lf |", *(double*)var);
+          break;
+      }
+    }
+    len += snprintf(dumpBuf + len, size - len, "\n");
+  }
+  len += snprintf(dumpBuf + len, size - len, "%s |end\n", flag);
+  return dumpBuf;
+}
+
 /**
  * @brief TODO: Assume that the final generated result it less than 3M
  *

source/dnode/snode/src/snode.c

@@ -257,7 +257,7 @@ int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) {
     case TDMT_STREAM_TASK_RECOVER_RSP:
       return sndProcessTaskRecoverRsp(pSnode, pMsg);
     case TDMT_STREAM_RETRIEVE_RSP:
-      return sndProcessTaskRecoverRsp(pSnode, pMsg);
+      return sndProcessTaskRetrieveRsp(pSnode, pMsg);
     default:
       ASSERT(0);
   }

source/dnode/vnode/src/tq/tqRead.c

@@ -109,11 +109,15 @@ int32_t tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t
 }
 
 bool tqNextDataBlock(STqReadHandle* pHandle) {
+  if (pHandle->pMsg == NULL) return false;
   while (1) {
     if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
       return false;
     }
-    if (pHandle->pBlock == NULL) return false;
+    if (pHandle->pBlock == NULL) {
+      pHandle->pMsg = NULL;
+      return false;
+    }
 
     if (pHandle->tbIdHash == NULL) {
       return true;

source/libs/executor/inc/executorimpl.h

@@ -293,6 +293,7 @@ typedef enum EStreamScanMode {
   STREAM_SCAN_FROM_RES,
   STREAM_SCAN_FROM_UPDATERES,
   STREAM_SCAN_FROM_DATAREADER,
+  STREAM_SCAN_FROM_DATAREADER_RETRIEVE,
 } EStreamScanMode;
 
 typedef struct SCatchSupporter {
@@ -348,7 +349,9 @@ typedef struct SStreamBlockScanInfo {
   SArray*        childIds;
   SessionWindowSupporter sessionSup;
   bool            assignBlockUid; // assign block uid to groupId, temporarily used for generating rollup SMA.
-  int32_t         scanWinIndex;
+  int32_t         scanWinIndex;   // for state operator
+  int32_t         pullDataResIndex;
+  SSDataBlock*    pPullDataRes;             // pull data SSDataBlock
 } SStreamBlockScanInfo;
 
 typedef struct SSysTableScanInfo {
@@ -427,8 +430,13 @@ typedef struct SStreamFinalIntervalOperatorInfo {
   STimeWindowAggSupp twAggSup;
   SArray*            pChildren;
   SSDataBlock*       pUpdateRes;
+  bool               returnUpdate;
   SPhysiNode*        pPhyNode;           // create new child
   bool               isFinal;
+  SHashObj*          pPullDataMap;
+  SArray*            pPullWins;          // SPullWindowInfo
+  int32_t            pullIndex;
+  SSDataBlock*       pPullDataRes;
 } SStreamFinalIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
@@ -851,6 +859,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
 void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
 
 int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey);
+SSDataBlock* createPullDataBlock();
 
 #ifdef __cplusplus
 }

source/libs/executor/src/scanoperator.c

@@ -776,15 +776,14 @@ static bool isStateWindow(SStreamBlockScanInfo* pInfo) {
   return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
 }
 
-static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
-  SSDataBlock* pSDB = pInfo->pUpdateRes;
+static bool prepareDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
   STimeWindow  win = {
        .skey = INT64_MIN,
        .ekey = INT64_MAX,
   };
   bool needRead = false;
-  if (!isStateWindow(pInfo) && pInfo->updateResIndex < pSDB->info.rows) {
-    SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, pInfo->primaryTsIndex);
+  if (!isStateWindow(pInfo) && (*pRowIndex) < pSDB->info.rows) {
+    SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, tsColIndex);
     TSKEY*           tsCols = (TSKEY*)pColDataInfo->pData;
     SResultRowInfo   dumyInfo;
     dumyInfo.cur.pageId = -1;
@@ -793,14 +792,14 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
       int64_t              gap = pInfo->sessionSup.gap;
       int32_t              winIndex = 0;
       SResultWindowInfo*   pCurWin =
-          getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
+          getSessionTimeWindow(pAggSup, tsCols[(*pRowIndex)], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
       win = pCurWin->win;
-      pInfo->updateResIndex +=
-          updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, pInfo->updateResIndex, gap, NULL);
+      (*pRowIndex) +=
+          updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, (*pRowIndex), gap, NULL);
     } else {
-      win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[pInfo->updateResIndex], &pInfo->interval,
+      win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[(*pRowIndex)], &pInfo->interval,
                                 pInfo->interval.precision, NULL);
-      pInfo->updateResIndex += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, pInfo->updateResIndex, win.ekey,
+      (*pRowIndex) += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, (*pRowIndex), win.ekey,
                                                         binarySearchForKey, NULL, TSDB_ORDER_ASC);
     }
     needRead = true;
@@ -823,6 +822,9 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
   pTableScanInfo->cond.twindows[0] = win;
   pTableScanInfo->curTWinIdx = 0;
 //  tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
+  // if (!pTableScanInfo->dataReader) {
+  //   return false;
+  // }
   pTableScanInfo->scanTimes = 0;
   pTableScanInfo->currentGroupId = -1;
   return true;
@@ -862,12 +864,12 @@ static uint64_t getGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_
   */
 }
 
-static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) {
+static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
   while (1) {
     SSDataBlock* pResult = NULL;
     pResult = doTableScan(pInfo->pSnapshotReadOp);
     if (pResult == NULL) {
-      if (prepareDataScan(pInfo)) {
+      if (prepareDataScan(pInfo, pSDB, tsColIndex, pRowIndex)) {
         // scan next window data
         pResult = doTableScan(pInfo->pSnapshotReadOp);
       }
@@ -916,7 +918,7 @@ static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDa
     pUpdateBlock->info.rows = i;
     pInfo->tsArrayIndex += i;
     pUpdateBlock->info.groupId = pInfo->groupId;
-    pUpdateBlock->info.type = STREAM_REPROCESS;
+    pUpdateBlock->info.type = STREAM_CLEAR;
     blockDataUpdateTsWindow(pUpdateBlock, 0);
   }
   // all rows have same group id
@@ -970,6 +972,14 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     int32_t      current = pInfo->validBlockIndex++;
     SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
     blockDataUpdateTsWindow(pBlock, 0);
+    if (pBlock->info.type == STREAM_RETRIEVE) {
+      pInfo->blockType = STREAM_DATA_TYPE_SUBMIT_BLOCK;
+      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;
+      copyDataBlock(pInfo->pPullDataRes, pBlock);
+      pInfo->pullDataResIndex = 0;
+      prepareDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
+      updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo);
+    }
     return pBlock;
   } else if (pInfo->blockType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
     if (pInfo->scanMode == STREAM_SCAN_FROM_RES) {
@@ -979,28 +989,39 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     } else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) {
       pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
       if (!isStateWindow(pInfo)) {
-        prepareDataScan(pInfo);
+        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
       }
       return pInfo->pUpdateRes;
+    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RETRIEVE) {
+      SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
+      if (pSDB != NULL) {
+        getUpdateDataBlock(pInfo, true, pSDB, NULL);
+        pSDB->info.type = STREAM_PUSH_DATA;
+        return pSDB;
+      }
+      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
     } else {
       if (isStateWindow(pInfo) && taosArrayGetSize(pInfo->sessionSup.pStreamAggSup->pScanWindow) > 0) {
         pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
         pInfo->updateResIndex = pInfo->pUpdateRes->info.rows;
-        prepareDataScan(pInfo);
+        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
       }
       if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
-        SSDataBlock* pSDB = doDataScan(pInfo);
+        SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
         if (pSDB == NULL) {
           setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
           if (pInfo->pUpdateRes->info.rows > 0) {
             if (!isStateWindow(pInfo)) {
-              prepareDataScan(pInfo);
+              // Todo(liuyao) mybe can delete this.
+              bool test = prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
+              ASSERT(test == false);
             }
             return pInfo->pUpdateRes;
           } else {
             pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
           }
         } else {
+          pSDB->info.type = STREAM_NORMAL;
           getUpdateDataBlock(pInfo, true, pSDB, NULL);
           return pSDB;
         }
@@ -1070,10 +1091,12 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
       taosArrayDestroy(block.pDataBlock);
       if (pInfo->pRes->pDataBlock == NULL) {
         // TODO add log
+        updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
         pOperator->status = OP_EXEC_DONE;
         pTaskInfo->code = terrno;
         return NULL;
       }
+
       // currently only the tbname pseudo column
       if (pInfo->numOfPseudoExpr > 0) {
         addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes);
@@ -1091,12 +1114,13 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     pOperator->resultInfo.totalRows += pBlockInfo->rows;
 
     if (pBlockInfo->rows == 0) {
+      updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
       pOperator->status = OP_EXEC_DONE;
     } else if (pInfo->pUpdateInfo) {
       pInfo->tsArrayIndex = 0;
       getUpdateDataBlock(pInfo, true, pInfo->pRes, pInfo->pUpdateRes);
       if (pInfo->pUpdateRes->info.rows > 0) {
-        if (pInfo->pUpdateRes->info.type == STREAM_REPROCESS) {
+        if (pInfo->pUpdateRes->info.type == STREAM_CLEAR) {
           pInfo->updateResIndex = 0;
           pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
         } else if (pInfo->pUpdateRes->info.type == STREAM_INVERT) {
@@ -1209,6 +1233,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle,
   pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
   pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1};
   pInfo->groupId = 0;
+  pInfo->pPullDataRes = createPullDataBlock();
 
   pOperator->name = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;

source/libs/stream/src/streamDispatch.c

@@ -115,6 +115,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
       .srcNodeId = pTask->nodeId,
       .srcTaskId = pTask->taskId,
       .pRetrieve = pRetrieve,
+      .retrieveLen = dataStrLen,
   };
 
   int32_t sz = taosArrayGetSize(pTask->childEpInfo);
@@ -146,7 +147,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
         .code = 0,
         .msgType = TDMT_STREAM_RETRIEVE,
         .pCont = buf,
-        .contLen = len,
+        .contLen = sizeof(SMsgHead) + len,
     };
 
     if (tmsgSendReq(&pEpInfo->epSet, &rpcMsg) < 0) {

source/libs/stream/src/streamExec.c

@@ -45,11 +45,16 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
       ASSERT(false);
     }
     if (output == NULL) {
-      if (pItem->type == STREAM_INPUT__DATA_RETRIEVE && !hasData) {
-        SSDataBlock block = {0};
-        block.info.type = STREAM_PUSH_DATA;
-        block.info.childId = pTask->selfChildId;
-        taosArrayPush(pRes, &block);
+      if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
+        //SSDataBlock block = {0};
+        //block.info.type = STREAM_PUSH_EMPTY;
+        //block.info.childId = pTask->selfChildId;
+        SStreamDataBlock* pRetrieveBlock = (SStreamDataBlock*)data;
+        ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);
+        SSDataBlock* pBlock = createOneDataBlock(taosArrayGet(pRetrieveBlock->blocks, 0), true);
+        pBlock->info.type = STREAM_PUSH_EMPTY;
+        pBlock->info.childId = pTask->selfChildId;
+        taosArrayPush(pRes, pBlock);
       }
       break;
     }

source/libs/stream/src/streamUpdate.c

@@ -119,6 +119,7 @@ SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t waterma
     taosArrayPush(pInfo->pTsBuckets, &dumy);
   }
   pInfo->numBuckets = DEFAULT_BUCKET_SIZE;
+  pInfo->pCloseWinSBF = NULL;
   return pInfo;
 }
 
@@ -154,6 +155,9 @@ bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts) {
   TSKEY maxTs = *(TSKEY *)taosArrayGet(pInfo->pTsBuckets, index);
   if (ts < maxTs - pInfo->watermark) {
     // this window has been closed.
+    if (pInfo->pCloseWinSBF) {
+      return tScalableBfPut(pInfo->pCloseWinSBF, &ts, sizeof(TSKEY));
+    }
     return true;
   }
 
@@ -193,3 +197,19 @@ void updateInfoDestroy(SUpdateInfo *pInfo) {
   taosArrayDestroy(pInfo->pTsSBFs);
   taosMemoryFree(pInfo);
 }
+
+void updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo) {
+  if (pInfo->pCloseWinSBF) {
+    return;
+  }
+  int64_t rows = adjustExpEntries(pInfo->interval * ROWS_PER_MILLISECOND);
+  pInfo->pCloseWinSBF = tScalableBfInit(rows, DEFAULT_FALSE_POSITIVE);
+}
+
+void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo) {
+  if (!pInfo || !pInfo->pCloseWinSBF) {
+    return;
+  }
+  tScalableBfDestroy(pInfo->pCloseWinSBF);
+  pInfo->pCloseWinSBF = NULL;
+}

tests/script/tsim/stream/distributeInterval0.sim

@@ -80,17 +80,17 @@ endi
 
 if $data03 != 4 then
   print ======$data03
-  return -1
+  goto loop1
 endi
 
 if $data04 != 52 then
   print ======$data04
-  return -1
+  goto loop1
 endi
 
 if $data05 != 13 then
   print ======$data05
-  return -1
+  goto loop1
 endi
 
 # row 1
@@ -179,7 +179,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 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 create stream stream_t2 trigger at_once  watermark 20s 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);

tests/script/tsim/stream/partitionby.sim

@@ -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  _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 create stream stream_t2 trigger at_once  watermark 20s 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);
@@ -83,16 +83,16 @@ 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);
+sql insert into ts1 values(1648791211000,1,2,3,7);
+sql insert into ts1 values(1648791222001,2,2,3,8);
+sql insert into ts2 values(1648791211000,1,2,3,9);
+sql insert into ts2 values(1648791222001,2,2,3,10);
 
 $loop_count = 0
 
 loop2:
 sleep 300
-sql select * from streamtST;
+sql select * from streamtST order by c7 asc;
 
 $loop_count = $loop_count + 1
 if $loop_count == 10 then
@@ -104,8 +104,18 @@ print =====data01=$data01
 goto loop2
 endi
 
-if $data02 != 1 then
-print =====data02=$data02
+if $data11 != 1 then
+print =====data11=$data11
+goto loop2
+endi
+
+if $data21 != 1 then
+print =====data21=$data21
+goto loop2
+endi
+
+if $data31 != 2 then
+print =====data31=$data31
 goto loop2
 endi
 
@@ -114,8 +124,18 @@ print =====data03=$data03
 goto loop2
 endi
 
-if $data04 != 2 then
-print =====data04=$data04
+if $data13 != 2 then
+print =====data13=$data13
+goto loop2
+endi
+
+if $data23 != 1 then
+print =====data23=$data23
+goto loop2
+endi
+
+if $data33 != 4 then
+print =====data33=$data33
 goto loop2
 endi
 

tests/script/tsim/stream/schedSnode.sim

@@ -79,17 +79,17 @@ endi
 
 if $data03 != 4 then
   print ======$data03
-  return -1
+  goto loop1
 endi
 
 if $data04 != 52 then
   print ======$data04
-  return -1
+  goto loop1
 endi
 
 if $data05 != 13 then
   print ======$data05
-  return -1
+  goto loop1
 endi
 
 # row 1