Merge pull request #16524 from taosdata/feature/stream

feat(tmq): support taosx

.clang-format

@@ -88,4 +88,3 @@ Standard:        Auto
 TabWidth:        8
 UseTab:          Never
 ...
-

include/common/tmsg.h

@@ -2626,6 +2626,22 @@ typedef struct {
   };
 } STqOffsetVal;
 
+static FORCE_INLINE void tqOffsetResetToData(STqOffsetVal* pOffsetVal, int64_t uid, int64_t ts) {
+  pOffsetVal->type = TMQ_OFFSET__SNAPSHOT_DATA;
+  pOffsetVal->uid = uid;
+  pOffsetVal->ts = ts;
+}
+
+static FORCE_INLINE void tqOffsetResetToMeta(STqOffsetVal* pOffsetVal, int64_t uid) {
+  pOffsetVal->type = TMQ_OFFSET__SNAPSHOT_META;
+  pOffsetVal->uid = uid;
+}
+
+static FORCE_INLINE void tqOffsetResetToLog(STqOffsetVal* pOffsetVal, int64_t ver) {
+  pOffsetVal->type = TMQ_OFFSET__LOG;
+  pOffsetVal->version = ver;
+}
+
 int32_t tEncodeSTqOffsetVal(SEncoder* pEncoder, const STqOffsetVal* pOffsetVal);
 int32_t tDecodeSTqOffsetVal(SDecoder* pDecoder, STqOffsetVal* pOffsetVal);
 int32_t tFormatOffset(char* buf, int32_t maxLen, const STqOffsetVal* pVal);
@@ -2957,6 +2973,7 @@ typedef struct {
 
 int32_t tEncodeSMqDataRsp(SEncoder* pEncoder, const SMqDataRsp* pRsp);
 int32_t tDecodeSMqDataRsp(SDecoder* pDecoder, SMqDataRsp* pRsp);
+void    tDeleteSMqDataRsp(SMqDataRsp* pRsp);
 
 typedef struct {
   SMqRspHead   head;

source/client/src/tmq.c

@@ -811,8 +811,19 @@ int32_t tmq_subscription(tmq_t* tmq, tmq_list_t** topics) {
 }
 
 int32_t tmq_unsubscribe(tmq_t* tmq) {
+  int32_t     rsp;
+  int32_t     retryCnt = 0;
   tmq_list_t* lst = tmq_list_new();
-  int32_t     rsp = tmq_subscribe(tmq, lst);
+  while (1) {
+    rsp = tmq_subscribe(tmq, lst);
+    if (rsp != TSDB_CODE_MND_CONSUMER_NOT_READY || retryCnt > 5) {
+      break;
+    } else {
+      retryCnt++;
+      taosMsleep(500);
+    }
+  }
+
   tmq_list_destroy(lst);
   return rsp;
 }

source/common/src/tmsg.c

@@ -5889,6 +5889,13 @@ int32_t tDecodeSMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
   return 0;
 }
 
+void tDeleteSMqDataRsp(SMqDataRsp *pRsp) {
+  taosArrayDestroy(pRsp->blockDataLen);
+  taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree);
+  taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper);
+  taosArrayDestroyP(pRsp->blockTbName, (FDelete)taosMemoryFree);
+}
+
 int32_t tEncodeSTaosxRsp(SEncoder *pEncoder, const STaosxRsp *pRsp) {
   if (tEncodeSTqOffsetVal(pEncoder, &pRsp->reqOffset) < 0) return -1;
   if (tEncodeSTqOffsetVal(pEncoder, &pRsp->rspOffset) < 0) return -1;

source/dnode/mnode/impl/src/mndTopic.c

@@ -763,8 +763,9 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
     int32_t          cols = 0;
 
     char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
-    tNameFromString(&n, pTopic->name, T_NAME_ACCT | T_NAME_DB);
+    strcpy(varDataVal(topicName), mndGetDbStr(pTopic->name));
-    tNameGetDbName(&n, varDataVal(topicName));
+    /*tNameFromString(&n, pTopic->name, T_NAME_ACCT | T_NAME_DB);*/
+    /*tNameGetDbName(&n, varDataVal(topicName));*/
     varDataSetLen(topicName, strlen(varDataVal(topicName)));
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)topicName, false);

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

