Merge pull request #22072 from taosdata/mark/tmq

fix:modify commit version to next validate version

include/client/taos.h

@@ -287,11 +287,20 @@ DLL_EXPORT TAOS_RES *tmq_consumer_poll(tmq_t *tmq, int64_t timeout);
 DLL_EXPORT int32_t   tmq_consumer_close(tmq_t *tmq);
 DLL_EXPORT int32_t   tmq_commit_sync(tmq_t *tmq, const TAOS_RES *msg);
 DLL_EXPORT void      tmq_commit_async(tmq_t *tmq, const TAOS_RES *msg, tmq_commit_cb *cb, void *param);
+DLL_EXPORT int32_t   tmq_commit_offset_sync(tmq_t *tmq, const char *pTopicName, int32_t vgId, int64_t offset);
+DLL_EXPORT int32_t   tmq_commit_offset_async(tmq_t *tmq, const char *pTopicName, int32_t vgId, int64_t offset, tmq_commit_cb *cb, void *param);
 DLL_EXPORT int32_t   tmq_get_topic_assignment(tmq_t *tmq, const char *pTopicName, tmq_topic_assignment **assignment,
                                               int32_t *numOfAssignment);
 DLL_EXPORT void      tmq_free_assignment(tmq_topic_assignment* pAssignment);
 DLL_EXPORT int32_t   tmq_offset_seek(tmq_t *tmq, const char *pTopicName, int32_t vgId, int64_t offset);
 
