feat(tmq): change default config

include/common/tmsg.h

@@ -1455,7 +1455,7 @@ typedef struct {
 static FORCE_INLINE SMqRebInfo* tNewSMqRebSubscribe(const char* key) {
   SMqRebInfo* pRebInfo = (SMqRebInfo*)taosMemoryCalloc(1, sizeof(SMqRebInfo));
   if (pRebInfo == NULL) {
-    goto _err;
+    return NULL;
   }
   strcpy(pRebInfo->key, key);
   pRebInfo->lostConsumers = taosArrayInit(0, sizeof(int64_t));

include/common/tname.h

@@ -16,8 +16,8 @@
 #ifndef _TD_COMMON_NAME_H_
 #define _TD_COMMON_NAME_H_
 
-#include "tdef.h"
 #include "tarray.h"
+#include "tdef.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -65,7 +65,7 @@ bool tNameDBNameEqual(SName* left, SName* right);
 
 typedef struct {
   // input
-  SArray         *tags;         // element is SSmlKV
+  SArray*     tags;           // element is SSmlKv
   const char* sTableName;     // super table name
   uint8_t     sTableNameLen;  // the length of super table name
 

include/libs/stream/tstream.h

@@ -16,6 +16,7 @@
 #include "tdatablock.h"
 #include "tmsg.h"
 #include "tmsgcb.h"
+#include "tqueue.h"
 #include "trpc.h"
 
 #ifdef __cplusplus
@@ -154,6 +155,10 @@ struct SStreamTask {
     STaskDispatcherShuffle shuffleDispatcher;
   };
 
+  // msg buffer
+  int32_t     memUsed;
+  STaosQueue* inputQ;
+
   // application storage
   void* ahandle;
 };
@@ -194,6 +199,8 @@ typedef struct {
   SArray* res;  // SArray<SSDataBlock>
 } SStreamSinkReq;
 
+int32_t streamEnqueueData(SStreamTask* pTask, const void* input, int32_t inputType);
+
 int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId);
 
 #ifdef __cplusplus

source/client/src/tmq.c

@@ -187,7 +187,7 @@ typedef struct {
 
 tmq_conf_t* tmq_conf_new() {
   tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
-  conf->autoCommit = false;
+  conf->autoCommit = true;
   conf->autoCommitInterval = 5000;
   conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
   return conf;

source/common/src/tname.c

@@ -20,9 +20,7 @@
 
 #define VALID_NAME_TYPE(x) ((x) == TSDB_DB_NAME_T || (x) == TSDB_TABLE_NAME_T)
 
-bool tscValidateTableNameLength(size_t len) {
+bool tscValidateTableNameLength(size_t len) { return len < TSDB_TABLE_NAME_LEN; }
-  return len < TSDB_TABLE_NAME_LEN;
-}
 
 #if 0
 // TODO refactor
@@ -200,9 +198,7 @@ const char* tNameGetTableName(const SName* name) {
   return &name->tname[0];
 }
 
-void tNameAssign(SName* dst, const SName* src) {
+void tNameAssign(SName* dst, const SName* src) { memcpy(dst, src, sizeof(SName)); }
-  memcpy(dst, src, sizeof(SName));
-}
 
 int32_t tNameSetDbName(SName* dst, int32_t acct, const char* dbName, size_t nameLen) {
   assert(dst != NULL && dbName != NULL && nameLen > 0);
@@ -244,7 +240,6 @@ bool tNameDBNameEqual(SName* left, SName* right) {
   return (0 == strcmp(left->dbname, right->dbname));
 }
 
-
 int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
   assert(dst != NULL && str != NULL && strlen(str) > 0);
 

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

@@ -196,7 +196,9 @@ SMqVgEp *tCloneSMqVgEp(const SMqVgEp *pVgEp) {
   return pVgEpNew;
 }
 
-void tDeleteSMqVgEp(SMqVgEp *pVgEp) { taosMemoryFree(pVgEp->qmsg); }
+void tDeleteSMqVgEp(SMqVgEp *pVgEp) {
+  if (pVgEp->qmsg) taosMemoryFree(pVgEp->qmsg);
+}
 
 int32_t tEncodeSMqVgEp(void **buf, const SMqVgEp *pVgEp) {
   int32_t tlen = 0;

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

@@ -298,6 +298,8 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
 
     SNode *pAst = NULL;
     if (nodesStringToNode(pCreate->ast, &pAst) != 0) {
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
       return -1;
     }
@@ -307,16 +309,22 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
     SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true};
     if (qCreateQueryPlan(&cxt, &pPlan, NULL) != 0) {
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       return -1;
     }
 
     if (qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema) != 0) {
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       return -1;
     }
 
     if (nodesNodeToString(pPlan, false, &topicObj.physicalPlan, NULL) != 0) {
       mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+      taosMemoryFree(topicObj.ast);
+      taosMemoryFree(topicObj.sql);
       return -1;
     }
   } else {
@@ -331,6 +339,8 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_CREATE_TOPIC, &pReq->rpcMsg);
   if (pTrans == NULL) {
     mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
+    taosMemoryFreeClear(topicObj.ast);
+    taosMemoryFreeClear(topicObj.sql);
     taosMemoryFreeClear(topicObj.physicalPlan);
     return -1;
   }

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

@@ -233,17 +233,19 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
 int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
   if (msgType != TDMT_VND_SUBMIT) return 0;
 
+  // make sure msgType == TDMT_VND_SUBMIT
+  if (tsdbUpdateSmaWindow(pTq->pVnode->pTsdb, msg, ver) != 0) {
+    return -1;
+  }
+
+  if (taosHashGetSize(pTq->pStreamTasks) == 0) return 0;
+
   void* data = taosMemoryMalloc(msgLen);
   if (data == NULL) {
     return -1;
   }
   memcpy(data, msg, msgLen);
 
-  // make sure msgType == TDMT_VND_SUBMIT
-  if (tsdbUpdateSmaWindow(pTq->pVnode->pTsdb, msg, ver) != 0) {
-    return -1;
-  }
-
   SRpcMsg req = {
       .msgType = TDMT_VND_STREAM_TRIGGER,
       .pCont = data,