Merge pull request #11051 from taosdata/feature/tq

stream support sma

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

@@ -19,15 +19,14 @@
 #include "tmallocator.h"
 // #include "sync.h"
 #include "tcoding.h"
+#include "tdatablock.h"
 #include "tfs.h"
 #include "tlist.h"
 #include "tlockfree.h"
 #include "tmacro.h"
-#include "wal.h"
-
 #include "vnode.h"
-
 #include "vnodeQuery.h"
+#include "wal.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -203,7 +202,7 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen);
 
 // sma
-void smaHandleRes(SVnode* pVnode, int64_t smaId, const SArray* data);
+void smaHandleRes(void* pVnode, int64_t smaId, const SArray* data);
 
 #ifdef __cplusplus
 }

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

@@ -473,10 +473,16 @@ int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
   }
   tCoderClear(&decoder);
 
+  // exec
   if (tqExpandTask(pTq, pTask, 4) < 0) {
     ASSERT(0);
   }
+
+  // sink
   pTask->ahandle = pTq->pVnode;
+  if (pTask->sinkType == TASK_SINK__SMA) {
+    pTask->smaSink.smaHandle = smaHandleRes;
+  }
 
   taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask));
 

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

@@ -15,6 +15,11 @@
 
 #include "vnd.h"
 
+void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
+  // TODO
+  blockDebugShowData(data);
+}
+
 void vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
   SNodeMsg *pMsg;
   SRpcMsg  *pRpc;