+DLL_EXPORT const char *tmq_get_topic_name(TAOS_RES *res);
+DLL_EXPORT const char *tmq_get_db_name(TAOS_RES *res);
+DLL_EXPORT int32_t     tmq_get_vgroup_id(TAOS_RES *res);
+DLL_EXPORT int64_t     tmq_get_vgroup_offset(TAOS_RES* res);
+DLL_EXPORT int64_t     tmq_position(tmq_t *tmq, const char *pTopicName, int32_t vgId);
+DLL_EXPORT int64_t     tmq_committed(tmq_t *tmq, const char *pTopicName, int32_t vgId);
+
 /* ----------------------TMQ CONFIGURATION INTERFACE---------------------- */
 
 enum tmq_conf_res_t {
@@ -309,11 +318,6 @@ DLL_EXPORT void           tmq_conf_set_auto_commit_cb(tmq_conf_t *conf, tmq_comm
 
 /* -------------------------TMQ MSG HANDLE INTERFACE---------------------- */
 
-DLL_EXPORT const char *tmq_get_topic_name(TAOS_RES *res);
-DLL_EXPORT const char *tmq_get_db_name(TAOS_RES *res);
-DLL_EXPORT int32_t     tmq_get_vgroup_id(TAOS_RES *res);
-DLL_EXPORT int64_t     tmq_get_vgroup_offset(TAOS_RES* res);
-
 /* ------------------------------ TAOSX -----------------------------------*/
 // note: following apis are unstable
 enum tmq_res_t {

include/common/tmsgdef.h

@@ -312,6 +312,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_CONSUME, "vnode-tmq-consume", SMqPollReq, SMqDataBlkRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_CONSUME_PUSH, "vnode-tmq-consume-push", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_VG_WALINFO, "vnode-tmq-vg-walinfo", SMqPollReq, SMqDataBlkRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_VG_COMMITTEDINFO, "vnode-tmq-committed-walinfo", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_TMQ_MAX_MSG, "vnd-tmq-max", NULL, NULL)
 
 

include/libs/executor/storageapi.h

@@ -228,7 +228,7 @@ typedef struct SStoreTqReader {
 } SStoreTqReader;
 
 typedef struct SStoreSnapshotFn {
-  int32_t (*createSnapshot)(SSnapContext* ctx, int64_t uid);
+  int32_t (*setForSnapShot)(SSnapContext* ctx, int64_t uid);
   int32_t (*destroySnapshot)(SSnapContext* ctx);
   SMetaTableInfo (*getMetaTableInfoFromSnapshot)(SSnapContext* ctx);
   int32_t (*getTableInfoFromSnapshot)(SSnapContext* ctx, void** pBuf, int32_t* contLen, int16_t* type, int64_t* uid);

include/util/taoserror.h

@@ -778,6 +778,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TMQ_VERSION_OUT_OF_RANGE       TAOS_DEF_ERROR_CODE(0, 0x4007)
 #define TSDB_CODE_TMQ_INVALID_VGID               TAOS_DEF_ERROR_CODE(0, 0x4008)
 #define TSDB_CODE_TMQ_INVALID_TOPIC              TAOS_DEF_ERROR_CODE(0, 0x4009)
+#define TSDB_CODE_TMQ_NEED_INITIALIZED           TAOS_DEF_ERROR_CODE(0, 0x4010)
+#define TSDB_CODE_TMQ_NO_COMMITTED               TAOS_DEF_ERROR_CODE(0, 0x4011)
 
 // stream
 #define TSDB_CODE_STREAM_TASK_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x4100)

source/client/src/clientRawBlockWrite.c

@@ -1327,6 +1327,9 @@ end:
 
 int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD* fields,
                                      int numFields) {
+  if (!taos || !pData || !tbname) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t     code = TSDB_CODE_SUCCESS;
   STableMeta* pTableMeta = NULL;
   SQuery*     pQuery = NULL;
@@ -1413,6 +1416,9 @@ end:
 }
 
 int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) {
+  if (!taos || !pData || !tbname) {
+    return TSDB_CODE_INVALID_PARA;
+  }
   int32_t     code = TSDB_CODE_SUCCESS;
   STableMeta* pTableMeta = NULL;
   SQuery*     pQuery = NULL;
@@ -1812,6 +1818,7 @@ end:
 }
 
 char* tmq_get_json_meta(TAOS_RES* res) {
+  if (res == NULL) return NULL;
   uDebug("tmq_get_json_meta called");
   if (!TD_RES_TMQ_META(res) && !TD_RES_TMQ_METADATA(res)) {
     return NULL;

source/client/test/clientTests.cpp

@@ -1075,6 +1075,98 @@ TEST(clientCase, sub_db_test) {
   fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
 }
 
+TEST(clientCase, tmq_commit) {
+//  taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
+
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  tmq_conf_t* conf = tmq_conf_new();
+
+  tmq_conf_set(conf, "enable.auto.commit", "false");
+  tmq_conf_set(conf, "auto.commit.interval.ms", "2000");
+  tmq_conf_set(conf, "group.id", "group_id_2");
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "auto.offset.reset", "earliest");
+  tmq_conf_set(conf, "msg.with.table.name", "true");
+
+  tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  tmq_conf_destroy(conf);
+
+  char topicName[128] = "tp";
+  // 创建订阅 topics 列表
+  tmq_list_t* topicList = tmq_list_new();
+  tmq_list_append(topicList, topicName);
+
+  // 启动订阅
+  tmq_subscribe(tmq, topicList);
+  tmq_list_destroy(topicList);
+
+  int32_t     totalRows = 0;
+  int32_t     msgCnt = 0;
+  int32_t     timeout = 2000;
+
+  tmq_topic_assignment* pAssign = NULL;
+  int32_t numOfAssign = 0;
+
+  int32_t code = tmq_get_topic_assignment(tmq, topicName, &pAssign, &numOfAssign);
+  if (code != 0) {
+    printf("error occurs:%s\n", tmq_err2str(code));
+    tmq_free_assignment(pAssign);
+    tmq_consumer_close(tmq);
+    taos_close(pConn);
+    fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+    return;
+  }
+
+  for(int i = 0; i < numOfAssign; i++){
+    printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssign[i].vgId, pAssign[i].currentOffset, pAssign[i].begin, pAssign[i].end);
+
+    int64_t committed = tmq_committed(tmq, topicName, pAssign[i].vgId);
+    printf("committed vgId:%d, committed:%lld\n", pAssign[i].vgId, committed);
+
+    int64_t position = tmq_position(tmq, topicName, pAssign[i].vgId);
+    printf("position vgId:%d, position:%lld\n", pAssign[i].vgId, position);
+    tmq_offset_seek(tmq, topicName, pAssign[i].vgId, 1);
+    position = tmq_position(tmq, topicName, pAssign[i].vgId);
+    printf("after seek 1, position vgId:%d, position:%lld\n", pAssign[i].vgId, position);
+  }
+
+  while (1) {
+    printf("start to poll\n");
+    TAOS_RES* pRes = tmq_consumer_poll(tmq, timeout);
+    if (pRes) {
+      printSubResults(pRes, &totalRows);
+    } else {
+      break;
+    }
+
+    tmq_commit_sync(tmq, pRes);
+    for(int i = 0; i < numOfAssign; i++) {
+      int64_t committed = tmq_committed(tmq, topicName, pAssign[i].vgId);
+      printf("committed vgId:%d, committed:%lld\n", pAssign[i].vgId, committed);
+      if(committed > 0){
+        int32_t code = tmq_commit_offset_sync(tmq, topicName, pAssign[i].vgId, 4);
+        printf("tmq_commit_offset_sync vgId:%d, offset:4, code:%d\n", pAssign[i].vgId, code);
+        int64_t committed = tmq_committed(tmq, topicName, pAssign[i].vgId);
+        printf("after tmq_commit_offset_sync, committed vgId:%d, committed:%lld\n", pAssign[i].vgId, committed);
+      }
+    }
+    if (pRes != NULL) {
+      taos_free_result(pRes);
+    }
+
+//    tmq_offset_seek(tmq, "tp", pAssign[0].vgId, pAssign[0].begin);
+  }
+
+  tmq_free_assignment(pAssign);
+
+  tmq_consumer_close(tmq);
+  taos_close(pConn);
+  fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+}
+
 TEST(clientCase, td_25129) {
 //  taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
 
@@ -1094,9 +1186,10 @@ TEST(clientCase, td_25129) {
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
   tmq_conf_destroy(conf);
 
+  char topicName[128] = "tp";
   // 创建订阅 topics 列表
   tmq_list_t* topicList = tmq_list_new();
-  tmq_list_append(topicList, "tp");
+  tmq_list_append(topicList, topicName);
 
   // 启动订阅
   tmq_subscribe(tmq, topicList);
@@ -1114,7 +1207,7 @@ TEST(clientCase, td_25129) {
   tmq_topic_assignment* pAssign = NULL;
   int32_t numOfAssign = 0;
 
-  int32_t code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+  int32_t code = tmq_get_topic_assignment(tmq, topicName, &pAssign, &numOfAssign);
   if (code != 0) {
     printf("error occurs:%s\n", tmq_err2str(code));
     tmq_free_assignment(pAssign);
@@ -1131,7 +1224,7 @@ TEST(clientCase, td_25129) {
 //  tmq_offset_seek(tmq, "tp", pAssign[0].vgId, 4);
   tmq_free_assignment(pAssign);
 
-  code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+  code = tmq_get_topic_assignment(tmq, topicName, &pAssign, &numOfAssign);
   if (code != 0) {
     printf("error occurs:%s\n", tmq_err2str(code));
     tmq_free_assignment(pAssign);
@@ -1147,7 +1240,7 @@ TEST(clientCase, td_25129) {
 
   tmq_free_assignment(pAssign);
 
-  code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+  code = tmq_get_topic_assignment(tmq, topicName, &pAssign, &numOfAssign);
   if (code != 0) {
     printf("error occurs:%s\n", tmq_err2str(code));
     tmq_free_assignment(pAssign);
@@ -1177,7 +1270,7 @@ TEST(clientCase, td_25129) {
 
       printSubResults(pRes, &totalRows);
 
-      code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+      code = tmq_get_topic_assignment(tmq, topicName, &pAssign, &numOfAssign);
       if (code != 0) {
         printf("error occurs:%s\n", tmq_err2str(code));
         tmq_free_assignment(pAssign);
@@ -1191,10 +1284,11 @@ TEST(clientCase, td_25129) {
         printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssign[i].vgId, pAssign[i].currentOffset, pAssign[i].begin, pAssign[i].end);
       }
     } else {
-      tmq_offset_seek(tmq, "tp", pAssign[0].vgId, pAssign[0].currentOffset);
-      tmq_offset_seek(tmq, "tp", pAssign[1].vgId, pAssign[1].currentOffset);
+      for(int i = 0; i < numOfAssign; i++) {
+        tmq_offset_seek(tmq, topicName, pAssign[i].vgId, pAssign[i].currentOffset);
+      }
       tmq_commit_sync(tmq, pRes);
-      continue;
+      break;
     }
 
 //    tmq_commit_sync(tmq, pRes);
@@ -1226,6 +1320,7 @@ TEST(clientCase, td_25129) {
     printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssign[i].vgId, pAssign[i].currentOffset, pAssign[i].begin, pAssign[i].end);
   }
 
+  tmq_free_assignment(pAssign);
   tmq_consumer_close(tmq);
   taos_close(pConn);
   fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -732,6 +732,7 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_CONSUME, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_CONSUME_PUSH, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_VG_WALINFO, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_VG_COMMITTEDINFO, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_DEL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

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

@@ -232,6 +232,7 @@ int32_t tqProcessSeekReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessPollPush(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessVgWalInfoReq(STQ* pTq, SRpcMsg* pMsg);
+int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg);
 
 // tq-stream
 int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);

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

@@ -85,9 +85,9 @@ void tqDestroyTqHandle(void* data) {
   }
 }
 
-static bool tqOffsetLessOrEqual(const STqOffset* pLeft, const STqOffset* pRight) {
+static bool tqOffsetEqual(const STqOffset* pLeft, const STqOffset* pRight) {
   return pLeft->val.type == TMQ_OFFSET__LOG && pRight->val.type == TMQ_OFFSET__LOG &&
-         pLeft->val.version <= pRight->val.version;
+         pLeft->val.version == pRight->val.version;
 }
 
 STQ* tqOpen(const char* path, SVnode* pVnode) {
@@ -302,10 +302,10 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t sversion, char* msg, int32_t
   STqOffset* pOffset = &vgOffset.offset;
 
   if (pOffset->val.type == TMQ_OFFSET__SNAPSHOT_DATA || pOffset->val.type == TMQ_OFFSET__SNAPSHOT_META) {
-    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:snapshot) uid:%" PRId64 ", ts:%" PRId64,
+    tqInfo("receive offset commit msg to %s on vgId:%d, offset(type:snapshot) uid:%" PRId64 ", ts:%" PRId64,
             pOffset->subKey, vgId, pOffset->val.uid, pOffset->val.ts);
   } else if (pOffset->val.type == TMQ_OFFSET__LOG) {
-    tqDebug("receive offset commit msg to %s on vgId:%d, offset(type:log) version:%" PRId64, pOffset->subKey, vgId,
+    tqInfo("receive offset commit msg to %s on vgId:%d, offset(type:log) version:%" PRId64, pOffset->subKey, vgId,
             pOffset->val.version);
     if (pOffset->val.version + 1 == sversion) {
       pOffset->val.version += 1;
@@ -316,8 +316,8 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t sversion, char* msg, int32_t
   }
 
   STqOffset* pSavedOffset = tqOffsetRead(pTq->pOffsetStore, pOffset->subKey);
-  if (pSavedOffset != NULL && tqOffsetLessOrEqual(pOffset, pSavedOffset)) {
-    tqDebug("not update the offset, vgId:%d sub:%s since committed:%" PRId64 " less than/equal to existed:%" PRId64,
+  if (pSavedOffset != NULL && tqOffsetEqual(pOffset, pSavedOffset)) {
+    tqInfo("not update the offset, vgId:%d sub:%s since committed:%" PRId64 " less than/equal to existed:%" PRId64,
             vgId, pOffset->subKey, pOffset->val.version, pSavedOffset->val.version);
     return 0;  // no need to update the offset value
   }
@@ -578,6 +578,49 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   return code;
 }
 
+int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg) {
+  void* data = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
+  int32_t len = pMsg->contLen - sizeof(SMsgHead);
+
+  SMqVgOffset vgOffset = {0};
+
+  SDecoder decoder;
+  tDecoderInit(&decoder, (uint8_t*)data, len);
+  if (tDecodeMqVgOffset(&decoder, &vgOffset) < 0) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  tDecoderClear(&decoder);
+
+  STqOffset* pOffset = &vgOffset.offset;
+  STqOffset* pSavedOffset = tqOffsetRead(pTq->pOffsetStore, pOffset->subKey);
+  if (pSavedOffset == NULL) {
+    return TSDB_CODE_TMQ_NO_COMMITTED;
+  }
+  vgOffset.offset = *pSavedOffset;
+
+  int32_t code = 0;
+  tEncodeSize(tEncodeMqVgOffset, &vgOffset, len, code);
+  if (code < 0) {
+    return TSDB_CODE_INVALID_PARA;
+  }
+
+  void* buf = rpcMallocCont(len);
+  if (buf == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  SEncoder encoder;
+  tEncoderInit(&encoder, buf, len);
+  tEncodeMqVgOffset(&encoder, &vgOffset);
+  tEncoderClear(&encoder);
+
+  SRpcMsg rsp = {.info = pMsg->info, .pCont = buf, .contLen = len, .code = 0};
+
+  tmsgSendRsp(&rsp);
+
+  return 0;
+}
+
 int32_t tqProcessVgWalInfoReq(STQ* pTq, SRpcMsg* pMsg) {
   SMqPollReq req = {0};
   if (tDeserializeSMqPollReq(pMsg->pCont, pMsg->contLen, &req) < 0) {

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

@@ -196,7 +196,7 @@ int32_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHea
       tqDebug("tmq poll: consumer:0x%" PRIx64 ", (epoch %d) vgId:%d offset %" PRId64
               ", no more log to return, reqId:0x%" PRIx64,
               pHandle->consumerId, pHandle->epoch, vgId, offset, reqId);
-      *fetchOffset = offset - 1;
+      *fetchOffset = offset;
       code = -1;
       goto END;
     }

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

@@ -119,7 +119,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
         }
       } else {
         walRefFirstVer(pTq->pVnode->pWal, pHandle->pRef);
-        tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer - 1);
+        tqOffsetResetToLog(pOffsetVal, pHandle->pRef->refVer);
       }
     } else if (reqOffset.type == TMQ_OFFSET__RESET_LATEST) {
       walRefLastVer(pTq->pVnode->pWal, pHandle->pRef);
@@ -127,7 +127,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
         SMqDataRsp dataRsp = {0};
         tqInitDataRsp(&dataRsp, pRequest);
 
-        tqOffsetResetToLog(&dataRsp.rspOffset, pHandle->pRef->refVer);
+        tqOffsetResetToLog(&dataRsp.rspOffset, pHandle->pRef->refVer + 1);
         tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, (latest) offset reset to %" PRId64, consumerId,
                 pHandle->subKey, vgId, dataRsp.rspOffset.version);
         int32_t code = tqSendDataRsp(pHandle, pMsg, pRequest, &dataRsp, TMQ_MSG_TYPE__POLL_RSP, vgId);
@@ -138,7 +138,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
       } else {
         STaosxRsp taosxRsp = {0};
         tqInitTaosxRsp(&taosxRsp, pRequest);
-        tqOffsetResetToLog(&taosxRsp.rspOffset, pHandle->pRef->refVer);
+        tqOffsetResetToLog(&taosxRsp.rspOffset, pHandle->pRef->refVer + 1);
         int32_t code = tqSendDataRsp(pHandle, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP, vgId);
         tDeleteSTaosxRsp(&taosxRsp);
 
@@ -246,7 +246,7 @@ static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle,
 
   if (offset->type == TMQ_OFFSET__LOG) {
     walReaderVerifyOffset(pHandle->pWalReader, offset);
-    int64_t fetchVer = offset->version + 1;
+    int64_t fetchVer = offset->version;
     pCkHead = taosMemoryMalloc(sizeof(SWalCkHead) + 2048);
     if (pCkHead == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -279,14 +279,14 @@ static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle,
       // process meta
       if (pHead->msgType != TDMT_VND_SUBMIT) {
         if (totalRows > 0) {
-          tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer - 1);
+          tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer);
           setRequestVersion(&taosxRsp.reqOffset, offset->version);
           code = tqSendDataRsp(pHandle, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP, vgId);
           goto end;
         }
 
         tqDebug("fetch meta msg, ver:%" PRId64 ", type:%s", pHead->version, TMSG_INFO(pHead->msgType));
-        tqOffsetResetToLog(&metaRsp.rspOffset, fetchVer);
+        tqOffsetResetToLog(&metaRsp.rspOffset, fetchVer + 1);
         metaRsp.resMsgType = pHead->msgType;
         metaRsp.metaRspLen = pHead->bodyLen;
         metaRsp.metaRsp = pHead->body;
@@ -309,7 +309,7 @@ static int32_t extractDataAndRspForDbStbSubscribe(STQ* pTq, STqHandle* pHandle,
       }
 
       if (totalRows >= 4096 || taosxRsp.createTableNum > 0) {
-        tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer);
+        tqOffsetResetToLog(&taosxRsp.rspOffset, fetchVer + 1);
         setRequestVersion(&taosxRsp.reqOffset, offset->version);
         code = tqSendDataRsp(pHandle, pMsg, pRequest, (SMqDataRsp*)&taosxRsp, TMQ_MSG_TYPE__TAOSX_RSP, vgId);
         goto end;

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

@@ -237,7 +237,7 @@ void initCacheFn(SStoreCacheReader* pCache) {
 }
 
 void initSnapshotFn(SStoreSnapshotFn* pSnapshot) {
-  pSnapshot->createSnapshot = setForSnapShot;
+  pSnapshot->setForSnapShot = setForSnapShot;
   pSnapshot->destroySnapshot = destroySnapContext;
   pSnapshot->getMetaTableInfoFromSnapshot = getMetaTableInfoFromSnapshot;
   pSnapshot->getTableInfoFromSnapshot = getTableInfoFromSnapshot;

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

@@ -462,7 +462,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg
       }
       break;
     case TDMT_VND_TMQ_COMMIT_OFFSET:
-      if (tqProcessOffsetCommitReq(pVnode->pTq, ver, pReq, pMsg->contLen - sizeof(SMsgHead)) < 0) {
+      if (tqProcessOffsetCommitReq(pVnode->pTq, ver, pReq, len) < 0) {
         goto _err;
       }
       break;
@@ -638,6 +638,8 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
       //      return tqProcessPollReq(pVnode->pTq, pMsg);
     case TDMT_VND_TMQ_VG_WALINFO:
       return tqProcessVgWalInfoReq(pVnode->pTq, pMsg);
+    case TDMT_VND_TMQ_VG_COMMITTEDINFO:
+      return tqProcessVgCommittedInfoReq(pVnode->pTq, pMsg);
     case TDMT_VND_TMQ_SEEK:
       return tqProcessSeekReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_TASK_RUN:

source/libs/executor/src/executor.c

@@ -1112,8 +1112,8 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
       SStoreTqReader* pReaderAPI = &pTaskInfo->storageAPI.tqReaderFn;
       SWalReader* pWalReader = pReaderAPI->tqReaderGetWalReader(pInfo->tqReader);
       walReaderVerifyOffset(pWalReader, pOffset);
-      if (pReaderAPI->tqReaderSeek(pInfo->tqReader, pOffset->version + 1, id) < 0) {
-        qError("tqReaderSeek failed ver:%" PRId64 ", %s", pOffset->version + 1, id);
+      if (pReaderAPI->tqReaderSeek(pInfo->tqReader, pOffset->version, id) < 0) {
+        qError("tqReaderSeek failed ver:%" PRId64 ", %s", pOffset->version, id);
         return -1;
       }
     } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
@@ -1202,7 +1202,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
       SOperatorInfo*  p = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, id);
       STableListInfo* pTableListInfo = ((SStreamRawScanInfo*)(p->info))->pTableListInfo;
 
-      if (pAPI->snapshotFn.createSnapshot(sContext, pOffset->uid) != 0) {
+      if (pAPI->snapshotFn.setForSnapShot(sContext, pOffset->uid) != 0) {
         qError("setDataForSnapShot error. uid:%" PRId64 " , %s", pOffset->uid, id);
         terrno = TSDB_CODE_PAR_INTERNAL_ERROR;
         return -1;
@@ -1239,7 +1239,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
     } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_META) {
       SStreamRawScanInfo* pInfo = pOperator->info;
       SSnapContext*       sContext = pInfo->sContext;
-      if (pTaskInfo->storageAPI.snapshotFn.createSnapshot(sContext, pOffset->uid) != 0) {
+      if (pTaskInfo->storageAPI.snapshotFn.setForSnapShot(sContext, pOffset->uid) != 0) {
         qError("setForSnapShot error. uid:%" PRIu64 " ,version:%" PRId64, pOffset->uid, pOffset->version);
         terrno = TSDB_CODE_PAR_INTERNAL_ERROR;
         return -1;

source/libs/executor/src/scanoperator.c

@@ -1645,12 +1645,13 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
     pAPI->tsdReader.tsdReaderClose(pTSInfo->base.dataReader);
 
     pTSInfo->base.dataReader = NULL;
-    qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", pTaskInfo->streamInfo.snapshotVer + 1);
-    if (pAPI->tqReaderFn.tqReaderSeek(pInfo->tqReader, pTaskInfo->streamInfo.snapshotVer + 1, pTaskInfo->id.str) < 0) {
+    int64_t validVer = pTaskInfo->streamInfo.snapshotVer + 1;
+    qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", validVer);
+    if (pAPI->tqReaderFn.tqReaderSeek(pInfo->tqReader, validVer, pTaskInfo->id.str) < 0) {
       return NULL;
     }
 
-    tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pTaskInfo->streamInfo.snapshotVer);
+    tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, validVer);
   }
 
   if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__LOG) {
@@ -1661,8 +1662,8 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
       SSDataBlock* pRes = pAPI->tqReaderFn.tqGetResultBlock(pInfo->tqReader);
       struct SWalReader* pWalReader = pAPI->tqReaderFn.tqReaderGetWalReader(pInfo->tqReader);
 
-      // curVersion move to next, so currentOffset = curVersion - 1
-      tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pWalReader->curVersion - 1);
+      // curVersion move to next
+      tqOffsetResetToLog(&pTaskInfo->streamInfo.currentOffset, pWalReader->curVersion);
 
       if (hasResult) {
         qDebug("doQueueScan get data from log %" PRId64 " rows, version:%" PRId64, pRes->info.rows,
@@ -2183,7 +2184,7 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
     STqOffsetVal   offset = {0};
     if (mtInfo.uid == 0) {  // read snapshot done, change to get data from wal
       qDebug("tmqsnap read snapshot done, change to get data from wal");
-      tqOffsetResetToLog(&offset, pInfo->sContext->snapVersion);
+      tqOffsetResetToLog(&offset, pInfo->sContext->snapVersion + 1);
     } else {
       tqOffsetResetToData(&offset, mtInfo.uid, INT64_MIN);
       qDebug("tmqsnap change get data uid:%" PRId64 "", mtInfo.uid);

source/libs/wal/src/walRead.c

@@ -135,8 +135,8 @@ void walReaderVerifyOffset(SWalReader *pWalReader, STqOffsetVal* pOffset){
   int64_t firstVer = walGetFirstVer((pWalReader)->pWal);
   taosThreadMutexUnlock(&pWalReader->pWal->mutex);
 
-  if (pOffset->version + 1 < firstVer){
-    pOffset->version = firstVer - 1;
+  if (pOffset->version < firstVer){
+    pOffset->version = firstVer;
   }
 }
 

source/util/src/terror.c

@@ -631,7 +631,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SCALAR_CONVERT_ERROR,           "Cannot convert to s
 
 //tmq
 TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_INVALID_MSG,                "Invalid message")
+TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_NEED_INITIALIZED,           "Assignment or poll interface need to be called first")
 TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_SNAPSHOT_ERROR,             "Can not operate in snapshot mode")
+TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_NO_COMMITTED,               "No committed info")
 TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_VERSION_OUT_OF_RANGE,       "Offset out of range")
 TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_INVALID_VGID,               "VgId does not belong to this consumer")
 TAOS_DEFINE_ERROR(TSDB_CODE_TMQ_INVALID_TOPIC,              "Topic does not belong to this consumer")