Merge remote-tracking branch 'origin/3.0' into feature/privilege

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 tu1";
-  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);
     }
   }
 
@@ -165,11 +159,45 @@ void sync_consume_loop(tmq_t *tmq,
     fprintf(stderr, "%% Consumer closed\n");
 }
 
+void perf_loop(tmq_t* tmq, tmq_list_t* topics) {
+  tmq_resp_err_t err;
+
+  if ((err = tmq_subscribe(tmq, topics))) {
+    fprintf(stderr, "%% Failed to start consuming topics: %s\n", tmq_err2str(err));
+    printf("subscribe err\n");
+    return;
+  }
+  int32_t batchCnt = 0;
+  int32_t skipLogNum = 0;
+  clock_t startTime = clock();
+  while (running) {
+    tmq_message_t* tmqmessage = tmq_consumer_poll(tmq, 500);
+    if (tmqmessage) {
+      batchCnt++;
+      skipLogNum += tmqGetSkipLogNum(tmqmessage);
+      /*msg_process(tmqmessage);*/
+      tmq_message_destroy(tmqmessage);
+    } else {
+      break;
+    }
+  }
+  clock_t endTime = clock();
+  printf("log batch cnt: %d, skip log cnt: %d, time used:%f s\n", batchCnt, skipLogNum,
+         (double)(endTime - startTime) / CLOCKS_PER_SEC);
+
+  err = tmq_consumer_close(tmq);
+  if (err)
+    fprintf(stderr, "%% Failed to close consumer: %s\n", tmq_err2str(err));
+  else
+    fprintf(stderr, "%% Consumer closed\n");
+}
+
 int main() {
   int code;
   code = init_env();
   tmq_t*      tmq = build_consumer();
   tmq_list_t* topic_list = build_topic_list();
-  basic_consume_loop(tmq, topic_list);
+  perf_loop(tmq, topic_list);
+  /*basic_consume_loop(tmq, topic_list);*/
   /*sync_consume_loop(tmq, topic_list);*/
 }

include/client/taos.h

@@ -16,8 +16,8 @@
 #ifndef TDENGINE_TAOS_H
 #define TDENGINE_TAOS_H
 
-#include <stdint.h>
 #include <stdbool.h>
+#include <stdint.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -128,7 +128,8 @@ typedef struct TAOS_MULTI_BIND {
 DLL_EXPORT void  taos_cleanup(void);
 DLL_EXPORT int   taos_options(TSDB_OPTION option, const void *arg, ...);
 DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
-DLL_EXPORT TAOS *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen, const char *db, int dbLen, uint16_t port);
+DLL_EXPORT TAOS *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen,
+                                const char *db, int dbLen, uint16_t port);
 DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port);
 DLL_EXPORT void  taos_close(TAOS *taos);
 
@@ -182,7 +183,8 @@ DLL_EXPORT void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp,
 DLL_EXPORT void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param);
 
 typedef void (*__taos_sub_fn_t)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
-DLL_EXPORT TAOS_SUB *taos_subscribe(TAOS* taos, int restart, const char* topic, const char *sql, __taos_sub_fn_t fp, void *param, int interval);
+DLL_EXPORT TAOS_SUB *taos_subscribe(TAOS *taos, int restart, const char *topic, const char *sql, __taos_sub_fn_t fp,
+                                    void *param, int interval);
 DLL_EXPORT TAOS_RES *taos_consume(TAOS_SUB *tsub);
 DLL_EXPORT void      taos_unsubscribe(TAOS_SUB *tsub, int keepProgress);
 
@@ -253,6 +255,7 @@ DLL_EXPORT void           tmq_conf_set_offset_commit_cb(tmq_conf_t *conf, tmq_co
 
 // temporary used function for demo only
 void    tmqShowMsg(tmq_message_t *tmq_message);
+int32_t tmqGetSkipLogNum(tmq_message_t *tmq_message);
 
 #ifdef __cplusplus
 }

include/common/common.h

@@ -17,8 +17,8 @@
 #define TDENGINE_COMMON_H
 
 #include "taosdef.h"
-#include "tmsg.h"
 #include "tarray.h"
+#include "tmsg.h"
 #include "tvariant.h"
 // typedef struct STimeWindow {
 //   TSKEY skey;
@@ -110,7 +110,7 @@ static FORCE_INLINE int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlo
   return tlen;
 }
 
-static FORCE_INLINE void* tDecodeDataBlock(void* buf, SSDataBlock* pBlock) {
+static FORCE_INLINE void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
   int32_t sz;
 
   buf = taosDecodeFixedI64(buf, &pBlock->info.uid);
@@ -127,7 +127,7 @@ static FORCE_INLINE void* tDecodeDataBlock(void* buf, SSDataBlock* pBlock) {
     buf = taosDecodeBinary(buf, (void**)&data.pData, colSz);
     taosArrayPush(pBlock->pDataBlock, &data);
   }
-  return buf;
+  return (void*)buf;
 }
 
 static FORCE_INLINE int32_t tEncodeSMqConsumeRsp(void** buf, const SMqConsumeRsp* pRsp) {
@@ -191,7 +191,6 @@ static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) {
   // tfree(pBlock);
 }
 
