add msg for vnode consume

include/common/tmsg.h

@@ -1552,6 +1552,25 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
   return buf;
 }
 
+typedef struct SMqCVConsumeReq {
+  int64_t reqId;
+  int64_t offset;
+  int64_t clientId;
+  char    topicName[TSDB_TOPIC_FNAME_LEN];
+  char    cgroup[TSDB_CONSUMER_GROUP_LEN];
+} SMqCVConsumeReq;
+
+typedef struct SMqCVConsumeRsp {
+  int64_t reqId;
+  int64_t clientId;
+  int64_t committedOffset;
+  int64_t receiveOffset;
+  int64_t rspOffset;
+  int32_t skipLogNum;
+  int32_t bodyLen;
+  char    topicName[TSDB_TOPIC_FNAME_LEN];
+  char    body[];
+} SMqCvConsumeRsp;
 
 #ifdef __cplusplus
 }

include/common/tmsgdef.h

@@ -175,6 +175,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SCHEDULE_DATA_SINK, "vnode-schedule-data-sink", NULL, NULL)
 
   TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp)
 
 
   // Requests handled by QNODE

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

@@ -542,7 +542,10 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
   return 0;
 }
 
-static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pMsg) { return 0; }
+static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pRsp) {
+  mndTransProcessRsp(pRsp);
+  return 0;
+}
 
 static int32_t mndProcessConsumerMetaMsg(SMnodeMsg *pMsg) {
   SMnode        *pMnode = pMsg->pMnode;

source/dnode/vnode/CMakeLists.txt

@@ -25,6 +25,7 @@ target_link_libraries(
     PUBLIC bdb
     PUBLIC tfs
     PUBLIC wal
+    PUBLIC scheduler
     PUBLIC qworker
 )
 

source/dnode/vnode/inc/tq.h

@@ -18,14 +18,15 @@
 
 #include "common.h"
 #include "mallocator.h"
+#include "meta.h"
 #include "os.h"
+#include "scheduler.h"
 #include "taoserror.h"
-#include "tmsg.h"
 #include "tlist.h"
+#include "tmsg.h"
 #include "trpc.h"
 #include "ttimer.h"
 #include "tutil.h"
-#include "meta.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -159,6 +160,31 @@ typedef struct STqGroup {
   STqRspHandle rspHandle;
 } STqGroup;
 
+typedef struct STqTaskItem {
+  int32_t       status;
+  void*         dst;
+  SSubQueryMsg* pMsg;
+} STqTaskItem;
+
+// new version
+typedef struct STqBuffer {
+  int64_t     firstOffset;
+  int64_t     lastOffset;
+  STqTaskItem output[TQ_BUFFER_SIZE];
+} STqBuffer;
+
+typedef struct STqClientHandle {
+  int64_t   clientId;
+  char      topicName[TSDB_TOPIC_FNAME_LEN];
+  char      cGroup[TSDB_TOPIC_FNAME_LEN];
+  char*     sql;
+  char*     logicalPlan;
+  char*     physicalPlan;
+  int64_t   committedOffset;
+  int64_t   currentOffset;
+  STqBuffer buffer;
+} STqClientHandle;
+
 typedef struct STqQueryMsg {
   STqMsgItem*         item;
   struct STqQueryMsg* next;
@@ -325,7 +351,6 @@ typedef struct STqReadHandle {
 } STqReadHandle;
 
 typedef struct SSubmitBlkScanInfo {
-  
 } SSubmitBlkScanInfo;
 
 STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta, SSubmitMsg* pMsg);
@@ -333,7 +358,6 @@ bool tqNextDataBlock(STqReadHandle* pHandle);
 int            tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo);
 // return SArray<SColumnInfoData>
 SArray* tqRetrieveDataBlock(STqReadHandle* pHandle, SArray* pColumnIdList);
-//int tqLoadDataBlock(SExecTaskInfo* pTaskInfo, SSubmitBlkScanInfo* pSubmitBlkScanInfo, SSDataBlock* pBlock, uint32_t status);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -57,6 +57,8 @@ int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       //      return qWorkerProcessShowFetchMsg(pVnode->pMeta, pVnode->pQuery, pMsg);
     case TDMT_VND_TABLE_META:
       return vnodeGetTableMeta(pVnode, pMsg, pRsp);
+    case TDMT_VND_CONSUME:
+      return 0;
     default:
       vError("unknown msg type:%d in fetch queue", pMsg->msgType);
       return TSDB_CODE_VND_APP_ERROR;

source/dnode/vnode/src/vnd/vnodeWrite.c

@@ -14,6 +14,7 @@
  */
 
 #include "vnd.h"
+#include "tq.h"
 
 int vnodeProcessNoWalWMsgs(SVnode *pVnode, SRpcMsg *pMsg) {
   switch (pMsg->msgType) {
@@ -111,9 +112,34 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
     case TDMT_VND_MQ_SET_CONN: {
       char* reqStr = ptr;
       SMqSetCVgReq req;
-      /*tDecodeSMqSetCVgReq(reqStr, &req);*/
-      // create topic if not exist
+      tDecodeSMqSetCVgReq(reqStr, &req);
+      STqClientHandle* pHandle = calloc(sizeof(STqClientHandle), 1);
+      if (pHandle == NULL) {
+        // TODO: handle error
+      }
+      strcpy(pHandle->topicName, req.topicName); 
+      strcpy(pHandle->cGroup, req.cGroup); 
+      strcpy(pHandle->sql, req.sql);
+      strcpy(pHandle->logicalPlan, req.logicalPlan);
+      strcpy(pHandle->physicalPlan, req.physicalPlan);
+      SArray *pArray;
+      //TODO: deserialize to SQueryDag
+      SQueryDag *pDag;
       // convert to task
+      if (schedulerConvertDagToTaskList(pDag, &pArray) < 0) {
+        // TODO: handle error
+      }
+      ASSERT(taosArrayGetSize(pArray) == 0);
+      STaskInfo *pInfo = taosArrayGet(pArray, 0);
+      SArray* pTasks;
+      schedulerCopyTask(pInfo, &pTasks, TQ_BUFFER_SIZE);
+      pHandle->buffer.firstOffset = -1;
+      pHandle->buffer.lastOffset = -1;
+      for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
+        SSubQueryMsg* pMsg = taosArrayGet(pTasks, i);
+        pHandle->buffer.output[i].pMsg = pMsg;
+        pHandle->buffer.output[i].status = 0;
+      }
       // write mq meta
     }
       break;