diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h
index 35a6838b2e0f36ba36ccd41c4e8dd175e2efe7c9..a08152714408280389b2d76d9dcb5029259109d6 100644
--- a/include/libs/wal/wal.h
+++ b/include/libs/wal/wal.h
@@ -149,6 +149,7 @@ struct SWalReader {
   TdFilePtr      pIdxFile;
   int64_t        curFileFirstVer;
   int64_t        curVersion;
+  int64_t        skipToVersion; // skip data and jump to destination version, usually used by stream resume ignoring untreated data
   int64_t        capacity;
   TdThreadMutex  mutex;
   SWalFilterCond cond;
@@ -200,6 +201,8 @@ int32_t     walReaderSeekVer(SWalReader *pRead, int64_t ver);
 int32_t     walNextValidMsg(SWalReader *pRead);
 int64_t     walReaderGetCurrentVer(const SWalReader *pReader);
 int64_t     walReaderGetValidFirstVer(const SWalReader *pReader);
+int64_t     walReaderGetSkipToVersion(SWalReader *pReader);
+void        walReaderSetSkipToVersion(SWalReader *pReader, int64_t ver);
 void        walReaderValidVersionRange(SWalReader *pReader, int64_t *sver, int64_t *ever);
 void        walReaderVerifyOffset(SWalReader *pWalReader, STqOffsetVal* pOffset);
 
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index c6077a3f8f013d03b5f24bf997922fb878cfc1ba..633b15a1fb82c43884083f15a4fab36e7a614010 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1303,19 +1303,22 @@ int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
 
 int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
   SVResumeStreamTaskReq* pReq = (SVResumeStreamTaskReq*)msg;
+
+  int32_t      vgId = pTq->pStreamMeta->vgId;
   SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->taskId);
   if (pTask) {
     atomic_store_8(&pTask->status.taskStatus, pTask->status.keepTaskStatus);
 
     // no lock needs to secure the access of the version
-    if (pReq->igUntreated && pTask->taskLevel == TASK_LEVEL__SOURCE) {  // discard all the data  when the stream task is suspended.
-      pTask->chkInfo.currentVer = sversion;
-      walReaderSeekVer(pTask->exec.pWalReader, sversion);
-      tqDebug("vgId:%d s-task:%s resume to normal from the latest version:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d", pTq->pStreamMeta->vgId,
-              pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus);
+    if (pReq->igUntreated && pTask->taskLevel == TASK_LEVEL__SOURCE) {
+      // discard all the data  when the stream task is suspended.
+      walReaderSetSkipToVersion(pTask->exec.pWalReader, sversion);
+      tqDebug("vgId:%d s-task:%s resume to exec, prev paused version:%" PRId64 ", start from vnode ver:%" PRId64
+              ", schedStatus:%d",
+              vgId, pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus);
     } else {  // from the previous paused version and go on
-      tqDebug("vgId:%d s-task:%s resume to normal from paused ver:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d", pTq->pStreamMeta->vgId,
-              pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus);
+      tqDebug("vgId:%d s-task:%s resume to exec, from paused ver:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d",
+              vgId, pTask->id.idStr, pTask->chkInfo.currentVer, sversion, pTask->status.schedStatus);
     }
 
     if (pTask->taskLevel == TASK_LEVEL__SOURCE && taosQueueItemSize(pTask->inputQueue->queue) == 0) {
@@ -1324,6 +1327,8 @@ int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t ms
       streamSchedExec(pTask);
     }
     streamMetaReleaseTask(pTq->pStreamMeta, pTask);
+  } else {
+    tqError("vgId:%d failed to find the s-task:0x%x for resume stream task", vgId, pReq->taskId);
   }
 
   return 0;
@@ -1432,7 +1437,7 @@ int32_t tqStartStreamTasks(STQ* pTq) {
   int32_t numOfTasks = taosArrayGetSize(pMeta->pTaskList);
   if (numOfTasks == 0) {
     tqInfo("vgId:%d no stream tasks exist", vgId);
-    taosWUnLockLatch(&pTq->pStreamMeta->lock);
+    taosWUnLockLatch(&pMeta->lock);
     return 0;
   }
 
@@ -1440,7 +1445,7 @@ int32_t tqStartStreamTasks(STQ* pTq) {
 
   if (pMeta->walScanCounter > 1) {
     tqDebug("vgId:%d wal read task has been launched, remain scan times:%d", vgId, pMeta->walScanCounter);
-    taosWUnLockLatch(&pTq->pStreamMeta->lock);
+    taosWUnLockLatch(&pMeta->lock);
     return 0;
   }
 
@@ -1448,7 +1453,7 @@ int32_t tqStartStreamTasks(STQ* pTq) {
   if (pRunReq == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     tqError("vgId:%d failed to create msg to start wal scanning to launch stream tasks, code:%s", vgId, terrstr());
-    taosWUnLockLatch(&pTq->pStreamMeta->lock);
+    taosWUnLockLatch(&pMeta->lock);
     return -1;
   }
 
@@ -1459,7 +1464,7 @@ int32_t tqStartStreamTasks(STQ* pTq) {
 
   SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
   tmsgPutToQueue(&pTq->pVnode->msgCb, STREAM_QUEUE, &msg);
-  taosWUnLockLatch(&pTq->pStreamMeta->lock);
+  taosWUnLockLatch(&pMeta->lock);
 
   return 0;
 }
diff --git a/source/dnode/vnode/src/tq/tqRestore.c b/source/dnode/vnode/src/tq/tqRestore.c
index 6e3036a66ba73c60b421ad73a2a6fb7335a3803f..fe80f486918413390ee7916fb97fe07c58a1b80d 100644
--- a/source/dnode/vnode/src/tq/tqRestore.c
+++ b/source/dnode/vnode/src/tq/tqRestore.c
@@ -87,6 +87,16 @@ static int32_t doSetOffsetForWalReader(SStreamTask *pTask, int32_t vgId) {
     }
   }
 
+  int64_t skipToVer = walReaderGetSkipToVersion(pTask->exec.pWalReader);
+  if (skipToVer != 0 && skipToVer > pTask->chkInfo.currentVer) {
+    int32_t code = walReaderSeekVer(pTask->exec.pWalReader, skipToVer);
+    if (code != TSDB_CODE_SUCCESS) {  // no data in wal, quit
+      return code;
+    }
+
+    tqDebug("vgId:%d s-task:%s wal reader jump to ver:%" PRId64, vgId, pTask->id.idStr, skipToVer);
+  }
+
   return TSDB_CODE_SUCCESS;
 }
 
diff --git a/source/libs/wal/src/walRead.c b/source/libs/wal/src/walRead.c
index c29d82bcf394b65beb5dcd94e6d35835e0a7cc60..0eb3d6ef09e8c81b981d69803b9bdb5f335b721c 100644
--- a/source/libs/wal/src/walRead.c
+++ b/source/libs/wal/src/walRead.c
@@ -108,6 +108,14 @@ int32_t walNextValidMsg(SWalReader *pReader) {
 
 int64_t walReaderGetCurrentVer(const SWalReader *pReader) { return pReader->curVersion; }
 int64_t walReaderGetValidFirstVer(const SWalReader *pReader) { return walGetFirstVer(pReader->pWal); }
+void    walReaderSetSkipToVersion(SWalReader *pReader, int64_t ver) { atomic_store_64(&pReader->skipToVersion, ver); }
+
+// this function is NOT multi-thread safe, and no need to be.
+int64_t walReaderGetSkipToVersion(SWalReader *pReader) {
+  int64_t newVersion = pReader->skipToVersion;
+  pReader->skipToVersion = 0;
+  return newVersion;
+}
 
 void walReaderValidVersionRange(SWalReader *pReader, int64_t *sver, int64_t *ever) {
   *sver = walGetFirstVer(pReader->pWal);