diff --git a/cmake/cmake.define b/cmake/cmake.define
index 44b36d0efa..7710c071eb 100644
--- a/cmake/cmake.define
+++ b/cmake/cmake.define
@@ -97,7 +97,15 @@ ENDIF()
 SET(JEMALLOC_ENABLED OFF)
 IF (TD_WINDOWS)
     MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
-    SET(COMMON_FLAGS "/w /D_WIN32 /DWIN32 /Zi /MTd")
+    IF (${CMAKE_BUILD_TYPE} MATCHES "Release")
+      MESSAGE("${Green} will build Release version! ${ColourReset}")
+      SET(COMMON_FLAGS "/W3 /D_WIN32 /DWIN32 /Zi- /O2 /GL /MD")
+
+    ELSE ()
+      MESSAGE("${Green} will build Debug version! ${ColourReset}")
+      SET(COMMON_FLAGS "/w /D_WIN32 /DWIN32 /Zi /MTd")
+    ENDIF()
+
     SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} /MANIFEST:NO")
     # IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
     #  SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")
diff --git a/contrib/CMakeLists.txt b/contrib/CMakeLists.txt
index c5715bd53f..1b0a091e9d 100644
--- a/contrib/CMakeLists.txt
+++ b/contrib/CMakeLists.txt
@@ -317,7 +317,8 @@ if (${BUILD_WITH_ROCKSDB})
                 SET(CMAKE_BUILD_TYPE Release)
             endif()
         endif(${TD_LINUX})
-        MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
+        MESSAGE(STATUS "ROCKSDB CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
+        MESSAGE(STATUS "ROCKSDB C STATUS CONFIG: " ${CMAKE_C_FLAGS})
 
         if(${TD_DARWIN})
             SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized")
@@ -329,8 +330,12 @@ if (${BUILD_WITH_ROCKSDB})
 
         if (${TD_WINDOWS})
             SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819")
+            SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} /wd4244 /wd4819")
             option(WITH_JNI "" OFF)
-            option(WITH_MD_LIBRARY "build with MD" OFF)
+            if(CMAKE_C_FLAGS MATCHES "/MT" OR CMAKE_C_FLAGS MATCHES "/MTd")
+               message("Rocksdb build runtime lib use /MT or /MTd")
+               option(WITH_MD_LIBRARY "build with MD" OFF)
+            endif()
             set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib)
         endif(${TD_WINDOWS})
 
@@ -361,9 +366,11 @@ if (${BUILD_WITH_ROCKSDB})
         )
     else() 
         if (NOT ${TD_LINUX})
-            MESSAGE(STATUS "CXXXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
+            MESSAGE(STATUS "ROCKSDB CXX STATUS CONFIG: " ${CMAKE_CXX_FLAGS})
+            MESSAGE(STATUS "ROCKSDB C STATUS CONFIG: " ${CMAKE_C_FLAGS})
             if(${TD_DARWIN})
                 SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-error=maybe-uninitialized")
+                SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Wno-error=maybe-uninitialized")
             endif(${TD_DARWIN})
 
             if (${TD_DARWIN_ARM64})
@@ -372,8 +379,12 @@ if (${BUILD_WITH_ROCKSDB})
 
             if (${TD_WINDOWS})
                 SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd4244 /wd4819")
+                SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} /wd4244 /wd4819")
                 option(WITH_JNI "" OFF)
-                option(WITH_MD_LIBRARY "build with MD" OFF)
+                if(CMAKE_C_FLAGS MATCHES "/MT" OR CMAKE_C_FLAGS MATCHES "/MTd")
+                    message("Rocksdb build runtime lib use /MT or /MTd")
+                    option(WITH_MD_LIBRARY "build with MD" OFF)
+                endif()
                 set(SYSTEM_LIBS ${SYSTEM_LIBS} shlwapi.lib rpcrt4.lib)
             endif(${TD_WINDOWS})
 
@@ -456,7 +467,9 @@ endif(${BUILD_WITH_NURAFT})
 
 # pthread
 if(${BUILD_PTHREAD})
-    set(CMAKE_BUILD_TYPE debug)
+    if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
+    SET(CMAKE_BUILD_TYPE Release)
+    endif()
     add_definitions(-DPTW32_STATIC_LIB)
     add_subdirectory(pthread EXCLUDE_FROM_ALL)
     set_target_properties(libpthreadVC3 PROPERTIES OUTPUT_NAME pthread)
@@ -640,13 +653,18 @@ if(${BUILD_GEOS})
     if(${TD_LINUX})
         set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS_REL}")
         set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS_REL}")
-        IF ("${CMAKE_BUILD_TYPE}" STREQUAL "")
+        if ("${CMAKE_BUILD_TYPE}" STREQUAL "")
             SET(CMAKE_BUILD_TYPE Release)
         endif()
     endif(${TD_LINUX})
     option(BUILD_SHARED_LIBS "Build GEOS with shared libraries" OFF)
     add_subdirectory(geos EXCLUDE_FROM_ALL)
-    unset(CMAKE_CXX_STANDARD CACHE) # undo libgeos's setting of global CMAKE_CXX_STANDARD
+    if (${TD_WINDOWS})
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS}")
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}")
+    else ()
+        unset(CMAKE_CXX_STANDARD CACHE) # undo libgeos's setting of global CMAKE_CXX_STANDARD
+    endif(${TD_WINDOWS})
     target_include_directories(
         geos_c
         PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/geos/include>
diff --git a/include/common/tmsg.h b/include/common/tmsg.h
index 86d34502c6..e17a72992c 100644
--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@@ -3112,7 +3112,7 @@ typedef struct {
 
 int32_t tSerializeSMDropStreamReq(void* buf, int32_t bufLen, const SMDropStreamReq* pReq);
 int32_t tDeserializeSMDropStreamReq(void* buf, int32_t bufLen, SMDropStreamReq* pReq);
-void    tFreeSMDropStreamReq(SMDropStreamReq* pReq);
+void    tFreeMDropStreamReq(SMDropStreamReq* pReq);
 
 typedef struct {
   char   name[TSDB_STREAM_FNAME_LEN];
diff --git a/include/libs/executor/storageapi.h b/include/libs/executor/storageapi.h
index d5f1da957d..045f2bad70 100644
--- a/include/libs/executor/storageapi.h
+++ b/include/libs/executor/storageapi.h
@@ -150,19 +150,6 @@ typedef struct {
   int32_t colNum;
 } SMetaStbStats;
 
-// void    tqReaderSetColIdList(STqReader *pReader, SArray *pColIdList);
-// int32_t tqReaderSetTbUidList(STqReader *pReader, const SArray *tbUidList);
-// int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *pTableUidList);
-// int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
-// bool    tqReaderIsQueriedTable(STqReader* pReader, uint64_t uid);
-// bool    tqCurrentBlockConsumed(const STqReader* pReader);
-// int32_t tqReaderSeek(STqReader *pReader, int64_t ver, const char *id);
-// bool    tqNextBlockInWal(STqReader* pReader, const char* idstr);
-// bool    tqNextBlockImpl(STqReader *pReader, const char* idstr);
-// int32_t        getTableInfoFromSnapshot(SSnapContext *ctx, void **pBuf, int32_t *contLen, int16_t *type, int64_t
-// *uid); SMetaTableInfo getMetaTableInfoFromSnapshot(SSnapContext *ctx); int32_t        setForSnapShot(SSnapContext
-// *ctx, int64_t uid); int32_t        destroySnapContext(SSnapContext *ctx);
-
 // clang-format off
 /*-------------------------------------------------new api format---------------------------------------------------*/
 typedef struct TsdReader {
@@ -197,27 +184,6 @@ typedef struct SStoreCacheReader {
 // clang-format on
 
 /*------------------------------------------------------------------------------------------------------------------*/
-/*
-void    tqReaderSetColIdList(STqReader *pReader, SArray *pColIdList);
-int32_t tqReaderSetTbUidList(STqReader *pReader, const SArray *tbUidList);
-int32_t tqReaderAddTbUidList(STqReader *pReader, const SArray *pTableUidList);
-int32_t tqReaderRemoveTbUidList(STqReader *pReader, const SArray *tbUidList);
-bool    tqReaderIsQueriedTable(STqReader* pReader, uint64_t uid);
-bool    tqCurrentBlockConsumed(const STqReader* pReader);
-
-int32_t tqReaderSeek(STqReader *pReader, int64_t ver, const char *id);
-bool    tqNextBlockInWal(STqReader* pReader, const char* idstr);
-bool    tqNextBlockImpl(STqReader *pReader, const char* idstr);
-
- int32_t    tqRetrieveDataBlock(STqReader *pReader, SSDataBlock **pRes, const char* idstr);
-STqReader *tqReaderOpen(void *pVnode);
-void       tqReaderClose(STqReader *);
-
-int32_t tqReaderSetSubmitMsg(STqReader *pReader, void *msgStr, int32_t msgLen, int64_t ver);
-bool    tqNextDataBlockFilterOut(STqReader *pReader, SHashObj *filterOutUids);
-SWalReader* tqGetWalReader(STqReader* pReader);
-int32_t tqRetrieveTaosxBlock(STqReader *pReader, SArray *blocks, SArray *schemas, SSubmitTbData **pSubmitTbDataRet);
-*/
 // todo rename
 typedef struct SStoreTqReader {
   struct STqReader* (*tqReaderOpen)();
@@ -281,28 +247,18 @@ typedef struct SStoreMeta {
 
   void* (*storeGetIndexInfo)();
   void* (*getInvertIndex)(void* pVnode);
-  int32_t (*getChildTableList)(
-      void* pVnode, int64_t suid,
-      SArray* list);  // support filter and non-filter cases. [vnodeGetCtbIdList & vnodeGetCtbIdListByFilter]
-  int32_t (*storeGetTableList)(void* pVnode, int8_t type, SArray* pList);  // vnodeGetStbIdList  & vnodeGetAllTableList
+  // support filter and non-filter cases. [vnodeGetCtbIdList & vnodeGetCtbIdListByFilter]
+  int32_t (*getChildTableList)( void* pVnode, int64_t suid, SArray* list);
+  int32_t (*storeGetTableList)(void* pVnode, int8_t type, SArray* pList);
   void* storeGetVersionRange;
   void* storeGetLastTimestamp;
 
   int32_t (*getTableSchema)(void* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid);  // tsdbGetTableSchema
+  int32_t (*getNumOfChildTables)( void* pVnode, int64_t uid, int64_t* numOfTables, int32_t* numOfCols);
+  void (*getBasicInfo)(void* pVnode, const char** dbname, int32_t* vgId, int64_t* numOfTables, int64_t* numOfNormalTables);
 
-  // db name, vgId, numOfTables, numOfSTables
-  int32_t (*getNumOfChildTables)(
-      void* pVnode, int64_t uid, int64_t* numOfTables,
-      int32_t* numOfCols);  // int32_t metaGetStbStats(SMeta *pMeta, int64_t uid, SMetaStbStats *pInfo);
-  void (*getBasicInfo)(void* pVnode, const char** dbname, int32_t* vgId, int64_t* numOfTables,
-                       int64_t* numOfNormalTables);  // vnodeGetInfo(void *pVnode, const char **dbname, int32_t *vgId) &
-                                                     // metaGetTbNum(SMeta *pMeta) & metaGetNtbNum(SMeta *pMeta);
   int64_t (*getNumOfRowsInMem)(void* pVnode);
-  /**
-int32_t vnodeGetCtbIdList(void *pVnode, int64_t suid, SArray *list);
-int32_t vnodeGetCtbIdListByFilter(void *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg), void *arg);
-int32_t vnodeGetStbIdList(void *pVnode, int64_t suid, SArray *list);
- */
+
   SMCtbCursor* (*openCtbCursor)(void *pVnode, tb_uid_t uid, int lock);
   int32_t      (*resumeCtbCursor)(SMCtbCursor* pCtbCur, int8_t first);
   void         (*pauseCtbCursor)(SMCtbCursor* pCtbCur);
diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
index 31837fbdf5..ad2e9514d1 100644
--- a/include/libs/stream/tstream.h
+++ b/include/libs/stream/tstream.h
@@ -304,6 +304,7 @@ typedef struct SCheckpointInfo {
   int64_t processedVer;    // already processed ver, that has generated results version.
   int64_t nextProcessVer;  // current offset in WAL, not serialize it
   int64_t failedId;        // record the latest failed checkpoint id
+  bool    dispatchCheckpointTrigger;
 } SCheckpointInfo;
 
 typedef struct SStreamStatus {
@@ -757,7 +758,8 @@ void initRpcMsg(SRpcMsg* pMsg, int32_t msgType, void* pCont, int32_t contLen);
 // recover and fill history
 void streamTaskCheckDownstream(SStreamTask* pTask);
 
-int32_t streamTaskCheckStatus(SStreamTask* pTask, int32_t upstreamTaskId, int32_t vgId, int64_t stage, int64_t* oldStage);
+int32_t streamTaskCheckStatus(SStreamTask* pTask, int32_t upstreamTaskId, int32_t vgId, int64_t stage,
+                              int64_t* oldStage);
 int32_t streamTaskUpdateEpsetInfo(SStreamTask* pTask, SArray* pNodeList);
 void    streamTaskResetUpstreamStageInfo(SStreamTask* pTask);
 bool    streamTaskAllUpstreamClosed(SStreamTask* pTask);
@@ -786,7 +788,6 @@ int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue);
 int32_t streamRestoreParam(SStreamTask* pTask);
 void    streamTaskPause(SStreamTask* pTask, SStreamMeta* pMeta);
 void    streamTaskResume(SStreamTask* pTask);
-void    streamTaskEnablePause(SStreamTask* pTask);
 int32_t streamTaskSetUpstreamInfo(SStreamTask* pTask, const SStreamTask* pUpstreamTask);
 void    streamTaskUpdateUpstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet);
 void    streamTaskUpdateDownstreamInfo(SStreamTask* pTask, int32_t nodeId, const SEpSet* pEpSet);
diff --git a/include/os/osFile.h b/include/os/osFile.h
index 63483dc906..e409936468 100644
--- a/include/os/osFile.h
+++ b/include/os/osFile.h
@@ -54,15 +54,16 @@ extern "C" {
 
 typedef struct TdFile *TdFilePtr;
 
-#define TD_FILE_CREATE   0x0001
-#define TD_FILE_WRITE    0x0002
-#define TD_FILE_READ     0x0004
-#define TD_FILE_TRUNC    0x0008
-#define TD_FILE_APPEND   0x0010
-#define TD_FILE_TEXT     0x0020
-#define TD_FILE_AUTO_DEL 0x0040
-#define TD_FILE_EXCL     0x0080
-#define TD_FILE_STREAM   0x0100  // Only support taosFprintfFile, taosGetLineFile, taosEOFFile
+#define TD_FILE_CREATE        0x0001
+#define TD_FILE_WRITE         0x0002
+#define TD_FILE_READ          0x0004
+#define TD_FILE_TRUNC         0x0008
+#define TD_FILE_APPEND        0x0010
+#define TD_FILE_TEXT          0x0020
+#define TD_FILE_AUTO_DEL      0x0040
+#define TD_FILE_EXCL          0x0080
+#define TD_FILE_STREAM        0x0100  // Only support taosFprintfFile, taosGetLineFile, taosEOFFile
+#define TD_FILE_WRITE_THROUGH 0x0200
 TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions);
 TdFilePtr taosCreateFile(const char *path, int32_t tdFileOptions);
 
diff --git a/source/common/src/tdataformat.c b/source/common/src/tdataformat.c
index d220da0d84..f97f9c0c11 100644
--- a/source/common/src/tdataformat.c
+++ b/source/common/src/tdataformat.c
@@ -3590,9 +3590,9 @@ void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_
     tColDataCalcSMAUInt,           // TSDB_DATA_TYPE_UINT
     tColDataCalcSMAUBigInt,        // TSDB_DATA_TYPE_UBIGINT
     tColDataCalcSMAVarType,        // TSDB_DATA_TYPE_JSON
-    NULL,                          // TSDB_DATA_TYPE_VARBINARY
-    NULL,                          // TSDB_DATA_TYPE_DECIMAL
-    NULL,                          // TSDB_DATA_TYPE_BLOB
+    tColDataCalcSMAVarType,        // TSDB_DATA_TYPE_VARBINARY
+    tColDataCalcSMAVarType,        // TSDB_DATA_TYPE_DECIMAL
+    tColDataCalcSMAVarType,        // TSDB_DATA_TYPE_BLOB
     NULL,                          // TSDB_DATA_TYPE_MEDIUMBLOB
     tColDataCalcSMAVarType         // TSDB_DATA_TYPE_GEOMETRY
 };
diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index 64ed02f369..d9aecd96d7 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -354,11 +354,16 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *input
   char cfgFile[PATH_MAX + 100] = {0};
 
   taosExpandDir(inputCfgDir, cfgDir, PATH_MAX);
+  char lastC = cfgDir[strlen(cfgDir) - 1];
+  char *tdDirsep = TD_DIRSEP;
+  if (lastC == '\\' || lastC == '/') {
+    tdDirsep = "";
+  }
   if (taosIsDir(cfgDir)) {
 #ifdef CUS_PROMPT
-    snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "%s.cfg", cfgDir, CUS_PROMPT);
+    snprintf(cfgFile, sizeof(cfgFile), "%s" "%s" "%s.cfg", cfgDir, tdDirsep, CUS_PROMPT);
 #else
-    snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "taos.cfg", cfgDir);
+    snprintf(cfgFile, sizeof(cfgFile), "%s" "%s" "taos.cfg", cfgDir, tdDirsep);
 #endif
   } else {
     tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
@@ -508,6 +513,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
 
   tsNumOfTaskQueueThreads = tsNumOfCores / 2;
   tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4);
+
   if (tsNumOfTaskQueueThreads >= 50) {
     tsNumOfTaskQueueThreads = 50;
   }
@@ -1738,4 +1744,4 @@ void taosSetAllDebugFlag(int32_t flag, bool rewrite) {
   uInfo("all debug flag are set to %d", flag);
 }
 
-int8_t taosGranted() { return atomic_load_8(&tsGrant); }
+int8_t taosGranted() { return atomic_load_8(&tsGrant); }
\ No newline at end of file
diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c
index dc3ba7934f..01b1df9d5f 100644
--- a/source/common/src/tmsg.c
+++ b/source/common/src/tmsg.c
@@ -7152,7 +7152,7 @@ int32_t tDeserializeSMDropStreamReq(void *buf, int32_t bufLen, SMDropStreamReq *
   return 0;
 }
 
-void tFreeSMDropStreamReq(SMDropStreamReq *pReq) {
+void tFreeMDropStreamReq(SMDropStreamReq *pReq) {
   FREESQL();
 }
 
diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmFile.c b/source/dnode/mgmt/mgmt_mnode/src/mmFile.c
index 64e18ef06d..27baa5ede5 100644
--- a/source/dnode/mgmt/mgmt_mnode/src/mmFile.c
+++ b/source/dnode/mgmt/mgmt_mnode/src/mmFile.c
@@ -169,7 +169,7 @@ int32_t mmWriteFile(const char *path, const SMnodeOpt *pOption) {
   if (buffer == NULL) goto _OVER;
   terrno = 0;
 
-  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) goto _OVER;
 
   int32_t len = strlen(buffer);
diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c
index ed32e75d18..53139330a3 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmFile.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmFile.c
@@ -200,7 +200,7 @@ int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
   if (buffer == NULL) goto _OVER;
   terrno = 0;
 
-  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) goto _OVER;
 
   int32_t len = strlen(buffer);
diff --git a/source/dnode/mgmt/node_util/src/dmEps.c b/source/dnode/mgmt/node_util/src/dmEps.c
index 3e948678a4..bee77528bd 100644
--- a/source/dnode/mgmt/node_util/src/dmEps.c
+++ b/source/dnode/mgmt/node_util/src/dmEps.c
@@ -232,7 +232,7 @@ int32_t dmWriteEps(SDnodeData *pData) {
   if (buffer == NULL) goto _OVER;
   terrno = 0;
 
-  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) goto _OVER;
 
   int32_t len = strlen(buffer);
diff --git a/source/dnode/mgmt/node_util/src/dmFile.c b/source/dnode/mgmt/node_util/src/dmFile.c
index c81efddcc1..03c6734e0c 100644
--- a/source/dnode/mgmt/node_util/src/dmFile.c
+++ b/source/dnode/mgmt/node_util/src/dmFile.c
@@ -120,7 +120,7 @@ int32_t dmWriteFile(const char *path, const char *name, bool deployed) {
   if (buffer == NULL) goto _OVER;
   terrno = 0;
 
-  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) goto _OVER;
 
   int32_t len = strlen(buffer);
diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h
index fc9086eebf..27fef1e81e 100644
--- a/source/dnode/mnode/impl/inc/mndDef.h
+++ b/source/dnode/mnode/impl/inc/mndDef.h
@@ -651,8 +651,7 @@ typedef struct SStreamConf {
 } SStreamConf;
 
 typedef struct {
-  char name[TSDB_STREAM_FNAME_LEN];
-  // ctl
+  char     name[TSDB_STREAM_FNAME_LEN];
   SRWLatch lock;
 
   // create info
diff --git a/source/dnode/mnode/impl/inc/mndStream.h b/source/dnode/mnode/impl/inc/mndStream.h
index 19fd2a3fd4..244a6d08dd 100644
--- a/source/dnode/mnode/impl/inc/mndStream.h
+++ b/source/dnode/mnode/impl/inc/mndStream.h
@@ -22,17 +22,37 @@
 extern "C" {
 #endif
 
-int32_t mndInitStream(SMnode *pMnode);
-void    mndCleanupStream(SMnode *pMnode);
+typedef struct SStreamTransInfo {
+  int64_t     startTime;
+  int32_t     transId;
+  const char *name;
+} SStreamTransInfo;
 
+typedef struct SStreamTransMgmt {
+  SHashObj *pDBTrans;
+} SStreamTransMgmt;
+
+typedef struct SStreamExecInfo {
+  SArray       *pNodeList;
+  int64_t       ts;                // snapshot ts
+  SStreamTransMgmt  transMgmt;
+  int64_t       activeCheckpoint;  // active check point id
+  SHashObj *    pTaskMap;
+  SArray *      pTaskList;
+  TdThreadMutex lock;
+} SStreamExecInfo;
+
+extern SStreamExecInfo execInfo;
+
+int32_t     mndInitStream(SMnode *pMnode);
+void        mndCleanupStream(SMnode *pMnode);
 SStreamObj *mndAcquireStream(SMnode *pMnode, char *streamName);
 void        mndReleaseStream(SMnode *pMnode, SStreamObj *pStream);
+int32_t     mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
+int32_t     mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
 
-SSdbRaw *mndStreamActionEncode(SStreamObj *pStream);
-SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw);
-
-int32_t mndDropStreamByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb);
-int32_t mndPersistStream(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream);
+int32_t mndStreamRegisterTrans(STrans* pTrans, const char* pName, const char* pSrcDb, const char* pDstDb);
+bool    streamTransConflictOtherTrans(SMnode *pMnode, const char *pSrcDb, const char *pDstDb);
 
 // for sma
 // TODO refactor
diff --git a/source/dnode/mnode/impl/src/mndDump.c b/source/dnode/mnode/impl/src/mndDump.c
index 481495cbe5..5efebbc16e 100644
--- a/source/dnode/mnode/impl/src/mndDump.c
+++ b/source/dnode/mnode/impl/src/mndDump.c
@@ -605,7 +605,7 @@ void mndDumpSdb() {
   char     *pCont = tjsonToString(json);
   int32_t   contLen = strlen(pCont);
   char      file[] = "sdb.json";
-  TdFilePtr pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  TdFilePtr pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC| TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
     mError("failed to write %s since %s", file, terrstr());
diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c
index 88a06cb513..3ef4c9a4d2 100644
--- a/source/dnode/mnode/impl/src/mndScheduler.c
+++ b/source/dnode/mnode/impl/src/mndScheduler.c
@@ -28,7 +28,7 @@
 extern bool tsDeployOnSnode;
 
 static int32_t doAddSinkTask(SStreamObj* pStream, SArray* pTaskList, SMnode* pMnode, int32_t vgId, SVgObj* pVgroup,
-                                  SEpSet* pEpset, bool isFillhistory);
+                             SEpSet* pEpset, bool isFillhistory);
 
 int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64_t uid, int8_t triggerType,
                            int64_t watermark, int64_t deleteMark) {
diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c
index 94c4a03277..d2f0a13038 100644
--- a/source/dnode/mnode/impl/src/mndStream.c
+++ b/source/dnode/mnode/impl/src/mndStream.c
@@ -34,7 +34,13 @@
 #define MND_STREAM_VER_NUMBER      4
 #define MND_STREAM_RESERVE_SIZE    64
 #define MND_STREAM_MAX_NUM         60
-#define MND_STREAM_CHECKPOINT_NAME "stream-checkpoint"
+
+#define MND_STREAM_CHECKPOINT_NAME  "stream-checkpoint"
+#define MND_STREAM_PAUSE_NAME       "stream-pause"
+#define MND_STREAM_RESUME_NAME      "stream-resume"
+#define MND_STREAM_DROP_NAME        "stream-drop"
+#define MND_STREAM_TASK_RESET_NAME  "stream-task-reset"
+#define MND_STREAM_TASK_UPDATE_NAME "stream-task-update"
 
 typedef struct SNodeEntry {
   int32_t nodeId;
@@ -43,22 +49,13 @@ typedef struct SNodeEntry {
   int64_t hbTimestamp;   // second
 } SNodeEntry;
 
-typedef struct SStreamExecInfo {
-  SArray       *pNodeList;
-  int64_t       ts;                // snapshot ts
-  int64_t       activeCheckpoint;  // active check point id
-  SHashObj *    pTaskMap;
-  SArray *      pTaskList;
-  TdThreadMutex lock;
-} SStreamExecInfo;
-
 typedef struct SVgroupChangeInfo {
   SHashObj *pDBMap;
   SArray *  pUpdateNodeList;  // SArray<SNodeUpdateInfo>
 } SVgroupChangeInfo;
 
-static int32_t         mndNodeCheckSentinel = 0;
-static SStreamExecInfo execInfo;
+static int32_t  mndNodeCheckSentinel = 0;
+SStreamExecInfo execInfo;
 
 static int32_t mndStreamActionInsert(SSdb *pSdb, SStreamObj *pStream);
 static int32_t mndStreamActionDelete(SSdb *pSdb, SStreamObj *pStream);
@@ -83,17 +80,20 @@ static SArray *mndTakeVgroupSnapshot(SMnode *pMnode, bool *allReady);
 
 static SVgroupChangeInfo mndFindChangedNodeInfo(SMnode *pMnode, const SArray *pPrevNodeList, const SArray *pNodeList);
 
-static STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, const char *name);
+static STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, SRpcMsg *pReq, const char *name, const char* pMsg);
 static int32_t mndPersistTransLog(SStreamObj *pStream, STrans *pTrans);
 static void initTransAction(STransAction *pAction, void *pCont, int32_t contLen, int32_t msgType, const SEpSet *pEpset,
                             int32_t retryCode);
 static int32_t createStreamUpdateTrans(SStreamObj *pStream, SVgroupChangeInfo *pInfo, STrans *pTrans);
 static void    removeStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode);
-static void    keepStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode);
+static void    saveStreamTasksInfo(SStreamObj *pStream, SStreamExecInfo *pExecNode);
 static int32_t removeExpirednodeEntryAndTask(SArray *pNodeSnapshot);
