diff --git a/include/common/tcommon.h b/include/common/tcommon.h
index c5404085bbb56d01b39866e515bf743349ced983..4eea744be19b51ebb321629c92c8d7e1f515fcda 100644
--- a/include/common/tcommon.h
+++ b/include/common/tcommon.h
@@ -56,7 +56,6 @@ enum {
   STREAM_INPUT__DATA_SUBMIT = 1,
   STREAM_INPUT__DATA_BLOCK,
   STREAM_INPUT__MERGED_SUBMIT,
-  // STREAM_INPUT__TABLE_SCAN,
   STREAM_INPUT__TQ_SCAN,
   STREAM_INPUT__DATA_RETRIEVE,
   STREAM_INPUT__GET_RES,
@@ -154,7 +153,7 @@ typedef struct SQueryTableDataCond {
   int32_t      order;  // desc|asc order to iterate the data block
   int32_t      numOfCols;
   SColumnInfo* colList;
-  int32_t      type;   // data block load type:
+  int32_t      type;  // data block load type:
   STimeWindow  twindows;
   int64_t      startVersion;
   int64_t      endVersion;
diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h
index 14173690967ffd26be92cf13a05af6f7508533fe..980e01a65c44dca7ed3ad446321ced38dbb96f34 100644
--- a/include/libs/wal/wal.h
+++ b/include/libs/wal/wal.h
@@ -41,7 +41,7 @@ extern "C" {
 #define WAL_REFRESH_MS    1000
 #define WAL_PATH_LEN      (TSDB_FILENAME_LEN + 12)
 #define WAL_FILE_LEN      (WAL_PATH_LEN + 32)
-#define WAL_MAGIC         0xFAFBFCFDULL
+#define WAL_MAGIC         0xFAFBFCFDF4F3F2F1ULL
 #define WAL_SCAN_BUF_SIZE (1024 * 1024 * 3)
 
 typedef enum {
diff --git a/source/dnode/vnode/src/inc/meta.h b/source/dnode/vnode/src/inc/meta.h
index e08925acc398f740c1bc4bcd205e14768cdfcc98..12c3f7f2e80c80d2fd0f63d8a2cad94aa503f869 100644
--- a/source/dnode/vnode/src/inc/meta.h
+++ b/source/dnode/vnode/src/inc/meta.h
@@ -78,7 +78,10 @@ struct SMeta {
   TTB*  pTagIdx;
   TTB*  pTtlIdx;
 
-  TTB*      pSmaIdx;
+  TTB* pSmaIdx;
+
+  TTB* pTaskIdx;
+
   SMetaIdx* pIdx;
 };
 
diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h
index ce83b335d78a198b9dc64668f59a5c7f0e5f4fa1..d47b1cc5cbe7dd47ecf9e5748721c44b9b974cb4 100644
--- a/source/dnode/vnode/src/inc/vnodeInt.h
+++ b/source/dnode/vnode/src/inc/vnodeInt.h
@@ -153,7 +153,7 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDeployReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessTaskDropReq(STQ* pTq, char* msg, int32_t msgLen);
-int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* data);
+int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* data, int64_t ver);
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg);
diff --git a/source/dnode/vnode/src/meta/metaOpen.c b/source/dnode/vnode/src/meta/metaOpen.c
index 396a58c988eb5c3beae43e973a3896049a2dbcc4..4910830c7dabf5e1a619dc157898505c92608535 100644
--- a/source/dnode/vnode/src/meta/metaOpen.c
+++ b/source/dnode/vnode/src/meta/metaOpen.c
@@ -22,6 +22,7 @@ static int tagIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kL
 static int ttlIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
 static int uidIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
 static int smaIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
+static int taskIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
 
 static int32_t metaInitLock(SMeta *pMeta) { return taosThreadRwlockInit(&pMeta->lock, NULL); }
 static int32_t metaDestroyLock(SMeta *pMeta) { return taosThreadRwlockDestroy(&pMeta->lock); }
@@ -130,6 +131,12 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
     goto _err;
   }
 
+  ret = tdbTbOpen("stream.task.db", sizeof(int64_t), -1, taskIdxKeyCmpr, pMeta->pEnv, &pMeta->pTaskIdx);
+  if (ret < 0) {
+    metaError("vgId: %d, failed to open meta stream task index since %s", TD_VID(pVnode), tstrerror(terrno));
+    goto _err;
+  }
+
   // open index
   if (metaOpenIdx(pMeta) < 0) {
     metaError("vgId:%d, failed to open meta index since %s", TD_VID(pVnode), tstrerror(terrno));
@@ -143,6 +150,7 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
 
 _err:
   if (pMeta->pIdx) metaCloseIdx(pMeta);
+  if (pMeta->pTaskIdx) tdbTbClose(pMeta->pTaskIdx);
   if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx);
   if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx);
   if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx);
@@ -162,6 +170,7 @@ _err:
 int metaClose(SMeta *pMeta) {
   if (pMeta) {
     if (pMeta->pIdx) metaCloseIdx(pMeta);
+    if (pMeta->pTaskIdx) tdbTbClose(pMeta->pTaskIdx);
     if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx);
     if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx);
     if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx);
@@ -378,3 +387,16 @@ static int smaIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kL
 
   return 0;
 }
+
+static int taskIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) {
+  int32_t uid1 = *(int32_t *)pKey1;
+  int32_t uid2 = *(int32_t *)pKey2;
+
+  if (uid1 > uid2) {
+    return 1;
+  } else if (uid1 < uid2) {
+    return -1;
+  }
+
+  return 0;
+}
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 86b016b8bee36c7cfae93d323dc7c47e72c83fe4..a05013d99694984d8e76ddefcb0aaf8dc6cf1d66 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -695,7 +695,7 @@ FAIL:
   return -1;
 }
 
