diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 6b5eb3b491fdbbd441eed703d5327efe67afcb64..31d13d6cf49bde537388bd96c9f7b7fde194b5c3 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -308,9 +308,11 @@ static FORCE_INLINE int32_t streamTaskOutput(SStreamTask* pTask, SStreamDataBloc
   if (pTask->sinkType == TASK_SINK__TABLE) {
     ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
     pTask->tbSink.tbSinkFunc(pTask, pTask->tbSink.vnode, 0, pBlock->blocks);
+    taosFreeQitem(pBlock);
   } else if (pTask->sinkType == TASK_SINK__SMA) {
     ASSERT(pTask->dispatchType == TASK_DISPATCH__NONE);
     pTask->smaSink.smaSink(pTask->smaSink.vnode, pTask->smaSink.smaId, pBlock->blocks);
+    taosFreeQitem(pBlock);
   } else {
     ASSERT(pTask->dispatchType != TASK_DISPATCH__NONE);
     taosWriteQitem(pTask->outputQueue->queue, pBlock);
diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h
index 95af8ac30666b67b0a933477ff8ca0764d2d0a43..c7d1ccd3de8f6e2e921cf605c96d6fa2b1fc231f 100644
--- a/include/libs/wal/wal.h
+++ b/include/libs/wal/wal.h
@@ -141,6 +141,8 @@ typedef struct SWal {
   // ctl
   int64_t       refId;
   TdThreadMutex mutex;
+  // ref
+  SHashObj *pRefHash;  // ref -> SWalRef
   // path
   char path[WAL_PATH_LEN];
   // reusable write head
@@ -184,7 +186,7 @@ int32_t walRollback(SWal *, int64_t ver);
 // notify that previous logs can be pruned safely
 int32_t walBeginSnapshot(SWal *, int64_t ver);
 int32_t walEndSnapshot(SWal *);
-void    walRestoreFromSnapshot(SWal *, int64_t ver);
+int32_t walRestoreFromSnapshot(SWal *, int64_t ver);
 // int32_t  walDataCorrupted(SWal*);
 
 // read
@@ -199,6 +201,16 @@ int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead);
 int32_t walFetchBody(SWalReadHandle *pRead, SWalHead **ppHead);
 int32_t walSkipFetchBody(SWalReadHandle *pRead, const SWalHead *pHead);
 
+typedef struct {
+  int64_t refId;
+  int64_t ver;
+} SWalRef;
+
+SWalRef *walOpenRef(SWal *);
+void     walCloseRef(SWalRef *);
+int32_t  walRefVer(SWalRef *, int64_t ver);
+int32_t  walUnrefVer(SWal *);
+
 // deprecated
 #if 0
 int32_t walRead(SWal *, SWalHead **, int64_t ver);
diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c
index 13a2a2950fa49695795e63c8fcec6476afcb473d..66b12678ab981c2e29162e170349d70f05cbf42d 100644
--- a/source/common/src/tdatablock.c
+++ b/source/common/src/tdatablock.c
@@ -1773,6 +1773,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
   }
 
   // assign data
+  // TODO
   ret = taosMemoryCalloc(1, cap + 46);
   ret = POINTER_SHIFT(ret, 46);
   ret->header.vgId = vgId;
diff --git a/source/libs/wal/inc/walInt.h b/source/libs/wal/inc/walInt.h
index 84fe2814ffa9377830bf5e2f02e8309aeda9f505..7ca105ff2bf5bff338fd3ce0a57d63c3569534dc 100644
--- a/source/libs/wal/inc/walInt.h
+++ b/source/libs/wal/inc/walInt.h
@@ -132,6 +132,7 @@ static inline void walResetVer(SWalVer* pVer) {
 
 int walLoadMeta(SWal* pWal);
 int walSaveMeta(SWal* pWal);
+int walRemoveMeta(SWal* pWal);
 int walRollFileInfo(SWal* pWal);
 
 int walCheckAndRepairMeta(SWal* pWal);
diff --git a/source/libs/wal/src/walMeta.c b/source/libs/wal/src/walMeta.c
index 9aa848a7bb2d514533e60195a49aaa9c3539e88a..8e9cb3a84b87233901cfdead73498486f5a70687 100644
--- a/source/libs/wal/src/walMeta.c
+++ b/source/libs/wal/src/walMeta.c
@@ -419,3 +419,12 @@ int walLoadMeta(SWal* pWal) {
   taosMemoryFree(buf);
   return code;
 }
+
+int walRemoveMeta(SWal* pWal) {
+  int metaVer = walFindCurMetaVer(pWal);
+  if (metaVer == -1) return 0;
+  char fnameStr[WAL_FILE_LEN];
+  walBuildMetaName(pWal, metaVer, fnameStr);
+  taosRemoveFile(fnameStr);
+  return 0;
+}
diff --git a/source/libs/wal/src/walMgmt.c b/source/libs/wal/src/walMgmt.c
index 71cd6de73f005cbc710caf0a22805d60495f55ee..9505b02806015cee989128c6adec2e433e0a17e9 100644
--- a/source/libs/wal/src/walMgmt.c
+++ b/source/libs/wal/src/walMgmt.c
@@ -75,7 +75,7 @@ void walCleanUp() {
 }
 
 SWal *walOpen(const char *path, SWalCfg *pCfg) {
-  SWal *pWal = taosMemoryMalloc(sizeof(SWal));
+  SWal *pWal = taosMemoryCalloc(1, sizeof(SWal));
   if (pWal == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     return NULL;
@@ -92,6 +92,13 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
     return NULL;
   }
 
+  // init ref
+  pWal->pRefHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
+  if (pWal->pRefHash == NULL) {
+    taosMemoryFree(pWal);
+    return NULL;
+  }
+
   // open meta
   walResetVer(&pWal->vers);
   pWal->pWriteLogTFile = NULL;
@@ -100,6 +107,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
   pWal->fileInfoSet = taosArrayInit(8, sizeof(SWalFileInfo));
   if (pWal->fileInfoSet == NULL) {
     wError("vgId:%d, path:%s, failed to init taosArray %s", pWal->cfg.vgId, pWal->path, strerror(errno));
+    taosHashCleanup(pWal->pRefHash);
     taosMemoryFree(pWal);
     return NULL;
   }
@@ -115,12 +123,14 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
 
   if (taosThreadMutexInit(&pWal->mutex, NULL) < 0) {
     taosArrayDestroy(pWal->fileInfoSet);
+    taosHashCleanup(pWal->pRefHash);
     taosMemoryFree(pWal);
     return NULL;
   }
 
   pWal->refId = taosAddRef(tsWal.refSetId, pWal);
   if (pWal->refId < 0) {
+    taosHashCleanup(pWal->pRefHash);
     taosThreadMutexDestroy(&pWal->mutex);
     taosArrayDestroy(pWal->fileInfoSet);
     taosMemoryFree(pWal);
@@ -130,6 +140,7 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
   walLoadMeta(pWal);
 
   if (walCheckAndRepairMeta(pWal) < 0) {
+    taosHashCleanup(pWal->pRefHash);
     taosRemoveRef(tsWal.refSetId, pWal->refId);
     taosThreadMutexDestroy(&pWal->mutex);
     taosArrayDestroy(pWal->fileInfoSet);
@@ -175,6 +186,7 @@ void walClose(SWal *pWal) {
   walSaveMeta(pWal);
   taosArrayDestroy(pWal->fileInfoSet);
   pWal->fileInfoSet = NULL;
+  taosHashCleanup(pWal->pRefHash);
   taosThreadMutexUnlock(&pWal->mutex);
 
   taosRemoveRef(tsWal.refSetId, pWal->refId);
diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c
index 793ab8b2fb3e676f0e2404bf6fc2f781b9c93677..d30e0b6844d84a061813811557a800bf8efb440f 100644
--- a/source/libs/wal/src/walWrite.c
+++ b/source/libs/wal/src/walWrite.c
@@ -18,12 +18,47 @@
 #include "tchecksum.h"
 #include "walInt.h"
 
-void walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
-  /*pWal->vers.firstVer = -1;*/
+int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) {
+  taosThreadMutexLock(&pWal->mutex);
+
+  void *pIter = NULL;
+  while (1) {
+    taosHashIterate(pWal->pRefHash, pIter);
+    if (pIter == NULL) break;
+    SWalRef *pRef = (SWalRef *)pIter;
+    if (pRef->ver != -1) {
+      taosHashCancelIterate(pWal->pRefHash, pIter);
+      return -1;
+    }
+  }
+
+  taosCloseFile(&pWal->pWriteLogTFile);
+  taosCloseFile(&pWal->pWriteIdxTFile);
+
+  if (pWal->vers.firstVer != -1) {
+    int32_t fileSetSize = taosArrayGetSize(pWal->fileInfoSet);
+    for (int32_t i = 0; i < fileSetSize; i++) {
+      SWalFileInfo *pFileInfo = taosArrayGet(pWal->fileInfoSet, i);
+      char          fnameStr[WAL_FILE_LEN];
+      walBuildLogName(pWal, pFileInfo->firstVer, fnameStr);
+      taosRemoveFile(fnameStr);
+    }
+  }
+  walRemoveMeta(pWal);
+
+  pWal->writeCur = -1;
+  pWal->totSize = 0;
+  pWal->lastRollSeq = -1;
+
+  taosArrayClear(pWal->fileInfoSet);
+  pWal->vers.firstVer = -1;
   pWal->vers.lastVer = ver;
   pWal->vers.commitVer = ver - 1;
   pWal->vers.snapshotVer = ver - 1;
   pWal->vers.verInSnapshotting = -1;
+
+  taosThreadMutexUnlock(&pWal->mutex);
+  return 0;
 }
 
 int32_t walCommit(SWal *pWal, int64_t ver) {