Merge branch '3.0' into feature/3.0_liaohj

example/src/tmq.c

@@ -13,16 +13,14 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include <assert.h>
 #include <stdio.h>
 #include <string.h>
-#include <assert.h>
 #include <time.h>
 #include "taos.h"
 
 static int  running = 1;
-static void msg_process(tmq_message_t* message) {
+static void msg_process(tmq_message_t* message) { tmqShowMsg(message); }
-  tmqShowMsg(message);
-}
 
 int32_t init_env() {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
@@ -44,29 +42,28 @@ int32_t init_env() {
   }
   taos_free_result(pRes);
 
-  /*pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)");*/
+  pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, k int) tags(a int)");
-  /*if (taos_errno(pRes) != 0) {*/
+  if (taos_errno(pRes) != 0) {
-    /*printf("failed to create super table 123_$^), reason:%s\n", taos_errstr(pRes));*/
+    printf("failed to create super table 123_$^), reason:%s\n", taos_errstr(pRes));
-    /*return -1;*/
+    return -1;
-  /*}*/
+  }
-  /*taos_free_result(pRes);*/
+  taos_free_result(pRes);
 
-  /*pRes = taos_query(pConn, "create table if not exists tu using st1 tags(1)");*/
+  pRes = taos_query(pConn, "create table if not exists tu1 using st1 tags(1)");
-  /*if (taos_errno(pRes) != 0) {*/
+  if (taos_errno(pRes) != 0) {
-    /*printf("failed to create child table tu, reason:%s\n", taos_errstr(pRes));*/
+    printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes));
-    /*return -1;*/
+    return -1;
-  /*}*/
+  }
-  /*taos_free_result(pRes);*/
+  taos_free_result(pRes);
 
-  /*pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags(2)");*/
+  pRes = taos_query(pConn, "create table if not exists tu2 using st1 tags(2)");
-  /*if (taos_errno(pRes) != 0) {*/
+  if (taos_errno(pRes) != 0) {
-    /*printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes));*/
+    printf("failed to create child table tu2, reason:%s\n", taos_errstr(pRes));
-    /*return -1;*/
+    return -1;
-  /*}*/
+  }
-  /*taos_free_result(pRes);*/
+  taos_free_result(pRes);
 
-
+  const char* sql = "select * from tu2";
-  const char* sql = "select * from st1";
   pRes = tmq_create_topic(pConn, "test_stb_topic_1", sql, strlen(sql));
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic test_stb_topic_1, reason:%s\n", taos_errstr(pRes));
@@ -104,8 +101,7 @@ tmq_list_t* build_topic_list() {
   return topic_list;
 }
 