-static int32_t doKillActiveCheckpointTrans(SMnode *pMnode);
+static int32_t killActiveCheckpointTrans(SMnode *pMnode, const char* pDbName, size_t len);
 static int32_t setNodeEpsetExpiredFlag(const SArray *pNodeList);
 
+static SSdbRaw *mndStreamActionEncode(SStreamObj *pStream);
+static SSdbRow *mndStreamActionDecode(SSdbRaw *pRaw);
+
 int32_t mndInitStream(SMnode *pMnode) {
   SSdbTable table = {
       .sdbType = SDB_STREAM,
@@ -133,8 +133,11 @@ int32_t mndInitStream(SMnode *pMnode) {
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STREAM_TASKS, mndCancelGetNextStreamTask);
 
   taosThreadMutexInit(&execInfo.lock, NULL);
-  execInfo.pTaskMap = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK);
+  _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR);
+
   execInfo.pTaskList = taosArrayInit(4, sizeof(STaskId));
+  execInfo.pTaskMap = taosHashInit(64, fn, true, HASH_NO_LOCK);
+  execInfo.transMgmt.pDBTrans = taosHashInit(32, fn, true, HASH_NO_LOCK);
 
   return sdbSetTable(pMnode->pSdb, table);
 }
@@ -142,6 +145,7 @@ int32_t mndInitStream(SMnode *pMnode) {
 void mndCleanupStream(SMnode *pMnode) {
   taosArrayDestroy(execInfo.pTaskList);
   taosHashCleanup(execInfo.pTaskMap);
+  taosHashCleanup(execInfo.transMgmt.pDBTrans);
   taosThreadMutexDestroy(&execInfo.lock);
   mDebug("mnd stream exec info cleanup");
 }
@@ -335,7 +339,7 @@ static int32_t mndStreamGetPlanString(const char *ast, int8_t triggerType, int64
         .pAstRoot = pAst,
         .topicQuery = false,
         .streamQuery = true,
-        .triggerType = triggerType == STREAM_TRIGGER_MAX_DELAY ? STREAM_TRIGGER_WINDOW_CLOSE : triggerType,
+        .triggerType = (triggerType == STREAM_TRIGGER_MAX_DELAY) ? STREAM_TRIGGER_WINDOW_CLOSE : triggerType,
         .watermark = watermark,
     };
     code = qCreateQueryPlan(&cxt, &pPlan, NULL);
@@ -720,6 +724,34 @@ int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream)
   return 0;
 }
 