@@ -144,7 +144,7 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
 int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
 
 // tqExec
-int32_t tqLogScanExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataRsp* pRsp);
+int32_t tqLogScanExec(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, SMqDataRsp* pRsp);
 int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp);
 
 // tqMeta
@@ -175,22 +175,6 @@ void tqTableSink(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
 char*   tqOffsetBuildFName(const char* path, int32_t ver);
 int32_t tqOffsetRestoreFromFile(STqOffsetStore* pStore, const char* fname);
 
-static FORCE_INLINE void tqOffsetResetToData(STqOffsetVal* pOffsetVal, int64_t uid, int64_t ts) {
-  pOffsetVal->type = TMQ_OFFSET__SNAPSHOT_DATA;
-  pOffsetVal->uid = uid;
-  pOffsetVal->ts = ts;
-}
-
-static FORCE_INLINE void tqOffsetResetToMeta(STqOffsetVal* pOffsetVal, int64_t uid) {
-  pOffsetVal->type = TMQ_OFFSET__SNAPSHOT_META;
-  pOffsetVal->uid = uid;
-}
-
-static FORCE_INLINE void tqOffsetResetToLog(STqOffsetVal* pOffsetVal, int64_t ver) {
-  pOffsetVal->type = TMQ_OFFSET__LOG;
-  pOffsetVal->version = ver;
-}
-
 // tqStream
 int32_t tqExpandTask(STQ* pTq, SStreamTask* pTask);
 

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

@@ -373,14 +373,16 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
         if (tqSendDataRsp(pTq, pMsg, pReq, &dataRsp) < 0) {
           code = -1;
         }
-        goto OVER;
+        tDeleteSMqDataRsp(&dataRsp);
+        return code;
       } else if (reqOffset.type == TMQ_OFFSET__RESET_NONE) {
         tqError("tmq poll: subkey %s, no offset committed for consumer %" PRId64
                 " in vg %d, subkey %s, reset none failed",
                 pHandle->subKey, consumerId, TD_VID(pTq->pVnode), pReq->subKey);
         terrno = TSDB_CODE_TQ_NO_COMMITTED_OFFSET;
         code = -1;
-        goto OVER;
+        tDeleteSMqDataRsp(&dataRsp);
+        return code;
       }
     }
   }
@@ -393,8 +395,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
       if (tqSendMetaPollRsp(pTq, pMsg, pReq, &metaRsp) < 0) {
         code = -1;
       }
-      tqDebug("tmq poll: consumer %ld, subkey %s, vg %d, send meta offset type:%d,uid:%ld,version:%ld", consumerId, pHandle->subKey,
+      tqDebug("tmq poll: consumer %ld, subkey %s, vg %d, send meta offset type:%d,uid:%ld,version:%ld", consumerId,
-              TD_VID(pTq->pVnode), metaRsp.rspOffset.type, metaRsp.rspOffset.uid, metaRsp.rspOffset.version);
+              pHandle->subKey, TD_VID(pTq->pVnode), metaRsp.rspOffset.type, metaRsp.rspOffset.uid,
+              metaRsp.rspOffset.version);
       taosMemoryFree(metaRsp.metaRsp);
       goto OVER;
     }
@@ -408,8 +411,9 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
       fetchOffsetNew = dataRsp.rspOffset;
     }
 