-void basic_consume_loop(tmq_t *tmq,
+void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
-                        tmq_list_t *topics) {
   tmq_resp_err_t err;
 
   if ((err = tmq_subscribe(tmq, topics))) {
@@ -135,8 +131,7 @@ void basic_consume_loop(tmq_t *tmq,
     fprintf(stderr, "%% Consumer closed\n");
 }
 
-void sync_consume_loop(tmq_t *tmq,
+void sync_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
-                  tmq_list_t *topics) {
   static const int MIN_COMMIT_COUNT = 1000;
 
   int            msg_count = 0;
@@ -153,8 +148,7 @@ void sync_consume_loop(tmq_t *tmq,
       msg_process(tmqmessage);
       tmq_message_destroy(tmqmessage);
 
-      if ((++msg_count % MIN_COMMIT_COUNT) == 0)
+      if ((++msg_count % MIN_COMMIT_COUNT) == 0) tmq_commit(tmq, NULL, 0);
-        tmq_commit(tmq, NULL, 0);
     }
   }
 

include/common/tmsg.h

@@ -25,9 +25,9 @@ extern "C" {
 #include "taoserror.h"
 #include "tarray.h"
 #include "tcoding.h"
-#include "trow.h"
 #include "thash.h"
 #include "tlist.h"
+#include "trow.h"
 
 /* ------------------------ MESSAGE DEFINITIONS ------------------------ */
 #define TD_MSG_NUMBER_
@@ -82,7 +82,6 @@ enum {
   HEARTBEAT_KEY_MQ_TMP,
 };
 
-
 typedef enum _mgmt_table {
   TSDB_MGMT_TABLE_START,
   TSDB_MGMT_TABLE_ACCT,
@@ -192,7 +191,7 @@ typedef struct {
 
 // Submit message for one table
 typedef struct SSubmitBlk {
-  uint64_t uid;        // table unique id
+  int64_t uid;        // table unique id
   int32_t tid;        // table id
   int32_t padding;    // TODO just for padding here
   int32_t sversion;   // data schema version
@@ -328,7 +327,6 @@ static FORCE_INLINE void* taosDecodeSEpSet(void* buf, SEpSet* pEp) {
   }
   return buf;
 }
-
 static FORCE_INLINE int32_t tEncodeSEpSet(SCoder* pEncoder, const SEpSet* pEp) {
   if (tEncodeI8(pEncoder, pEp->inUse) < 0) return -1;
   if (tEncodeI8(pEncoder, pEp->numOfEps) < 0) return -1;
@@ -1043,7 +1041,6 @@ typedef struct {
   uint64_t taskId;
 } SQueryContinueReq;
 
-
 typedef struct {
   SMsgHead header;
   uint64_t sId;
@@ -1277,7 +1274,8 @@ _err:
   return NULL;
 }
 
-// this message is sent from mnode to mnode(read thread to write thread), so there is no need for serialization / deserialization
+// this message is sent from mnode to mnode(read thread to write thread), so there is no need for serialization /
+// deserialization
 typedef struct {
   // SArray* rebSubscribes;     //SArray<SMqRebSubscribe>
   SHashObj* rebSubHash;  // SHashObj<key, SMqRebSubscribe>
@@ -1941,9 +1939,7 @@ typedef struct {
   SArray* topics;  // SArray<SMqSubTopicEp>
 } SMqCMGetSubEpRsp;
 
-static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) {
+static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { taosArrayDestroy(pSubTopicEp->vgs); }
-  taosArrayDestroy(pSubTopicEp->vgs);
-}
 
 static FORCE_INLINE int32_t tEncodeSMqSubVgEp(void** buf, const SMqSubVgEp* pVgEp) {
   int32_t tlen = 0;

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

@@ -205,6 +205,117 @@ int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsu
 }
 
 int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
+  SMqConsumeReq* pReq = pMsg->pCont;
+  int64_t        consumerId = pReq->consumerId;
+  int64_t        fetchOffset = pReq->offset;
+  /*int64_t        blockingTime = pReq->blockingTime;*/
+
+  SMqConsumeRsp rsp = {.consumerId = consumerId, .numOfTopics = 0, .pBlockData = NULL};
+
+  STqConsumer* pConsumer = tqHandleGet(pTq->tqMeta, consumerId);
+  if (pConsumer == NULL) {
+    pMsg->pCont = NULL;
+    pMsg->contLen = 0;
+    pMsg->code = -1;
+    rpcSendResponse(pMsg);
+    return 0;
+  }
+  int sz = taosArrayGetSize(pConsumer->topics);
+  ASSERT(sz == 1);
+  STqTopic* pTopic = taosArrayGet(pConsumer->topics, 0);
+  ASSERT(strcmp(pTopic->topicName, pReq->topic) == 0);
+  ASSERT(pConsumer->consumerId == consumerId);
+
+  if (pReq->reqType == TMQ_REQ_TYPE_COMMIT_ONLY) {
+    pTopic->committedOffset = pReq->offset;
+    pMsg->pCont = NULL;
+    pMsg->contLen = 0;
+    pMsg->code = 0;
+    rpcSendResponse(pMsg);
+    return 0;
+  }
+
+  if (pReq->reqType == TMQ_REQ_TYPE_CONSUME_AND_COMMIT) {
+    pTopic->committedOffset = pReq->offset - 1;
+  }
+
+  rsp.committedOffset = pTopic->committedOffset;
+  rsp.reqOffset = pReq->offset;
+  rsp.skipLogNum = 0;
+
+  SWalHead* pHead;
+  while (1) {
+    int8_t pos = fetchOffset % TQ_BUFFER_SIZE;
+    if (walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) {
+      break;
+    }
+    pHead = pTopic->pReadhandle->pHead;
+    if (pHead->head.msgType == TDMT_VND_SUBMIT) {
+      SSubmitMsg* pCont = (SSubmitMsg*)&pHead->head.body;
+      qTaskInfo_t task = pTopic->buffer.output[pos].task;
+      qSetStreamInput(task, pCont);
+      SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
+      while (1) {
+        SSDataBlock* pDataBlock;
+        uint64_t     ts;
+        if (qExecTask(task, &pDataBlock, &ts) < 0) {
+          ASSERT(false);
+        }
+        if (pDataBlock == NULL) {
+          fetchOffset++;
+          rsp.skipLogNum++;
+          break;
+        }
+
+        taosArrayPush(pRes, pDataBlock);
+        rsp.schemas = pTopic->buffer.output[pos].pReadHandle->pSchemaWrapper;
+        rsp.rspOffset = fetchOffset;
+        pTopic->currentOffset = fetchOffset;
+
+        rsp.numOfTopics = 1;
+        rsp.pBlockData = pRes;
+
+        int32_t tlen = tEncodeSMqConsumeRsp(NULL, &rsp);
+        void*   buf = rpcMallocCont(tlen);
+        if (buf == NULL) {
+          pMsg->code = -1;
+          return -1;
+        }
+
+        void* abuf = buf;
+        tEncodeSMqConsumeRsp(&abuf, &rsp);
+        taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);
+        pMsg->pCont = buf;
+        pMsg->contLen = tlen;
+        pMsg->code = 0;
+        rpcSendResponse(pMsg);
+        return 0;
+      }
+    } else {
+      fetchOffset++;
+      rsp.skipLogNum++;
+    }
+  }
+
+  int32_t tlen = tEncodeSMqConsumeRsp(NULL, &rsp);
+  void*   buf = rpcMallocCont(tlen);
+  if (buf == NULL) {
+    pMsg->code = -1;
+    return -1;
+  }
+
+  void* abuf = buf;
+  tEncodeSMqConsumeRsp(&abuf, &rsp);
+  rsp.pBlockData = NULL;
+  pMsg->pCont = buf;
+  pMsg->contLen = tlen;
+  pMsg->code = 0;
+  rpcSendResponse(pMsg);
+  return 0;
+}
+
+#if 0
+int32_t tqProcessConsumeReqV0(STQ* pTq, SRpcMsg* pMsg) {
   SMqConsumeReq* pReq = pMsg->pCont;
   int64_t        reqId = pReq->reqId;
   int64_t        consumerId = pReq->consumerId;
@@ -265,6 +376,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
         break;
       }
       if (walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) {
+        printf("read offset %ld\n", fetchOffset);
         // check err
         atomic_store_8(&pTopic->buffer.output[pos].status, 0);
         skip = 1;
@@ -273,10 +385,10 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
       // read until find TDMT_VND_SUBMIT
       pHead = pTopic->pReadhandle->pHead;
       if (pHead->head.msgType == TDMT_VND_SUBMIT) {
-        break;
       }
       rsp.skipLogNum++;
       if (walReadWithHandle(pTopic->pReadhandle, fetchOffset) < 0) {
+        printf("read offset %ld\n", fetchOffset);
         atomic_store_8(&pTopic->buffer.output[pos].status, 0);
         skip = 1;
         break;
@@ -288,6 +400,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
     SSubmitMsg* pCont = (SSubmitMsg*)&pHead->head.body;
     qTaskInfo_t task = pTopic->buffer.output[pos].task;
 
+    printf("current fetch offset %ld\n", fetchOffset);
     qSetStreamInput(task, pCont);
 
     // SArray<SSDataBlock>
@@ -307,6 +420,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
     // TODO copy
     rsp.schemas = pTopic->buffer.output[pos].pReadHandle->pSchemaWrapper;
     rsp.rspOffset = fetchOffset;
+    pTopic->currentOffset = fetchOffset;
 
     atomic_store_8(&pTopic->buffer.output[pos].status, 0);
 
@@ -350,6 +464,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
   rpcSendResponse(pMsg);
   return 0;
 }
+#endif
 
 int32_t tqProcessRebReq(STQ* pTq, char* msg) {
   SMqMVRebReq req = {0};

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

@@ -52,12 +52,15 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
     ASSERT(pHandle->tbIdHash);
     void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->pBlock->uid, sizeof(int64_t));
     if (ret != NULL) {
+      /*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/
       pHandle->pBlock->tid = htonl(pHandle->pBlock->tid);
       pHandle->pBlock->sversion = htonl(pHandle->pBlock->sversion);
       pHandle->pBlock->dataLen = htonl(pHandle->pBlock->dataLen);
       pHandle->pBlock->schemaLen = htonl(pHandle->pBlock->schemaLen);
       pHandle->pBlock->numOfRows = htons(pHandle->pBlock->numOfRows);
       return true;
+    } else {
+      /*printf("skip one tb %ld\n", pHandle->pBlock->uid);*/
     }
   }
   return false;