+static int32_t checkForNumOfStreams(SMnode *pMnode, SStreamObj *pStreamObj) {  // check for number of existed tasks
+  int32_t     numOfStream = 0;
+  SStreamObj *pStream = NULL;
+  void       *pIter = NULL;
+
+  while ((pIter = sdbFetch(pMnode->pSdb, SDB_STREAM, pIter, (void **)&pStream)) != NULL) {
+    if (pStream->sourceDbUid == pStreamObj->sourceDbUid) {
+      ++numOfStream;
+    }
+
+    sdbRelease(pMnode->pSdb, pStream);
+
+    if (numOfStream > MND_STREAM_MAX_NUM) {
+      mError("too many streams, no more than %d for each database", MND_STREAM_MAX_NUM);
+      sdbCancelFetch(pMnode->pSdb, pIter);
+      return TSDB_CODE_MND_TOO_MANY_STREAMS;
+    }
+
+    if (pStream->targetStbUid == pStreamObj->targetStbUid) {
+      mError("Cannot write the same stable as other stream:%s", pStream->name);
+      sdbCancelFetch(pMnode->pSdb, pIter);
+      return TSDB_CODE_MND_INVALID_TARGET_TABLE;
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
   SMnode *           pMnode = pReq->info.node;
   int32_t            code = -1;
@@ -732,6 +764,7 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
     terrno = TSDB_CODE_INVALID_MSG;
     goto _OVER;
   }
+
 #ifdef WINDOWS
   terrno = TSDB_CODE_MND_INVALID_PLATFORM;
   goto _OVER;
@@ -772,42 +805,9 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
-  {
-    int32_t numOfStream = 0;
-
-    SStreamObj *pStream = NULL;
-    void *      pIter = NULL;
-
-    while (1) {
-      pIter = sdbFetch(pMnode->pSdb, SDB_STREAM, pIter, (void **)&pStream);
-      if (pIter == NULL) {
-        if (numOfStream > MND_STREAM_MAX_NUM) {
-          mError("too many streams, no more than %d for each database", MND_STREAM_MAX_NUM);
-          terrno = TSDB_CODE_MND_TOO_MANY_STREAMS;
-          goto _OVER;
-        }
-        break;
-      }
-
-      if (pStream->sourceDbUid == streamObj.sourceDbUid) {
-        ++numOfStream;
-      }
-
-      sdbRelease(pMnode->pSdb, pStream);
-      if (numOfStream > MND_STREAM_MAX_NUM) {
-        mError("too many streams, no more than %d for each database", MND_STREAM_MAX_NUM);
-        terrno = TSDB_CODE_MND_TOO_MANY_STREAMS;
-        sdbCancelFetch(pMnode->pSdb, pIter);
-        goto _OVER;
-      }
-
-      if (pStream->targetStbUid == streamObj.targetStbUid) {
-        mError("Cannot write the same stable as other stream:%s", pStream->name);
-        terrno = TSDB_CODE_MND_INVALID_TARGET_TABLE;
-        sdbCancelFetch(pMnode->pSdb, pIter);
-        goto _OVER;
-      }
-    }
+  code = checkForNumOfStreams(pMnode, &streamObj);
+  if (code != TSDB_CODE_SUCCESS) {
+    goto _OVER;
   }
 
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "create-stream");
@@ -866,7 +866,7 @@ static int32_t mndProcessCreateStreamReq(SRpcMsg *pReq) {
 
   taosThreadMutexLock(&execInfo.lock);
   mDebug("stream tasks register into node list");
-  keepStreamTasksInBuf(&streamObj, &execInfo);
+  saveStreamTasksInfo(&streamObj, &execInfo);
   taosThreadMutexUnlock(&execInfo.lock);
 
   code = TSDB_CODE_ACTION_IN_PROGRESS;
@@ -893,7 +893,6 @@ _OVER:
   }
 
   mndReleaseStream(pMnode, pStream);
-
   tFreeSCMCreateStreamReq(&createStreamReq);
   tFreeStreamObj(&streamObj);
   if(sql != NULL){
@@ -1268,7 +1267,7 @@ static int32_t mndProcessStreamDoCheckpoint(SRpcMsg *pReq) {
   SMStreamDoCheckpointMsg *pMsg = (SMStreamDoCheckpointMsg *)pReq->pCont;
   int64_t                  checkpointId = pMsg->checkpointId;
 
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, NULL, MND_STREAM_CHECKPOINT_NAME);
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, NULL, MND_STREAM_CHECKPOINT_NAME);
   if (pTrans == NULL) {
     mError("failed to trigger checkpoint, reason: %s", tstrerror(TSDB_CODE_OUT_OF_MEMORY));
     return -1;
@@ -1277,7 +1276,8 @@ static int32_t mndProcessStreamDoCheckpoint(SRpcMsg *pReq) {
   mDebug("start to trigger checkpoint, checkpointId: %" PRId64, checkpointId);
 
   const char *pDb = mndGetStreamDB(pMnode);
-  mndTransSetDbName(pTrans, pDb, "checkpoint");
+  mndTransSetDbName(pTrans, pDb, pDb);
+  mndStreamRegisterTrans(pTrans, MND_STREAM_CHECKPOINT_NAME, pDb, pDb);
   taosMemoryFree((void *)pDb);
 
   if (mndTransCheckConflict(pMnode, pTrans) != 0) {
@@ -1329,46 +1329,56 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
     if (dropReq.igNotExists) {
       mInfo("stream:%s not exist, ignore not exist is set", dropReq.name);
       sdbRelease(pMnode->pSdb, pStream);
-      tFreeSMDropStreamReq(&dropReq);
+      tFreeMDropStreamReq(&dropReq);
       return 0;
     } else {
       terrno = TSDB_CODE_MND_STREAM_NOT_EXIST;
       mError("stream:%s not exist failed to drop", dropReq.name);
-      tFreeSMDropStreamReq(&dropReq);
+      tFreeMDropStreamReq(&dropReq);
       return -1;
     }
   }
 
   if (mndCheckDbPrivilegeByName(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pStream->targetDb) != 0) {
     sdbRelease(pMnode->pSdb, pStream);
-    tFreeSMDropStreamReq(&dropReq);
+    tFreeMDropStreamReq(&dropReq);
     return -1;
   }
 
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "drop-stream");
+  // check if it is conflict with other trans in both sourceDb and targetDb.
+  bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
+  if (conflict) {
+    sdbRelease(pMnode->pSdb, pStream);
+    tFreeMDropStreamReq(&dropReq);
+    return -1;
+  }
+
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, MND_STREAM_DROP_NAME);
   if (pTrans == NULL) {
     mError("stream:%s, failed to drop since %s", dropReq.name, terrstr());
     sdbRelease(pMnode->pSdb, pStream);
-    tFreeSMDropStreamReq(&dropReq);
+    tFreeMDropStreamReq(&dropReq);
     return -1;
   }
 
-  mInfo("trans:%d, used to drop stream:%s", pTrans->id, dropReq.name);
+  mInfo("trans:%d used to drop stream:%s", pTrans->id, dropReq.name);
 
   mndTransSetDbName(pTrans, pStream->sourceDb, pStream->targetDb);
   if (mndTransCheckConflict(pMnode, pTrans) != 0) {
     sdbRelease(pMnode->pSdb, pStream);
     mndTransDrop(pTrans);
-    tFreeSMDropStreamReq(&dropReq);
+    tFreeMDropStreamReq(&dropReq);
     return -1;
   }
 
+  int32_t code = mndStreamRegisterTrans(pTrans, MND_STREAM_DROP_NAME, pStream->sourceDb, pStream->targetDb);
+
   // drop all tasks
   if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
     mError("stream:%s, failed to drop task since %s", dropReq.name, terrstr());
     sdbRelease(pMnode->pSdb, pStream);
     mndTransDrop(pTrans);
-    tFreeSMDropStreamReq(&dropReq);
+    tFreeMDropStreamReq(&dropReq);
     return -1;
   }
 
@@ -1376,7 +1386,7 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
   if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
     sdbRelease(pMnode->pSdb, pStream);
     mndTransDrop(pTrans);
-    tFreeSMDropStreamReq(&dropReq);
+    tFreeMDropStreamReq(&dropReq);
     return -1;
   }
 
@@ -1384,7 +1394,7 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
     mError("trans:%d, failed to prepare drop stream trans since %s", pTrans->id, terrstr());
     sdbRelease(pMnode->pSdb, pStream);
     mndTransDrop(pTrans);
-    tFreeSMDropStreamReq(&dropReq);
+    tFreeMDropStreamReq(&dropReq);
     return -1;
   }
 
@@ -1392,13 +1402,12 @@ static int32_t mndProcessDropStreamReq(SRpcMsg *pReq) {
 
   SName name = {0};
   tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
-  // reuse this function for stream
 
   auditRecord(pReq, pMnode->clusterId, "dropStream", "", name.dbname, dropReq.sql, dropReq.sqlLen);
 
   sdbRelease(pMnode->pSdb, pStream);
   mndTransDrop(pTrans);
-  tFreeSMDropStreamReq(&dropReq);
+  tFreeMDropStreamReq(&dropReq);
 
   return TSDB_CODE_ACTION_IN_PROGRESS;
 }
@@ -1814,6 +1823,13 @@ static int32_t mndProcessPauseStreamReq(SRpcMsg *pReq) {
     return -1;
   }
 