-    tqDebug("tmq poll: consumer %ld, subkey %s, vg %d, send data blockNum:%d, offset type:%d,uid:%ld,version:%ld", consumerId, pHandle->subKey,
+    tqDebug("tmq poll: consumer %ld, subkey %s, vg %d, send data blockNum:%d, offset type:%d,uid:%ld,version:%ld",
-            TD_VID(pTq->pVnode), dataRsp.blockNum, dataRsp.rspOffset.type, dataRsp.rspOffset.uid, dataRsp.rspOffset.version);
+            consumerId, pHandle->subKey, TD_VID(pTq->pVnode), dataRsp.blockNum, dataRsp.rspOffset.type,
+            dataRsp.rspOffset.uid, dataRsp.rspOffset.version);
   }
 
   if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__COLUMN && fetchOffsetNew.type == TMQ_OFFSET__LOG) {
@@ -449,7 +453,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
       if (pHead->msgType == TDMT_VND_SUBMIT) {
         SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
 
-        if (tqLogScanExec(pTq, &pHandle->execHandle, pCont, &dataRsp) < 0) {
+        if (tqLogScanExec(pTq, pHandle, pCont, &dataRsp) < 0) {
           /*ASSERT(0);*/
         }
         // TODO batch optimization:
@@ -490,18 +494,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
 
 OVER:
   if (pCkHead) taosMemoryFree(pCkHead);
-  // TODO wrap in destroy func
+  tDeleteSMqDataRsp(&dataRsp);
-  taosArrayDestroy(dataRsp.blockDataLen);
-  taosArrayDestroyP(dataRsp.blockData, (FDelete)taosMemoryFree);
-
-  if (dataRsp.withSchema) {
-    taosArrayDestroyP(dataRsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
-  }
-
-  if (dataRsp.withTbName) {
-    taosArrayDestroyP(dataRsp.blockTbName, (FDelete)taosMemoryFree);
-  }
-
   return code;
 }
 
@@ -629,9 +622,9 @@ int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLe
       tqReaderSetTbUidList(pHandle->execHandle.pExecReader, tbUidList);
       taosArrayDestroy(tbUidList);
 
-      buildSnapContext(handle.meta, handle.version, req.suid, pHandle->execHandle.subType, pHandle->fetchMeta, (SSnapContext **)(&handle.sContext));
+      buildSnapContext(handle.meta, handle.version, req.suid, pHandle->execHandle.subType, pHandle->fetchMeta,
-      pHandle->execHandle.task =
+                       (SSnapContext**)(&handle.sContext));
-          qCreateQueueExecTaskInfo(NULL, &handle, NULL, NULL);
+      pHandle->execHandle.task = qCreateQueueExecTaskInfo(NULL, &handle, NULL, NULL);
     }
     taosHashPut(pTq->pHandle, req.subKey, strlen(req.subKey), pHandle, sizeof(STqHandle));
     tqDebug("try to persist handle %s consumer %" PRId64, req.subKey, pHandle->consumerId);

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

@@ -60,6 +60,46 @@ static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp) {
   return 0;
 }
 
+int64_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset) {
+  const STqExecHandle* pExec = &pHandle->execHandle;
+  ASSERT(pExec->subType == TOPIC_SUB_TYPE__COLUMN);
+
+  qTaskInfo_t task = pExec->task;
+
+  if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
+    tqDebug("prepare scan failed, return");
+    if (pOffset->type == TMQ_OFFSET__LOG) {
+      pRsp->rspOffset = *pOffset;
+      return 0;
+    } else {
+      tqOffsetResetToLog(pOffset, pHandle->snapshotVer);
+      if (qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType) < 0) {
+        tqDebug("prepare scan failed, return");
+        pRsp->rspOffset = *pOffset;
+        return 0;
+      }
+    }
+  }
+
+  int32_t rowCnt = 0;
+  while (1) {
+    SSDataBlock* pDataBlock = NULL;
+    uint64_t     ts = 0;
+    tqDebug("tmq task start to execute");
+    if (qExecTask(task, &pDataBlock, &ts) < 0) {
+      ASSERT(0);
+    }
+    tqDebug("tmq task execute end, get %p", pDataBlock);
+
+    if (pDataBlock) {
+      tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols);
+      pRsp->blockNum++;
+    }
+  }
+
+  return 0;
+}
+
 int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* pOffset) {
   const STqExecHandle* pExec = &pHandle->execHandle;
   qTaskInfo_t          task = pExec->task;
@@ -125,15 +165,7 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
       }
     }
 
