make snapshot compile

include/libs/transport/trpc.h

@@ -27,7 +27,7 @@ extern "C" {
 
 #define TAOS_CONN_SERVER 0
 #define TAOS_CONN_CLIENT 1
-#define IsReq(pMsg) (pMsg->msgType & 1U)
+#define IsReq(pMsg)      (pMsg->msgType & 1U)
 
 extern int32_t tsRpcHeadSize;
 

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

@@ -212,7 +212,9 @@ typedef struct {
 struct SVState {
   int64_t committed;
   int64_t applied;
+  int64_t applyTerm;
   int64_t commitID;
+  int64_t commitTerm;
 };
 
 struct SVnodeInfo {

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

@@ -223,6 +223,7 @@ int vnodeCommit(SVnode *pVnode) {
   // save info
   info.config = pVnode->config;
   info.state.committed = pVnode->state.applied;
+  info.state.commitTerm = pVnode->state.applyTerm;
   info.state.commitID = pVnode->state.commitID;
   snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
   if (vnodeSaveInfo(dir, &info) < 0) {
@@ -270,7 +271,7 @@ int vnodeCommit(SVnode *pVnode) {
     ASSERT(0);
     return -1;
   }
-  
+
   pVnode->state.committed = info.state.committed;
 
   // postCommit
@@ -316,6 +317,7 @@ static int vnodeEncodeState(const void *pObj, SJson *pJson) {
 
   if (tjsonAddIntegerToObject(pJson, "commit version", pState->committed) < 0) return -1;
   if (tjsonAddIntegerToObject(pJson, "commit ID", pState->commitID) < 0) return -1;
+  if (tjsonAddIntegerToObject(pJson, "commit term", pState->commitTerm) < 0) return -1;
 
   return 0;
 }
@@ -328,6 +330,8 @@ static int vnodeDecodeState(const SJson *pJson, void *pObj) {
   if (code < 0) return -1;
   tjsonGetNumberValue(pJson, "commit ID", pState->commitID, code);
   if (code < 0) return -1;
+  tjsonGetNumberValue(pJson, "commit term", pState->commitTerm, code);
+  if (code < 0) return -1;
 
   return 0;
 }

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

@@ -79,8 +79,10 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
   strcpy(pVnode->path, path);
   pVnode->config = info.config;
   pVnode->state.committed = info.state.committed;
+  pVnode->state.commitTerm = info.state.commitTerm;
   pVnode->state.applied = info.state.committed;
   pVnode->state.commitID = info.state.commitID;
+  pVnode->state.commitTerm = info.state.commitTerm;
   pVnode->pTfs = pTfs;
   pVnode->msgCb = msgCb;
   pVnode->blockCount = 0;
@@ -194,4 +196,9 @@ void vnodeStop(SVnode *pVnode) {}
 
 int64_t vnodeGetSyncHandle(SVnode *pVnode) { return pVnode->sync; }
 
-void vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot) { pSnapshot->lastApplyIndex = pVnode->state.committed; }
+void vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot) {
+  pSnapshot->data = NULL;
+  pSnapshot->lastApplyIndex = pVnode->state.committed;
+  pSnapshot->lastApplyTerm = pVnode->state.commitTerm;
+  pSnapshot->lastConfigIndex = -1;
+}

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

@@ -143,6 +143,7 @@ int32_t vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
          version);
 
   pVnode->state.applied = version;
+  pVnode->state.applyTerm = pMsg->info.conn.applyTerm;
 
   // skip header
   pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
@@ -169,7 +170,7 @@ int32_t vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
       if (vnodeProcessDropTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
       break;
     case TDMT_VND_DROP_TTL_TABLE:
-      //if (vnodeProcessDropTtlTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
+      // if (vnodeProcessDropTtlTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
       break;
     case TDMT_VND_CREATE_SMA: {
       if (vnodeProcessCreateTSmaReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
@@ -799,7 +800,8 @@ _exit:
   taosArrayDestroy(submitRsp.pArray);
 
   // TODO: the partial success scenario and the error case
-  // => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level 1/level 2.
+  // => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level
+  // 1/level 2.
   // TODO: refactor
   if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) {
     tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT);

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

@@ -432,20 +432,19 @@ static void vnodeSyncRollBackMsg(SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta
 static int32_t vnodeSnapshotStartRead(struct SSyncFSM *pFsm, void *pParam, void **ppReader) {
   SVnode         *pVnode = pFsm->data;
   SSnapshotParam *pSnapshotParam = pParam;
-  int32_t         code =
-      vnodeSnapshotReaderOpen(pVnode, (SVSnapshotReader **)ppReader, pSnapshotParam->start, pSnapshotParam->end);
+  int32_t code = vnodeSnapReaderOpen(pVnode, pSnapshotParam->start, pSnapshotParam->end, (SVSnapReader **)ppReader);
   return code;
 }
 
 static int32_t vnodeSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) {
   SVnode *pVnode = pFsm->data;
-  int32_t code = vnodeSnapshotReaderClose(pReader);
+  int32_t code = vnodeSnapReaderClose(pReader);
   return code;
 }
 
 static int32_t vnodeSnapshotDoRead(struct SSyncFSM *pFsm, void *pReader, void **ppBuf, int32_t *len) {
   SVnode *pVnode = pFsm->data;
-  int32_t code = vnodeSnapshotRead(pReader, (const void **)ppBuf, len);
+  int32_t code = vnodeSnapRead(pReader, (uint8_t **)ppBuf, len);
   return code;
 }