+  // check if it is conflict with other trans in both sourceDb and targetDb.
+  bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
+  if (conflict) {
+    sdbRelease(pMnode->pSdb, pStream);
+    return -1;
+  }
+
   bool updated = taskNodeIsUpdated(pMnode);
   if (updated) {
     mError("tasks are not ready for pause, node update detected");
@@ -1822,7 +1838,7 @@ static int32_t mndProcessPauseStreamReq(SRpcMsg *pReq) {
 
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "pause-stream");
   if (pTrans == NULL) {
-    mError("stream:%s, failed to pause stream since %s", pauseReq.name, terrstr());
+    mError("stream:%s failed to pause stream since %s", pauseReq.name, terrstr());
     sdbRelease(pMnode->pSdb, pStream);
     return -1;
   }
@@ -1836,7 +1852,9 @@ static int32_t mndProcessPauseStreamReq(SRpcMsg *pReq) {
     return -1;
   }
 
-  // pause all tasks
+  int32_t code = mndStreamRegisterTrans(pTrans, MND_STREAM_PAUSE_NAME, pStream->sourceDb, pStream->targetDb);
+
+  // if nodeUpdate happened, not send pause trans
   if (mndPauseAllStreamTasks(pMnode, pTrans, pStream) < 0) {
     mError("stream:%s, failed to pause task since %s", pauseReq.name, terrstr());
     sdbRelease(pMnode->pSdb, pStream);
@@ -1940,13 +1958,21 @@ static int32_t mndProcessResumeStreamReq(SRpcMsg *pReq) {
     return -1;
   }
 
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "pause-stream");
-  if (pTrans == NULL) {
-    mError("stream:%s, failed to pause stream since %s", pauseReq.name, terrstr());
+  // check if it is conflict with other trans in both sourceDb and targetDb.
+  bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
+  if (conflict) {
     sdbRelease(pMnode->pSdb, pStream);
     return -1;
   }
-  mInfo("trans:%d, used to pause stream:%s", pTrans->id, pauseReq.name);
+
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, MND_STREAM_RESUME_NAME);
+  if (pTrans == NULL) {
+    mError("stream:%s, failed to resume stream since %s", pauseReq.name, terrstr());
+    sdbRelease(pMnode->pSdb, pStream);
+    return -1;
+  }
+
+  mInfo("trans:%d used to resume stream:%s", pTrans->id, pauseReq.name);
 
   mndTransSetDbName(pTrans, pStream->sourceDb, pStream->targetDb);
   if (mndTransCheckConflict(pMnode, pTrans) != 0) {
@@ -1955,6 +1981,8 @@ static int32_t mndProcessResumeStreamReq(SRpcMsg *pReq) {
     return -1;
   }
 
+  int32_t code = mndStreamRegisterTrans(pTrans, MND_STREAM_RESUME_NAME, pStream->sourceDb, pStream->targetDb);
+
   // resume all tasks
   if (mndResumeAllStreamTasks(pTrans, pMnode, pStream, pauseReq.igUntreated) < 0) {
     mError("stream:%s, failed to drop task since %s", pauseReq.name, terrstr());
@@ -2240,7 +2268,7 @@ static int32_t mndProcessVgroupChange(SMnode *pMnode, SVgroupChangeInfo *pChange
 
     // here create only one trans
     if (pTrans == NULL) {
-      pTrans = doCreateTrans(pMnode, pStream, "stream-task-update");
+      pTrans = doCreateTrans(pMnode, pStream, NULL, MND_STREAM_TASK_UPDATE_NAME, "update task epsets");
       if (pTrans == NULL) {
         sdbRelease(pSdb, pStream);
         sdbCancelFetch(pSdb, pIter);
@@ -2350,7 +2378,7 @@ static void doExtractTasksFromStream(SMnode *pMnode) {
       break;
     }
 
-    keepStreamTasksInBuf(pStream, &execInfo);
+    saveStreamTasksInfo(pStream, &execInfo);
     sdbRelease(pSdb, pStream);
   }
 }
@@ -2436,6 +2464,17 @@ int32_t removeExpirednodeEntryAndTask(SArray *pNodeSnapshot) {
   return 0;
 }
 
+static void killAllCheckpointTrans(SMnode* pMnode, SVgroupChangeInfo* pChangeInfo) {
+  void* pIter = NULL;
+  while((pIter = taosHashIterate(pChangeInfo->pDBMap, pIter)) != NULL) {
+    char* pDb = (char*) pIter;
+
+    size_t len = 0;
+    void* pKey = taosHashGetKey(pDb, &len);
+    killActiveCheckpointTrans(pMnode, pKey, len);
+  }
+}
+
 // this function runs by only one thread, so it is not multi-thread safe
 static int32_t mndProcessNodeCheckReq(SRpcMsg *pMsg) {
   int32_t code = 0;
@@ -2477,7 +2516,8 @@ static int32_t mndProcessNodeCheckReq(SRpcMsg *pMsg) {
   SVgroupChangeInfo changeInfo = mndFindChangedNodeInfo(pMnode, execInfo.pNodeList, pNodeSnapshot);
   if (taosArrayGetSize(changeInfo.pUpdateNodeList) > 0) {
     // kill current active checkpoint transaction, since the transaction is vnode wide.
-    doKillActiveCheckpointTrans(pMnode);
+    killAllCheckpointTrans(pMnode, &changeInfo);
+
     code = mndProcessVgroupChange(pMnode, &changeInfo);
 
     // keep the new vnode snapshot
@@ -2523,7 +2563,7 @@ static int32_t mndProcessNodeCheck(SRpcMsg *pReq) {
   return 0;
 }
 
-void keepStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode) {
+void saveStreamTasksInfo(SStreamObj *pStream, SStreamExecInfo *pExecNode) {
   int32_t level = taosArrayGetSize(pStream->tasks);
 
   for (int32_t i = 0; i < level; i++) {
@@ -2566,8 +2606,9 @@ void removeStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode) {
           STaskId *pId = taosArrayGet(pExecNode->pTaskList, k);
           if (pId->taskId == id.taskId && pId->streamId == id.streamId) {
             taosArrayRemove(pExecNode->pTaskList, k);
-            mInfo("s-task:0x%x removed from buffer, remain:%d", (int32_t)id.taskId,
-                  (int32_t)taosArrayGetSize(pExecNode->pTaskList));
+
+            int32_t num = taosArrayGetSize(pExecNode->pTaskList);
+            mInfo("s-task:0x%x removed from buffer, remain:%d", (int32_t)id.taskId, num);
             break;
           }
         }
@@ -2578,15 +2619,15 @@ void removeStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode) {
   ASSERT(taosHashGetSize(pExecNode->pTaskMap) == taosArrayGetSize(pExecNode->pTaskList));
 }
 
-STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, const char *name) {
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, NULL, name);
+STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, SRpcMsg *pReq, const char *name, const char* pMsg) {
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, name);
   if (pTrans == NULL) {
     mError("failed to build trans:%s, reason: %s", name, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
 
-  mDebug("start to build stream:0x%" PRIx64 " task DAG update", pStream->uid);
+  mDebug("s-task:0x%"PRIx64" start to build trans %s", pStream->uid, pMsg);
 
   mndTransSetDbName(pTrans, pStream->sourceDb, pStream->targetDb);
   if (mndTransCheckConflict(pMnode, pTrans) != 0) {
@@ -2601,7 +2642,7 @@ STrans *doCreateTrans(SMnode *pMnode, SStreamObj *pStream, const char *name) {
 }
 
 int32_t createStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream) {
-  STrans *pTrans = doCreateTrans(pMnode, pStream, "stream-task-reset");
+  STrans *pTrans = doCreateTrans(pMnode, pStream, NULL, MND_STREAM_TASK_RESET_NAME, " reset from failed checkpoint");
   if (pTrans == NULL) {
     return terrno;
   }
@@ -2665,43 +2706,36 @@ int32_t createStreamResetStatusTrans(SMnode *pMnode, SStreamObj *pStream) {
   return TSDB_CODE_ACTION_IN_PROGRESS;
 }
 
-int32_t doKillActiveCheckpointTrans(SMnode *pMnode) {
-  int32_t transId = 0;
-  SSdb *  pSdb = pMnode->pSdb;
-  STrans *pTrans = NULL;
-  void *  pIter = NULL;
-
-  while (1) {
-    pIter = sdbFetch(pSdb, SDB_TRANS, pIter, (void **)&pTrans);
-    if (pIter == NULL) {
-      break;
-    }
-
-    if (strncmp(pTrans->opername, MND_STREAM_CHECKPOINT_NAME, tListLen(pTrans->opername) - 1) == 0) {
-      transId = pTrans->id;
-      sdbRelease(pSdb, pTrans);
-      sdbCancelFetch(pSdb, pIter);
-      break;
-    }
-
-    sdbRelease(pSdb, pTrans);
-  }
-
-  if (transId == 0) {
-    mDebug("failed to find the checkpoint trans, reset not executed");
+int32_t killActiveCheckpointTrans(SMnode *pMnode, const char* pDBName, size_t len) {
+  // data in the hash table will be removed automatically, no need to remove it here.
+  SStreamTransInfo* pTransInfo = taosHashGet(execInfo.transMgmt.pDBTrans, pDBName, len);
+  if (pTransInfo == NULL) {
     return TSDB_CODE_SUCCESS;
   }
 
-  pTrans = mndAcquireTrans(pMnode, transId);
-  mInfo("kill checkpoint trans:%d", transId);
+  // not checkpoint trans, ignore
+  if (strcmp(pTransInfo->name, MND_STREAM_CHECKPOINT_NAME) != 0) {
+    mDebug("not checkpoint trans, not kill it, name:%s, transId:%d", pTransInfo->name, pTransInfo->transId);
+    return TSDB_CODE_SUCCESS;
+  }
+
+  STrans* pTrans = mndAcquireTrans(pMnode, pTransInfo->transId);
+  if (pTrans != NULL) {
+    mInfo("kill checkpoint transId:%d in Db:%s", pTransInfo->transId, pDBName);
+    mndKillTrans(pMnode, pTrans);
+    mndReleaseTrans(pMnode, pTrans);
+  }
 
-  mndKillTrans(pMnode, pTrans);
-  mndReleaseTrans(pMnode, pTrans);
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t mndResetFromCheckpoint(SMnode *pMnode) {
-  doKillActiveCheckpointTrans(pMnode);
+int32_t mndResetStatusFromCheckpoint(SMnode *pMnode, int32_t transId) {
+  STrans* pTrans = mndAcquireTrans(pMnode, transId);
+  if (pTrans != NULL) {
+    mInfo("kill checkpoint transId:%d to reset task status", transId);
+    mndKillTrans(pMnode, pTrans);
+    mndReleaseTrans(pMnode, pTrans);
+  }
 
   // set all tasks status to be normal, refactor later to be stream level, instead of vnode level.
   SSdb *      pSdb = pMnode->pSdb;
@@ -2713,7 +2747,13 @@ int32_t mndResetFromCheckpoint(SMnode *pMnode) {
       break;
     }
 
-    // todo this transaction should exist be only one
+    bool conflict = streamTransConflictOtherTrans(pMnode, pStream->sourceDb, pStream->targetDb);
+    if (conflict) {
+      mError("stream:%s other trans exists in DB:%s & %s failed to start reset-status trans",
+             pStream->name, pStream->sourceDb, pStream->targetDb);
+      continue;
+    }
+
     mDebug("stream:%s (0x%" PRIx64 ") reset checkpoint procedure, create reset trans", pStream->name, pStream->uid);
     int32_t code = createStreamResetStatusTrans(pMnode, pStream);
     if (code != TSDB_CODE_SUCCESS) {
@@ -2842,7 +2882,7 @@ int32_t mndProcessStreamHb(SRpcMsg *pReq) {
       // if the execInfo.activeCheckpoint == 0, the checkpoint is restoring from wal
       mInfo("checkpointId:%" PRId64 " failed, issue task-reset trans to reset all tasks status",
             execInfo.activeCheckpoint);
-      mndResetFromCheckpoint(pMnode);
+      mndResetStatusFromCheckpoint(pMnode, activeCheckpointId);
     } else {
       mInfo("not all vgroups are ready, wait for next HB from stream tasks");
     }
diff --git a/source/dnode/mnode/impl/src/mndStreamTrans.c b/source/dnode/mnode/impl/src/mndStreamTrans.c
new file mode 100644
index 0000000000..2345de290a
--- /dev/null
+++ b/source/dnode/mnode/impl/src/mndStreamTrans.c
@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "mndTrans.h"
+#include "mndStream.h"
+
+typedef struct SKeyInfo {
+  void*   pKey;
+  int32_t keyLen;
+} SKeyInfo;
+
+static int32_t clearFinishedTrans(SMnode* pMnode);
+
+int32_t mndStreamRegisterTrans(STrans* pTrans, const char* pName, const char* pSrcDb, const char* pDstDb) {
+  SStreamTransInfo info = {.transId = pTrans->id, .startTime = taosGetTimestampMs(), .name = pName};
+  taosHashPut(execInfo.transMgmt.pDBTrans, pSrcDb, strlen(pSrcDb), &info, sizeof(SStreamTransInfo));
+
+  if (strcmp(pSrcDb, pDstDb) != 0) {
+    taosHashPut(execInfo.transMgmt.pDBTrans, pDstDb, strlen(pDstDb), &info, sizeof(SStreamTransInfo));
+  }
+
+  return 0;
+}
+
+int32_t clearFinishedTrans(SMnode* pMnode) {
+  SArray* pList = taosArrayInit(4, sizeof(SKeyInfo));
+  size_t  keyLen = 0;
+
+  taosThreadMutexLock(&execInfo.lock);
+
+  void* pIter = NULL;
+  while ((pIter = taosHashIterate(execInfo.transMgmt.pDBTrans, pIter)) != NULL) {
+    SStreamTransInfo *pEntry = (SStreamTransInfo *)pIter;
+    STrans* pTrans = mndAcquireTrans(pMnode, pEntry->transId);
+
+    // let's clear the finished trans
+    if (pTrans == NULL) {
+      void* pKey = taosHashGetKey(pEntry, &keyLen);
+      // key is the name of src/dst db name
+      SKeyInfo info = {.pKey = pKey, .keyLen = keyLen};
+
+      mDebug("transId:%d %s startTs:%" PRId64 "cleared due to finished", pEntry->transId, pEntry->name,
+             pEntry->startTime);
+      taosArrayPush(pList, &info);
+    } else {
+      mndReleaseTrans(pMnode, pTrans);
+    }
+  }
+
+  size_t num = taosArrayGetSize(pList);
+  for(int32_t i = 0; i < num; ++i) {
+    SKeyInfo* pKey = taosArrayGet(pList, i);
+    taosHashRemove(execInfo.transMgmt.pDBTrans, pKey->pKey, pKey->keyLen);
+  }
+
+  mDebug("clear %d finished stream-trans, remained:%d", (int32_t) num, taosHashGetSize(execInfo.transMgmt.pDBTrans));
+  taosThreadMutexUnlock(&execInfo.lock);
+
+  terrno = TSDB_CODE_SUCCESS;
+  taosArrayDestroy(pList);
+  return 0;
+}
+
+bool streamTransConflictOtherTrans(SMnode* pMnode, const char* pSrcDb, const char* pDstDb) {
+  clearFinishedTrans(pMnode);
+
+  taosThreadMutexLock(&execInfo.lock);
+  int32_t num = taosHashGetSize(execInfo.transMgmt.pDBTrans);
+  if (num <= 0) {
+    taosThreadMutexUnlock(&execInfo.lock);
+    return false;
+  }
+
+  SStreamTransInfo *pEntry = taosHashGet(execInfo.transMgmt.pDBTrans, pSrcDb, strlen(pSrcDb));
+  if (pEntry != NULL) {
+    taosThreadMutexUnlock(&execInfo.lock);
+    mWarn("conflict with other transId:%d in Db:%s, trans:%s", pEntry->transId, pSrcDb, pEntry->name);
+    return true;
+  }
+
+  pEntry = taosHashGet(execInfo.transMgmt.pDBTrans, pDstDb, strlen(pDstDb));
+  if (pEntry != NULL) {
+    taosThreadMutexUnlock(&execInfo.lock);
+    mWarn("conflict with other transId:%d in Db:%s, trans:%s", pEntry->transId, pSrcDb, pEntry->name);
+    return true;
+  }
+
+  taosThreadMutexUnlock(&execInfo.lock);
+  return false;
+}
+
+
+
diff --git a/source/dnode/mnode/impl/test/trans/trans1.cpp b/source/dnode/mnode/impl/test/trans/trans1.cpp
index 92a442aa5e..aff1156449 100644
--- a/source/dnode/mnode/impl/test/trans/trans1.cpp
+++ b/source/dnode/mnode/impl/test/trans/trans1.cpp
@@ -38,7 +38,7 @@ class MndTestTrans1 : public ::testing::Test {
 
     test.ServerStop();
 
-    pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+    pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
     int32_t writeLen = taosWriteFile(pFile, buffer, readLen);
     if (writeLen < 0 || writeLen == readLen) {
       ASSERT(1);
diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c
index 007243192c..9a14258752 100644
--- a/source/dnode/snode/src/snode.c
+++ b/source/dnode/snode/src/snode.c
@@ -324,6 +324,7 @@ int32_t sndInit(SSnode * pSnode) {
 }
 
 void sndClose(SSnode *pSnode) {
+  stopRsync();
   streamMetaNotifyClose(pSnode->pMeta);
   streamMetaCommit(pSnode->pMeta);
   streamMetaClose(pSnode->pMeta);
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 6c639da872..7be2a357d3 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1134,16 +1134,10 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
 
   // let's decide which step should be executed now
   if (pTask->execInfo.step1Start == 0) {
-    ASSERT(pTask->status.pauseAllowed == false);
     int64_t ts = taosGetTimestampMs();
 
     pTask->execInfo.step1Start = ts;
     tqDebug("s-task:%s start scan-history stage(step 1), status:%s, step1 startTs:%" PRId64, id, pStatus, ts);
-
-    // NOTE: in case of stream task, scan-history data in wal is not allowed to pause
-    if (pTask->info.fillHistory == 1) {
-      streamTaskEnablePause(pTask);
-    }
   } else {
     if (pTask->execInfo.step2Start == 0) {
       tqDebug("s-task:%s continue exec scan-history(step1), original step1 startTs:%" PRId64 ", already elapsed:%.2fs",
diff --git a/source/dnode/vnode/src/tq/tqStreamTask.c b/source/dnode/vnode/src/tq/tqStreamTask.c
index e069d6c4b7..56dcdb2abc 100644
--- a/source/dnode/vnode/src/tq/tqStreamTask.c
+++ b/source/dnode/vnode/src/tq/tqStreamTask.c
@@ -213,8 +213,10 @@ int32_t tqScanWalAsync(STQ* pTq, bool ckPause) {
   int32_t      vgId = TD_VID(pTq->pVnode);
   SStreamMeta* pMeta = pTq->pStreamMeta;
 
+  bool alreadyRestored = pTq->pVnode->restored;
+
   // do not launch the stream tasks, if it is a follower or not restored vnode.
-  if (!(vnodeIsRoleLeader(pTq->pVnode) && pTq->pVnode->restored)) {
+  if (!(vnodeIsRoleLeader(pTq->pVnode) && alreadyRestored)) {
     return TSDB_CODE_SUCCESS;
   }
 
@@ -256,7 +258,9 @@ int32_t tqScanWalAsync(STQ* pTq, bool ckPause) {
     return -1;
   }
 
-  tqDebug("vgId:%d create msg to start wal scan to launch stream tasks, numOfTasks:%d", vgId, numOfTasks);
+  tqDebug("vgId:%d create msg to start wal scan to launch stream tasks, numOfTasks:%d, restored:%d", vgId, numOfTasks,
+          alreadyRestored);
+
   pRunReq->head.vgId = vgId;
   pRunReq->streamId = 0;
   pRunReq->taskId = STREAM_EXEC_EXTRACT_DATA_IN_WAL_ID;
diff --git a/source/dnode/vnode/src/tsdb/tsdbFS.c b/source/dnode/vnode/src/tsdb/tsdbFS.c
index c0c74d6b87..732f46467e 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFS.c
+++ b/source/dnode/vnode/src/tsdb/tsdbFS.c
@@ -104,7 +104,7 @@ static int32_t tsdbSaveFSToFile(STsdbFS *pFS, const char *fname) {
   taosCalcChecksumAppend(0, pData, size);
 
   // save to file
-  TdFilePtr pFD = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
+  TdFilePtr pFD = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFD == NULL) {
     code = TAOS_SYSTEM_ERROR(errno);
     TSDB_CHECK_CODE(code, lino, _exit);
diff --git a/source/dnode/vnode/src/tsdb/tsdbFS2.c b/source/dnode/vnode/src/tsdb/tsdbFS2.c
index 70a83ebdbb..c9ecff8890 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFS2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbFS2.c
@@ -85,7 +85,7 @@ static int32_t save_json(const cJSON *json, const char *fname) {
   char *data = cJSON_PrintUnformatted(json);
   if (data == NULL) return TSDB_CODE_OUT_OF_MEMORY;
 
-  TdFilePtr fp = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
+  TdFilePtr fp = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (fp == NULL) {
     code = TAOS_SYSTEM_ERROR(code);
     goto _exit;
@@ -1186,11 +1186,6 @@ const char *gFSBgTaskName[] = {NULL, "MERGE", "RETENTION", "COMPACT"};
 static int32_t tsdbFSRunBgTask(void *arg) {
   STFSBgTask   *task = (STFSBgTask *)arg;
   STFileSystem *fs = task->fs;
-  STFileSet    *fset;
-
-  tsdbFSGetFSet(fs, task->fid, &fset);
-
-  ASSERT(fset != NULL && fset->bgTaskRunning == task);
 
   task->launchTime = taosGetTimestampMs();
   task->run(task->arg);
@@ -1203,6 +1198,10 @@ static int32_t tsdbFSRunBgTask(void *arg) {
 
   taosThreadMutexLock(&fs->tsdb->mutex);
 
+  STFileSet *fset = NULL;
+  tsdbFSGetFSet(fs, task->fid, &fset);
+  ASSERT(fset != NULL && fset->bgTaskRunning == task);
+
   // free last
   tsdbDoDoneBgTask(fs, task);
   fset->bgTaskRunning = NULL;
diff --git a/source/dnode/vnode/src/tsdb/tsdbFile.c b/source/dnode/vnode/src/tsdb/tsdbFile.c
index 62b37cd0a6..3ee0c482a7 100644
--- a/source/dnode/vnode/src/tsdb/tsdbFile.c
+++ b/source/dnode/vnode/src/tsdb/tsdbFile.c
@@ -177,7 +177,7 @@ int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype) {
   taosCalcChecksumAppend(0, hdr, TSDB_FHDR_SIZE);
 
   // open
-  pFD = taosOpenFile(fname, TD_FILE_WRITE);
+  pFD = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_WRITE_THROUGH);
   if (pFD == NULL) {
     code = TAOS_SYSTEM_ERROR(errno);
     goto _err;
diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c
index 6169014d9f..751df706ab 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c
@@ -572,7 +572,12 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SBlockN
 
     if (isEmptyQueryTimeWindow(&w)) {
       k += 1;
-      continue;
+
+      if (k >= numOfTables) {
+        break;
+      } else {
+        continue;
+      }
     }
 
     // 1. time range check
diff --git a/source/dnode/vnode/src/vnd/vnodeCommit.c b/source/dnode/vnode/src/vnd/vnodeCommit.c
index 50ca2f5d03..f9b5e9168c 100644
--- a/source/dnode/vnode/src/vnd/vnodeCommit.c
+++ b/source/dnode/vnode/src/vnd/vnodeCommit.c
@@ -177,7 +177,7 @@ int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
   }
 
   // save info to a vnode_tmp.json
-  pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) {
     vError("failed to open info file:%s for write:%s", fname, terrstr());
     terrno = TAOS_SYSTEM_ERROR(errno);
diff --git a/source/libs/index/src/indexFilter.c b/source/libs/index/src/indexFilter.c
index bfdcd2b030..7ed36fbf9e 100644
--- a/source/libs/index/src/indexFilter.c
+++ b/source/libs/index/src/indexFilter.c
@@ -328,6 +328,7 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
       SIF_ERR_RET(scalarGenerateSetFromList((void **)&param->pFilter, node, nl->node.resType.type));
       if (taosHashPut(ctx->pRes, &node, POINTER_BYTES, param, sizeof(*param))) {
         taosHashCleanup(param->pFilter);
+        param->pFilter = NULL;
         indexError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param));
         SIF_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
       }
diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index 1a65a29259..1f8af22241 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -1433,14 +1433,18 @@ static int32_t dataTypeComp(const SDataType* l, const SDataType* r) {
 
 static EDealRes translateOperator(STranslateContext* pCxt, SOperatorNode* pOp) {
   if (isMultiResFunc(pOp->pLeft)) {
-    return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName);
+    generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pLeft))->aliasName);
+    return DEAL_RES_ERROR;
   }
   if (isMultiResFunc(pOp->pRight)) {
-    return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pRight))->aliasName);
+    generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, ((SExprNode*)(pOp->pRight))->aliasName);
+    return DEAL_RES_ERROR;
   }
 
-  if (TSDB_CODE_SUCCESS != scalarGetOperatorResultType(pOp)) {
-    return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pOp->node.aliasName);
+  int32_t res = scalarGetOperatorResultType(pOp);
+  if (TSDB_CODE_SUCCESS != res) {
+    pCxt->errCode = res;
+    return DEAL_RES_ERROR;
   }
 
   return DEAL_RES_CONTINUE;
@@ -7958,7 +7962,7 @@ static int32_t translateDropStream(STranslateContext* pCxt, SDropStreamStmt* pSt
   tNameGetFullDbName(&name, dropReq.name);
   dropReq.igNotExists = pStmt->ignoreNotExists;
   int32_t code = buildCmdMsg(pCxt, TDMT_MND_DROP_STREAM, (FSerializeFunc)tSerializeSMDropStreamReq, &dropReq);
-  tFreeSMDropStreamReq(&dropReq);
+  tFreeMDropStreamReq(&dropReq);
   return code;
 }
 
diff --git a/source/libs/scalar/src/scalar.c b/source/libs/scalar/src/scalar.c
index 3e003234cf..90cec4522f 100644
--- a/source/libs/scalar/src/scalar.c
+++ b/source/libs/scalar/src/scalar.c
@@ -208,6 +208,7 @@ void sclFreeParam(SScalarParam *param) {
   if (param->columnData != NULL) {
     colDataDestroy(param->columnData);
     taosMemoryFreeClear(param->columnData);
+    param->columnData = NULL;
   }
 
   if (param->pHashFilter != NULL) {
@@ -845,6 +846,7 @@ int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *outp
   SScalarParam *params = NULL;
   int32_t       rowNum = 0;
   int32_t       code = 0;
+  int32_t       paramNum = 0;
 
   // json not support in in operator
   if (nodeType(node->pLeft) == QUERY_NODE_VALUE) {
@@ -865,7 +867,7 @@ int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *outp
 
   _bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(node->opType);
 
-  int32_t       paramNum = scalarGetOperatorParamNum(node->opType);
+  paramNum = scalarGetOperatorParamNum(node->opType);
   SScalarParam *pLeft = &params[0];
   SScalarParam *pRight = paramNum > 1 ? &params[1] : NULL;
 
diff --git a/source/libs/stream/inc/streamInt.h b/source/libs/stream/inc/streamInt.h
index b9e9db8662..4800e62109 100644
--- a/source/libs/stream/inc/streamInt.h
+++ b/source/libs/stream/inc/streamInt.h
@@ -75,7 +75,8 @@ struct STokenBucket {
   double  quotaCapacity;  // available capacity for maximum input size, KiloBytes per Second
   double  quotaRemain;    // not consumed bytes per second
   double  quotaRate;      // number of token per second
-  int64_t fillTimestamp;  // fill timestamp
+  int64_t tokenFillTimestamp;  // fill timestamp
+  int64_t quotaFillTimestamp;  // fill timestamp
 };
 
 struct SStreamQueue {
diff --git a/source/libs/stream/src/streamCheckpoint.c b/source/libs/stream/src/streamCheckpoint.c
index c392b17c74..9391cbcd9b 100644
--- a/source/libs/stream/src/streamCheckpoint.c
+++ b/source/libs/stream/src/streamCheckpoint.c
@@ -158,6 +158,7 @@ static int32_t continueDispatchCheckpointBlock(SStreamDataBlock* pBlock, SStream
 
   int32_t code = taosWriteQitem(pTask->outputq.queue->pQueue, pBlock);
   if (code == 0) {
+    ASSERT(pTask->chkInfo.dispatchCheckpointTrigger == false);
     streamDispatchStreamBlock(pTask);
   } else {
     stError("s-task:%s failed to put checkpoint into outputQ, code:%s", pTask->id.idStr, tstrerror(code));
@@ -280,8 +281,8 @@ void streamTaskClearCheckInfo(SStreamTask* pTask, bool clearChkpReadyMsg) {
   pTask->chkInfo.startTs = 0;  // clear the recorded start time
   pTask->checkpointNotReadyTasks = 0;
   pTask->checkpointAlignCnt = 0;
+  pTask->chkInfo.dispatchCheckpointTrigger = false;
   streamTaskOpenAllUpstreamInput(pTask);  // open inputQ for all upstream tasks
-
   if (clearChkpReadyMsg) {
     streamClearChkptReadyMsg(pTask);
   }
diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c
index 9d546ab9ce..da077c67f8 100644
--- a/source/libs/stream/src/streamDispatch.c
+++ b/source/libs/stream/src/streamDispatch.c
@@ -601,6 +601,12 @@ int32_t streamDispatchStreamBlock(SStreamTask* pTask) {
     return 0;
   }
 
+  if (pTask->chkInfo.dispatchCheckpointTrigger) {
+    stDebug("s-task:%s already send checkpoint trigger, not dispatch anymore", id);
+    atomic_store_8(&pTask->outputq.status, TASK_OUTPUT_STATUS__NORMAL);
+    return 0;
+  }
+
   ASSERT(pTask->msgInfo.pData == NULL);
   stDebug("s-task:%s start to dispatch msg, set output status:%d", id, pTask->outputq.status);
 
@@ -1064,30 +1070,14 @@ int32_t streamNotifyUpstreamContinue(SStreamTask* pTask) {
   return 0;
 }
 
-static void dispatchDataInFuture(void* param, void* tmrId) {
-  SStreamTask* pTask = param;
-  if (streamTaskShouldStop(pTask)) {
-    int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
-    stDebug("s-task:%s should stop, abort from timer, ref:%d", pTask->id.idStr, ref);
-    return;
-  }
-
-  ETaskStatus status = streamTaskGetStatus(pTask, NULL);
-  if (status == TASK_STATUS__CK) {
-    stDebug("s-task:%s in checkpoint status, wait for 500ms to dispatch data downstream", pTask->id.idStr);
-    taosTmrReset(dispatchDataInFuture, 500, pTask, streamEnv.timer, &pTask->msgInfo.pTimer);
-  } else {
-    int32_t ref = atomic_sub_fetch_32(&pTask->status.timerActive, 1);
-    stDebug("s-task:%s start to dispatch data, jump out of timer, ref:%d", pTask->id.idStr, ref);
-    streamDispatchStreamBlock(pTask);
-  }
-}
-
 // this message has been sent successfully, let's try next one.
 static int32_t handleDispatchSuccessRsp(SStreamTask* pTask, int32_t downstreamId) {
   destroyDispatchMsg(pTask->msgInfo.pData, getNumOfDispatchBranch(pTask));
 
   bool delayDispatch = (pTask->msgInfo.dispatchMsgType == STREAM_INPUT__CHECKPOINT_TRIGGER);
+  if (delayDispatch) {
+    pTask->chkInfo.dispatchCheckpointTrigger = true;
+  }
 
   pTask->msgInfo.pData = NULL;
   pTask->msgInfo.dispatchMsgType = 0;
@@ -1108,13 +1098,7 @@ static int32_t handleDispatchSuccessRsp(SStreamTask* pTask, int32_t downstreamId
 
   // otherwise, continue dispatch the first block to down stream task in pipeline
   if (delayDispatch) {
-    int32_t ref = atomic_add_fetch_32(&pTask->status.timerActive, 1);
-    stDebug("s-task:%s in checkpoint status, add in timer, try dispatch data in 500ms, ref:%d", pTask->id.idStr, ref);
-    if (pTask->msgInfo.pTimer != NULL) {
-      taosTmrReset(dispatchDataInFuture, 500, pTask, streamEnv.timer, &pTask->msgInfo.pTimer);
-    } else {
-      pTask->msgInfo.pTimer = taosTmrStart(dispatchDataInFuture, 500, pTask, streamEnv.timer);
-    }
+    return 0;
   } else {
     streamDispatchStreamBlock(pTask);
   }
@@ -1127,6 +1111,13 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
   int32_t     vgId = pTask->pMeta->vgId;
   int32_t     msgId = pTask->execInfo.dispatch;
 
+#if 0
+  // for test purpose, build  the failure case
+  if (pTask->msgInfo.dispatchMsgType == STREAM_INPUT__CHECKPOINT_TRIGGER) {
+    pRsp->inputStatus = TASK_INPUT_STATUS__REFUSED;
+  }
+#endif
+
   // follower not handle the dispatch rsp
   if ((pTask->pMeta->role == NODE_ROLE_FOLLOWER) || (pTask->status.downstreamReady != 1)) {
     stError("s-task:%s vgId:%d is follower or task just re-launched, not handle the dispatch rsp, discard it", id,
@@ -1169,8 +1160,21 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
       stWarn("s-task:%s inputQ of downstream task:0x%x(vgId:%d) is full, wait for %dms and retry dispatch", id,
              pRsp->downstreamTaskId, pRsp->downstreamNodeId, DISPATCH_RETRY_INTERVAL_MS);
     } else if (pRsp->inputStatus == TASK_INPUT_STATUS__REFUSED) {
-      stError("s-task:%s downstream task:0x%x(vgId:%d) refused the dispatch msg, treat it as success", id,
-              pRsp->downstreamTaskId, pRsp->downstreamNodeId);
+      // todo handle the agg task failure, add test case
+      if (pTask->msgInfo.dispatchMsgType == STREAM_INPUT__CHECKPOINT_TRIGGER &&
+          pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
+        stError("s-task:%s failed to dispatch checkpoint-trigger msg, checkpointId:%" PRId64
+                ", set the current checkpoint failed, and send rsp to mnode",
+                id, pTask->checkpointingId);
+        { // send checkpoint failure msg to mnode directly
+          pTask->chkInfo.failedId = pTask->checkpointingId;   // record the latest failed checkpoint id
+          pTask->checkpointingId = pTask->checkpointingId;
+          streamTaskSendCheckpointSourceRsp(pTask);
+        }
+      } else {
+        stError("s-task:%s downstream task:0x%x(vgId:%d) refused the dispatch msg, treat it as success", id,
+                pRsp->downstreamTaskId, pRsp->downstreamNodeId);
+      }
     }
   }
 
diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c
index cae537a860..43875319b7 100644
--- a/source/libs/stream/src/streamExec.c
+++ b/source/libs/stream/src/streamExec.c
@@ -48,6 +48,7 @@ static int32_t doOutputResultBlockImpl(SStreamTask* pTask, SStreamDataBlock* pBl
       return code;
     }
 
+    // checkpoint trigger will be checked
     streamDispatchStreamBlock(pTask);
   }
 
diff --git a/source/libs/stream/src/streamQueue.c b/source/libs/stream/src/streamQueue.c
index 556de169b4..d19dfc13bf 100644
--- a/source/libs/stream/src/streamQueue.c
+++ b/source/libs/stream/src/streamQueue.c
@@ -388,32 +388,36 @@ int32_t streamTaskInitTokenBucket(STokenBucket* pBucket, int32_t numCap, int32_t
   pBucket->quotaCapacity = quotaRate * MAX_SMOOTH_BURST_RATIO;
   pBucket->quotaRemain = pBucket->quotaCapacity;
 
-  pBucket->fillTimestamp = taosGetTimestampMs();
+  pBucket->tokenFillTimestamp = taosGetTimestampMs();
+  pBucket->quotaFillTimestamp = taosGetTimestampMs();
   stDebug("s-task:%s sink quotaRate:%.2fMiB, numRate:%d", id, quotaRate, numRate);
   return TSDB_CODE_SUCCESS;
 }
 
 static void fillTokenBucket(STokenBucket* pBucket, const char* id) {
   int64_t now = taosGetTimestampMs();
-  int64_t delta = now - pBucket->fillTimestamp;
+
+  int64_t deltaToken = now - pBucket->tokenFillTimestamp;
   ASSERT(pBucket->numOfToken >= 0);
 
-  int32_t incNum = (delta / 1000.0) * pBucket->numRate;
+  int32_t incNum = (deltaToken / 1000.0) * pBucket->numRate;
   if (incNum > 0) {
     pBucket->numOfToken = TMIN(pBucket->numOfToken + incNum, pBucket->numCapacity);
-    pBucket->fillTimestamp = now;
+    pBucket->tokenFillTimestamp = now;
   }
 
   // increase the new available quota as time goes on
-  double incSize = (delta / 1000.0) * pBucket->quotaRate;
+  int64_t deltaQuota = now - pBucket->quotaFillTimestamp;
+  double incSize = (deltaQuota / 1000.0) * pBucket->quotaRate;
   if (incSize > 0) {
     pBucket->quotaRemain = TMIN(pBucket->quotaRemain + incSize, pBucket->quotaCapacity);
-    pBucket->fillTimestamp = now;
+    pBucket->quotaFillTimestamp = now;
   }
 
   if (incNum > 0 || incSize > 0) {
-    stTrace("token/quota available, token:%d inc:%d, quota:%.2fMiB inc:%.3fMiB, ts:%" PRId64 " idle:%" PRId64 "ms, %s",
-            pBucket->numOfToken, incNum, pBucket->quotaRemain, incSize, now, delta, id);
+    stTrace("token/quota available, token:%d inc:%d, token_TsDelta:%" PRId64
+            ", quota:%.2fMiB inc:%.3fMiB quotaTs:%" PRId64 " now:%" PRId64 "ms, %s",
+            pBucket->numOfToken, incNum, deltaToken, pBucket->quotaRemain, incSize, deltaQuota, now, id);
   }
 }
 
diff --git a/source/libs/stream/src/streamStart.c b/source/libs/stream/src/streamStart.c
index a1c0fa2040..f042687942 100644
--- a/source/libs/stream/src/streamStart.c
+++ b/source/libs/stream/src/streamStart.c
@@ -168,7 +168,6 @@ int32_t streamTaskStartScanHistory(SStreamTask* pTask) {
   } else if (level == TASK_LEVEL__AGG) {
     if (pTask->info.fillHistory) {
       streamSetParamForScanHistory(pTask);
-      streamTaskEnablePause(pTask);
     }
   } else if (level == TASK_LEVEL__SINK) {
     stDebug("s-task:%s sink task do nothing to handle scan-history", pTask->id.idStr);
@@ -346,7 +345,6 @@ int32_t onNormalTaskReady(SStreamTask* pTask) {
     stDebug("s-task:%s level:%d status:%s sched-status:%d", id, pTask->info.taskLevel, p, pTask->status.schedStatus);
   }
 
-  streamTaskEnablePause(pTask);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -660,9 +658,6 @@ int32_t streamProcessScanHistoryFinishRsp(SStreamTask* pTask) {
   streamMetaCommit(pMeta);
   streamMetaWUnLock(pMeta);
 
-  // history data scan in the stream time window finished, now let's enable the pause
-  streamTaskEnablePause(pTask);
-
   // for source tasks, let's continue execute.
   if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
     streamSchedExec(pTask);
@@ -1041,11 +1036,6 @@ void streamTaskResume(SStreamTask* pTask) {
   }
 }
 
-void streamTaskEnablePause(SStreamTask* pTask) {
-  stDebug("s-task:%s enable task pause", pTask->id.idStr);
-  pTask->status.pauseAllowed = 1;
-}
-
 static void displayStatusInfo(SStreamMeta* pMeta, SHashObj* pTaskSet, bool succ) {
   int32_t vgId = pMeta->vgId;
   void*   pIter = NULL;
diff --git a/source/libs/stream/src/streamState.c b/source/libs/stream/src/streamState.c
index 2e51200fe4..0f32fd6879 100644
--- a/source/libs/stream/src/streamState.c
+++ b/source/libs/stream/src/streamState.c
@@ -1086,7 +1086,6 @@ _end:
 }
 
 int32_t streamStatePutParName(SStreamState* pState, int64_t groupId, const char tbname[TSDB_TABLE_NAME_LEN]) {
-  stDebug("try to write to cf parname");
 #ifdef USE_ROCKSDB
   if (tSimpleHashGetSize(pState->parNameMap) > MAX_TABLE_NAME_NUM) {
     if (tSimpleHashGet(pState->parNameMap, &groupId, sizeof(int64_t)) == NULL) {
diff --git a/source/libs/stream/src/streamTaskSm.c b/source/libs/stream/src/streamTaskSm.c
index 04b449aaaf..1c951e1452 100644
--- a/source/libs/stream/src/streamTaskSm.c
+++ b/source/libs/stream/src/streamTaskSm.c
@@ -269,6 +269,7 @@ int32_t streamTaskHandleEvent(SStreamTaskSM* pSM, EStreamTaskEvent event) {
               pTask->id.idStr, pSM->current.name, GET_EVT_NAME(evt));
       taosMsleep(100);
     } else {
+      // no active event trans exists, handle this event directly
       pTrans = streamTaskFindTransform(pSM->current.state, event);
       if (pTrans == NULL) {
         stDebug("s-task:%s failed to handle event:%s", pTask->id.idStr, GET_EVT_NAME(event));
@@ -451,60 +452,43 @@ int32_t initStateTransferTable() {
   return TSDB_CODE_SUCCESS;
 }
 
+//clang-format off
 void doInitStateTransferTable(void) {
   streamTaskSMTrans = taosArrayInit(8, sizeof(STaskStateTrans));
 
   // initialization event handle
-  STaskStateTrans trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__READY, TASK_EVENT_INIT,
-                                               streamTaskInitStatus, onNormalTaskReady, false, false);
+  STaskStateTrans trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__READY, TASK_EVENT_INIT, streamTaskInitStatus, onNormalTaskReady, false, false);
   taosArrayPush(streamTaskSMTrans, &trans);
-
-  trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__SCAN_HISTORY, TASK_EVENT_INIT_SCANHIST,
-                               streamTaskInitStatus, onScanhistoryTaskReady, false, false);
+  trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__SCAN_HISTORY, TASK_EVENT_INIT_SCANHIST, streamTaskInitStatus, onScanhistoryTaskReady, false, false);
   taosArrayPush(streamTaskSMTrans, &trans);
-
-  trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__STREAM_SCAN_HISTORY, TASK_EVENT_INIT_STREAM_SCANHIST,
-                               streamTaskInitStatus, onScanhistoryTaskReady, false, false);
+  trans = createStateTransform(TASK_STATUS__UNINIT, TASK_STATUS__STREAM_SCAN_HISTORY, TASK_EVENT_INIT_STREAM_SCANHIST, streamTaskInitStatus, onScanhistoryTaskReady, false, false);
   taosArrayPush(streamTaskSMTrans, &trans);
 
   // scan-history related event
-  trans = createStateTransform(TASK_STATUS__SCAN_HISTORY, TASK_STATUS__READY, TASK_EVENT_SCANHIST_DONE, NULL, NULL,
-                               NULL, true);
+  trans = createStateTransform(TASK_STATUS__SCAN_HISTORY, TASK_STATUS__READY, TASK_EVENT_SCANHIST_DONE, NULL, NULL, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
-
-  trans = createStateTransform(TASK_STATUS__STREAM_SCAN_HISTORY, TASK_STATUS__READY, TASK_EVENT_SCANHIST_DONE, NULL,
-                               NULL, NULL, true);
+  trans = createStateTransform(TASK_STATUS__STREAM_SCAN_HISTORY, TASK_STATUS__READY, TASK_EVENT_SCANHIST_DONE, NULL, NULL, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
 
   // halt stream task, from other task status
-  trans = createStateTransform(TASK_STATUS__READY, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL,
-                               streamTaskKeepCurrentVerInWal, NULL, true);
+  trans = createStateTransform(TASK_STATUS__READY, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL, streamTaskKeepCurrentVerInWal, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
-
-  trans = createStateTransform(TASK_STATUS__HALT, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL,
-                               streamTaskKeepCurrentVerInWal, NULL, true);
+  trans = createStateTransform(TASK_STATUS__HALT, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL, streamTaskKeepCurrentVerInWal, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
 
   SAttachedEventInfo info = {.status = TASK_STATUS__READY, .event = TASK_EVENT_HALT};
-  trans = createStateTransform(TASK_STATUS__STREAM_SCAN_HISTORY, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL,
-                               streamTaskKeepCurrentVerInWal, &info, true);
-  taosArrayPush(streamTaskSMTrans, &trans);
 
-  trans = createStateTransform(TASK_STATUS__CK, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL, streamTaskKeepCurrentVerInWal,
-                               &info, true);
+  trans = createStateTransform(TASK_STATUS__STREAM_SCAN_HISTORY, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL, streamTaskKeepCurrentVerInWal, &info, true);
   taosArrayPush(streamTaskSMTrans, &trans);
-
-  trans = createStateTransform(TASK_STATUS__PAUSE, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL,
-                               streamTaskKeepCurrentVerInWal, NULL, true);
+  trans = createStateTransform(TASK_STATUS__CK, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL, streamTaskKeepCurrentVerInWal, &info, true);
+  taosArrayPush(streamTaskSMTrans, &trans);
+  trans = createStateTransform(TASK_STATUS__PAUSE, TASK_STATUS__HALT, TASK_EVENT_HALT, NULL, streamTaskKeepCurrentVerInWal, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
 
   // checkpoint related event
-  trans = createStateTransform(TASK_STATUS__READY, TASK_STATUS__CK, TASK_EVENT_GEN_CHECKPOINT, NULL,
-                               streamTaskDoCheckpoint, NULL, true);
+  trans = createStateTransform(TASK_STATUS__READY, TASK_STATUS__CK, TASK_EVENT_GEN_CHECKPOINT, NULL, streamTaskDoCheckpoint, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
-
-  trans =
-      createStateTransform(TASK_STATUS__CK, TASK_STATUS__READY, TASK_EVENT_CHECKPOINT_DONE, NULL, NULL, NULL, true);
+  trans = createStateTransform(TASK_STATUS__CK, TASK_STATUS__READY, TASK_EVENT_CHECKPOINT_DONE, NULL, NULL, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
 
   // pause & resume related event handle
@@ -571,4 +555,5 @@ void doInitStateTransferTable(void) {
   taosArrayPush(streamTaskSMTrans, &trans);
   trans = createStateTransform(TASK_STATUS__STREAM_SCAN_HISTORY, TASK_STATUS__DROPPING, TASK_EVENT_DROPPING, NULL, NULL, NULL, true);
   taosArrayPush(streamTaskSMTrans, &trans);
-}
\ No newline at end of file
+}
+//clang-format on
\ No newline at end of file
diff --git a/source/libs/sync/src/syncRaftCfg.c b/source/libs/sync/src/syncRaftCfg.c
index 0dcc3eee29..0e98fe94eb 100644
--- a/source/libs/sync/src/syncRaftCfg.c
+++ b/source/libs/sync/src/syncRaftCfg.c
@@ -103,7 +103,7 @@ int32_t syncWriteCfgFile(SSyncNode *pNode) {
   if (buffer == NULL) goto _OVER;
   terrno = 0;
 
-  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) goto _OVER;
 
   int32_t len = strlen(buffer);
diff --git a/source/libs/sync/src/syncRaftStore.c b/source/libs/sync/src/syncRaftStore.c
index 051106b99d..c200c6cb4b 100644
--- a/source/libs/sync/src/syncRaftStore.c
+++ b/source/libs/sync/src/syncRaftStore.c
@@ -128,7 +128,7 @@ int32_t raftStoreWriteFile(SSyncNode *pNode) {
   if (buffer == NULL) goto _OVER;
   terrno = 0;
 
-  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pFile == NULL) goto _OVER;
 
   int32_t len = strlen(buffer);
diff --git a/source/libs/wal/src/walMeta.c b/source/libs/wal/src/walMeta.c
index f5e5427c68..933014466a 100644
--- a/source/libs/wal/src/walMeta.c
+++ b/source/libs/wal/src/walMeta.c
@@ -873,7 +873,7 @@ int walSaveMeta(SWal* pWal) {
     return -1;
   }
 
-  TdFilePtr pMetaFile = taosOpenFile(tmpFnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
+  TdFilePtr pMetaFile = taosOpenFile(tmpFnameStr, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
   if (pMetaFile == NULL) {
     wError("vgId:%d, failed to open file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), tmpFnameStr);
     terrno = TAOS_SYSTEM_ERROR(errno);
diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c
index ef97bff896..33d8d34514 100644
--- a/source/libs/wal/src/walWrite.c
+++ b/source/libs/wal/src/walWrite.c
@@ -133,6 +133,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   }
 
   walBuildIdxName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
+  taosCloseFile(&pWal->pIdxFile);
   TdFilePtr pIdxFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
 
   if (pIdxFile == NULL) {
@@ -153,6 +154,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   }
 
   walBuildLogName(pWal, walGetCurFileFirstVer(pWal), fnameStr);
+  taosCloseFile(&pWal->pLogFile);
   TdFilePtr pLogFile = taosOpenFile(fnameStr, TD_FILE_WRITE | TD_FILE_READ | TD_FILE_APPEND);
   wDebug("vgId:%d, wal truncate file %s", pWal->cfg.vgId, fnameStr);
   if (pLogFile == NULL) {
@@ -204,6 +206,7 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   pWal->vers.lastVer = ver - 1;
   ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1;
   ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset;
+
   taosCloseFile(&pIdxFile);
   taosCloseFile(&pLogFile);
 
@@ -605,7 +608,7 @@ int32_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SWalSync
     return -1;
   }
 
-  if (pWal->pIdxFile == NULL || pWal->pIdxFile == NULL || pWal->writeCur < 0) {
+  if (pWal->pIdxFile == NULL || pWal->pLogFile == NULL || pWal->writeCur < 0) {
     if (walInitWriteFile(pWal) < 0) {
       taosThreadMutexUnlock(&pWal->mutex);
       return -1;
diff --git a/source/os/src/osFile.c b/source/os/src/osFile.c
index 30f079d10d..cf3bce1ad4 100644
--- a/source/os/src/osFile.c
+++ b/source/os/src/osFile.c
@@ -21,11 +21,12 @@
 #include <io.h>
 #include <WinBase.h>
 #include <ktmw32.h>
+#include <windows.h>
 #define F_OK 0
 #define W_OK 2
 #define R_OK 4
 
-#define _SEND_FILE_STEP_ 1000
+#define _SEND_FILE_STEP_ 1024
 
 #else
 #include <fcntl.h>
@@ -44,12 +45,22 @@
 
 typedef int32_t FileFd;
 
+#ifdef WINDOWS
+typedef struct TdFile {
+  TdThreadRwlock rwlock;
+  int            refId;
+  HANDLE         hFile;
+  FILE*          fp;
+  int32_t        tdFileOptions;
+} TdFile;
+#else
 typedef struct TdFile {
   TdThreadRwlock rwlock;
   int            refId;
   FileFd         fd;
   FILE          *fp;
 } TdFile;
+#endif  // WINDOWS
 
 #define FILE_WITH_LOCK 1
 
@@ -240,15 +251,12 @@ int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime, int32_t *a
   return 0;
 }
 int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
-  if (pFile == NULL || pFile->fd < 0) {
+#ifdef WINDOWS
+  if (pFile == NULL || pFile->hFile == NULL) {
     return -1;
   }
-
-#ifdef WINDOWS
-
   BY_HANDLE_FILE_INFORMATION bhfi;
-  HANDLE                     handle = (HANDLE)_get_osfhandle(pFile->fd);
-  if (GetFileInformationByHandle(handle, &bhfi) == FALSE) {
+  if (GetFileInformationByHandle(pFile->hFile, &bhfi) == FALSE) {
     printf("taosFStatFile get file info fail.");
     return -1;
   }
@@ -262,7 +270,9 @@ int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
   }
 
 #else
-
+  if (pFile == NULL || pFile->fd < 0) {
+    return -1;
+  }
   struct stat fileStat;
   int32_t code = fstat(pFile->fd, &fileStat);
   if (code < 0) {
@@ -282,116 +292,363 @@ int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
   return 0;
 }
 
-TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
-  int   fd = -1;
-  FILE *fp = NULL;
-  if (tdFileOptions & TD_FILE_STREAM) {
-    char *mode = NULL;
-    if (tdFileOptions & TD_FILE_APPEND) {
-      mode = (tdFileOptions & TD_FILE_TEXT) ? "at+" : "ab+";
-    } else if (tdFileOptions & TD_FILE_TRUNC) {
-      mode = (tdFileOptions & TD_FILE_TEXT) ? "wt+" : "wb+";
-    } else if ((tdFileOptions & TD_FILE_READ) && !(tdFileOptions & TD_FILE_WRITE)) {
-      mode = (tdFileOptions & TD_FILE_TEXT) ? "rt" : "rb";
-    } else {
-      mode = (tdFileOptions & TD_FILE_TEXT) ? "rt+" : "rb+";
-    }
-    ASSERT(!(tdFileOptions & TD_FILE_EXCL));
-    if (tdFileOptions & TD_FILE_EXCL) {
-      return NULL;
-    }
-    fp = fopen(path, mode);
-    if (fp == NULL) {
-      return NULL;
-    }
+FILE *taosOpenFileForStream(const char *path, int32_t tdFileOptions) {
+  char *mode = NULL;
+  if (tdFileOptions & TD_FILE_APPEND) {
+    mode = (tdFileOptions & TD_FILE_TEXT) ? "at+" : "ab+";
+  } else if (tdFileOptions & TD_FILE_TRUNC) {
+    mode = (tdFileOptions & TD_FILE_TEXT) ? "wt+" : "wb+";
+  } else if ((tdFileOptions & TD_FILE_READ) && !(tdFileOptions & TD_FILE_WRITE)) {
+    mode = (tdFileOptions & TD_FILE_TEXT) ? "rt" : "rb";
   } else {
-    int access = O_BINARY;
-    access |= (tdFileOptions & TD_FILE_CREATE) ? O_CREAT : 0;
-    if ((tdFileOptions & TD_FILE_WRITE) && (tdFileOptions & TD_FILE_READ)) {
-      access |= O_RDWR;
-    } else if (tdFileOptions & TD_FILE_WRITE) {
-      access |= O_WRONLY;
-    } else if (tdFileOptions & TD_FILE_READ) {
-      access |= O_RDONLY;
-    }
-    access |= (tdFileOptions & TD_FILE_TRUNC) ? O_TRUNC : 0;
-    access |= (tdFileOptions & TD_FILE_APPEND) ? O_APPEND : 0;
-    access |= (tdFileOptions & TD_FILE_TEXT) ? O_TEXT : 0;
-    access |= (tdFileOptions & TD_FILE_EXCL) ? O_EXCL : 0;
-#ifdef WINDOWS
-    int32_t pmode = _S_IREAD | _S_IWRITE;
-    if (tdFileOptions & TD_FILE_AUTO_DEL) {
-      pmode |= _O_TEMPORARY;
-    }
-    fd = _open(path, access, pmode);
-#else
-    fd = open(path, access, S_IRWXU | S_IRWXG | S_IRWXO);
-#endif
-    if (fd == -1) {
-      return NULL;
-    }
+    mode = (tdFileOptions & TD_FILE_TEXT) ? "rt+" : "rb+";
   }
-
-  TdFilePtr pFile = (TdFilePtr)taosMemoryMalloc(sizeof(TdFile));
-  if (pFile == NULL) {
-    if (fd >= 0) close(fd);
-    if (fp != NULL) fclose(fp);
+  ASSERT(!(tdFileOptions & TD_FILE_EXCL));
+  if (tdFileOptions & TD_FILE_EXCL) {
     return NULL;
   }
-
-#if FILE_WITH_LOCK
-  taosThreadRwlockInit(&(pFile->rwlock), NULL);
-#endif
-  pFile->fd = fd;
-  pFile->fp = fp;
-  pFile->refId = 0;
-
-  if (tdFileOptions & TD_FILE_AUTO_DEL) {
-#ifdef WINDOWS
-    // do nothing, since the property of pmode is set with _O_TEMPORARY; the OS will recycle
-    // the file handle, as well as the space on disk.
-#else
-    // Remove it instantly, so when the program exits normally/abnormally, the file
-    // will be automatically remove by OS.
-    unlink(path);
-#endif
-  }
-
-  return pFile;
+  return fopen(path, mode);
 }
 
-int32_t taosCloseFile(TdFilePtr *ppFile) {
-  int32_t code = 0;
-  if (ppFile == NULL || *ppFile == NULL) {
+#ifdef WINDOWS
+HANDLE taosOpenFileNotStream(const char *path, int32_t tdFileOptions) {
+  DWORD openMode = 0;
+  DWORD access = 0;
+  DWORD fileFlag = FILE_ATTRIBUTE_NORMAL;
+  DWORD shareMode = FILE_SHARE_READ;
+
+  openMode = OPEN_EXISTING;
+  if (tdFileOptions & TD_FILE_CREATE) {
+    openMode = OPEN_ALWAYS;
+  } else if (tdFileOptions & TD_FILE_EXCL) {
+    openMode = CREATE_NEW;
+  } else if ((tdFileOptions & TD_FILE_TRUNC)) {
+    openMode = TRUNCATE_EXISTING;
+    access |= GENERIC_WRITE;
+  }
+  if (tdFileOptions & TD_FILE_APPEND) {
+    access |= FILE_APPEND_DATA;
+  }
+  if (tdFileOptions & TD_FILE_WRITE) {
+    access |= GENERIC_WRITE;
+  }
+
+  shareMode |= FILE_SHARE_WRITE;
+
+  access |= GENERIC_READ;
+
+  if (tdFileOptions & TD_FILE_AUTO_DEL) {
+    fileFlag |= FILE_ATTRIBUTE_TEMPORARY;
+  }
+  if (tdFileOptions & TD_FILE_WRITE_THROUGH) {
+    fileFlag |= FILE_FLAG_WRITE_THROUGH;
+  }
+
+  HANDLE h = CreateFile(path, access, shareMode, NULL, openMode, fileFlag, NULL);
+  if (h != INVALID_HANDLE_VALUE && (tdFileOptions & TD_FILE_APPEND) && (tdFileOptions & TD_FILE_WRITE)) {
+    SetFilePointer(h, 0, NULL, FILE_END);
+  }
+  if (h == INVALID_HANDLE_VALUE) {
+    DWORD dwError = GetLastError();
+    LPVOID lpMsgBuf;
+    FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM, NULL, dwError,
+                  0,
+                  (LPTSTR)&lpMsgBuf, 0, NULL);
+    printf("CreateFile failed with error %d: %s", dwError, (char*)lpMsgBuf);
+    LocalFree(lpMsgBuf);
+  }
+  return h;
+}
+
+int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
+#if FILE_WITH_LOCK
+  taosThreadRwlockRdlock(&(pFile->rwlock));
+#endif
+  if (pFile->hFile == NULL) {
+#if FILE_WITH_LOCK
+    taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+    return -1;
+  }
+
+  DWORD bytesRead;
+  if (!ReadFile(pFile->hFile, buf, count, &bytesRead, NULL)) {
+    bytesRead = -1;
+  } 
+#if FILE_WITH_LOCK
+  taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+  return bytesRead;
+}
+
+int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
+  if (pFile == NULL || pFile->hFile == NULL) {
     return 0;
   }
 #if FILE_WITH_LOCK
-  taosThreadRwlockWrlock(&((*ppFile)->rwlock));
+  taosThreadRwlockWrlock(&(pFile->rwlock));
 #endif
-  if ((*ppFile)->fp != NULL) {
-    fflush((*ppFile)->fp);
-    fclose((*ppFile)->fp);
-    (*ppFile)->fp = NULL;
+
+  DWORD   bytesWritten;
+  if (!WriteFile(pFile->hFile, buf, count, &bytesWritten, NULL)) {
+    bytesWritten = -1;
   }
-  if ((*ppFile)->fd >= 0) {
-#ifdef WINDOWS
-    HANDLE h = (HANDLE)_get_osfhandle((*ppFile)->fd);
-    !FlushFileBuffers(h);
-#else
-    // warning: never fsync silently in base lib
-    /*fsync((*ppFile)->fd);*/
-#endif
-    code = close((*ppFile)->fd);
-    (*ppFile)->fd = -1;
-  }
-  (*ppFile)->refId = 0;
+
 #if FILE_WITH_LOCK
-  taosThreadRwlockUnlock(&((*ppFile)->rwlock));
-  taosThreadRwlockDestroy(&((*ppFile)->rwlock));
+  taosThreadRwlockUnlock(&(pFile->rwlock));
 #endif
-  taosMemoryFree(*ppFile);
-  *ppFile = NULL;
-  return code;
+  return bytesWritten;
+}
+
+int64_t taosPWriteFile(TdFilePtr pFile, const void *buf, int64_t count, int64_t offset) {
+  if (pFile == NULL) {
+    return 0;
+  }
+#if FILE_WITH_LOCK
+  taosThreadRwlockWrlock(&(pFile->rwlock));
+#endif
+  ASSERT(pFile->hFile != NULL);  // Please check if you have closed the file.
+  if (pFile->hFile == NULL) {
+#if FILE_WITH_LOCK
+    taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+    return 0;
+  }
+
+  DWORD      ret = 0;
+  OVERLAPPED ol = {0};
+  ol.OffsetHigh = (uint32_t)((offset & 0xFFFFFFFF00000000LL) >> 0x20);
+  ol.Offset = (uint32_t)(offset & 0xFFFFFFFFLL);
+
+  SetLastError(0);
+  BOOL result = WriteFile(pFile->hFile, buf, count, &ret, &ol);
+  if (!result) {
+    errno = GetLastError();
+    ret = -1;
+  }
+
+#if FILE_WITH_LOCK
+  taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+  return ret;
+}
+
+int64_t taosLSeekFile(TdFilePtr pFile, int64_t offset, int32_t whence) {
+  if (pFile == NULL || pFile->hFile == NULL) {
+    return -1;
+  }
+#if FILE_WITH_LOCK
+  taosThreadRwlockRdlock(&(pFile->rwlock));
+#endif
+
+  LARGE_INTEGER liOffset;
+  liOffset.QuadPart = offset;
+  if (!SetFilePointerEx(pFile->hFile, liOffset, NULL, whence)) {
+    return -1;
+  }
+
+  liOffset.QuadPart = 0;
+  if (!SetFilePointerEx(pFile->hFile, liOffset, &liOffset, FILE_CURRENT)) {
+    return -1;
+  }
+#if FILE_WITH_LOCK
+  taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+  return liOffset.QuadPart;
+}
+
+int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime) {
+  if (pFile == NULL || pFile->hFile == NULL) {
+    return 0;
+  }
+
+  if (size != NULL) {
+    LARGE_INTEGER fileSize;
+    if (!GetFileSizeEx(pFile->hFile, &fileSize)) {
+      return -1;  // Error getting file size
+    }
+    *size = fileSize.QuadPart;
+  }
+
+  if (mtime != NULL) {
+    FILETIME creationTime, lastAccessTime, lastWriteTime;
+    if (!GetFileTime(pFile->hFile, &creationTime, &lastAccessTime, &lastWriteTime)) {
+      return -1;  // Error getting file time
+    }
+    // Convert the FILETIME structure to a time_t value
+    ULARGE_INTEGER ull;
+    ull.LowPart = lastWriteTime.dwLowDateTime;
+    ull.HighPart = lastWriteTime.dwHighDateTime;
+    *mtime = (int32_t)((ull.QuadPart - 116444736000000000ULL) / 10000000ULL);
+  }
+  return 0;
+}
+
+int32_t taosLockFile(TdFilePtr pFile) {
+  if (pFile == NULL || pFile->hFile == NULL) {
+    return -1;
+  }
+
+  BOOL          fSuccess = FALSE;
+  LARGE_INTEGER fileSize;
+  OVERLAPPED    overlapped = {0};
+
+  fSuccess = LockFileEx(pFile->hFile, LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY,
+                        0,           // reserved
+                        ~0,          // number of bytes to lock low
+                        ~0,          // number of bytes to lock high
+                        &overlapped  // overlapped structure
+  );
+  if (!fSuccess) {
+    return GetLastError();
+  }
+  return 0;
+}
+
+int32_t taosUnLockFile(TdFilePtr pFile) {
+  if (pFile == NULL || pFile->hFile == NULL) {
+    return 0;
+  }
+  BOOL       fSuccess = FALSE;
+  OVERLAPPED overlapped = {0};
+
+  fSuccess = UnlockFileEx(pFile->hFile, 0, ~0, ~0, &overlapped);
+  if (!fSuccess) {
+    return GetLastError();
+  }
+  return 0;
+}
+
+int32_t taosFtruncateFile(TdFilePtr pFile, int64_t l_size) {
+  if (pFile == NULL) {
+    return 0;
+  }
+  if (pFile->hFile == NULL) {
+    printf("Ftruncate file error, hFile was null\n");
+    return -1;
+  }
+
+  LARGE_INTEGER li_0;
+  li_0.QuadPart = (int64_t)0;
+  BOOL cur = SetFilePointerEx(pFile->hFile, li_0, NULL, FILE_CURRENT);
+  if (!cur) {
+    printf("SetFilePointerEx Error getting current position in file.\n");
+    return -1;
+  }
+
+  LARGE_INTEGER li_size;
+  li_size.QuadPart = l_size;
+  BOOL cur2 = SetFilePointerEx(pFile->hFile, li_size, NULL, FILE_BEGIN);
+  if (cur2 == 0) {
+    int error = GetLastError();
+    printf("SetFilePointerEx GetLastError is: %d\n", error);
+    switch (error) {
+      case ERROR_INVALID_HANDLE:
+        errno = EBADF;
+        break;
+      default:
+        errno = EIO;
+        break;
+    }
+    return -1;
+  }
+
+  if (!SetEndOfFile(pFile->hFile)) {
+    int error = GetLastError();
+    printf("SetEndOfFile GetLastError is:%d", error);
+    switch (error) {
+      case ERROR_INVALID_HANDLE:
+        errno = EBADF;
+        break;
+      default:
+        errno = EIO;
+        break;
+    }
+    return -1;
+  }
+  return 0;
+}
+
+int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size) {
+  if (pFileOut == NULL || pFileIn == NULL) {
+    return 0;
+  }
+  if (pFileIn->hFile == NULL || pFileOut->hFile == NULL) {
+    return 0;
+  }
+
+  LARGE_INTEGER fileOffset;
+  fileOffset.QuadPart = *offset;
+
+  if (!SetFilePointerEx(pFileIn->hFile, fileOffset, &fileOffset, FILE_BEGIN)) {
+    return -1;
+  }
+
+  int64_t writeLen = 0;
+  uint8_t buffer[_SEND_FILE_STEP_] = {0};
+
+  DWORD bytesRead;
+  DWORD bytesWritten;
+  for (int64_t len = 0; len < (size - _SEND_FILE_STEP_); len += _SEND_FILE_STEP_) {
+    if (!ReadFile(pFileIn->hFile, buffer, _SEND_FILE_STEP_, &bytesRead, NULL)) {
+      return writeLen;
+    }
+
+    if (bytesRead <= 0) {
+      return writeLen;
+    } else if (bytesRead < _SEND_FILE_STEP_) {
+      if (!WriteFile(pFileOut->hFile, buffer, bytesRead, &bytesWritten, NULL)) {
+        return -1;
+      } else {
+        return (int64_t)(writeLen + bytesRead);
+      }
+    } else {
+      if (!WriteFile(pFileOut->hFile, buffer, _SEND_FILE_STEP_, &bytesWritten, NULL)) {
+        return -1;
+      } else {
+        writeLen += _SEND_FILE_STEP_;
+      }
+    }
+  }
+
+  int64_t remain = size - writeLen;
+  if (remain > 0) {
+    DWORD bytesRead;
+    if (!ReadFile(pFileIn->hFile, buffer, (DWORD)remain, &bytesRead, NULL)) {
+      return -1;
+    }
+
+    if (bytesRead <= 0) {
+      return writeLen;
+    } else {
+      DWORD bytesWritten;
+      if (!WriteFile(pFileOut->hFile, buffer, bytesRead, &bytesWritten, NULL)) {
+        return -1;
+      } else {
+        writeLen += bytesWritten;
+      }
+    }
+  }
+  return writeLen;
+}
+
+#else
+int taosOpenFileNotStream(const char *path, int32_t tdFileOptions) {
+  int access = O_BINARY;
+  access |= (tdFileOptions & TD_FILE_CREATE) ? O_CREAT : 0;
+  if ((tdFileOptions & TD_FILE_WRITE) && (tdFileOptions & TD_FILE_READ)) {
+    access |= O_RDWR;
+  } else if (tdFileOptions & TD_FILE_WRITE) {
+    access |= O_WRONLY;
+  } else if (tdFileOptions & TD_FILE_READ) {
+    access |= O_RDONLY;
+  }
+  access |= (tdFileOptions & TD_FILE_TRUNC) ? O_TRUNC : 0;
+  access |= (tdFileOptions & TD_FILE_APPEND) ? O_APPEND : 0;
+  access |= (tdFileOptions & TD_FILE_TEXT) ? O_TEXT : 0;
+  access |= (tdFileOptions & TD_FILE_EXCL) ? O_EXCL : 0;
+  int fd = open(path, access, S_IRWXU | S_IRWXG | S_IRWXO);
+  return fd;
 }
 
 int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
@@ -407,7 +664,7 @@ int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
   }
   int64_t leftbytes = count;
   int64_t readbytes;
-  char   *tbuf = (char *)buf;
+  char *  tbuf = (char *)buf;
 
   while (leftbytes > 0) {
 #ifdef WINDOWS
@@ -441,42 +698,6 @@ int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count) {
   return count;
 }
 
-int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset) {
-  if (pFile == NULL) {
-    return 0;
-  }
-#if FILE_WITH_LOCK
-  taosThreadRwlockRdlock(&(pFile->rwlock));
-#endif
-  ASSERT(pFile->fd >= 0);  // Please check if you have closed the file.
-  if (pFile->fd < 0) {
-#if FILE_WITH_LOCK
-    taosThreadRwlockUnlock(&(pFile->rwlock));
-#endif
-    return -1;
-  }
-#ifdef WINDOWS
-  DWORD     ret = 0;
-  OVERLAPPED ol = {0};
-  ol.OffsetHigh = (uint32_t)((offset & 0xFFFFFFFF00000000LL) >> 0x20);
-  ol.Offset = (uint32_t)(offset & 0xFFFFFFFFLL);
-
-  HANDLE handle = (HANDLE)_get_osfhandle(pFile->fd);
-  SetLastError(0);
-  BOOL result = ReadFile(handle, buf, count, &ret, &ol);
-  if (!result && GetLastError() != ERROR_HANDLE_EOF) {
-    errno = GetLastError();
-    ret = -1;
-  }
-#else
-  int64_t ret = pread(pFile->fd, buf, count, offset);
-#endif
-#if FILE_WITH_LOCK
-  taosThreadRwlockUnlock(&(pFile->rwlock));
-#endif
-  return ret;
-}
-
 int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
   if (pFile == NULL) {
     return 0;
@@ -493,7 +714,7 @@ int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count) {
 
   int64_t nleft = count;
   int64_t nwritten = 0;
-  char   *tbuf = (char *)buf;
+  char *  tbuf = (char *)buf;
 
   while (nleft > 0) {
     nwritten = write(pFile->fd, (void *)tbuf, (uint32_t)nleft);
@@ -706,25 +927,6 @@ int32_t taosFtruncateFile(TdFilePtr pFile, int64_t l_size) {
 #endif
 }
 
-int32_t taosFsyncFile(TdFilePtr pFile) {
-  if (pFile == NULL) {
-    return 0;
-  }
-
-  // this implementation is WRONG
-  // fflush is not a replacement of fsync
-  if (pFile->fp != NULL) return fflush(pFile->fp);
-  if (pFile->fd >= 0) {
-#ifdef WINDOWS
-    HANDLE h = (HANDLE)_get_osfhandle(pFile->fd);
-    return !FlushFileBuffers(h);
-#else
-    return fsync(pFile->fd);
-#endif
-  }
-  return 0;
-}
-
 int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size) {
   if (pFileOut == NULL || pFileIn == NULL) {
     return 0;
@@ -824,6 +1026,167 @@ int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, in
 #endif
 }
 
+#endif   // WINDOWS
+
+TdFilePtr taosOpenFile(const char *path, int32_t tdFileOptions) {
+  FILE *fp = NULL;
+#ifdef WINDOWS
+  HANDLE hFile = NULL;
+#else
+  int fd = -1;
+#endif
+  if (tdFileOptions & TD_FILE_STREAM) {
+    fp = taosOpenFileForStream(path, tdFileOptions);
+    if (fp == NULL) return NULL;
+  } else {
+#ifdef WINDOWS
+    hFile = taosOpenFileNotStream(path, tdFileOptions);
+    if (hFile == INVALID_HANDLE_VALUE) return NULL;
+#else
+    fd = taosOpenFileNotStream(path, tdFileOptions);
+    if (fd == -1) return NULL;
+#endif
+  }
+
+  TdFilePtr pFile = (TdFilePtr)taosMemoryMalloc(sizeof(TdFile));
+  if (pFile == NULL) {
+#ifdef WINDOWS
+    if (hFile != NULL) CloseHandle(hFile);
+#else
+    if (fd >= 0) close(fd);
+#endif   
+    if (fp != NULL) fclose(fp);
+    return NULL;
+  }
+
+#if FILE_WITH_LOCK
+  taosThreadRwlockInit(&(pFile->rwlock), NULL);
+#endif
+  pFile->fp = fp;
+  pFile->refId = 0;
+
+  #ifdef WINDOWS
+  pFile->hFile = hFile;
+  pFile->tdFileOptions = tdFileOptions;
+  // do nothing, since the property of pmode is set with _O_TEMPORARY; the OS will recycle
+  // the file handle, as well as the space on disk.
+#else
+  pFile->fd = fd;
+  // Remove it instantly, so when the program exits normally/abnormally, the file
+  // will be automatically remove by OS.
+  if (tdFileOptions & TD_FILE_AUTO_DEL) {
+    unlink(path);
+  }
+#endif
+  return pFile;
+}
+
+int32_t taosCloseFile(TdFilePtr *ppFile) {
+  int32_t code = 0;
+  if (ppFile == NULL || *ppFile == NULL) {
+    return 0;
+  }
+#if FILE_WITH_LOCK
+  taosThreadRwlockWrlock(&((*ppFile)->rwlock));
+#endif
+  if ((*ppFile)->fp != NULL) {
+    fflush((*ppFile)->fp);
+    fclose((*ppFile)->fp);
+    (*ppFile)->fp = NULL;
+  }
+#ifdef WINDOWS
+  if ((*ppFile)->hFile != NULL) {
+    // FlushFileBuffers((*ppFile)->hFile);
+    if (!CloseHandle((*ppFile)->hFile)) {
+      code = -1;
+    }
+    (*ppFile)->hFile = NULL;
+#else
+  if ((*ppFile)->fd >= 0) {
+    // warning: never fsync silently in base lib
+    /*fsync((*ppFile)->fd);*/
+    code = close((*ppFile)->fd);
+    (*ppFile)->fd = -1;
+#endif
+  }
+  (*ppFile)->refId = 0;
+#if FILE_WITH_LOCK
+  taosThreadRwlockUnlock(&((*ppFile)->rwlock));
+  taosThreadRwlockDestroy(&((*ppFile)->rwlock));
+#endif
+  taosMemoryFree(*ppFile);
+  *ppFile = NULL;
+  return code;
+}
+
+int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset) {
+  if (pFile == NULL) {
+    return 0;
+  }
+
+#ifdef WINDOWS
+#if FILE_WITH_LOCK
+  taosThreadRwlockRdlock(&(pFile->rwlock));
+#endif
+  ASSERT(pFile->hFile != NULL);  // Please check if you have closed the file.
+  if (pFile->hFile == NULL) {
+#if FILE_WITH_LOCK
+    taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+    return -1;
+  }
+  DWORD     ret = 0;
+  OVERLAPPED ol = {0};
+  ol.OffsetHigh = (uint32_t)((offset & 0xFFFFFFFF00000000LL) >> 0x20);
+  ol.Offset = (uint32_t)(offset & 0xFFFFFFFFLL);
+
+  SetLastError(0);
+  BOOL result = ReadFile(pFile->hFile, buf, count, &ret, &ol);
+  if (!result && GetLastError() != ERROR_HANDLE_EOF) {
+    errno = GetLastError();
+    ret = -1;
+  }
+#else
+#if FILE_WITH_LOCK
+  taosThreadRwlockRdlock(&(pFile->rwlock));
+#endif
+  ASSERT(pFile->fd >= 0);  // Please check if you have closed the file.
+  if (pFile->fd < 0) {
+#if FILE_WITH_LOCK
+    taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+    return -1;
+  }
+  int64_t ret = pread(pFile->fd, buf, count, offset);
+#endif
+#if FILE_WITH_LOCK
+  taosThreadRwlockUnlock(&(pFile->rwlock));
+#endif
+  return ret;
+}
+
+int32_t taosFsyncFile(TdFilePtr pFile) {
+  if (pFile == NULL) {
+    return 0;
+  }
+
+  // this implementation is WRONG
+  // fflush is not a replacement of fsync
+  if (pFile->fp != NULL) return fflush(pFile->fp);
+#ifdef WINDOWS
+  if (pFile->hFile != NULL) {
+    if (pFile->tdFileOptions & TD_FILE_WRITE_THROUGH) {
+      return 0;
+    } 
+    return !FlushFileBuffers(pFile->hFile);
+#else
+  if (pFile->fd >= 0) {
+    return fsync(pFile->fd);
+#endif
+  }
+  return 0;
+}
+
 void taosFprintfFile(TdFilePtr pFile, const char *format, ...) {
   if (pFile == NULL || pFile->fp == NULL) {
     return;
@@ -834,7 +1197,13 @@ void taosFprintfFile(TdFilePtr pFile, const char *format, ...) {
   va_end(ap);
 }
 
-bool taosValidFile(TdFilePtr pFile) { return pFile != NULL && pFile->fd > 0; }
+bool taosValidFile(TdFilePtr pFile) {
+#ifdef WINDOWS
+  return pFile != NULL && pFile->hFile != NULL;
+#else
+  return pFile != NULL && pFile->fd > 0;
+#endif   
+}
 
 int32_t taosUmaskFile(int32_t maskVal) {
 #ifdef WINDOWS
@@ -960,14 +1329,20 @@ int32_t taosCompressFile(char *srcFileName, char *destFileName) {
     goto cmp_end;
   }
 
-  pFile = taosOpenFile(destFileName, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
-  if (pFile == NULL) {
+  int access = O_BINARY | O_WRONLY | O_TRUNC | O_CREAT;
+#ifdef WINDOWS
+  int32_t pmode = _S_IREAD | _S_IWRITE;
+#else
+  int32_t pmode = S_IRWXU | S_IRWXG | S_IRWXO;
+#endif
+  int fd = open(destFileName, access, pmode);
+  if (fd < 0) {
     ret = -2;
     goto cmp_end;
   }
 
   // Both gzclose() and fclose() will close the associated fd, so they need to have different fds.
-  FileFd gzFd = dup(pFile->fd);
+  FileFd gzFd = dup(fd);
   if (gzFd < 0) {
     ret = -4;
     goto cmp_end;
diff --git a/source/os/test/osTests.cpp b/source/os/test/osTests.cpp
index a2ccc4de02..e2185aeac2 100644
--- a/source/os/test/osTests.cpp
+++ b/source/os/test/osTests.cpp
@@ -15,6 +15,7 @@
 
 #include <gtest/gtest.h>
 #include <iostream>
+#include <inttypes.h>
 
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wwrite-strings"
@@ -29,6 +30,10 @@
 #include "os.h"
 #include "tlog.h"
 
+#ifdef WINDOWS
+#include <windows.h>
+#endif  //  WINDOWS
+
 TEST(osTest, osSystem) {
   const char *flags = "UTL FATAL ";
   ELogLevel   level = DEBUG_FATAL;
@@ -68,7 +73,8 @@ void fileOperateOnBusy(void *param) {
   char *    fname = (char *)param;
   TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
   printf("On busy thread open file\n");
-  ASSERT_NE(pFile, nullptr);
+  if (pFile == NULL) return;
+  // ASSERT_NE(pFile, nullptr);
 
   int ret = taosLockFile(pFile);
   printf("On busy thread lock file ret:%d\n", ret);
@@ -97,6 +103,7 @@ TEST(osTest, osFile) {
   TdFilePtr pOutFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
   ASSERT_NE(pOutFD, nullptr);
   printf("create file success\n");
+  taosCloseFile(&pOutFD);
 
   TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
   printf("open file\n");
@@ -135,4 +142,178 @@ TEST(osTest, osFile) {
   //printf("remove file success");
 }
 
+#ifndef OSFILE_PERFORMANCE_TEST
+
+#define MAX_WORDS          100
+#define MAX_WORD_LENGTH    20
+#define MAX_TEST_FILE_SIZE 100000
+#define TESTTIMES          1000
+
+char *getRandomWord() {
+  static char words[][MAX_WORD_LENGTH] = {
+        "Lorem", "ipsum", "dolor", "sit", "amet", "consectetur", "adipiscing", "elit",
+        "sed", "do", "eiusmod", "tempor", "incididunt", "ut", "labore", "et", "dolore", "magna",
+        "aliqua", "Ut", "enim", "ad", "minim", "veniam", "quis", "nostrud", "exercitation", "ullamco",
+        "Why", "do", "programmers", "prefer", "using", "dark", "mode?", "Because", "light", "attracts",
+        "bugs", "and", "they", "want", "to", "code", "in", "peace,", "like", "a", "ninja", "in", "the", "shadows."
+        "aliqua", "Ut", "enim", "ad", "minim", "veniam", "quis", "nostrud", "exercitation", "ullamco",
+        "laboris", "nisi", "ut", "aliquip", "ex", "ea", "commodo", "consequat", "Duis", "aute", "irure",
+        "dolor", "in", "reprehenderit", "in", "voluptate", "velit", "esse", "cillum", "dolore", "eu",
+        "fugiat", "nulla", "pariatur", "Excepteur", "sint", "occaecat", "cupidatat", "non", "proident",
+        "sunt", "in", "culpa", "qui", "officia", "deserunt", "mollit", "anim", "id", "est", "laborum"
+    };
+
+  return words[taosRand() % MAX_WORDS];
+}
+
+int64_t fillBufferWithRandomWords(char *buffer, int64_t maxBufferSize) {
+  int64_t len = 0;
+  while (len < maxBufferSize) {
+    char * word = getRandomWord();
+    size_t wordLen = strlen(word);
+
+    if (len + wordLen + 1 < maxBufferSize) {
+      strcat(buffer, word);
+      strcat(buffer, " ");
+      len += wordLen + 1;
+    } else {
+      break;
+    }
+  }
+  return len;
+}
+
+int64_t calculateAverage(int64_t arr[], int size) {
+  int64_t sum = 0;
+  for (int i = 0; i < size; i++) {
+    sum += arr[i];
+  }
+  return sum / size;
+}
+
+int64_t calculateMax(int64_t arr[], int size) {
+  int64_t max = arr[0];
+  for (int i = 1; i < size; i++) {
+    if (arr[i] > max) {
+      max = arr[i];
+    }
+  }
+  return max;
+}
+
+int64_t calculateMin(int64_t arr[], int size) {
+  int64_t min = arr[0];
+  for (int i = 1; i < size; i++) {
+    if (arr[i] < min) {
+      min = arr[i];
+    }
+  }
+  return min;
+}
+
+TEST(osTest, osFilePerformance) {
+  printf("os file performance testting...\n");
+  int64_t WriteFileCost;
+  int64_t ReadFileCost;
+  int64_t OpenForWriteCloseFileCost;
+  int64_t OpenForReadCloseFileCost;
+
+  char *  buffer;
+  char *  writeBuffer = (char *)taosMemoryCalloc(1, MAX_TEST_FILE_SIZE);
+  char *  readBuffer = (char *)taosMemoryCalloc(1, MAX_TEST_FILE_SIZE);
+  int64_t size = fillBufferWithRandomWords(writeBuffer, MAX_TEST_FILE_SIZE);
+  char *  fname = "./osFilePerformanceTest.txt";
+
+  TdFilePtr pOutFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
+  ASSERT_NE(pOutFD, nullptr);
+  taosCloseFile(&pOutFD);
+
+  printf("os file performance start write...\n");
+  int64_t t1 = taosGetTimestampUs();
+  for (int i = 0; i < TESTTIMES; ++i) {
+    TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_WRITE_THROUGH);
+    ASSERT_NE(pFile, nullptr);
+    taosWriteFile(pFile, writeBuffer, size);
+    taosFsyncFile(pFile);
+    taosCloseFile(&pFile);
+  }
+
+  int64_t t2 = taosGetTimestampUs();
+  WriteFileCost = t2 - t1;
+
+  printf("os file performance start read...\n");
+  for (int i = 0; i < TESTTIMES; ++i) {
+    TdFilePtr pFile = taosOpenFile(fname, TD_FILE_READ);
+    ASSERT_NE(pFile, nullptr);
+    taosReadFile(pFile, readBuffer, size);
+    taosCloseFile(&pFile);
+    int readLine = strlen(readBuffer);
+    ASSERT_EQ(size, readLine);
+  }
+  int64_t t3 = taosGetTimestampUs();
+  ReadFileCost = t3 - t2;
+
+  printf("os file performance start open1...\n");
+  for (int i = 0; i < TESTTIMES; ++i) {
+    TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
+    ASSERT_NE(pFile, nullptr);
+    taosCloseFile(&pFile);
+  }
+  int64_t t4 = taosGetTimestampUs();
+  OpenForWriteCloseFileCost = t4 - t3;
+
+  printf("os file performance start open2...\n");
+  for (int i = 0; i < TESTTIMES; ++i) {
+    TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_READ);
+    ASSERT_NE(pFile, nullptr);
+    taosCloseFile(&pFile);
+  }
+  int64_t t5 = taosGetTimestampUs();
+  OpenForReadCloseFileCost = t5 - t4;
+
+#ifdef WINDOWS
+  printf("os file performance start window native...\n");
+  for (int i = 0; i < TESTTIMES; ++i) {
+    HANDLE hFile = CreateFile(fname,                    // 文件名
+                              GENERIC_WRITE,            // 写权限
+                              FILE_SHARE_READ,          // 不共享
+                              NULL,                     // 默认安全描述符
+                              OPEN_ALWAYS,              // 打开已存在的文件
+                              FILE_FLAG_WRITE_THROUGH,  // 文件标志，可以根据实际需求调整
+                              NULL                      // 模板文件句柄，对于创建新文件不需要
+    );
+
+    if (hFile == INVALID_HANDLE_VALUE) {
+      printf("Error opening file\n");
+      break;
+    }
+
+    // 写入数据
+    DWORD bytesWritten;
+    if (!WriteFile(hFile, writeBuffer, size, &bytesWritten, NULL)) {
+      // 处理错误
+      printf("Error writing to file\n");
+      CloseHandle(hFile);
+      break;
+    }
+    // 关闭文件
+    CloseHandle(hFile);
+  }
+  int64_t t6 = taosGetTimestampUs();
+  int64_t nativeWritCost = t6 - t5;
+
+  printf("Test Write file using native API %d times, cost: %" PRId64 "us\n", TESTTIMES, nativeWritCost);
+#endif  // WINDOWS
+
+  taosMemoryFree(writeBuffer);
+  taosMemoryFree(readBuffer);
+
+  printf("Test Write file %d times, cost: %" PRId64 "us\n", TESTTIMES, WriteFileCost);
+  printf("Test Read file %d times, cost: %" PRId64 "us\n", TESTTIMES, ReadFileCost);
+  printf("Test OpenForWrite & Close file %d times, cost: %" PRId64 "us\n", TESTTIMES, OpenForWriteCloseFileCost);
+  printf("Test OpenForRead & Close file %d times, cost: %" PRId64 "us\n", TESTTIMES, OpenForReadCloseFileCost);
+}
+
+#endif OSFILE_PERFORMANCE_TEST
+
 #pragma GCC diagnostic pop
diff --git a/source/util/src/tlog.c b/source/util/src/tlog.c
index 46b25dee6a..184e18fc67 100644
--- a/source/util/src/tlog.c
+++ b/source/util/src/tlog.c
@@ -154,7 +154,12 @@ int32_t taosInitSlowLog() {
 #endif
 
   if (strlen(tsLogDir) != 0) {
-    snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logFileName);
+    char lastC = tsLogDir[strlen(tsLogDir) - 1];
+    if (lastC == '\\' || lastC == '/') {
+      snprintf(fullName, PATH_MAX, "%s" "%s", tsLogDir, logFileName);
+    } else {
+      snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logFileName);
+    }
   } else {
     snprintf(fullName, PATH_MAX, "%s", logFileName);
   }
@@ -178,7 +183,12 @@ int32_t taosInitLog(const char *logName, int32_t maxFiles) {
 
   char fullName[PATH_MAX] = {0};
   if (strlen(tsLogDir) != 0) {
-    snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logName);
+    char lastC = tsLogDir[strlen(tsLogDir) - 1];
+    if (lastC == '\\' || lastC == '/') {
+      snprintf(fullName, PATH_MAX, "%s" "%s", tsLogDir, logName);
+    } else {
+      snprintf(fullName, PATH_MAX, "%s" TD_DIRSEP "%s", tsLogDir, logName);
+    }
   } else {
     snprintf(fullName, PATH_MAX, "%s", logName);
   }
diff --git a/tests/script/tsim/stream/pauseAndResume.sim b/tests/script/tsim/stream/pauseAndResume.sim
index 673bc77c0f..5eb9eef010 100644
--- a/tests/script/tsim/stream/pauseAndResume.sim
+++ b/tests/script/tsim/stream/pauseAndResume.sim
@@ -16,9 +16,8 @@ sql create table ts2 using st tags(2,2,2);
 sql create table ts3 using st tags(3,2,2);
 sql create table ts4 using st tags(4,2,2);
 sql create stream streams1 trigger at_once IGNORE EXPIRED 0 IGNORE UPDATE 0 watermark 1d into streamt1 as select  _wstart, count(*) c1, sum(a) c3 from st interval(10s);
-sleep 1000
+sleep 2000
 
-sleep 1000
 sql pause stream streams1;
 
 sql insert into ts1 values(1648791213001,1,12,3,1.0);
diff --git a/tests/system-test/0-others/view/non_marterial_view/test_view.py b/tests/system-test/0-others/view/non_marterial_view/test_view.py
index afb2476305..4b829b4049 100644
--- a/tests/system-test/0-others/view/non_marterial_view/test_view.py
+++ b/tests/system-test/0-others/view/non_marterial_view/test_view.py
@@ -162,7 +162,7 @@ class TDTestCase:
         assert('TIMESTAMP' in data_type_list and 'INT' in data_type_list and 'INT UNSIGNED' in data_type_list and 'BIGINT' in data_type_list and 'BIGINT UNSIGNED' in data_type_list and 'FLOAT' in data_type_list and 'DOUBLE' in data_type_list and 'VARCHAR' in data_type_list and 'SMALLINT' in data_type_list and 'SMALLINT UNSIGNED' in data_type_list and 'TINYINT' in data_type_list and 'TINYINT UNSIGNED' in data_type_list and 'BOOL' in data_type_list and 'VARCHAR' in data_type_list and 'NCHAR' in data_type_list and 'GEOMETRY' in data_type_list and 'VARBINARY' in data_type_list)
         tdSql.execute("create view v2 as select * from tb where c1 >5 and c7 like '%ab%';")
         self.check_view_num(2)
-        tdSql.error("create view v3 as select * from tb where c1 like '%ab%';", expectErrInfo='Invalid value type')
+        tdSql.error("create view v3 as select * from tb where c1 like '%ab%';", expectErrInfo='Invalid operation')
         tdSql.execute("create view v3 as select first(ts), sum(c1) from tb group by c2 having avg(c4) > 0;")
         tdSql.execute("create view v4 as select _wstart,sum(c6) from tb interval(10s);")
         tdSql.execute("create view v5 as select * from tb join v2 on tb.ts = v2.ts;")
diff --git a/utils/test/c/tmq_taosx_ci.c b/utils/test/c/tmq_taosx_ci.c
index ff89bb1f75..8a7074844a 100644
--- a/utils/test/c/tmq_taosx_ci.c
+++ b/utils/test/c/tmq_taosx_ci.c
@@ -30,7 +30,7 @@ typedef struct {
   int  meta;
   int  srcVgroups;
   int  dstVgroups;
-  char dir[64];
+  char dir[256];
 } Config;
 
 Config g_conf = {0};
@@ -409,6 +409,30 @@ int buildStable(TAOS* pConn, TAOS_RES* pRes) {
   }
   taos_free_result(pRes);
 
+#ifdef WINDOWS
+  pRes = taos_query(pConn,
+                    "CREATE STABLE `meters_summary` (`_wstart` TIMESTAMP, `current` FLOAT, `groupid` INT, `location` VARCHAR(16)) TAGS (`group_id` BIGINT UNSIGNED)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table meters_summary, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn,
+                    "  CREATE TABLE `t_d2a450ee819dcf7576f0282d9ac22dbc` USING `meters_summary` (`group_id`) TAGS (13135550082773579308)");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create super table meters_summary, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "insert into t_d2a450ee819dcf7576f0282d9ac22dbc values (now, 120, 1, 'San Francisco')");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to insert into table d0, reason:%s\n", taos_errstr(pRes));
+    return -1;
+  }
+  taos_free_result(pRes);
+#else
   pRes = taos_query(pConn,
                     "create stream meters_summary_s trigger at_once IGNORE EXPIRED 0 into meters_summary as select _wstart, max(current) as current, "
                     "groupid, location from meters partition by groupid, location interval(10m)");
@@ -417,6 +441,7 @@ int buildStable(TAOS* pConn, TAOS_RES* pRes) {
     return -1;
   }
   taos_free_result(pRes);
+#endif
 
   pRes = taos_query(pConn, "insert into d0 (ts, current) values (now, 120)");
   if (taos_errno(pRes) != 0) {
@@ -598,8 +623,8 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
 }
 
 void initLogFile() {
-  char f1[256] = {0};
-  char f2[256] = {0};
+  char f1[1024] = {0};
+  char f2[1024] = {0};
 
   if (g_conf.snapShot) {
     sprintf(f1, "%s/../log/tmq_taosx_tmp_snapshot.source", g_conf.dir);