-    if(pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN){
+    if (pHandle->execHandle.subType != TOPIC_SUB_TYPE__COLUMN) {
-      if (pRsp->blockNum == 0 && pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
-        tqDebug("vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode),
-                pHandle->snapshotVer + 1);
-        tqOffsetResetToLog(pOffset, pHandle->snapshotVer);
-        qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);
-        continue;
-      }
-    }else{
       if (pDataBlock == NULL && pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
         if (qStreamExtractPrepareUid(task) != 0) {
           continue;
@@ -173,57 +205,8 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqMetaRsp*
   return 0;
 }
 
-#if 0
+int32_t tqLogScanExec(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, SMqDataRsp* pRsp) {
-int32_t tqScanSnapshot(STQ* pTq, const STqExecHandle* pExec, SMqDataRsp* pRsp, STqOffsetVal offset, int32_t workerId) {
+  STqExecHandle* pExec = &pHandle->execHandle;
-  ASSERT(pExec->subType == TOPIC_SUB_TYPE__COLUMN);
-  qTaskInfo_t task = pExec->execCol.task[workerId];
-
-  if (qStreamPrepareTsdbScan(task, offset.uid, offset.ts) < 0) {
-    ASSERT(0);
-  }
-
-  int32_t rowCnt = 0;
-  while (1) {
-    SSDataBlock* pDataBlock = NULL;
-    uint64_t     ts = 0;
-    if (qExecTask(task, &pDataBlock, &ts) < 0) {
-      ASSERT(0);
-    }
-    if (pDataBlock == NULL) break;
-
-    ASSERT(pDataBlock->info.rows != 0);
-    ASSERT(taosArrayGetSize(pDataBlock->pDataBlock) != 0);
-
-    tqAddBlockDataToRsp(pDataBlock, pRsp);
-
-    if (pRsp->withTbName) {
-      pRsp->withTbName = 0;
-#if 0
-      int64_t uid;
-      int64_t ts;
-      if (qGetStreamScanStatus(task, &uid, &ts) < 0) {
-        ASSERT(0);
-      }
-      tqAddTbNameToRsp(pTq, uid, pRsp);
-#endif
-    }
-    pRsp->blockNum++;
-
-    rowCnt += pDataBlock->info.rows;
-    if (rowCnt >= 4096) break;
-  }
-  int64_t uid;
-  int64_t ts;
-  if (qGetStreamScanStatus(task, &uid, &ts) < 0) {
-    ASSERT(0);
-  }
-  tqOffsetResetToData(&pRsp->rspOffset, uid, ts);
-
-  return 0;
-}
-#endif
-
-int32_t tqLogScanExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataRsp* pRsp) {
   ASSERT(pExec->subType != TOPIC_SUB_TYPE__COLUMN);
 
   if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
@@ -268,6 +251,28 @@ int32_t tqLogScanExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataR
       tqAddBlockSchemaToRsp(pExec, pRsp);
       pRsp->blockNum++;
     }
+#if 0
+    if (pHandle->fetchMeta && pRsp->blockNum) {
+      SSubmitMsgIter iter = {0};
+      tInitSubmitMsgIter(pReq, &iter);
+      STaosxRsp* pXrsp = (STaosxRsp*)pRsp;
+      while (1) {
+        SSubmitBlk* pBlk = NULL;
+        if (tGetSubmitMsgNext(&iter, &pBlk) < 0) return -1;
+        if (pBlk->schemaLen > 0) {
+          if (pXrsp->createTableNum == 0) {
+            pXrsp->createTableLen = taosArrayInit(0, sizeof(int32_t));
+            pXrsp->createTableReq = taosArrayInit(0, sizeof(void*));
+          }
+          void* createReq = taosMemoryCalloc(1, pBlk->schemaLen);
+          memcpy(createReq, pBlk->data, pBlk->schemaLen);
+          taosArrayPush(pXrsp->createTableLen, &pBlk->schemaLen);
+          taosArrayPush(pXrsp->createTableReq, &createReq);
+          pXrsp->createTableNum++;
+        }
+      }
+    }
+#endif
   }
 
   if (pRsp->blockNum == 0) {

source/libs/executor/inc/executorimpl.h

@@ -143,6 +143,8 @@ typedef struct {
   STqOffsetVal   prepareStatus; // for tmq
   STqOffsetVal   lastStatus;    // for tmq
   SMqMetaRsp     metaRsp;       // for tmq fetching meta
+  int64_t snapshotVer;
+
   SSchemaWrapper *schema;
   char           tbName[TSDB_TABLE_NAME_LEN];
   SSDataBlock*   pullOverBlk;   // for streaming
@@ -486,13 +488,12 @@ typedef struct SStreamScanInfo {
   STimeWindowAggSupp     twAggSup;
   SSDataBlock*           pUpdateDataRes;
   // status for tmq
-  // SSchemaWrapper schema;
   SNodeList* pGroupTags;
   SNode*     pTagCond;
   SNode*     pTagIndexCond;
 } SStreamScanInfo;
 
-typedef struct SStreamRawScanInfo{
+typedef struct {
   //  int8_t    subType;
   //  bool      withMeta;
   //  int64_t   suid;

source/libs/executor/src/executor.c

@@ -139,7 +139,7 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO
 
 qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchema) {
   if (msg == NULL) {
-    // TODO create raw scan
+    // create raw scan
 
     SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo));
     if (NULL == pTaskInfo) {
@@ -853,11 +853,13 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
     pTaskInfo->streamInfo.tableCond.twindows.skey = pOffset->ts;
     pTaskInfo->tableqinfoList.pTableList = taosArrayInit(1, sizeof(STableKeyInfo));
     taosArrayPush(pTaskInfo->tableqinfoList.pTableList, &(STableKeyInfo){.uid = mtInfo.uid, .groupId = 0});
-    tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pTaskInfo->tableqinfoList.pTableList, &pInfo->dataReader, NULL);
+    tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pTaskInfo->tableqinfoList.pTableList,
+                   &pInfo->dataReader, NULL);
 
     strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName);
     tDeleteSSchemaWrapper(pTaskInfo->streamInfo.schema);
     pTaskInfo->streamInfo.schema = mtInfo.schema;
+
     qDebug("tmqsnap qStreamPrepareScan snapshot data uid %ld ts %ld", mtInfo.uid, pOffset->ts);
   } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_META) {
     SStreamRawScanInfo* pInfo = pOperator->info;

source/libs/executor/src/executorimpl.c

@@ -2815,92 +2815,6 @@ int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scan
     }
   }
 }
