feat:add checkpoint

source/common/src/tglobal.c

@@ -49,7 +49,7 @@ int32_t tsNumOfMnodeQueryThreads = 4;
 int32_t tsNumOfMnodeFetchThreads = 1;
 int32_t tsNumOfMnodeReadThreads = 1;
 int32_t tsNumOfVnodeQueryThreads = 4;
-float   tsRatioOfVnodeStreamThreads = 1.0;
+float   tsRatioOfVnodeStreamThreads = 2.0;
 int32_t tsNumOfVnodeFetchThreads = 4;
 int32_t tsNumOfVnodeRsmaThreads = 2;
 int32_t tsNumOfQnodeQueryThreads = 4;

source/libs/stream/src/tstreamFileState.c

@@ -279,9 +279,9 @@ void streamFileStateDecode(SStreamFileState* pFileState, void* pBuff, int32_t le
 
 void streamFileStateEncode(SStreamFileState* pFileState, void** pVal, int32_t* pLen) {
   *pLen = sizeof(TSKEY);
-  *pVal = taosMemoryCalloc(1, *pLen);
-  void** buff = pVal; 
-  taosEncodeFixedI64(buff, pFileState->flushMark);
+  (*pVal) = taosMemoryCalloc(1, *pLen);
+  void* buff = *pVal; 
+  taosEncodeFixedI64(&buff, pFileState->flushMark);
 }
 
 int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState) {
@@ -300,6 +300,7 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot,
     streamFileStateEncode(pFileState, &buff, &len);
     SWinKey key = {.ts = -1, .groupId = 0}; // dengyihao
     streamStatePut_rocksdb(pFileState->pFileStore, &key, buff, len);
+    taosMemoryFree(buff);
   }
   return code;
 }