refactor: do some internal refactor.

include/common/tcommon.h

@@ -208,11 +208,6 @@ typedef struct SSDataBlock {
   SDataBlockInfo   info;
 } SSDataBlock;
 
-enum {
-  FETCH_TYPE__DATA = 0,
-  FETCH_TYPE__NONE,
-};
-
 typedef struct SVarColAttr {
   int32_t* offset;    // start position for each entry in the list
   uint32_t length;    // used buffer size that contain the valid data

source/dnode/vnode/inc/vnode.h

@@ -267,7 +267,7 @@ int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *pTableUidList);
 int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
 
 int32_t tqSeekVer(STqReader *pReader, int64_t ver, const char *id);
-int32_t tqNextBlockInWal(STqReader* pReader);
+bool    tqNextBlockInWal(STqReader* pReader, const char* idstr);
 bool    tqNextBlockImpl(STqReader *pReader, const char* idstr);
 
 int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, const char* id);

source/dnode/vnode/src/inc/vnodeInt.h

@@ -219,8 +219,6 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t version, char* msg, int32_t msgL
 int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
 int32_t tqProcessStreamTaskCheckReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessStreamTaskCheckRsp(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
-int32_t tqProcessSubmitReqForSubscribe(STQ* pTq);
-int32_t tqProcessDeleteDataReq(STQ* pTq, void* pReq, int32_t len, int64_t ver);
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec);
 int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg);

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