-#if 0
-int32_t doPrepareScan(SOperatorInfo* pOperator, uint64_t uid, int64_t ts) {
-  uint8_t type = pOperator->operatorType;
-
-  pOperator->status = OP_OPENED;
-
-  if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
-    SStreamScanInfo* pScanInfo = pOperator->info;
-    pScanInfo->blockType = STREAM_INPUT__TABLE_SCAN;
-
-    pScanInfo->pTableScanOp->status = OP_OPENED;
-
-    STableScanInfo* pInfo = pScanInfo->pTableScanOp->info;
-    ASSERT(pInfo->scanMode == TABLE_SCAN__TABLE_ORDER);
-
-    if (uid == 0) {
-      pInfo->noTable = 1;
-      return TSDB_CODE_SUCCESS;
-    }
-
-    /*if (pSnapShotScanInfo->dataReader == NULL) {*/
-    /*pSnapShotScanInfo->dataReader = tsdbReaderOpen(pHandle->vnode, &pSTInfo->cond, tableList, 0, 0);*/
-    /*pSnapShotScanInfo->scanMode = TABLE_SCAN__TABLE_ORDER;*/
-    /*}*/
-
-    pInfo->noTable = 0;
-
-    if (pInfo->lastStatus.uid != uid || pInfo->lastStatus.ts != ts) {
-      SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
-
-      int32_t tableSz = taosArrayGetSize(pTaskInfo->tableqinfoList.pTableList);
-      bool    found = false;
-      for (int32_t i = 0; i < tableSz; i++) {
-        STableKeyInfo* pTableInfo = taosArrayGet(pTaskInfo->tableqinfoList.pTableList, i);
-        if (pTableInfo->uid == uid) {
-          found = true;
-          pInfo->currentTable = i;
-        }
-      }
-      // TODO after processing drop, found can be false
-      ASSERT(found);
-
-      tsdbSetTableId(pInfo->dataReader, uid);
-      int64_t oldSkey = pInfo->cond.twindows.skey;
-      pInfo->cond.twindows.skey = ts + 1;
-      tsdbReaderReset(pInfo->dataReader, &pInfo->cond);
-      pInfo->cond.twindows.skey = oldSkey;
-      pInfo->scanTimes = 0;
-
-      qDebug("tsdb reader offset seek to uid %" PRId64 " ts %" PRId64 ", table cur set to %d , all table num %d", uid, ts,
-             pInfo->currentTable, tableSz);
-    }
-
-    return TSDB_CODE_SUCCESS;
-
-  } else {
-    if (pOperator->numOfDownstream == 1) {
-      return doPrepareScan(pOperator->pDownstream[0], uid, ts);
-    } else if (pOperator->numOfDownstream == 0) {
-      qError("failed to find stream scan operator to set the input data block");
-      return TSDB_CODE_QRY_APP_ERROR;
-    } else {
-      qError("join not supported for stream block scan");
-      return TSDB_CODE_QRY_APP_ERROR;
-    }
-  }
-}
-
-int32_t doGetScanStatus(SOperatorInfo* pOperator, uint64_t* uid, int64_t* ts) {
-  int32_t type = pOperator->operatorType;
-  if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
-    SStreamScanInfo* pScanInfo = pOperator->info;
-    STableScanInfo*  pSnapShotScanInfo = pScanInfo->pTableScanOp->info;
-    *uid = pSnapShotScanInfo->lastStatus.uid;
-    *ts = pSnapShotScanInfo->lastStatus.ts;
-  } else {
-    if (pOperator->pDownstream[0] == NULL) {
-      return TSDB_CODE_INVALID_PARA;
-    } else {
-      doGetScanStatus(pOperator->pDownstream[0], uid, ts);
-    }
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-#endif
 
 // this is a blocking operator
 static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
@@ -4294,42 +4208,6 @@ SArray* extractColumnInfo(SNodeList* pNodeList) {
   return pList;
 }
 
-#if 0
-STsdbReader* doCreateDataReader(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle,
-                                STableListInfo* pTableListInfo, const char* idstr) {
-  int32_t code = getTableList(pHandle->meta, pHandle->vnode, &pTableScanNode->scan, pTableListInfo);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  if (taosArrayGetSize(pTableListInfo->pTableList) == 0) {
-    code = 0;
-    qDebug("no table qualified for query, %s", idstr);
-    goto _error;
-  }
-
-  SQueryTableDataCond cond = {0};
-  code = initQueryTableDataCond(&cond, pTableScanNode);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  STsdbReader* pReader;
-  code = tsdbReaderOpen(pHandle->vnode, &cond, pTableListInfo->pTableList, &pReader, idstr);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  cleanupQueryTableDataCond(&cond);
-
-  return pReader;
-
-_error:
-  terrno = code;
-  return NULL;
-}
-#endif
-
 static int32_t extractTbscanInStreamOpTree(SOperatorInfo* pOperator, STableScanInfo** ppInfo) {
   if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
     if (pOperator->numOfDownstream == 0) {

source/libs/executor/src/scanoperator.c

@@ -1219,7 +1219,7 @@ static void setBlockGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32
 static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock) {
   SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
   SOperatorInfo*  pOperator = pInfo->pStreamScanOp;
-  SExecTaskInfo*  pTaskInfo = pInfo->pStreamScanOp->pTaskInfo;
+  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
 
   blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);
 
@@ -1228,7 +1228,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
   pInfo->pRes->info.type = STREAM_NORMAL;
   pInfo->pRes->info.version = pBlock->info.version;
 
-  uint64_t* groupIdPre = taosHashGet(pOperator->pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
+  uint64_t* groupIdPre = taosHashGet(pTaskInfo->tableqinfoList.map, &pBlock->info.uid, sizeof(int64_t));
   if (groupIdPre) {
     pInfo->pRes->info.groupId = *groupIdPre;
   } else {
@@ -1276,11 +1276,80 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
   return 0;
 }
 
+static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
+  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;
+  SStreamScanInfo* pInfo = pOperator->info;
+
+  qDebug("queue scan called");
+  if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
+    SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
+    if (pResult && pResult->info.rows > 0) {
+      qDebug("queue scan tsdb return %d rows", pResult->info.rows);
+      return pResult;
+    } else {
+      STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
+      tsdbReaderClose(pTSInfo->dataReader);
+      pTSInfo->dataReader = NULL;
+      tqOffsetResetToLog(&pTaskInfo->streamInfo.prepareStatus, pTaskInfo->streamInfo.snapshotVer);
+      qDebug("queue scan tsdb over, switch to wal ver %d", pTaskInfo->streamInfo.snapshotVer + 1);
+      if (tqSeekVer(pInfo->tqReader, pTaskInfo->streamInfo.snapshotVer + 1) < 0) {
+        return NULL;
+      }
+      ASSERT(pInfo->tqReader->pWalReader->curVersion == pTaskInfo->streamInfo.snapshotVer + 1);
+    }
+  }
+
+  if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) {
+    while (1) {
+      SFetchRet ret = {0};
+      tqNextBlock(pInfo->tqReader, &ret);
+      if (ret.fetchType == FETCH_TYPE__DATA) {
+        blockDataCleanup(pInfo->pRes);
+        if (setBlockIntoRes(pInfo, &ret.data) < 0) {
+          ASSERT(0);
+        }
+        // TODO clean data block
+        if (pInfo->pRes->info.rows > 0) {
+          qDebug("queue scan log return %d rows", pInfo->pRes->info.rows);
+          return pInfo->pRes;
+        }
+      } else if (ret.fetchType == FETCH_TYPE__META) {
+        ASSERT(0);
+        //        pTaskInfo->streamInfo.lastStatus = ret.offset;
+        //        pTaskInfo->streamInfo.metaBlk = ret.meta;
+        //        return NULL;
+      } else if (ret.fetchType == FETCH_TYPE__NONE) {
+        pTaskInfo->streamInfo.lastStatus = ret.offset;
+        ASSERT(pTaskInfo->streamInfo.lastStatus.version >= pTaskInfo->streamInfo.prepareStatus.version);
+        ASSERT(pTaskInfo->streamInfo.lastStatus.version + 1 == pInfo->tqReader->pWalReader->curVersion);
+        char formatBuf[80];
+        tFormatOffset(formatBuf, 80, &ret.offset);
+        qDebug("queue scan log return null, offset %s", formatBuf);
+        return NULL;
+      } else {
+        ASSERT(0);
+      }
+    }
+  } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
+    SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
+    if (pResult && pResult->info.rows > 0) {
+      qDebug("stream scan tsdb return %d rows", pResult->info.rows);
+      return pResult;
+    }
+    qDebug("stream scan tsdb return null");
+    return NULL;
+  } else {
+    ASSERT(0);
+    return NULL;
+  }
+}
+
 static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
   // NOTE: this operator does never check if current status is done or not
   SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;
   SStreamScanInfo* pInfo = pOperator->info;
 
