fix stream state transfer

include/libs/stream/tstream.h

@@ -368,6 +368,8 @@ typedef struct SStreamMeta {
   TdThreadMutex backendMutex;
 
   int32_t chkptNotReadyTasks;
+
+  int64_t checkpointId;
   SArray*  checkpointSaved;
   SArray*  checkpointInUse;
   int32_t  checkpointCap;

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

@@ -41,13 +41,25 @@ int32_t streamStateSnapReaderOpen(STQ* pTq, int64_t sver, int64_t ever, SStreamS
     code = TSDB_CODE_OUT_OF_MEMORY;
     goto _err;
   }
+
+  SStreamMeta* meta = pTq->pStreamMeta;
   pReader->pTq = pTq;
   pReader->sver = sver;
   pReader->ever = ever;
 
+  int64_t checkpointId = meta ? meta->checkpointId : 0;
+
   SStreamSnapReader* pSnapReader = NULL;
-  sprintf(tdir, "%s%s%s%s%s", pTq->path, TD_DIRSEP, VNODE_TQ_STREAM, TD_DIRSEP, "checkpoints");
-  if (streamSnapReaderOpen(pTq, sver, ever, tdir, &pSnapReader) == 0) {
+
+  // restore from checkpoint if checkpointid != 0
+  if (checkpointId != 0) {
+    sprintf(tdir, "%s%s%s%s%s%scheckpoint%" PRId64 "", pTq->path, TD_DIRSEP, VNODE_TQ_STREAM, TD_DIRSEP, "checkpoints",
+            TD_DIRSEP, checkpointId);
+  } else {
+    sprintf(tdir, "%s%s%s%s%s", pTq->path, TD_DIRSEP, VNODE_TQ_STREAM, TD_DIRSEP, "state");
+  }
+
+  if (streamSnapReaderOpen(pTq, sver, checkpointId, tdir, &pSnapReader) == 0) {
     pReader->complete = 1;
   } else {
     code = -1;

source/libs/stream/src/streamBackendRocksdb.c

@@ -163,8 +163,8 @@ int32_t copyFiles(const char* src, const char* dst) {
     char* name = taosGetDirEntryName(de);
     if (strcmp(name, ".") == 0 || strcmp(name, "..") == 0) continue;
 
-    sprintf(absSrcPath, "%s/%s", src, name);
-    sprintf(absDstPath, "%s/%s", dst, name);
+    sprintf(absSrcPath, "%s%s%s", src, TD_DIRSEP, name);
+    sprintf(absDstPath, "%s%s%s", dst, TD_DIRSEP, name);
     if (!taosDirEntryIsDir(de)) {
       code = taosCopyFile(absSrcPath, absDstPath);
       if (code == -1) {

source/libs/stream/src/streamSnapshot.c

@@ -64,6 +64,7 @@ struct SStreamSnapReader {
   int64_t           sver;
   int64_t           ever;
   SStreamSnapHandle handle;
+  int64_t           checkpointId;
 };
 struct SStreamSnapWriter {
   void*             pMeta;
@@ -217,6 +218,7 @@ int32_t streamSnapReaderOpen(void* pMeta, int64_t sver, int64_t ever, char* path
   if (pReader == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
+  pReader->handle.checkpointId = ever;
   // const char* path = NULL;
   if (streamSnapHandleInit(&pReader->handle, (char*)path) < 0) {
     taosMemoryFree(pReader);

source/libs/stream/src/streamState.c

@@ -108,7 +108,7 @@ SStreamState* streamStateOpen(char* path, void* pTask, bool specPath, int32_t sz
   SStreamTask* pStreamTask = pTask;
   char         statePath[1024];
   if (!specPath) {
-    sprintf(statePath, "%s/%d", path, pStreamTask->id.taskId);
+    sprintf(statePath, "%s%s%d", path, TD_DIRSEP, pStreamTask->id.taskId);
   } else {
     memset(statePath, 0, 1024);
     tstrncpy(statePath, path, 1024);
@@ -762,7 +762,8 @@ int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVa
 
 int32_t streamStateSessionDel(SStreamState* pState, const SSessionKey* key) {
 #ifdef USE_ROCKSDB
-  qDebug("===stream===delete skey:%" PRId64 ", ekey:%" PRId64 ", groupId:%" PRIu64, key->win.skey,key->win.ekey, key->groupId);
+  qDebug("===stream===delete skey:%" PRId64 ", ekey:%" PRId64 ", groupId:%" PRIu64, key->win.skey, key->win.ekey,
+         key->groupId);
   return streamStateSessionDel_rocksdb(pState, key);
 #else
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};