@@ -1195,159 +1195,6 @@ int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStream
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t tqProcessDeleteDataReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
-  bool        failed = false;
-  SDecoder*   pCoder = &(SDecoder){0};
-  SDeleteRes* pRes = &(SDeleteRes){0};
-
-  pRes->uidList = taosArrayInit(0, sizeof(tb_uid_t));
-  if (pRes->uidList == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    failed = true;
-  }
-
-  tDecoderInit(pCoder, pReq, len);
-  tDecodeDeleteRes(pCoder, pRes);
-  tDecoderClear(pCoder);
-
-  int32_t sz = taosArrayGetSize(pRes->uidList);
-  if (sz == 0 || pRes->affectedRows == 0) {
-    taosArrayDestroy(pRes->uidList);
-    return 0;
-  }
-
-  SSDataBlock* pDelBlock = createSpecialDataBlock(STREAM_DELETE_DATA);
-  blockDataEnsureCapacity(pDelBlock, sz);
-  pDelBlock->info.rows = sz;
-  pDelBlock->info.version = ver;
-
-  for (int32_t i = 0; i < sz; i++) {
-    // start key column
-    SColumnInfoData* pStartCol = taosArrayGet(pDelBlock->pDataBlock, START_TS_COLUMN_INDEX);
-    colDataSetVal(pStartCol, i, (const char*)&pRes->skey, false);  // end key column
-    SColumnInfoData* pEndCol = taosArrayGet(pDelBlock->pDataBlock, END_TS_COLUMN_INDEX);
-    colDataSetVal(pEndCol, i, (const char*)&pRes->ekey, false);
-    // uid column
-    SColumnInfoData* pUidCol = taosArrayGet(pDelBlock->pDataBlock, UID_COLUMN_INDEX);
-    int64_t*         pUid = taosArrayGet(pRes->uidList, i);
-    colDataSetVal(pUidCol, i, (const char*)pUid, false);
-
-    colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, GROUPID_COLUMN_INDEX), i);
-    colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX), i);
-    colDataSetNULL(taosArrayGet(pDelBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX), i);
-  }
-
-  taosArrayDestroy(pRes->uidList);
-
-  int32_t* pRef = taosMemoryMalloc(sizeof(int32_t));
-  *pRef = 1;
-
-  taosWLockLatch(&pTq->pStreamMeta->lock);
-
-  void* pIter = NULL;
-  while (1) {
-    pIter = taosHashIterate(pTq->pStreamMeta->pTasks, pIter);
-    if (pIter == NULL) {
-      break;
-    }
-
-    SStreamTask* pTask = *(SStreamTask**)pIter;
-    if (pTask->taskLevel != TASK_LEVEL__SOURCE) {
-      continue;
-    }
-
-    qDebug("s-task:%s delete req enqueue, ver: %" PRId64, pTask->id.idStr, ver);
-
-    if (!failed) {
-      SStreamRefDataBlock* pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0);
-      pRefBlock->type = STREAM_INPUT__REF_DATA_BLOCK;
-      pRefBlock->pBlock = pDelBlock;
-
-      if (tAppendDataToInputQueue(pTask, (SStreamQueueItem*)pRefBlock) < 0) {
-        atomic_sub_fetch_32(pRef, 1);
-        taosFreeQitem(pRefBlock);
-        continue;
-      }
-
-      if (streamSchedExec(pTask) < 0) {
-        qError("s-task:%s stream task launch failed", pTask->id.idStr);
-        continue;
-      }
-
-    } else {
-      streamTaskInputFail(pTask);
-    }
-  }
-
-  taosWUnLockLatch(&pTq->pStreamMeta->lock);
-
-  int32_t ref = atomic_sub_fetch_32(pRef, 1);
-  if (ref == 0) {
-    blockDataDestroy(pDelBlock);
-    taosMemoryFree(pRef);
-  }
-
-#if 0
-    SStreamDataBlock* pStreamBlock = taosAllocateQitem(sizeof(SStreamDataBlock), DEF_QITEM, 0);
-    pStreamBlock->type = STREAM_INPUT__DATA_BLOCK;
-    pStreamBlock->blocks = taosArrayInit(0, sizeof(SSDataBlock));
-    SSDataBlock block = {0};
-    assignOneDataBlock(&block, pDelBlock);
-    block.info.type = STREAM_DELETE_DATA;
-    taosArrayPush(pStreamBlock->blocks, &block);
-
-    if (!failed) {
-      if (tAppendDataToInputQueue(pTask, (SStreamQueueItem*)pStreamBlock) < 0) {
-        qError("stream task input del failed, task id %d", pTask->id.taskId);
-        continue;
-      }
-
-      if (streamSchedExec(pTask) < 0) {
-        qError("stream task launch failed, task id %d", pTask->id.taskId);
-        continue;
-      }
-    } else {
-      streamTaskInputFail(pTask);
-    }
-  }
-  blockDataDestroy(pDelBlock);
-#endif
-  return 0;
-}
-
-int32_t tqProcessSubmitReqForSubscribe(STQ* pTq) {
-  int32_t vgId = TD_VID(pTq->pVnode);
-
-  taosWLockLatch(&pTq->lock);
-
-  if (taosHashGetSize(pTq->pPushMgr) > 0) {
-    void* pIter = taosHashIterate(pTq->pPushMgr, NULL);
-
-    while (pIter) {
-      STqHandle* pHandle = *(STqHandle**)pIter;
-      tqDebug("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId);
-
-      if (ASSERT(pHandle->msg != NULL)) {
-        tqError("pHandle->msg should not be null");
-        break;
-      }else{
-        SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME, .pCont = pHandle->msg->pCont, .contLen = pHandle->msg->contLen, .info = pHandle->msg->info};
-        tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg);
-        taosMemoryFree(pHandle->msg);
-        pHandle->msg = NULL;
-      }
-
-      pIter = taosHashIterate(pTq->pPushMgr, pIter);
-    }
-
-    taosHashClear(pTq->pPushMgr);
-  }
-
-  // unlock
-  taosWUnLockLatch(&pTq->lock);
-  return 0;
-}
-
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
   SStreamTaskRunReq* pReq = pMsg->pCont;
 

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

@@ -16,8 +16,40 @@
 #include "tq.h"
 #include "vnd.h"
 
-int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
+int32_t tqProcessSubmitReqForSubscribe(STQ* pTq) {
+  int32_t vgId = TD_VID(pTq->pVnode);
+
+  taosWLockLatch(&pTq->lock);
+
+  if (taosHashGetSize(pTq->pPushMgr) > 0) {
+    void* pIter = taosHashIterate(pTq->pPushMgr, NULL);
+
+    while (pIter) {
+      STqHandle* pHandle = *(STqHandle**)pIter;
+      tqDebug("vgId:%d start set submit for pHandle:%p, consumer:0x%" PRIx64, vgId, pHandle, pHandle->consumerId);
 
+      if (ASSERT(pHandle->msg != NULL)) {
+        tqError("pHandle->msg should not be null");
+        break;
+      }else{
+        SRpcMsg msg = {.msgType = TDMT_VND_TMQ_CONSUME, .pCont = pHandle->msg->pCont, .contLen = pHandle->msg->contLen, .info = pHandle->msg->info};
+        tmsgPutToQueue(&pTq->pVnode->msgCb, QUERY_QUEUE, &msg);
+        taosMemoryFree(pHandle->msg);
+        pHandle->msg = NULL;
+      }
+
+      pIter = taosHashIterate(pTq->pPushMgr, pIter);
+    }
+
+    taosHashClear(pTq->pPushMgr);
+  }
+
+  // unlock
+  taosWUnLockLatch(&pTq->lock);
+  return 0;
+}
+
+int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
   if (msgType == TDMT_VND_SUBMIT) {
     tqProcessSubmitReqForSubscribe(pTq);
   }