+  qDebug("stream scan called");
 #if 0
   SStreamState* pState = pTaskInfo->streamInfo.pState;
   if (pState) {
@@ -1317,48 +1386,6 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
   }
 #endif
 
-  qDebug("stream scan called");
-  if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__LOG) {
-    while (1) {
-      SFetchRet ret = {0};
-      tqNextBlock(pInfo->tqReader, &ret);
-      if (ret.fetchType == FETCH_TYPE__DATA) {
-        blockDataCleanup(pInfo->pRes);
-        if (setBlockIntoRes(pInfo, &ret.data) < 0) {
-          ASSERT(0);
-        }
-        // TODO clean data block
-        if (pInfo->pRes->info.rows > 0) {
-          qDebug("stream scan log return %d rows", pInfo->pRes->info.rows);
-          return pInfo->pRes;
-        }
-      } else if (ret.fetchType == FETCH_TYPE__META) {
-        ASSERT(0);
-//        pTaskInfo->streamInfo.lastStatus = ret.offset;
-//        pTaskInfo->streamInfo.metaBlk = ret.meta;
-//        return NULL;
-      } else if (ret.fetchType == FETCH_TYPE__NONE) {
-        pTaskInfo->streamInfo.lastStatus = ret.offset;
-        ASSERT(pTaskInfo->streamInfo.lastStatus.version >= pTaskInfo->streamInfo.prepareStatus.version);
-        ASSERT(pTaskInfo->streamInfo.lastStatus.version + 1 == pInfo->tqReader->pWalReader->curVersion);
-        char formatBuf[80];
-        tFormatOffset(formatBuf, 80, &ret.offset);
-        qDebug("stream scan log return null, offset %s", formatBuf);
-        return NULL;
-      } else {
-        ASSERT(0);
-      }
-    }
-  } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) {
-    SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp);
-    if (pResult && pResult->info.rows > 0) {
-      qDebug("stream scan tsdb return %d rows", pResult->info.rows);
-      return pResult;
-    }
-    qDebug("stream scan tsdb return null");
-    return NULL;
-  }
-
   if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__PREPARE) {
     STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
     memcpy(&pTSInfo->cond, &pTaskInfo->streamInfo.tableCond, sizeof(SQueryTableDataCond));
@@ -1590,15 +1617,11 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
       pTaskInfo->streamInfo.prepareStatus.uid = mtInfo.uid;
       pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__LOG;
       pTaskInfo->streamInfo.lastStatus.version = pInfo->sContext->snapVersion;
-      tDeleteSSchemaWrapper(pTaskInfo->streamInfo.schema);
     } else {
       pTaskInfo->streamInfo.prepareStatus.uid = mtInfo.uid;
       pTaskInfo->streamInfo.prepareStatus.ts = INT64_MIN;
       qDebug("tmqsnap change get data uid:%ld", mtInfo.uid);
       qStreamPrepareScan(pTaskInfo, &pTaskInfo->streamInfo.prepareStatus, pInfo->sContext->subType);
-      strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName);
-      tDeleteSSchemaWrapper(pTaskInfo->streamInfo.schema);
-      pTaskInfo->streamInfo.schema = mtInfo.schema;
     }
     qDebug("tmqsnap stream scan tsdb return null");
     return NULL;