-int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
+int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq, int64_t ver) {
   void*              pIter = NULL;
   bool               failed = false;
   SStreamDataSubmit* pSubmit = NULL;
@@ -713,7 +713,7 @@ int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* pReq) {
     SStreamTask* pTask = *(SStreamTask**)pIter;
     if (!pTask->isDataScan) continue;
 
-    qDebug("data submit enqueue stream task: %d", pTask->taskId);
+    qDebug("data submit enqueue stream task: %d, ver: %ld", pTask->taskId, ver);
 
     if (!failed) {
       if (streamTaskInput(pTask, (SStreamQueueItem*)pSubmit) < 0) {
diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c
index 290ffe5c8d6a16865df7ee1296c871ccebf0a847..e866112250c516e09f359df9d3ca917b805d5f5f 100644
--- a/source/dnode/vnode/src/tq/tqMeta.c
+++ b/source/dnode/vnode/src/tq/tqMeta.c
@@ -48,7 +48,7 @@ int32_t tqMetaOpen(STQ* pTq) {
     ASSERT(0);
   }
 
-  if (tdbTbOpen("handles", -1, -1, 0, pTq->pMetaStore, &pTq->pExecStore) < 0) {
+  if (tdbTbOpen("handles", -1, -1, NULL, pTq->pMetaStore, &pTq->pExecStore) < 0) {
     ASSERT(0);
   }
 
diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c
index 6097ddd49eb504a2d5e8f6b41a632952c389992d..0720cf015bc30aa36f1f5f8de432b79d7586e28d 100644
--- a/source/dnode/vnode/src/tq/tqPush.c
+++ b/source/dnode/vnode/src/tq/tqPush.c
@@ -252,7 +252,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
     SSubmitReq* pReq = (SSubmitReq*)data;
     pReq->version = ver;
 
-    tqProcessStreamTrigger(pTq, data);
+    tqProcessStreamTrigger(pTq, data, ver);
   }
 
   return 0;
diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c
index 68d8c54e28807b2da90bd2c41b4df96195979b27..a29c7a130902285b3ebb58294dc75d556237219f 100644
--- a/source/libs/wal/src/walWrite.c
+++ b/source/libs/wal/src/walWrite.c
@@ -116,15 +116,15 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   }
 
   walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
-  TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
+  TdFilePtr pIdxFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
 
-  if (pIdxTFile == NULL) {
+  if (pIdxFile == NULL) {
     ASSERT(0);
     taosThreadMutexUnlock(&pWal->mutex);
     return -1;
   }
   int64_t idxOff = walGetVerIdxOffset(pWal, ver);
-  code = taosLSeekFile(pIdxTFile, idxOff, SEEK_SET);
+  code = taosLSeekFile(pIdxFile, idxOff, SEEK_SET);
   if (code < 0) {
     ASSERT(0);
     taosThreadMutexUnlock(&pWal->mutex);
@@ -132,7 +132,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   }
   // read idx file and get log file pos
   SWalIdxEntry entry;
-  if (taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry)) != sizeof(SWalIdxEntry)) {
+  if (taosReadFile(pIdxFile, &entry, sizeof(SWalIdxEntry)) != sizeof(SWalIdxEntry)) {
     ASSERT(0);
     taosThreadMutexUnlock(&pWal->mutex);
     return -1;
@@ -140,24 +140,24 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   ASSERT(entry.ver == ver);
 
   walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
-  TdFilePtr pLogTFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
-  if (pLogTFile == NULL) {
-    ASSERT(0);
+  TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
+  if (pLogFile == NULL) {
     // TODO
+    terrno = TAOS_SYSTEM_ERROR(errno);
     taosThreadMutexUnlock(&pWal->mutex);
     return -1;
   }
-  code = taosLSeekFile(pLogTFile, entry.offset, SEEK_SET);
+  code = taosLSeekFile(pLogFile, entry.offset, SEEK_SET);
   if (code < 0) {
-    ASSERT(0);
     // TODO
+    terrno = TAOS_SYSTEM_ERROR(errno);
     taosThreadMutexUnlock(&pWal->mutex);
     return -1;
   }
   // validate offset
   SWalCkHead head;
-  ASSERT(taosValidFile(pLogTFile));
-  int64_t size = taosReadFile(pLogTFile, &head, sizeof(SWalCkHead));
+  ASSERT(taosValidFile(pLogFile));
+  int64_t size = taosReadFile(pLogFile, &head, sizeof(SWalCkHead));
   if (size != sizeof(SWalCkHead)) {
     ASSERT(0);
     taosThreadMutexUnlock(&pWal->mutex);
@@ -180,14 +180,14 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   }
 
   // truncate old files
-  code = taosFtruncateFile(pLogTFile, entry.offset);
+  code = taosFtruncateFile(pLogFile, entry.offset);
   if (code < 0) {
     ASSERT(0);
     terrno = TAOS_SYSTEM_ERROR(errno);
     taosThreadMutexUnlock(&pWal->mutex);
     return -1;
   }
-  code = taosFtruncateFile(pIdxTFile, idxOff);
+  code = taosFtruncateFile(pIdxFile, idxOff);
   if (code < 0) {
     ASSERT(0);
     terrno = TAOS_SYSTEM_ERROR(errno);
@@ -205,8 +205,10 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
     ASSERT(((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize == 0);
     ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->firstVer = -1;
   }
-  taosCloseFile(&pIdxTFile);
-  taosCloseFile(&pLogTFile);
+  taosCloseFile(&pIdxFile);
+  taosCloseFile(&pLogFile);
+
+  walSaveMeta(pWal);
 
   // unlock
   taosThreadMutexUnlock(&pWal->mutex);