@@ -37,10 +69,6 @@ int32_t tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t v
     if (msgType == TDMT_VND_SUBMIT || msgType == TDMT_VND_DELETE) {
       tqStartStreamTasks(pTq);
     }
-
-    if (msgType == TDMT_VND_DELETE) {
-//      tqProcessDeleteDataReq(pTq, POINTER_SHIFT(msg, sizeof(SMsgHead)), msgLen - sizeof(SMsgHead), ver);
-    }
   }
 
   return 0;

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

@@ -344,7 +344,7 @@ int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, const char* id) {
 }
 
 // todo ignore the error in wal?
-int32_t tqNextBlockInWal(STqReader* pReader) {
+bool tqNextBlockInWal(STqReader* pReader, const char* id) {
   SWalReader* pWalReader = pReader->pWalReader;
 
   while (1) {
@@ -353,7 +353,7 @@ int32_t tqNextBlockInWal(STqReader* pReader) {
       // try next message in wal file
       // todo always retry to avoid read failure caused by wal file deletion
       if (walNextValidMsg(pWalReader) < 0) {
-        return FETCH_TYPE__NONE;
+        return false;
       }
 
       void*   pBody = POINTER_SHIFT(pWalReader->pHead->head.body, sizeof(SSubmitReq2Msg));
@@ -379,7 +379,7 @@ int32_t tqNextBlockInWal(STqReader* pReader) {
       if (tDecodeSubmitReq(&decoder, &pReader->submit) < 0) {
         tDecoderClear(&decoder);
         tqError("decode wal file error, msgLen:%d, ver:%" PRId64, bodyLen, ver);
-        return FETCH_TYPE__NONE;
+        return false;
       }
 
       tDecoderClear(&decoder);
@@ -388,15 +388,15 @@ int32_t tqNextBlockInWal(STqReader* pReader) {
 
     size_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData);
     while (pReader->nextBlk < numOfBlocks) {
-      tqDebug("tq reader next data block %p, %d %" PRId64 " %d", pReader->msg.msgStr, pReader->msg.msgLen,
-              pReader->msg.ver, pReader->nextBlk);
+      tqDebug("tq reader next data block %d/%d, len:%d %" PRId64 " %d", pReader->nextBlk,
+          numOfBlocks, pReader->msg.msgLen, pReader->msg.ver, pReader->nextBlk);
 
       SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
 
       if (pReader->tbIdHash == NULL) {
         int32_t code = tqRetrieveDataBlock(pReader, NULL);
         if (code == TSDB_CODE_SUCCESS && pReader->pResBlock->info.rows > 0) {
-          return FETCH_TYPE__DATA;
+          return true;
         }
       }
 
@@ -406,7 +406,7 @@ int32_t tqNextBlockInWal(STqReader* pReader) {
 
         int32_t code = tqRetrieveDataBlock(pReader, NULL);
         if (code == TSDB_CODE_SUCCESS && pReader->pResBlock->info.rows > 0) {
-          return FETCH_TYPE__DATA;
+          return true;
         }
       } else {
         pReader->nextBlk += 1;
@@ -414,7 +414,9 @@ int32_t tqNextBlockInWal(STqReader* pReader) {
       }
     }
 
+    qDebug("stream scan return empty, all %d submit blocks consumed, %s", numOfBlocks, id);
     tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE);
+
     pReader->msg.msgStr = NULL;
   }
 }
@@ -729,13 +731,17 @@ int32_t tqRetrieveDataBlock(STqReader* pReader, const char* id) {
   return 0;
 }
 
+// todo refactor:
 int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas, SSubmitTbData** pSubmitTbDataRet) {
   tqDebug("tq reader retrieve data block %p, %d", pReader->msg.msgStr, pReader->nextBlk);
 
   SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk);
   pReader->nextBlk++;
 
-  if (pSubmitTbDataRet) *pSubmitTbDataRet = pSubmitTbData;
+  if (pSubmitTbDataRet) {
+    *pSubmitTbDataRet = pSubmitTbData;
+  }
+
   int32_t sversion = pSubmitTbData->sver;
   int64_t suid = pSubmitTbData->suid;
   int64_t uid = pSubmitTbData->uid;

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

@@ -15,7 +15,7 @@
 
 #include "tq.h"
 
-static int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle);
+static int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle);
 
 // this function should be executed by stream threads.
 // extract submit block from WAL, and add them into the input queue for the sources tasks.