@@ -1647,8 +1670,8 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
   //      if (pHead->msgType == TDMT_VND_SUBMIT) {
   //        SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
   //        tqReaderSetDataMsg(pInfo->tqReader, pCont, 0);
-//        SSDataBlock* block = tqLogScanExec(pInfo->sContext->subType, pInfo->tqReader, pInfo->pFilterOutTbUid, &pInfo->pRes);
+  //        SSDataBlock* block = tqLogScanExec(pInfo->sContext->subType, pInfo->tqReader, pInfo->pFilterOutTbUid,
-//        if(block){
+  //        &pInfo->pRes); if(block){
   //          pTaskInfo->streamInfo.lastStatus.type = TMQ_OFFSET__LOG;
   //          pTaskInfo->streamInfo.lastStatus.version = fetchVer;
   //          qDebug("tmqsnap fetch data msg, ver:%" PRId64 ", type:%d", pHead->version, pHead->msgType);
@@ -1704,8 +1727,7 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT
   pOperator->info = pInfo;
   pOperator->pTaskInfo = pTaskInfo;
 
-  pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, NULL, destroyRawScanOperatorInfo,
+  pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, NULL, destroyRawScanOperatorInfo, NULL, NULL, NULL);
-                                         NULL, NULL, NULL);
   return pOperator;
 }
 