-
 static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqConsumeRsp* pRsp) {
   if (pRsp->schemas) {
     if (pRsp->schemas->nCols) {

include/common/taosdef.h

@@ -23,7 +23,7 @@ extern "C" {
 #include "taos.h"
 #include "tdef.h"
 
-typedef uint64_t tb_uid_t;
+typedef int64_t tb_uid_t;
 
 #define TSWINDOW_INITIALIZER ((STimeWindow){INT64_MIN, INT64_MAX})
 #define TSWINDOW_DESC_INITIALIZER ((STimeWindow){INT64_MAX, INT64_MIN})

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;
@@ -1055,7 +1053,6 @@ typedef struct {
   uint64_t taskId;
 } SQueryContinueReq;
 
-
 typedef struct {
   SMsgHead header;
   uint64_t sId;
@@ -1289,7 +1286,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>
@@ -1896,6 +1894,7 @@ static FORCE_INLINE void* tDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW)
   if (pSW->pSchema == NULL) {
     return NULL;
   }
+
   for (int32_t i = 0; i < pSW->nCols; i++) {
     buf = taosDecodeSSchema(buf, &pSW->pSchema[i]);
   }
@@ -1964,9 +1963,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;

include/util/tcoding.h

@@ -37,7 +37,7 @@ static FORCE_INLINE int taosEncodeFixedU8(void **buf, uint8_t value) {
   return (int)sizeof(value);
 }
 
-static FORCE_INLINE void *taosDecodeFixedU8(void *buf, uint8_t *value) {
+static FORCE_INLINE void *taosDecodeFixedU8(const void *buf, uint8_t *value) {
   *value = ((uint8_t *)buf)[0];
   return POINTER_SHIFT(buf, sizeof(*value));
 }
@@ -51,7 +51,7 @@ static FORCE_INLINE int taosEncodeFixedI8(void **buf, int8_t value) {
   return (int)sizeof(value);
 }
 
-static FORCE_INLINE void *taosDecodeFixedI8(void *buf, int8_t *value) {
+static FORCE_INLINE void *taosDecodeFixedI8(const void *buf, int8_t *value) {
   *value = ((int8_t *)buf)[0];
   return POINTER_SHIFT(buf, sizeof(*value));
 }
@@ -71,7 +71,7 @@ static FORCE_INLINE int taosEncodeFixedU16(void **buf, uint16_t value) {
   return (int)sizeof(value);
 }
 
-static FORCE_INLINE void *taosDecodeFixedU16(void *buf, uint16_t *value) {
+static FORCE_INLINE void *taosDecodeFixedU16(const void *buf, uint16_t *value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(value, buf, sizeof(*value));
   } else {
@@ -87,7 +87,7 @@ static FORCE_INLINE int taosEncodeFixedI16(void **buf, int16_t value) {
   return taosEncodeFixedU16(buf, ZIGZAGE(int16_t, value));
 }
 
-static FORCE_INLINE void *taosDecodeFixedI16(void *buf, int16_t *value) {
+static FORCE_INLINE void *taosDecodeFixedI16(const void *buf, int16_t *value) {
   uint16_t tvalue = 0;
   void    *ret = taosDecodeFixedU16(buf, &tvalue);
   *value = ZIGZAGD(int16_t, tvalue);
@@ -111,7 +111,7 @@ static FORCE_INLINE int taosEncodeFixedU32(void **buf, uint32_t value) {
   return (int)sizeof(value);
 }
 
-static FORCE_INLINE void *taosDecodeFixedU32(void *buf, uint32_t *value) {
+static FORCE_INLINE void *taosDecodeFixedU32(const void *buf, uint32_t *value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(value, buf, sizeof(*value));
   } else {
@@ -129,7 +129,7 @@ static FORCE_INLINE int taosEncodeFixedI32(void **buf, int32_t value) {
   return taosEncodeFixedU32(buf, ZIGZAGE(int32_t, value));
 }
 
-static FORCE_INLINE void *taosDecodeFixedI32(void *buf, int32_t *value) {
+static FORCE_INLINE void *taosDecodeFixedI32(const void *buf, int32_t *value) {
   uint32_t tvalue = 0;
   void    *ret = taosDecodeFixedU32(buf, &tvalue);
   *value = ZIGZAGD(int32_t, tvalue);
@@ -158,7 +158,7 @@ static FORCE_INLINE int taosEncodeFixedU64(void **buf, uint64_t value) {
   return (int)sizeof(value);
 }
 
-static FORCE_INLINE void *taosDecodeFixedU64(void *buf, uint64_t *value) {
+static FORCE_INLINE void *taosDecodeFixedU64(const void *buf, uint64_t *value) {
   if (IS_LITTLE_ENDIAN()) {
     memcpy(value, buf, sizeof(*value));
   } else {
@@ -180,7 +180,7 @@ static FORCE_INLINE int taosEncodeFixedI64(void **buf, int64_t value) {
   return taosEncodeFixedU64(buf, ZIGZAGE(int64_t, value));
 }
 
-static FORCE_INLINE void *taosDecodeFixedI64(void *buf, int64_t *value) {
+static FORCE_INLINE void *taosDecodeFixedI64(const void *buf, int64_t *value) {
   uint64_t tvalue = 0;
   void    *ret = taosDecodeFixedU64(buf, &tvalue);
   *value = ZIGZAGD(int64_t, tvalue);
@@ -205,7 +205,7 @@ static FORCE_INLINE int taosEncodeVariantU16(void **buf, uint16_t value) {
   return i + 1;
 }
 
-static FORCE_INLINE void *taosDecodeVariantU16(void *buf, uint16_t *value) {
+static FORCE_INLINE void *taosDecodeVariantU16(const void *buf, uint16_t *value) {
   int      i = 0;
   uint16_t tval = 0;
   *value = 0;
@@ -228,7 +228,7 @@ static FORCE_INLINE int taosEncodeVariantI16(void **buf, int16_t value) {
   return taosEncodeVariantU16(buf, ZIGZAGE(int16_t, value));
 }
 
-static FORCE_INLINE void *taosDecodeVariantI16(void *buf, int16_t *value) {
+static FORCE_INLINE void *taosDecodeVariantI16(const void *buf, int16_t *value) {
   uint16_t tvalue = 0;
   void    *ret = taosDecodeVariantU16(buf, &tvalue);
   *value = ZIGZAGD(int16_t, tvalue);
@@ -253,7 +253,7 @@ static FORCE_INLINE int taosEncodeVariantU32(void **buf, uint32_t value) {
   return i + 1;
 }
 
-static FORCE_INLINE void *taosDecodeVariantU32(void *buf, uint32_t *value) {
+static FORCE_INLINE void *taosDecodeVariantU32(const void *buf, uint32_t *value) {
   int      i = 0;
   uint32_t tval = 0;
   *value = 0;
@@ -276,7 +276,7 @@ static FORCE_INLINE int taosEncodeVariantI32(void **buf, int32_t value) {
   return taosEncodeVariantU32(buf, ZIGZAGE(int32_t, value));
 }
 
-static FORCE_INLINE void *taosDecodeVariantI32(void *buf, int32_t *value) {
+static FORCE_INLINE void *taosDecodeVariantI32(const void *buf, int32_t *value) {
   uint32_t tvalue = 0;
   void    *ret = taosDecodeVariantU32(buf, &tvalue);
   *value = ZIGZAGD(int32_t, tvalue);
@@ -301,7 +301,7 @@ static FORCE_INLINE int taosEncodeVariantU64(void **buf, uint64_t value) {
   return i + 1;
 }
 
-static FORCE_INLINE void *taosDecodeVariantU64(void *buf, uint64_t *value) {
+static FORCE_INLINE void *taosDecodeVariantU64(const void *buf, uint64_t *value) {
   int      i = 0;
   uint64_t tval = 0;
   *value = 0;
@@ -324,7 +324,7 @@ static FORCE_INLINE int taosEncodeVariantI64(void **buf, int64_t value) {
   return taosEncodeVariantU64(buf, ZIGZAGE(int64_t, value));
 }
 
-static FORCE_INLINE void *taosDecodeVariantI64(void *buf, int64_t *value) {
+static FORCE_INLINE void *taosDecodeVariantI64(const void *buf, int64_t *value) {
   uint64_t tvalue = 0;
   void    *ret = taosDecodeVariantU64(buf, &tvalue);
   *value = ZIGZAGD(int64_t, tvalue);
@@ -346,7 +346,7 @@ static FORCE_INLINE int taosEncodeString(void **buf, const char *value) {
   return tlen;
 }
 
-static FORCE_INLINE void *taosDecodeString(void *buf, char **value) {
+static FORCE_INLINE void *taosDecodeString(const void *buf, char **value) {
   uint64_t size = 0;
 
   buf = taosDecodeVariantU64(buf, &size);
@@ -360,7 +360,7 @@ static FORCE_INLINE void *taosDecodeString(void *buf, char **value) {
   return POINTER_SHIFT(buf, size);
 }
 
-static FORCE_INLINE void *taosDecodeStringTo(void *buf, char *value) {
+static FORCE_INLINE void *taosDecodeStringTo(const void *buf, char *value) {
   uint64_t size = 0;
 
   buf = taosDecodeVariantU64(buf, &size);
@@ -384,8 +384,7 @@ static FORCE_INLINE int taosEncodeBinary(void **buf, const void *value, int32_t
   return tlen;
 }
 
-static FORCE_INLINE void *taosDecodeBinary(void *buf, void **value, int32_t valueLen) {
+static FORCE_INLINE void *taosDecodeBinary(const void *buf, void **value, int32_t valueLen) {
-
   *value = malloc((size_t)valueLen);
   if (*value == NULL) return NULL;
   memcpy(*value, buf, (size_t)valueLen);
@@ -393,8 +392,7 @@ static FORCE_INLINE void *taosDecodeBinary(void *buf, void **value, int32_t valu
   return POINTER_SHIFT(buf, valueLen);
 }
 
-static FORCE_INLINE void *taosDecodeBinaryTo(void *buf, void *value, int32_t valueLen) {
+static FORCE_INLINE void *taosDecodeBinaryTo(const void *buf, void *value, int32_t valueLen) {
-
   memcpy(value, buf, (size_t)valueLen);
   return POINTER_SHIFT(buf, valueLen);
 }

source/CMakeLists.txt

@@ -4,4 +4,3 @@ add_subdirectory(common)
 add_subdirectory(libs)
 add_subdirectory(client)
 add_subdirectory(dnode)
-add_subdirectory(nodes)

source/client/src/tmq.c

@@ -452,6 +452,12 @@ static char* formatTimestamp(char* buf, int64_t val, int precision) {
   return buf;
 }
 
+int32_t tmqGetSkipLogNum(tmq_message_t* tmq_message) {
+  if (tmq_message == NULL) return 0;
+  SMqConsumeRsp* pRsp = (SMqConsumeRsp*)tmq_message;
+  return pRsp->skipLogNum;
+}
+
 void tmqShowMsg(tmq_message_t* tmq_message) {
   if (tmq_message == NULL) return;
 

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

@@ -248,6 +248,15 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCMCreateTopicReq
   SSdbRaw *pTopicRaw = mndTopicActionEncode(&topicObj);
   if (pTopicRaw == NULL) return -1;
   if (sdbSetRawStatus(pTopicRaw, SDB_STATUS_READY) != 0) return -1;
+  /*STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, &pMsg->rpcMsg);*/
+  /*mndTransAppendRedolog(pTrans, pTopicRaw);*/
+  /*if (mndTransPrepare(pMnode, pTrans) != 0) {*/
+  /*mError("mq-createTopic-trans:%d, failed to prepare since %s", pTrans->id, terrstr());*/
+  /*mndTransDrop(pTrans);*/
+  /*return -1;*/
+  /*}*/
+  /*mndTransDrop(pTrans);*/
+  /*return 0;*/
   return sdbWrite(pMnode->pSdb, pTopicRaw);
 }
 
@@ -288,13 +297,13 @@ static int32_t mndProcessCreateTopicMsg(SMnodeMsg *pMsg) {
   int32_t code = mndCreateTopic(pMnode, pMsg, &createTopicReq, pDb);
   mndReleaseDb(pMnode, pDb);
 
-  if (code != 0) {
+  if (code != TSDB_CODE_SUCCESS) {
     terrno = code;
     mError("topic:%s, failed to create since %s", createTopicReq.name, terrstr());
     return -1;
   }
 
-  return TSDB_CODE_SUCCESS;
+  return 0;
 }
 
 static int32_t mndDropTopic(SMnode *pMnode, SMnodeMsg *pMsg, SMqTopicObj *pTopic) { return 0; }

source/dnode/vnode/inc/tq.h

@@ -38,7 +38,7 @@ extern "C" {
 typedef struct STQ STQ;
 
 // memory allocator provided by vnode
-typedef struct STqMemRef {
+typedef struct {
   SMemAllocatorFactory* pAllocatorFactory;
   SMemAllocator*        pAllocator;
 } STqMemRef;

source/dnode/vnode/inc/vnode.h

@@ -35,12 +35,12 @@ typedef struct SDnode SDnode;
 typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
 typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
 
-typedef struct STqCfg {
+typedef struct {
   // TODO
   int32_t reserved;
 } STqCfg;
 
-typedef struct SVnodeCfg {
+typedef struct {
   int32_t  vgId;
   uint64_t dbId;
   SDnode  *pDnode;
@@ -68,9 +68,9 @@ typedef struct {
   SendReqToDnodeFp  sendReqToDnodeFp;
 } SVnodeOpt;
 
-typedef struct STqReadHandle {
+typedef struct {
   int64_t           ver;
-  uint64_t          tbUid;
+  int64_t           tbUid;
   SHashObj         *tbIdHash;
   const SSubmitMsg *pMsg;
   SSubmitBlk       *pBlock;
@@ -200,7 +200,7 @@ int32_t vnodeCompact(SVnode *pVnode);
 int32_t vnodeSync(SVnode *pVnode);
 int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad);
 
-/* ------------------------- TQ QUERY -------------------------- */
+/* ------------------------- TQ READ --------------------------- */
 
 STqReadHandle *tqInitSubmitMsgScanner(SMeta *pMeta);
 
@@ -208,12 +208,12 @@ static FORCE_INLINE void tqReadHandleSetColIdList(STqReadHandle *pReadHandle, SA
   pReadHandle->pColIdList = pColIdList;
 }
 
-// static FORCE_INLINE void tqReadHandleSetTbUid(STqReadHandle* pHandle, const SArray* pTableIdList) {
+// static FORCE_INLINE void tqReadHandleSetTbUid(STqReadHandle* pHandle, int64_t tbUid) {
-// pHandle->tbUid = pTableIdList;
+// pHandle->tbUid = tbUid;
 //}
 
 static FORCE_INLINE int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList) {
-  pHandle->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_NO_LOCK);
+  pHandle->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
   if (pHandle->tbIdHash == NULL) {
     return -1;
   }

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

@@ -20,6 +20,7 @@
 #include "tlog.h"
 #include "tq.h"
 #include "trpc.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -110,10 +111,11 @@ typedef struct {
   char    content[];
 } STqSerializedHead;
 
-typedef int (*FTqSerialize)(const void* pObj, STqSerializedHead** ppHead);
+typedef int32_t (*FTqSerialize)(const void* pObj, STqSerializedHead** ppHead);
-typedef const void* (*FTqDeserialize)(const STqSerializedHead* pHead, void** ppObj);
+typedef int32_t (*FTqDeserialize)(void* self, const STqSerializedHead* pHead, void** ppObj);
 typedef void (*FTqDelete)(void*);
-typedef struct STqMetaHandle {
+
+typedef struct {
   int64_t key;
   int64_t offset;
   int64_t serializedSize;
@@ -131,6 +133,7 @@ typedef struct STqMetaList {
 } STqMetaList;
 
 typedef struct {
+  STQ*         pTq;
   STqMetaList* bucket[TQ_BUCKET_SIZE];
   // a table head
   STqMetaList* unpersistHead;
@@ -187,21 +190,22 @@ typedef struct {
   char*           logicalPlan;
   char*           physicalPlan;
   char*           qmsg;
+  int64_t         persistedOffset;
   int64_t         committedOffset;
   int64_t         currentOffset;
   STqBuffer       buffer;
   SWalReadHandle* pReadhandle;
-} STqTopicHandle;
+} STqTopic;
 
 typedef struct {
   int64_t consumerId;
   int64_t epoch;
   char    cgroup[TSDB_TOPIC_FNAME_LEN];
-  SArray* topics;  // SArray<STqClientTopic>
+  SArray* topics;  // SArray<STqTopicHandle>
-} STqConsumerHandle;
+} STqConsumer;
 
-int         tqSerializeConsumer(const STqConsumerHandle*, STqSerializedHead**);
+int32_t tqSerializeConsumer(const STqConsumer*, STqSerializedHead**);
-const void* tqDeserializeConsumer(const STqSerializedHead* pHead, STqConsumerHandle**);
+int32_t tqDeserializeConsumer(STQ*, const STqSerializedHead*, STqConsumer**);
 
 static int FORCE_INLINE tqQueryExecuting(int32_t status) { return status; }
 

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

@@ -23,8 +23,8 @@
 extern "C" {
 #endif
 
-STqMetaStore* tqStoreOpen(const char* path, FTqSerialize pSerializer, FTqDeserialize pDeserializer, FTqDelete pDeleter,
+STqMetaStore* tqStoreOpen(STQ* pTq, const char* path, FTqSerialize pSerializer, FTqDeserialize pDeserializer,
-                          int32_t tqConfigFlag);
+                          FTqDelete pDeleter, int32_t tqConfigFlag);
 int32_t       tqStoreClose(STqMetaStore*);
 // int32_t       tqStoreDelete(TqMetaStore*);
 // int32_t       tqStoreCommitAll(TqMetaStore*);

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

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_TQ_OFFSET_H_
+#define _TD_TQ_OFFSET_H_
+
+#include "tqInt.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct STqOffsetCfg   STqOffsetCfg;
+typedef struct STqOffsetStore STqOffsetStore;
+
+STqOffsetStore* STqOffsetOpen(STqOffsetCfg*);
+void            STqOffsetClose(STqOffsetStore*);
+
+int64_t tqOffsetFetch(STqOffsetStore* pStore, const char* subscribeKey);
+int32_t tqOffsetCommit(STqOffsetStore* pStore, const char* subscribeKey, int64_t offset);
+int32_t tqOffsetPersist(STqOffsetStore* pStore, const char* subscribeKey);
+int32_t tqOffsetPersistAll(STqOffsetStore* pStore);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_TQ_OFFSET_H_*/

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

@@ -50,7 +50,8 @@ STQ* tqOpen(const char* path, SWal* pWal, SMeta* pMeta, STqCfg* tqConfig, SMemAl
     // TODO: error code of buffer pool
   }
 #endif
-  pTq->tqMeta = tqStoreOpen(path, (FTqSerialize)tqSerializeConsumer, (FTqDeserialize)tqDeserializeConsumer, free, 0);
+  pTq->tqMeta =
+      tqStoreOpen(pTq, path, (FTqSerialize)tqSerializeConsumer, (FTqDeserialize)tqDeserializeConsumer, free, 0);
   if (pTq->tqMeta == NULL) {
     free(pTq);
 #if 0
@@ -76,19 +77,89 @@ int tqPushMsg(STQ* pTq, void* p, int64_t version) {
   return 0;
 }
 
-int tqCommit(STQ* pTq) {
+int tqCommit(STQ* pTq) { return tqStorePersist(pTq->tqMeta); }
-  // do nothing
+
-  return 0;
+int32_t tqGetTopicHandleSize(const STqTopic* pTopic) {
+  return strlen(pTopic->topicName) + strlen(pTopic->sql) + strlen(pTopic->logicalPlan) + strlen(pTopic->physicalPlan) +
+         strlen(pTopic->qmsg) + sizeof(int64_t) * 3;
 }
 
-int tqSerializeConsumer(const STqConsumerHandle* pConsumer, STqSerializedHead** ppHead) {
+int32_t tqGetConsumerHandleSize(const STqConsumer* pConsumer) {
-  int32_t num = taosArrayGetSize(pConsumer->topics);
+  int     num = taosArrayGetSize(pConsumer->topics);
-  int32_t sz = sizeof(STqSerializedHead) + sizeof(int64_t) * 2 + TSDB_TOPIC_FNAME_LEN +
+  int32_t sz = 0;
-               num * (sizeof(int64_t) + TSDB_TOPIC_FNAME_LEN);
+  for (int i = 0; i < num; i++) {
+    STqTopic* pTopic = taosArrayGet(pConsumer->topics, i);
+    sz += tqGetTopicHandleSize(pTopic);
+  }
+  return sz;
+}
+
+static FORCE_INLINE int32_t tEncodeSTqTopic(void** buf, const STqTopic* pTopic) {
+  int32_t tlen = 0;
+  tlen += taosEncodeString(buf, pTopic->topicName);
+  /*tlen += taosEncodeString(buf, pTopic->sql);*/
+  /*tlen += taosEncodeString(buf, pTopic->logicalPlan);*/
+  /*tlen += taosEncodeString(buf, pTopic->physicalPlan);*/
+  tlen += taosEncodeString(buf, pTopic->qmsg);
+  tlen += taosEncodeFixedI64(buf, pTopic->persistedOffset);
+  tlen += taosEncodeFixedI64(buf, pTopic->committedOffset);
+  tlen += taosEncodeFixedI64(buf, pTopic->currentOffset);
+  return tlen;
+}
+
+static FORCE_INLINE const void* tDecodeSTqTopic(const void* buf, STqTopic* pTopic) {
+  buf = taosDecodeStringTo(buf, pTopic->topicName);
+  /*buf = taosDecodeString(buf, &pTopic->sql);*/
+  /*buf = taosDecodeString(buf, &pTopic->logicalPlan);*/
+  /*buf = taosDecodeString(buf, &pTopic->physicalPlan);*/
+  buf = taosDecodeString(buf, &pTopic->qmsg);
+  buf = taosDecodeFixedI64(buf, &pTopic->persistedOffset);
+  buf = taosDecodeFixedI64(buf, &pTopic->committedOffset);
+  buf = taosDecodeFixedI64(buf, &pTopic->currentOffset);
+  return buf;
+}
+
+static FORCE_INLINE int32_t tEncodeSTqConsumer(void** buf, const STqConsumer* pConsumer) {
+  int32_t sz;
+
+  int32_t tlen = 0;
+  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
+  tlen += taosEncodeFixedI64(buf, pConsumer->epoch);
+  tlen += taosEncodeString(buf, pConsumer->cgroup);
+  sz = taosArrayGetSize(pConsumer->topics);
+  tlen += taosEncodeFixedI32(buf, sz);
+  for (int32_t i = 0; i < sz; i++) {
+    STqTopic* pTopic = taosArrayGet(pConsumer->topics, i);
+    tlen += tEncodeSTqTopic(buf, pTopic);
+  }
+  return tlen;
+}
+
+static FORCE_INLINE const void* tDecodeSTqConsumer(const void* buf, STqConsumer* pConsumer) {
+  int32_t sz;
+
+  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
+  buf = taosDecodeFixedI64(buf, &pConsumer->epoch);
+  buf = taosDecodeStringTo(buf, pConsumer->cgroup);
+  buf = taosDecodeFixedI32(buf, &sz);
+  pConsumer->topics = taosArrayInit(sz, sizeof(STqTopic));
+  if (pConsumer->topics == NULL) return NULL;
+  for (int32_t i = 0; i < sz; i++) {
+    STqTopic pTopic;
+    buf = tDecodeSTqTopic(buf, &pTopic);
+    taosArrayPush(pConsumer->topics, &pTopic);
+  }
+  return buf;
+}
+
+int tqSerializeConsumer(const STqConsumer* pConsumer, STqSerializedHead** ppHead) {
+  int32_t sz = tEncodeSTqConsumer(NULL, pConsumer);
+
   if (sz > (*ppHead)->ssize) {
-    void* tmpPtr = realloc(*ppHead, sz);
+    void* tmpPtr = realloc(*ppHead, sizeof(STqSerializedHead) + sz);
     if (tmpPtr == NULL) {
       free(*ppHead);
+      terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
       return -1;
     }
     *ppHead = tmpPtr;
@@ -96,45 +167,155 @@ int tqSerializeConsumer(const STqConsumerHandle* pConsumer, STqSerializedHead**
   }
 
   void* ptr = (*ppHead)->content;
-  *(int64_t*)ptr = pConsumer->consumerId;
+  void* abuf = ptr;
-  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  tEncodeSTqConsumer(&abuf, pConsumer);
-  *(int64_t*)ptr = pConsumer->epoch;
+
-  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  return 0;
-  memcpy(ptr, pConsumer->topics, TSDB_TOPIC_FNAME_LEN);
+}
-  ptr = POINTER_SHIFT(ptr, TSDB_TOPIC_FNAME_LEN);
+
-  *(int32_t*)ptr = num;
+int32_t tqDeserializeConsumer(STQ* pTq, const STqSerializedHead* pHead, STqConsumer** ppConsumer) {
-  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+  const void* str = pHead->content;
-  for (int32_t i = 0; i < num; i++) {
+  *ppConsumer = calloc(1, sizeof(STqConsumer));
-    STqTopicHandle* pTopic = taosArrayGet(pConsumer->topics, i);
+  if (*ppConsumer == NULL) {
-    memcpy(ptr, pTopic->topicName, TSDB_TOPIC_FNAME_LEN);
+    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
-    ptr = POINTER_SHIFT(ptr, TSDB_TOPIC_FNAME_LEN);
+    return -1;
-    *(int64_t*)ptr = pTopic->committedOffset;
+  }
-    POINTER_SHIFT(ptr, sizeof(int64_t));
+  if (tDecodeSTqConsumer(str, *ppConsumer) == NULL) {
+    terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
+    return -1;
+  }
+  STqConsumer* pConsumer = *ppConsumer;
+  int32_t      sz = taosArrayGetSize(pConsumer->topics);
+  for (int32_t i = 0; i < sz; i++) {
+    STqTopic* pTopic = taosArrayGet(pConsumer->topics, i);
+    pTopic->pReadhandle = walOpenReadHandle(pTq->pWal);
+    if (pTopic->pReadhandle == NULL) {
+      ASSERT(false);
+    }
+    for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
+      pTopic->buffer.output[i].status = 0;
+      STqReadHandle* pReadHandle = tqInitSubmitMsgScanner(pTq->pMeta);
+      SReadHandle    handle = {.reader = pReadHandle, .meta = pTq->pMeta};
+      pTopic->buffer.output[i].pReadHandle = pReadHandle;
+      pTopic->buffer.output[i].task = qCreateStreamExecTaskInfo(pTopic->qmsg, &handle);
+    }
   }
 
   return 0;
 }
 
-const void* tqDeserializeConsumer(const STqSerializedHead* pHead, STqConsumerHandle** ppConsumer) {
+int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
-  STqConsumerHandle* pConsumer = *ppConsumer;
+  SMqConsumeReq* pReq = pMsg->pCont;
-  const void*        ptr = pHead->content;
+  int64_t        consumerId = pReq->consumerId;
-  pConsumer->consumerId = *(int64_t*)ptr;
+  int64_t        fetchOffset = pReq->offset;
-  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  /*int64_t        blockingTime = pReq->blockingTime;*/
-  pConsumer->epoch = *(int64_t*)ptr;
+
-  ptr = POINTER_SHIFT(ptr, sizeof(int64_t));
+  SMqConsumeRsp rsp = {.consumerId = consumerId, .numOfTopics = 0, .pBlockData = NULL};
-  memcpy(pConsumer->cgroup, ptr, TSDB_TOPIC_FNAME_LEN);
+
-  ptr = POINTER_SHIFT(ptr, TSDB_TOPIC_FNAME_LEN);
+  STqConsumer* pConsumer = tqHandleGet(pTq->tqMeta, consumerId);
-  int32_t sz = *(int32_t*)ptr;
+  if (pConsumer == NULL) {
-  ptr = POINTER_SHIFT(ptr, sizeof(int32_t));
+    pMsg->pCont = NULL;
-  pConsumer->topics = taosArrayInit(sz, sizeof(STqTopicHandle));
+    pMsg->contLen = 0;
-  for (int32_t i = 0; i < sz; i++) {
+    pMsg->code = -1;
-    /*STqTopicHandle* topicHandle = */
+    rpcSendResponse(pMsg);
-    /*taosArrayPush(pConsumer->topics, );*/
+    return 0;
   }
-  return NULL;
+  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;
   }
 
-int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
+  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;
@@ -145,7 +326,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
 
   /*printf("vg %d get consume req\n", pReq->head.vgId);*/
 
-  STqConsumerHandle* pConsumer = tqHandleGet(pTq->tqMeta, consumerId);
+  STqConsumer* pConsumer = tqHandleGet(pTq->tqMeta, consumerId);
   if (pConsumer == NULL) {
     pMsg->pCont = NULL;
     pMsg->contLen = 0;
@@ -156,10 +337,10 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
   int sz = taosArrayGetSize(pConsumer->topics);
 
   for (int i = 0; i < sz; i++) {
-    STqTopicHandle* pTopic = taosArrayGet(pConsumer->topics, i);
+    STqTopic* pTopic = taosArrayGet(pConsumer->topics, i);
     // TODO: support multiple topic in one req
     if (strcmp(pTopic->topicName, pReq->topic) != 0) {
-      /*ASSERT(false);*/
+      ASSERT(false);
       continue;
     }
 
@@ -195,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;
@@ -203,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;
@@ -218,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>
@@ -237,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);
 
@@ -268,28 +452,27 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
   }
   void* abuf = buf;
   tEncodeSMqConsumeRsp(&abuf, &rsp);
+
   if (rsp.pBlockData) {
     taosArrayDestroyEx(rsp.pBlockData, (void (*)(void*))tDeleteSSDataBlock);
     rsp.pBlockData = NULL;
-    /*for (int i = 0; i < taosArrayGetSize(rsp.pBlockData); i++) {*/
-    /*SSDataBlock* pBlock = taosArrayGet(rsp.pBlockData, i);*/
-    /*tDeleteSSDataBlock(pBlock);*/
-    /*}*/
-    /*taosArrayDestroy(rsp.pBlockData);*/
   }
+
   pMsg->pCont = buf;
   pMsg->contLen = tlen;
   pMsg->code = 0;
   rpcSendResponse(pMsg);
   return 0;
 }
+#endif
 
 int32_t tqProcessRebReq(STQ* pTq, char* msg) {
   SMqMVRebReq req = {0};
   tDecodeSMqMVRebReq(msg, &req);
 
-  STqConsumerHandle* pConsumer = tqHandleGet(pTq->tqMeta, req.oldConsumerId);
+  STqConsumer* pConsumer = tqHandleGet(pTq->tqMeta, req.oldConsumerId);
   ASSERT(pConsumer);
+  pConsumer->consumerId = req.newConsumerId;
   tqHandleMovePut(pTq->tqMeta, req.newConsumerId, pConsumer);
   tqHandleCommit(pTq->tqMeta, req.newConsumerId);
   tqHandlePurge(pTq->tqMeta, req.oldConsumerId);
@@ -302,19 +485,20 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   tDecodeSMqSetCVgReq(msg, &req);
 
   /*printf("vg %d set to consumer from %ld to %ld\n", req.vgId, req.oldConsumerId, req.newConsumerId);*/
-  STqConsumerHandle* pConsumer = calloc(1, sizeof(STqConsumerHandle));
+  STqConsumer* pConsumer = calloc(1, sizeof(STqConsumer));
   if (pConsumer == NULL) {
     terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
     return -1;
   }
 
   strcpy(pConsumer->cgroup, req.cgroup);
-  pConsumer->topics = taosArrayInit(0, sizeof(STqTopicHandle));
+  pConsumer->topics = taosArrayInit(0, sizeof(STqTopic));
   pConsumer->consumerId = req.consumerId;
   pConsumer->epoch = 0;
 
-  STqTopicHandle* pTopic = calloc(1, sizeof(STqTopicHandle));
+  STqTopic* pTopic = calloc(1, sizeof(STqTopic));
   if (pTopic == NULL) {
+    taosArrayDestroy(pConsumer->topics);
     free(pConsumer);
     return -1;
   }
@@ -330,6 +514,7 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   pTopic->buffer.lastOffset = -1;
   pTopic->pReadhandle = walOpenReadHandle(pTq->pWal);
   if (pTopic->pReadhandle == NULL) {
+    ASSERT(false);
   }
   for (int i = 0; i < TQ_BUFFER_SIZE; i++) {
     pTopic->buffer.output[i].status = 0;
@@ -344,143 +529,3 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   terrno = TSDB_CODE_SUCCESS;
   return 0;
 }
-
-STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
-  STqReadHandle* pReadHandle = malloc(sizeof(STqReadHandle));
-  if (pReadHandle == NULL) {
-    return NULL;
-  }
-  pReadHandle->pVnodeMeta = pMeta;
-  pReadHandle->pMsg = NULL;
-  pReadHandle->ver = -1;
-  pReadHandle->pColIdList = NULL;
-  pReadHandle->sver = -1;
-  pReadHandle->pSchema = NULL;
-  pReadHandle->pSchemaWrapper = NULL;
-  return pReadHandle;
-}
-
-void tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitMsg* pMsg, int64_t ver) {
-  pReadHandle->pMsg = pMsg;
-  pMsg->length = htonl(pMsg->length);
-  pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
-  tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter);
-  pReadHandle->ver = ver;
-  memset(&pReadHandle->blkIter, 0, sizeof(SSubmitBlkIter));
-}
-
-bool tqNextDataBlock(STqReadHandle* pHandle) {
-  while (1) {
-    if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
-      return false;
-    }
-    if (pHandle->pBlock == NULL) return false;
-
-    pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);
-    /*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
-    ASSERT(pHandle->tbIdHash);
-    void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->pBlock->uid, sizeof(int64_t));
-    if (ret != NULL) {
-      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;
-    }
-  }
-  return false;
-}
-
-int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo) {
-  /*int32_t         sversion = pHandle->pBlock->sversion;*/
-  /*SSchemaWrapper* pSchema = metaGetTableSchema(pHandle->pMeta, pHandle->pBlock->uid, sversion, false);*/
-  pBlockInfo->numOfCols = taosArrayGetSize(pHandle->pColIdList);
-  pBlockInfo->rows = pHandle->pBlock->numOfRows;
-  pBlockInfo->uid = pHandle->pBlock->uid;
-  return 0;
-}
-
-SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
-  /*int32_t         sversion = pHandle->pBlock->sversion;*/
-  // TODO set to real sversion
-  int32_t sversion = 0;
-  if (pHandle->sver != sversion) {
-    pHandle->pSchema = metaGetTbTSchema(pHandle->pVnodeMeta, pHandle->pBlock->uid, sversion);
-
-    tb_uid_t quid;
-    STbCfg*  pTbCfg = metaGetTbInfoByUid(pHandle->pVnodeMeta, pHandle->pBlock->uid);
-    if (pTbCfg->type == META_CHILD_TABLE) {
-      quid = pTbCfg->ctbCfg.suid;
-    } else {
-      quid = pHandle->pBlock->uid;
-    }
-    pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, quid, sversion, true);
-    pHandle->sver = sversion;
-  }
-
-  STSchema*       pTschema = pHandle->pSchema;
-  SSchemaWrapper* pSchemaWrapper = pHandle->pSchemaWrapper;
-
-  int32_t numOfRows = pHandle->pBlock->numOfRows;
-  int32_t numOfCols = pHandle->pSchema->numOfCols;
-  int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
-
-  // TODO: stable case
-  if (colNumNeed > pSchemaWrapper->nCols) {
-    colNumNeed = pSchemaWrapper->nCols;
-  }
-
-  SArray* pArray = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
-  if (pArray == NULL) {
-    return NULL;
-  }
-
-  int j = 0;
-  for (int32_t i = 0; i < colNumNeed; i++) {
-    int32_t colId = *(int32_t*)taosArrayGet(pHandle->pColIdList, i);
-    while (j < pSchemaWrapper->nCols && pSchemaWrapper->pSchema[j].colId < colId) {
-      j++;
-    }
-    SSchema*        pColSchema = &pSchemaWrapper->pSchema[j];
-    SColumnInfoData colInfo = {0};
-    int             sz = numOfRows * pColSchema->bytes;
-    colInfo.info.bytes = pColSchema->bytes;
-    colInfo.info.colId = colId;
-    colInfo.info.type = pColSchema->type;
-
-    colInfo.pData = calloc(1, sz);
-    if (colInfo.pData == NULL) {
-      // TODO free
-      taosArrayDestroy(pArray);
-      return NULL;
-    }
-    taosArrayPush(pArray, &colInfo);
-  }
-
-  STSRowIter iter = {0};
-  tdSTSRowIterInit(&iter, pTschema);
-  STSRow* row;
-  // int32_t kvIdx = 0;
-  int32_t curRow = 0;
-  tInitSubmitBlkIter(pHandle->pBlock, &pHandle->blkIter);
-  while ((row = tGetSubmitBlkNext(&pHandle->blkIter)) != NULL) {
-    tdSTSRowIterReset(&iter, row);
-    // get all wanted col of that block
-    for (int32_t i = 0; i < colNumNeed; i++) {
-      SColumnInfoData* pColData = taosArrayGet(pArray, i);
-      STColumn*        pCol = schemaColAt(pTschema, i);
-      // TODO
-      ASSERT(pCol->colId == pColData->info.colId);
-      // void* val = tdGetMemRowDataOfColEx(row, pCol->colId, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset, &kvIdx);
-      SCellVal sVal = {0};
-      if (!tdSTSRowIterNext(&iter, pCol->colId, pCol->type, &sVal)) {
-        // TODO: reach end
-        break;
-      }
-      memcpy(POINTER_SHIFT(pColData->pData, curRow * pCol->bytes), sVal.val, pCol->bytes);
-    }
-    curRow++;
-  }
-  return pArray;
-}

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

@@ -68,20 +68,21 @@ static inline int tqReadLastPage(int fd, STqIdxPageBuf* pBuf) {
   return lseek(fd, offset, SEEK_SET);
 }
 
-STqMetaStore* tqStoreOpen(const char* path, FTqSerialize serializer, FTqDeserialize deserializer, FTqDelete deleter,
+STqMetaStore* tqStoreOpen(STQ* pTq, const char* path, FTqSerialize serializer, FTqDeserialize deserializer,
-                          int32_t tqConfigFlag) {
+                          FTqDelete deleter, int32_t tqConfigFlag) {
-  STqMetaStore* pMeta = malloc(sizeof(STqMetaStore));
+  STqMetaStore* pMeta = calloc(1, sizeof(STqMetaStore));
   if (pMeta == NULL) {
     terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
     return NULL;
   }
-  memset(pMeta, 0, sizeof(STqMetaStore));
+  pMeta->pTq = pTq;
 
   // concat data file name and index file name
   size_t pathLen = strlen(path);
   pMeta->dirPath = malloc(pathLen + 1);
   if (pMeta->dirPath == NULL) {
     terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
+    free(pMeta);
     return NULL;
   }
   strcpy(pMeta->dirPath, path);
@@ -103,12 +104,11 @@ STqMetaStore* tqStoreOpen(const char* path, FTqSerialize serializer, FTqDeserial
   }
 
   pMeta->idxFd = idxFd;
-  pMeta->unpersistHead = malloc(sizeof(STqMetaList));
+  pMeta->unpersistHead = calloc(1, sizeof(STqMetaList));
   if (pMeta->unpersistHead == NULL) {
     terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
     return NULL;
   }
-  memset(pMeta->unpersistHead, 0, sizeof(STqMetaList));
   pMeta->unpersistHead->unpersistNext = pMeta->unpersistHead->unpersistPrev = pMeta->unpersistHead;
 
   strcpy(name, path);
@@ -145,12 +145,11 @@ STqMetaStore* tqStoreOpen(const char* path, FTqSerialize serializer, FTqDeserial
     ASSERT(idxBuf.head.writeOffset == idxRead);
     // loop read every entry
     for (int i = 0; i < idxBuf.head.writeOffset - TQ_IDX_PAGE_HEAD_SIZE; i += TQ_IDX_SIZE) {
-      STqMetaList* pNode = malloc(sizeof(STqMetaList));
+      STqMetaList* pNode = calloc(1, sizeof(STqMetaList));
       if (pNode == NULL) {
         terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
         // TODO: free memory
       }
-      memset(pNode, 0, sizeof(STqMetaList));
       memcpy(&pNode->handle, &idxBuf.buffer[i], TQ_IDX_SIZE);
 
       lseek(fileFd, pNode->handle.offset, SEEK_SET);
@@ -169,25 +168,25 @@ STqMetaStore* tqStoreOpen(const char* path, FTqSerialize serializer, FTqDeserial
       }
       if (serializedObj->action == TQ_ACTION_INUSE) {
         if (serializedObj->ssize != sizeof(STqSerializedHead)) {
-          pMeta->pDeserializer(serializedObj, &pNode->handle.valueInUse);
+          pMeta->pDeserializer(pTq, serializedObj, &pNode->handle.valueInUse);
         } else {
           pNode->handle.valueInUse = TQ_DELETE_TOKEN;
         }
       } else if (serializedObj->action == TQ_ACTION_INTXN) {
         if (serializedObj->ssize != sizeof(STqSerializedHead)) {
-          pMeta->pDeserializer(serializedObj, &pNode->handle.valueInTxn);
+          pMeta->pDeserializer(pTq, serializedObj, &pNode->handle.valueInTxn);
         } else {
           pNode->handle.valueInTxn = TQ_DELETE_TOKEN;
         }
       } else if (serializedObj->action == TQ_ACTION_INUSE_CONT) {
         if (serializedObj->ssize != sizeof(STqSerializedHead)) {
-          pMeta->pDeserializer(serializedObj, &pNode->handle.valueInUse);
+          pMeta->pDeserializer(pTq, serializedObj, &pNode->handle.valueInUse);
         } else {
           pNode->handle.valueInUse = TQ_DELETE_TOKEN;
         }
         STqSerializedHead* ptr = POINTER_SHIFT(serializedObj, serializedObj->ssize);
         if (ptr->ssize != sizeof(STqSerializedHead)) {
-          pMeta->pDeserializer(ptr, &pNode->handle.valueInTxn);
+          pMeta->pDeserializer(pTq, ptr, &pNode->handle.valueInTxn);
         } else {
           pNode->handle.valueInTxn = TQ_DELETE_TOKEN;
         }
@@ -302,7 +301,7 @@ int32_t tqStorePersist(STqMetaStore* pMeta) {
   pSHead->ver = TQ_SVER;
   pSHead->checksum = 0;
   pSHead->ssize = sizeof(STqSerializedHead);
-  int allocatedSize = sizeof(STqSerializedHead);
+  /*int allocatedSize = sizeof(STqSerializedHead);*/
   int offset = lseek(pMeta->fileFd, 0, SEEK_CUR);
 
   tqReadLastPage(pMeta->idxFd, &idxBuf);
@@ -417,14 +416,14 @@ static int32_t tqHandlePutCommitted(STqMetaStore* pMeta, int64_t key, void* valu
       pNode = pNode->next;
     }
   }
-  STqMetaList* pNewNode = malloc(sizeof(STqMetaList));
+  STqMetaList* pNewNode = calloc(1, sizeof(STqMetaList));
   if (pNewNode == NULL) {
     terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
     return -1;
   }
-  memset(pNewNode, 0, sizeof(STqMetaList));
   pNewNode->handle.key = key;
   pNewNode->handle.valueInUse = value;
+  pNewNode->next = pMeta->bucket[bucketKey];
   // put into unpersist list
   pNewNode->unpersistPrev = pMeta->unpersistHead;
   pNewNode->unpersistNext = pMeta->unpersistHead->unpersistNext;
@@ -489,12 +488,11 @@ static inline int32_t tqHandlePutImpl(STqMetaStore* pMeta, int64_t key, void* va
       pNode = pNode->next;
     }
   }
-  STqMetaList* pNewNode = malloc(sizeof(STqMetaList));
+  STqMetaList* pNewNode = calloc(1, sizeof(STqMetaList));
   if (pNewNode == NULL) {
     terrno = TSDB_CODE_TQ_OUT_OF_MEMORY;
     return -1;
   }
-  memset(pNewNode, 0, sizeof(STqMetaList));
   pNewNode->handle.key = key;
   pNewNode->handle.valueInTxn = value;
   pNewNode->next = pMeta->bucket[bucketKey];

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+#define _DEFAULT_SOURCE
+
+#include "tqOffset.h"
+
+enum ETqOffsetPersist {
+  TQ_OFFSET_PERSIST__LAZY = 1,
+  TQ_OFFSET_PERSIST__EAGER,
+};
+
+struct STqOffsetCfg {
+  int8_t persistPolicy;
+};
+
+struct STqOffsetStore {
+  STqOffsetCfg cfg;
+  SHashObj*    pHash;  // SHashObj<subscribeKey, offset>
+};
+
+STqOffsetStore* STqOffsetOpen(STqOffsetCfg* pCfg) {
+  STqOffsetStore* pStore = malloc(sizeof(STqOffsetStore));
+  if (pStore == NULL) {
+    return NULL;
+  }
+  memcpy(&pStore->cfg, pCfg, sizeof(STqOffsetCfg));
+  pStore->pHash = taosHashInit(64, MurmurHash3_32, true, HASH_NO_LOCK);
+  return pStore;
+}
+

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

@@ -0,0 +1,160 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+#define _DEFAULT_SOURCE
+
+#include "vnode.h"
+
+STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
+  STqReadHandle* pReadHandle = malloc(sizeof(STqReadHandle));
+  if (pReadHandle == NULL) {
+    return NULL;
+  }
+  pReadHandle->pVnodeMeta = pMeta;
+  pReadHandle->pMsg = NULL;
+  pReadHandle->ver = -1;
+  pReadHandle->pColIdList = NULL;
+  pReadHandle->sver = -1;
+  pReadHandle->pSchema = NULL;
+  pReadHandle->pSchemaWrapper = NULL;
+  return pReadHandle;
+}
+
+void tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitMsg* pMsg, int64_t ver) {
+  pReadHandle->pMsg = pMsg;
+  pMsg->length = htonl(pMsg->length);
+  pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
+  tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter);
+  pReadHandle->ver = ver;
+  memset(&pReadHandle->blkIter, 0, sizeof(SSubmitBlkIter));
+}
+
+bool tqNextDataBlock(STqReadHandle* pHandle) {
+  while (1) {
+    if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
+      return false;
+    }
+    if (pHandle->pBlock == NULL) return false;
+
+    pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);
+    /*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
+    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;
+}
+
+int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo) {
+  /*int32_t         sversion = pHandle->pBlock->sversion;*/
+  /*SSchemaWrapper* pSchema = metaGetTableSchema(pHandle->pMeta, pHandle->pBlock->uid, sversion, false);*/
+  pBlockInfo->numOfCols = taosArrayGetSize(pHandle->pColIdList);
+  pBlockInfo->rows = pHandle->pBlock->numOfRows;
+  pBlockInfo->uid = pHandle->pBlock->uid;
+  return 0;
+}
+
+SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
+  /*int32_t         sversion = pHandle->pBlock->sversion;*/
+  // TODO set to real sversion
+  int32_t sversion = 0;
+  if (pHandle->sver != sversion) {
+    pHandle->pSchema = metaGetTbTSchema(pHandle->pVnodeMeta, pHandle->pBlock->uid, sversion);
+
+    tb_uid_t quid;
+    STbCfg*  pTbCfg = metaGetTbInfoByUid(pHandle->pVnodeMeta, pHandle->pBlock->uid);
+    if (pTbCfg->type == META_CHILD_TABLE) {
+      quid = pTbCfg->ctbCfg.suid;
+    } else {
+      quid = pHandle->pBlock->uid;
+    }
+    pHandle->pSchemaWrapper = metaGetTableSchema(pHandle->pVnodeMeta, quid, sversion, true);
+    pHandle->sver = sversion;
+  }
+
+  STSchema*       pTschema = pHandle->pSchema;
+  SSchemaWrapper* pSchemaWrapper = pHandle->pSchemaWrapper;
+
+  int32_t numOfRows = pHandle->pBlock->numOfRows;
+  int32_t numOfCols = pHandle->pSchema->numOfCols;
+  int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
+
+  // TODO: stable case
+  if (colNumNeed > pSchemaWrapper->nCols) {
+    colNumNeed = pSchemaWrapper->nCols;
+  }
+
+  SArray* pArray = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
+  if (pArray == NULL) {
+    return NULL;
+  }
+
+  int j = 0;
+  for (int32_t i = 0; i < colNumNeed; i++) {
+    int32_t colId = *(int32_t*)taosArrayGet(pHandle->pColIdList, i);
+    while (j < pSchemaWrapper->nCols && pSchemaWrapper->pSchema[j].colId < colId) {
+      j++;
+    }
+    SSchema*        pColSchema = &pSchemaWrapper->pSchema[j];
+    SColumnInfoData colInfo = {0};
+    int             sz = numOfRows * pColSchema->bytes;
+    colInfo.info.bytes = pColSchema->bytes;
+    colInfo.info.colId = colId;
+    colInfo.info.type = pColSchema->type;
+
+    colInfo.pData = calloc(1, sz);
+    if (colInfo.pData == NULL) {
+      // TODO free
+      taosArrayDestroy(pArray);
+      return NULL;
+    }
+    taosArrayPush(pArray, &colInfo);
+  }
+
+  STSRowIter iter = {0};
+  tdSTSRowIterInit(&iter, pTschema);
+  STSRow* row;
+  // int32_t kvIdx = 0;
+  int32_t curRow = 0;
+  tInitSubmitBlkIter(pHandle->pBlock, &pHandle->blkIter);
+  while ((row = tGetSubmitBlkNext(&pHandle->blkIter)) != NULL) {
+    tdSTSRowIterReset(&iter, row);
+    // get all wanted col of that block
+    for (int32_t i = 0; i < colNumNeed; i++) {
+      SColumnInfoData* pColData = taosArrayGet(pArray, i);
+      STColumn*        pCol = schemaColAt(pTschema, i);
+      // TODO
+      ASSERT(pCol->colId == pColData->info.colId);
+      // void* val = tdGetMemRowDataOfColEx(row, pCol->colId, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset, &kvIdx);
+      SCellVal sVal = {0};
+      if (!tdSTSRowIterNext(&iter, pCol->colId, pCol->type, &sVal)) {
+        // TODO: reach end
+        break;
+      }
+      memcpy(POINTER_SHIFT(pColData->pData, curRow * pCol->bytes), sVal.val, pCol->bytes);
+    }
+    curRow++;
+  }
+  return pArray;
+}

source/dnode/vnode/src/tsdb/tsdbCommit.c

@@ -135,6 +135,7 @@ int tsdbPrepareCommit(STsdb *pTsdb) {
 
   pTsdb->imem = pTsdb->mem;
   pTsdb->mem = NULL;
+  return 0;
 }
 
 int tsdbCommit(STsdb *pRepo) {

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

@@ -16,19 +16,6 @@
 #include "tq.h"
 #include "vnd.h"
 
-#if 0
-int vnodeProcessNoWalWMsgs(SVnode *pVnode, SRpcMsg *pMsg) {
-  switch (pMsg->msgType) {
-    case TDMT_VND_MQ_SET_CUR:
-      if (tqSetCursor(pVnode->pTq, pMsg->pCont) < 0) {
-        // TODO: handle error
-      }
-      break;
-  }
-  return 0;
-}
-#endif
-
 int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
   SRpcMsg *pMsg;
 
@@ -38,9 +25,10 @@ int vnodeProcessWMsgs(SVnode *pVnode, SArray *pMsgs) {
     // ser request version
     void   *pBuf = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
     int64_t ver = pVnode->state.processed++;
-    taosEncodeFixedU64(&pBuf, ver);
+    taosEncodeFixedI64(&pBuf, ver);
 
     if (walWrite(pVnode->pWal, ver, pMsg->msgType, pMsg->pCont, pMsg->contLen) < 0) {
+      /*ASSERT(false);*/
       // TODO: handle error
     }
   }
@@ -64,8 +52,8 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
   memcpy(ptr, pMsg->pCont, pMsg->contLen);
 
   // todo: change the interface here
-  uint64_t ver;
+  int64_t ver;
-  taosDecodeFixedU64(POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), &ver);
+  taosDecodeFixedI64(POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), &ver);
   if (tqPushMsg(pVnode->pTq, ptr, ver) < 0) {
     // TODO: handle error
   }
@@ -132,7 +120,6 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
     } break;
     case TDMT_VND_MQ_REB: {
       if (tqProcessRebReq(pVnode->pTq, POINTER_SHIFT(ptr, sizeof(SMsgHead))) < 0) {
-
       }
     } break;
     default:

source/libs/CMakeLists.txt

@@ -13,3 +13,4 @@ add_subdirectory(function)
 add_subdirectory(qcom)
 add_subdirectory(qworker)
 add_subdirectory(tfs)
+add_subdirectory(nodes)

source/libs/nodes/CMakeLists.txt

@@ -2,7 +2,7 @@ aux_source_directory(src NODES_SRC)
 add_library(nodes STATIC ${NODES_SRC})
 target_include_directories(
     nodes
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/nodes"
+    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/nodes"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 target_link_libraries(