@@ -30,7 +30,7 @@ int32_t tqStreamTasksScanWal(STQ* pTq) {
 
     // check all restore tasks
     bool shouldIdle = true;
-    createStreamRunReq(pTq->pStreamMeta, &shouldIdle);
+    createStreamTaskRunReq(pTq->pStreamMeta, &shouldIdle);
 
     int32_t times = 0;
 
@@ -57,7 +57,39 @@ int32_t tqStreamTasksScanWal(STQ* pTq) {
   return 0;
 }
 
-int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
+static int32_t doSetOffsetForWalReader(SStreamTask *pTask, int32_t vgId) {
+  // seek the stored version and extract data from WAL
+  int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader);
+  if (pTask->chkInfo.currentVer < firstVer) {
+    pTask->chkInfo.currentVer = firstVer;
+    tqWarn("vgId:%d s-task:%s ver earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64, vgId,
+           pTask->id.idStr, firstVer, pTask->chkInfo.currentVer);
+
+    // todo need retry if failed
+    int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
+
+    // append the data for the stream
+    tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer);
+  } else {
+    int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);
+    if (currentVer == -1) {  // we only seek the read for the first time
+      int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer);
+      if (code != TSDB_CODE_SUCCESS) {  // no data in wal, quit
+        return code;
+      }
+
+      // append the data for the stream
+      tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer);
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
   *pScanIdle = true;
   bool    noNewDataInWal = true;
   int32_t vgId = pStreamMeta->vgId;
@@ -67,6 +99,7 @@ int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
     return TSDB_CODE_SUCCESS;
   }
 
+  // clone the task list, to avoid the task update during scan wal files
   SArray* pTaskList = NULL;
   taosWLockLatch(&pStreamMeta->lock);
   pTaskList = taosArrayDup(pStreamMeta->pTaskList, NULL);
@@ -107,38 +140,15 @@ int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
     *pScanIdle = false;
 
     // seek the stored version and extract data from WAL
-    int64_t firstVer = walReaderGetValidFirstVer(pTask->exec.pWalReader);
-    if (pTask->chkInfo.currentVer < firstVer) {
-      pTask->chkInfo.currentVer = firstVer;
-      tqWarn("vgId:%d s-task:%s ver earlier than the first ver of wal range %" PRId64 ", forward to %" PRId64, vgId,
-             pTask->id.idStr, firstVer, pTask->chkInfo.currentVer);
-
-      // todo need retry if failed
-      int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer);
-      if (code != TSDB_CODE_SUCCESS) {
-        streamMetaReleaseTask(pStreamMeta, pTask);
-        continue;
-      }
-
-      // append the data for the stream
-      tqDebug("vgId:%d s-task:%s wal reader seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer);
-    } else {
-      int64_t currentVer = walReaderGetCurrentVer(pTask->exec.pWalReader);
-      if (currentVer == -1) {
-        int32_t code = walReaderSeekVer(pTask->exec.pWalReader, pTask->chkInfo.currentVer);
-        if (code != TSDB_CODE_SUCCESS) {  // no data in wal, quit
-          streamMetaReleaseTask(pStreamMeta, pTask);
-          continue;
-        }
-
-        // append the data for the stream
-        tqDebug("vgId:%d s-task:%s wal reader initial seek to ver:%" PRId64, vgId, pTask->id.idStr, pTask->chkInfo.currentVer);
-      }
+    int32_t code = doSetOffsetForWalReader(pTask, vgId);
+    if (code != TSDB_CODE_SUCCESS) {
+      streamMetaReleaseTask(pStreamMeta, pTask);
+      continue;
     }
 
     // append the data for the stream
     SStreamQueueItem* pItem = NULL;
-    int32_t code = extractMsgFromWal(pTask->exec.pWalReader, (void**) &pItem, pTask->id.idStr);
+    code = extractMsgFromWal(pTask->exec.pWalReader, (void**) &pItem, pTask->id.idStr);
     if (code != TSDB_CODE_SUCCESS) {  // failed, continue
       streamMetaReleaseTask(pStreamMeta, pTask);
       continue;
@@ -161,7 +171,6 @@ int32_t createStreamRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
       tqError("s-task:%s append input queue failed, ver:%" PRId64, pTask->id.idStr, pTask->chkInfo.currentVer);
     }
 
-
     streamMetaReleaseTask(pStreamMeta, pTask);
   }