@@ -1724,7 +1746,7 @@ static void destroyStreamScanOperatorInfo(void* param) {
   }
   if (pStreamScan->pPseudoExpr) {
     destroyExprInfo(pStreamScan->pPseudoExpr, pStreamScan->numOfPseudoExpr);
-    taosMemoryFreeClear(pStreamScan->pPseudoExpr);
+    taosMemoryFree(pStreamScan->pPseudoExpr);
   }
 
   updateInfoDestroy(pStreamScan->pUpdateInfo);
@@ -1815,6 +1837,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
 
     pInfo->readHandle = *pHandle;
     pInfo->tableUid = pScanPhyNode->uid;
+    pTaskInfo->streamInfo.snapshotVer = pHandle->version;
 
     // set the extract column id to streamHandle
     tqReaderSetColIdList(pInfo->tqReader, pColIds);
@@ -1858,8 +1881,9 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
   pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
   pOperator->pTaskInfo = pTaskInfo;
 
-  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamScan, NULL, NULL, destroyStreamScanOperatorInfo,
+  __optr_fn_t nextFn = pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM ? doStreamScan : doQueueScan;
-                                         NULL, NULL, NULL);
+  pOperator->fpSet =
+      createOperatorFpSet(operatorDummyOpenFn, nextFn, NULL, NULL, destroyStreamScanOperatorInfo, NULL, NULL, NULL);
 
   return pOperator;