Merge pull request #14734 from taosdata/feature/stream

feat(tmq): support consume from tsdb then wal

include/libs/executor/executor.h

@@ -36,6 +36,7 @@ typedef struct SReadHandle {
   void*   vnode;
   void*   mnd;
   SMsgCb* pMsgCb;
+  int64_t version;
   bool    initMetaReader;
   bool    initTableReader;
   bool    initTqReader;

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

@@ -89,6 +89,8 @@ typedef struct {
     STqExecTb  execTb;
     STqExecDb  execDb;
   };
+  // TODO remove it
+  int64_t tsdbEndVer;
 
 } STqExecHandle;
 

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

@@ -483,6 +483,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
     /*for (int32_t i = 0; i < 5; i++) {*/
     /*pHandle->execHandle.pExecReader[i] = tqOpenReader(pTq->pVnode);*/
     /*}*/
+    int64_t ver = walGetCommittedVer(pTq->pVnode->pWal);
     if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
       pHandle->execHandle.execCol.qmsg = req.qmsg;
       req.qmsg = NULL;
@@ -493,6 +494,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
             .vnode = pTq->pVnode,
             .initTableReader = true,
             .initTqReader = true,
+            .version = ver,
         };
         pHandle->execHandle.execCol.task[i] = qCreateStreamExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle);
         ASSERT(pHandle->execHandle.execCol.task[i]);
@@ -501,6 +503,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
         ASSERT(scanner);
         pHandle->execHandle.pExecReader[i] = qExtractReaderFromStreamScanner(scanner);
         ASSERT(pHandle->execHandle.pExecReader[i]);
+        pHandle->execHandle.tsdbEndVer = ver;
       }
     } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__DB) {
       for (int32_t i = 0; i < 5; i++) {

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

@@ -96,6 +96,12 @@ int64_t tqScan(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffset
       }
     }
 
+    if (pRsp->blockNum == 0 && pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
+      tqOffsetResetToLog(pOffset, pExec->tsdbEndVer + 1);
+      qStreamPrepareScan(task, pOffset);
+      continue;
+    }
+
     void* meta = qStreamExtractMetaMsg(task);
     if (meta != NULL) {
       // tq add meta to rsp
@@ -107,7 +113,7 @@ int64_t tqScan(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffset
 
     ASSERT(pRsp->rspOffset.type != 0);
 
-    if (pRsp->rspOffset.type == TMQ_OFFSET__LOG) {
+    if (pRsp->reqOffset.type == TMQ_OFFSET__LOG) {
       ASSERT(pRsp->rspOffset.version + 1 >= pRsp->reqOffset.version);
     }
 

source/libs/executor/src/scanoperator.c

@@ -1493,6 +1493,9 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
   if (pHandle) {
     SOperatorInfo*  pTableScanOp = createTableScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo);
     STableScanInfo* pSTInfo = (STableScanInfo*)pTableScanOp->info;
+    if (pHandle->version > 0) {
+      pSTInfo->cond.endVersion = pHandle->version;
+    }
 
     SArray* tableList = taosArrayGetP(pTaskInfo->tableqinfoList.pGroupList, 0);
     if (pHandle->initTableReader) {