Merge branch 'develop' into feature/query

cmake/version.inc

@@ -4,7 +4,7 @@ PROJECT(TDengine)
 IF (DEFINED VERNUMBER)
   SET(TD_VER_NUMBER ${VERNUMBER})
 ELSE ()
-  SET(TD_VER_NUMBER "2.0.6.0")
+  SET(TD_VER_NUMBER "2.0.7.0")
 ENDIF ()
 
 IF (DEFINED VERCOMPATIBLE)

packaging/deb/makedeb.sh

@@ -48,7 +48,7 @@ cp ${compile_dir}/../packaging/deb/taosd            ${pkg_dir}${install_home_pat
 cp ${compile_dir}/../packaging/tools/post.sh        ${pkg_dir}${install_home_path}/script
 cp ${compile_dir}/../packaging/tools/preun.sh       ${pkg_dir}${install_home_path}/script
 cp ${compile_dir}/build/bin/taosdemo                ${pkg_dir}${install_home_path}/bin
-#cp ${compile_dir}/build/bin/taosdump                ${pkg_dir}${install_home_path}/bin
+cp ${compile_dir}/build/bin/taosdump                ${pkg_dir}${install_home_path}/bin
 cp ${compile_dir}/build/bin/taosd                   ${pkg_dir}${install_home_path}/bin
 cp ${compile_dir}/build/bin/taos                    ${pkg_dir}${install_home_path}/bin
 cp ${compile_dir}/build/lib/${libfile}              ${pkg_dir}${install_home_path}/driver 

packaging/rpm/tdengine.spec

@@ -58,7 +58,7 @@ cp %{_compiledir}/../packaging/tools/preun.sh       %{buildroot}%{homepath}/scri
 cp %{_compiledir}/build/bin/taos                    %{buildroot}%{homepath}/bin
 cp %{_compiledir}/build/bin/taosd                   %{buildroot}%{homepath}/bin
 cp %{_compiledir}/build/bin/taosdemo                %{buildroot}%{homepath}/bin
-#cp %{_compiledir}/build/bin/taosdump                %{buildroot}%{homepath}/bin
+cp %{_compiledir}/build/bin/taosdump                %{buildroot}%{homepath}/bin
 cp %{_compiledir}/build/lib/${libfile}              %{buildroot}%{homepath}/driver
 cp %{_compiledir}/../src/inc/taos.h                 %{buildroot}%{homepath}/include
 cp %{_compiledir}/../src/inc/taoserror.h            %{buildroot}%{homepath}/include

packaging/tools/get_client.sh

+#!/bin/bash
+#
+
+log_dir=$1
+result_file=$2
+
+if [ ! -n "$1" ];then
+  echo "Pleas input the director of taosdlog."
+  echo "usage: ./get_client.sh <taosdlog directory> <result file>"
+  exit 1
+else
+  log_dir=$1
+fi
+
+if [ ! -n "$2" ];then
+  result_file=clientInfo.txt
+else
+  result_file=$2
+fi
+
+grep "new TCP connection" ${log_dir}/taosdlog.* | sed -e "s/0x.* from / /"|sed -e "s/,.*$//"|sed -e "s/:[0-9]*$//"|sort -r|uniq -f 2|sort -k 3 -r|uniq -f 2 > ${result_file}

packaging/tools/makeclient.sh

@@ -45,8 +45,7 @@ if [ "$osType" != "Darwin" ]; then
     strip ${build_dir}/bin/taos
     bin_files="${build_dir}/bin/taos ${script_dir}/remove_client.sh"
   else  
-    #bin_files="${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${script_dir}/remove_client.sh ${script_dir}/set_core.sh"
-    bin_files="${build_dir}/bin/taos ${build_dir}/bin/taosdemo ${script_dir}/remove_client.sh ${script_dir}/set_core.sh"
+    bin_files="${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${script_dir}/remove_client.sh ${script_dir}/set_core.sh ${script_dir}/get_client.sh"
   fi
   lib_files="${build_dir}/lib/libtaos.so.${version}"
 else

packaging/tools/makeclient_power.sh

@@ -76,8 +76,10 @@ if [ "$osType" != "Darwin" ]; then
   else 
     cp ${build_dir}/bin/taos          ${install_dir}/bin/power
     cp ${script_dir}/remove_power.sh  ${install_dir}/bin
-    cp ${build_dir}/bin/taosdemo      ${install_dir}/bin/powerdemo   
+    cp ${build_dir}/bin/taosdemo      ${install_dir}/bin/powerdemo 
+    cp ${build_dir}/bin/taosdump      ${install_dir}/bin/powerdump  
     cp ${script_dir}/set_core.sh      ${install_dir}/bin
+    cp ${script_dir}/get_client.sh    ${install_dir}/bin
   fi
 else
   cp ${bin_files} ${install_dir}/bin

packaging/tools/makepkg.sh

@@ -36,8 +36,7 @@ if [ "$pagMode" == "lite" ]; then
   strip ${build_dir}/bin/taos
   bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${script_dir}/remove.sh"
 else 
-  #bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${build_dir}/bin/tarbitrator ${script_dir}/remove.sh ${script_dir}/set_core.sh"
-  bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${build_dir}/bin/taosdemo ${build_dir}/bin/tarbitrator ${script_dir}/remove.sh ${script_dir}/set_core.sh"
+  bin_files="${build_dir}/bin/taosd ${build_dir}/bin/taos ${build_dir}/bin/taosdump ${build_dir}/bin/taosdemo ${build_dir}/bin/tarbitrator ${script_dir}/remove.sh ${script_dir}/set_core.sh ${script_dir}/get_client.sh"
 fi
 
 lib_files="${build_dir}/lib/libtaos.so.${version}"

packaging/tools/makepkg_power.sh

@@ -77,8 +77,10 @@ else
   cp ${build_dir}/bin/taosd         ${install_dir}/bin/powerd
   cp ${script_dir}/remove_power.sh  ${install_dir}/bin
   cp ${build_dir}/bin/taosdemo      ${install_dir}/bin/powerdemo   
+  cp ${build_dir}/bin/taosdump      ${install_dir}/bin/powerdump  
   cp ${build_dir}/bin/tarbitrator   ${install_dir}/bin
   cp ${script_dir}/set_core.sh      ${install_dir}/bin
+  cp ${script_dir}/get_client.sh    ${install_dir}/bin
 fi
 chmod a+x ${install_dir}/bin/* || :
 

snap/snapcraft.yaml

 name: tdengine
 base: core18
-version: '2.0.6.0'
+version: '2.0.7.0'
 icon: snap/gui/t-dengine.svg
 summary: an open-source big data platform designed and optimized for IoT. 
 description: |

src/client/src/tscServer.c

@@ -241,11 +241,7 @@ int tscSendMsgToServer(SSqlObj *pSql) {
       .code    = 0
   };
 
-  // NOTE: the rpc context should be acquired before sending data to server.
-  // Otherwise, the pSql object may have been released already during the response function, which is
-  // processMsgFromServer function. In the meanwhile, the assignment of the rpc context to sql object will absolutely
-  // cause crash.
-  pSql->rpcRid = rpcSendRequest(pObj->pDnodeConn, &pSql->epSet, &rpcMsg);
+  rpcSendRequest(pObj->pDnodeConn, &pSql->epSet, &rpcMsg, &pSql->rpcRid);
   return TSDB_CODE_SUCCESS;
 }
 

src/client/src/tscSystem.c

@@ -105,6 +105,7 @@ void taos_init_imp(void) {
     taosReadGlobalCfg();
     taosCheckGlobalCfg();
 
+    rpcInit();
     tscDebug("starting to initialize TAOS client ...");
     tscDebug("Local End Point is:%s", tsLocalEp);
   }
@@ -179,6 +180,7 @@ void taos_cleanup(void) {
   taosCloseRef(tscRefId);
   taosCleanupKeywordsTable();
   taosCloseLog();
+  if (tscEmbedded == 0) rpcCleanup();
 
   m = tscTmr;
   if (m != NULL && atomic_val_compare_exchange_ptr(&tscTmr, m, 0) == m) {

src/dnode/src/dnodeMain.c

@@ -20,6 +20,7 @@
 #include "tconfig.h"
 #include "tglobal.h"
 #include "twal.h"
+#include "trpc.h"
 #include "dnode.h"
 #include "dnodeInt.h"
 #include "dnodeMgmt.h"
@@ -54,6 +55,7 @@ typedef struct {
 } SDnodeComponent;
 
 static const SDnodeComponent tsDnodeComponents[] = {
+  {"rpc",       rpcInit,             rpcCleanup},
   {"storage",   dnodeInitStorage,    dnodeCleanupStorage},
   {"dnodecfg",  dnodeInitCfg,        dnodeCleanupCfg},
   {"dnodeeps",  dnodeInitEps,        dnodeCleanupEps},

src/dnode/src/dnodePeer.c

@@ -169,7 +169,7 @@ void dnodeAddClientRspHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg)) {
 }
 
 void dnodeSendMsgToDnode(SRpcEpSet *epSet, SRpcMsg *rpcMsg) {
-  rpcSendRequest(tsClientRpc, epSet, rpcMsg);
+  rpcSendRequest(tsClientRpc, epSet, rpcMsg, NULL);
 }
 
 void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
@@ -180,4 +180,4 @@ void dnodeSendMsgToMnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp) {
 
 void dnodeSendMsgToDnodeRecv(SRpcMsg *rpcMsg, SRpcMsg *rpcRsp, SRpcEpSet *epSet) {
   rpcSendRecv(tsClientRpc, epSet, rpcMsg, rpcRsp);
-}
\ No newline at end of file
+}

src/inc/trpc.h

@@ -78,12 +78,14 @@ typedef struct SRpcInit {
   int  (*afp)(char *tableId, char *spi, char *encrypt, char *secret, char *ckey);
 } SRpcInit;
 
+int32_t rpcInit();
+void  rpcCleanup();
 void *rpcOpen(const SRpcInit *pRpc);
 void  rpcClose(void *);
 void *rpcMallocCont(int contLen);
 void  rpcFreeCont(void *pCont);
 void *rpcReallocCont(void *ptr, int contLen);
-int64_t rpcSendRequest(void *thandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg);
+void  rpcSendRequest(void *thandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg, int64_t *rid);
 void  rpcSendResponse(const SRpcMsg *pMsg);
 void  rpcSendRedirectRsp(void *pConn, const SRpcEpSet *pEpSet); 
 int   rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);

src/kit/CMakeLists.txt

@@ -3,3 +3,4 @@ PROJECT(TDengine)
 
 ADD_SUBDIRECTORY(shell)
 ADD_SUBDIRECTORY(taosdemo)
+ADD_SUBDIRECTORY(taosdump)

src/kit/shell/src/shellLinux.c

@@ -46,7 +46,7 @@ static struct argp_option options[] = {
   {"thread",     'T', "THREADNUM",  0,                   "Number of threads when using multi-thread to import data."},
   {"database",   'd', "DATABASE",   0,                   "Database to use when connecting to the server."},
   {"timezone",   't', "TIMEZONE",   0,                   "Time zone of the shell, default is local."},
-  {"netrole",    'n', "NETROLE",    0,                   "Net role when network connectivity test, default is NULL, valid option: client | server."},
+  {"netrole",    'n', "NETROLE",    0,                   "Net role when network connectivity test, default is NULL, options: client|clients|server."},
   {"endport",    'e', "ENDPORT",    0,                   "Net test end port, default is 6042."},
   {"pktlen",     'l', "PKTLEN",     0,                   "Packet length used for net test, default is 1000 bytes."},
   {0}};

src/kit/taosdump/CMakeLists.txt

+CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
+PROJECT(TDengine)
+
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
+INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
+INCLUDE_DIRECTORIES(inc)
+AUX_SOURCE_DIRECTORY(. SRC)
+
+IF (TD_LINUX)
+  ADD_EXECUTABLE(taosdump ${SRC})
+  IF (TD_SOMODE_STATIC)
+    TARGET_LINK_LIBRARIES(taosdump taos_static)
+  ELSE ()
+    TARGET_LINK_LIBRARIES(taosdump taos)
+  ENDIF ()  
+ENDIF ()

src/kit/taosdump/taosdump.sh

+taos1_6="/root/mnt/work/test/td1.6/build/bin/taos"
+taosdump1_6="/root/mnt/work/test/td1.6/build/bin/taosdump"
+taoscfg1_6="/root/mnt/work/test/td1.6/test/cfg"
+
+taos2_0="/root/mnt/work/test/td2.0/build/bin/taos"
+taosdump2_0="/root/mnt/work/test/td2.0/build/bin/taosdump"
+taoscfg2_0="/root/mnt/work/test/td2.0/test/cfg"
+
+data_dir="/root/mnt/work/test/td1.6/output"
+table_list="/root/mnt/work/test/td1.6/tables"
+
+DBNAME="test"
+NTABLES=$(wc -l ${table_list} | awk '{print $1;}')
+NTABLES_PER_DUMP=101
+
+mkdir -p ${data_dir}
+i=0
+round=0
+command="${taosdump1_6} -c ${taoscfg1_6} -o ${data_dir} -N 100 -T 20 ${DBNAME}"
+while IFS= read -r line
+do
+  i=$((i+1))
+
+  command="${command} ${line}"
+
+  if [[ "$i" -eq ${NTABLES_PER_DUMP} ]]; then
+    round=$((round+1))
+    echo "Starting round ${round} dump out..."
+    rm -f ${data_dir}/*
+    ${command}
+    echo "Starting round ${round} dump in..."
+    ${taosdump2_0} -c ${taoscfg2_0} -i ${data_dir}
+
+    # Reset variables
+    # command="${taosdump1_6} -c ${taoscfg1_6} -o ${data_dir} -N 100 ${DBNAME}"
+    command="${taosdump1_6} -c ${taoscfg1_6} -o ${data_dir} -N 100 -T 20 ${DBNAME}"
+    i=0
+  fi
+done < "${table_list}"
+
+if [[ ${i} -ne "0" ]]; then
+  round=$((round+1))
+  echo "Starting round ${round} dump out..."
+  rm -f ${data_dir}/*
+  ${command}
+  echo "Starting round ${round} dump in..."
+  ${taosdump2_0} -c ${taoscfg2_0} -i ${data_dir}
+fi

src/rpc/src/rpcMain.c

@@ -135,7 +135,7 @@ int tsRpcOverhead;
 
 static int     tsRpcRefId = -1;
 static int32_t tsRpcNum = 0;
-static pthread_once_t tsRpcInit = PTHREAD_ONCE_INIT;
+//static pthread_once_t tsRpcInit = PTHREAD_ONCE_INIT;
 
 // server:0 client:1  tcp:2 udp:0
 #define RPC_CONN_UDPS   0
@@ -221,13 +221,15 @@ static void rpcFree(void *p) {
   free(p);
 }
 
-void rpcInit(void) {
+int32_t rpcInit(void) {
   tsProgressTimer = tsRpcTimer/2; 
   tsRpcMaxRetry = tsRpcMaxTime * 1000/tsProgressTimer;
   tsRpcHeadSize = RPC_MSG_OVERHEAD; 
   tsRpcOverhead = sizeof(SRpcReqContext);
 
   tsRpcRefId = taosOpenRef(200, rpcFree);
+
+  return 0;
 }
  
 void rpcCleanup(void) {
@@ -238,7 +240,7 @@ void rpcCleanup(void) {
 void *rpcOpen(const SRpcInit *pInit) {
   SRpcInfo *pRpc;
 
-  pthread_once(&tsRpcInit, rpcInit);
+  //pthread_once(&tsRpcInit, rpcInit);
 
   pRpc = (SRpcInfo *)calloc(1, sizeof(SRpcInfo));
   if (pRpc == NULL) return NULL;
@@ -379,7 +381,7 @@ void *rpcReallocCont(void *ptr, int contLen) {
   return start + sizeof(SRpcReqContext) + sizeof(SRpcHead);
 }
 
-int64_t rpcSendRequest(void *shandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg) {
+void rpcSendRequest(void *shandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg, int64_t *pRid) {
   SRpcInfo       *pRpc = (SRpcInfo *)shandle;
   SRpcReqContext *pContext;
 
@@ -405,14 +407,10 @@ int64_t rpcSendRequest(void *shandle, const SRpcEpSet *pEpSet, SRpcMsg *pMsg) {
     || type == TSDB_MSG_TYPE_CM_SHOW )
     pContext->connType = RPC_CONN_TCPC;
   
-  // set the handle to pContext, so app can cancel the request
-  if (pMsg->handle) *((void **)pMsg->handle) = pContext;
-
   pContext->rid = taosAddRef(tsRpcRefId, pContext);
+  if (pRid) *pRid = pContext->rid;
 
   rpcSendReqToServer(pRpc, pContext);
-
-  return pContext->rid;
 }
 
 void rpcSendResponse(const SRpcMsg *pRsp) {
@@ -528,7 +526,7 @@ void rpcSendRecv(void *shandle, SRpcEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp)
   pContext->pRsp = pRsp;
   pContext->pSet = pEpSet;
 
-  rpcSendRequest(shandle, pEpSet, pMsg);
+  rpcSendRequest(shandle, pEpSet, pMsg, NULL);
 
   tsem_wait(&sem);
   tsem_destroy(&sem);

src/rpc/test/rclient.c

@@ -57,7 +57,7 @@ static void *sendRequest(void *param) {
     rpcMsg.ahandle = pInfo;
     rpcMsg.msgType = 1;
     tDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
-    rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg);
+    rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg, NULL);
     if ( pInfo->num % 20000 == 0 ) 
       tInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num);
     tsem_wait(&pInfo->rspSem);

src/sync/src/syncMain.c

@@ -578,6 +578,7 @@ static void syncChooseMaster(SSyncNode *pNode) {
 
 #if 0
       for (int32_t i = 0; i < pNode->replica; ++i) {
+        if (i == index) continue;
         pPeer = pNode->peerInfo[i];
         if (pPeer->version == nodeVersion) {
           pPeer->role = TAOS_SYNC_ROLE_SLAVE;

src/sync/src/syncRetrieve.c

@@ -182,6 +182,8 @@ static int32_t syncReadOneWalRecord(int32_t sfd, SWalHead *pHead, uint32_t *pEve
     return 0;
   }
 
+  assert(pHead->len <= TSDB_MAX_WAL_SIZE);
+
   ret = read(sfd, pHead->cont, pHead->len);
   if (ret < 0) return -1;
 

src/sync/test/syncClient.c

@@ -57,7 +57,7 @@ void *sendRequest(void *param) {
     rpcMsg.ahandle = pInfo;
     rpcMsg.msgType = 1;
     uDebug("thread:%d, send request, contLen:%d num:%d", pInfo->index, pInfo->msgSize, pInfo->num);
-    rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg);
+    rpcSendRequest(pInfo->pRpc, &pInfo->epSet, &rpcMsg, NULL);
     if (pInfo->num % 20000 == 0) {
       uInfo("thread:%d, %d requests have been sent", pInfo->index, pInfo->num);
     }

src/tsdb/inc/tsdbMain.h

@@ -208,6 +208,18 @@ typedef struct {
 } SFileGroupIter;
 
 // ------------------ tsdbMain.c
+typedef struct {
+  int32_t  totalLen;
+  int32_t  len;
+  SDataRow row;
+} SSubmitBlkIter;
+
+typedef struct {
+  int32_t totalLen;
+  int32_t len;
+  void *  pMsg;
+} SSubmitMsgIter;
+
 typedef struct {
   int8_t state;
 
@@ -430,7 +442,6 @@ void          tsdbCloseBufPool(STsdbRepo* pRepo);
 SListNode*    tsdbAllocBufBlockFromPool(STsdbRepo* pRepo);
 
 // ------------------ tsdbMemTable.c
-int   tsdbUpdateRowInMem(STsdbRepo* pRepo, SDataRow row, STable* pTable);
 int   tsdbRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbUnRefMemTable(STsdbRepo* pRepo, SMemTable* pMemTable);
 int   tsdbTakeMemSnapshot(STsdbRepo* pRepo, SMemTable** pMem, SMemTable** pIMem);

src/tsdb/src/tsdbFile.c

@@ -516,7 +516,7 @@ void tsdbGetFileInfoImpl(char *fname, uint32_t *magic, int64_t *size) {
   SFile         file;
   SFile *       pFile = &file;
 
-  strncpy(pFile->fname, fname, TSDB_FILENAME_LEN);
+  strncpy(pFile->fname, fname, TSDB_FILENAME_LEN - 1);
   pFile->fd = -1;
 
   if (tsdbOpenFile(pFile, O_RDONLY) < 0) goto _err;

src/tsdb/src/tsdbMain.c

@@ -32,18 +32,6 @@
 #define TSDB_DEFAULT_COMPRESSION TWO_STAGE_COMP
 #define IS_VALID_COMPRESSION(compression) (((compression) >= NO_COMPRESSION) && ((compression) <= TWO_STAGE_COMP))
 
-typedef struct {
-  int32_t  totalLen;
-  int32_t  len;
-  SDataRow row;
-} SSubmitBlkIter;
-
-typedef struct {
-  int32_t totalLen;
-  int32_t len;
-  void *  pMsg;
-} SSubmitMsgIter;
-
 static int32_t     tsdbCheckAndSetDefaultCfg(STsdbCfg *pCfg);
 static int32_t     tsdbSetRepoEnv(char *rootDir, STsdbCfg *pCfg);
 static int32_t     tsdbUnsetRepoEnv(char *rootDir);
@@ -52,20 +40,13 @@ static int         tsdbLoadConfig(char *rootDir, STsdbCfg *pCfg);
 static char *      tsdbGetCfgFname(char *rootDir);
 static STsdbRepo * tsdbNewRepo(char *rootDir, STsdbAppH *pAppH, STsdbCfg *pCfg);
 static void        tsdbFreeRepo(STsdbRepo *pRepo);
-static int         tsdbInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter);
-static int32_t     tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, TSKEY now, int32_t *affectedrows);
-static int         tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock);
-static SDataRow    tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter);
 static int         tsdbRestoreInfo(STsdbRepo *pRepo);
-static int         tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter);
 static void        tsdbAlterCompression(STsdbRepo *pRepo, int8_t compression);
 static int         tsdbAlterKeep(STsdbRepo *pRepo, int32_t keep);
 static int         tsdbAlterCacheTotalBlocks(STsdbRepo *pRepo, int totalBlocks);
 static int         keyFGroupCompFunc(const void *key, const void *fgroup);
 static int         tsdbEncodeCfg(void **buf, STsdbCfg *pCfg);
 static void *      tsdbDecodeCfg(void *buf, STsdbCfg *pCfg);
-static int         tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable);
-static int         tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg);
 static void        tsdbStartStream(STsdbRepo *pRepo);
 static void        tsdbStopStream(STsdbRepo *pRepo);
 
@@ -177,40 +158,6 @@ void tsdbCloseRepo(TSDB_REPO_T *repo, int toCommit) {
   tsdbDebug("vgId:%d repository is closed", vgId);
 }
 
-int32_t tsdbInsertData(TSDB_REPO_T *repo, SSubmitMsg *pMsg, SShellSubmitRspMsg *pRsp) {
-  STsdbRepo *    pRepo = (STsdbRepo *)repo;
-  SSubmitMsgIter msgIter = {0};
-
-  if (tsdbScanAndConvertSubmitMsg(pRepo, pMsg) < 0) {
-    if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-      tsdbError("vgId:%d failed to insert data since %s", REPO_ID(pRepo), tstrerror(terrno));
-    }
-    return -1;
-  }
-
-  if (tsdbInitSubmitMsgIter(pMsg, &msgIter) < 0) {
-    tsdbError("vgId:%d failed to insert data since %s", REPO_ID(pRepo), tstrerror(terrno));
-    return -1;
-  }
-
-  SSubmitBlk *pBlock = NULL;
-  int32_t     affectedrows = 0;
-
-  TSKEY now = taosGetTimestamp(pRepo->config.precision);
-  while (true) {
-    tsdbGetSubmitMsgNext(&msgIter, &pBlock);
-    if (pBlock == NULL) break;
-    if (tsdbInsertDataToTable(pRepo, pBlock, now, &affectedrows) < 0) {
-      return -1;
-    }
-  }
-
-  if (pRsp != NULL) pRsp->affectedRows = htonl(affectedrows);
-
-  if (tsdbCheckCommit(pRepo) < 0) return -1;
-  return 0;
-}
-
 uint32_t tsdbGetFileInfo(TSDB_REPO_T *repo, char *name, uint32_t *index, uint32_t eindex, int64_t *size) {
   STsdbRepo *pRepo = (STsdbRepo *)repo;
   // STsdbMeta *pMeta = pRepo->tsdbMeta;
@@ -735,93 +682,6 @@ static void tsdbFreeRepo(STsdbRepo *pRepo) {
   }
 }
 
-static int tsdbInitSubmitMsgIter(SSubmitMsg *pMsg, SSubmitMsgIter *pIter) {
-  if (pMsg == NULL) {
-    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
-    return -1;
-  }
-
-  pIter->totalLen = pMsg->length;
-  pIter->len = 0;
-  pIter->pMsg = pMsg;
-  if (pMsg->length <= TSDB_SUBMIT_MSG_HEAD_SIZE) {
-    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
-    return -1;
-  }
-
-  return 0;
-}
-
-static int32_t tsdbInsertDataToTable(STsdbRepo *pRepo, SSubmitBlk *pBlock, TSKEY now, int32_t *affectedrows) {
-  STsdbMeta *pMeta = pRepo->tsdbMeta;
-  int64_t    points = 0;
-
-  ASSERT(pBlock->tid < pMeta->maxTables);
-  STable *pTable = pMeta->tables[pBlock->tid];
-  ASSERT(pTable != NULL && TABLE_UID(pTable) == pBlock->uid);
-
-  SSubmitBlkIter blkIter = {0};
-  SDataRow       row = NULL;
-
-  TSKEY minKey = now - tsMsPerDay[pRepo->config.precision] * pRepo->config.keep;
-  TSKEY maxKey = now + tsMsPerDay[pRepo->config.precision] * pRepo->config.daysPerFile;
-
-  tsdbInitSubmitBlkIter(pBlock, &blkIter);
-  while ((row = tsdbGetSubmitBlkNext(&blkIter)) != NULL) {
-    if (dataRowKey(row) < minKey || dataRowKey(row) > maxKey) {
-      tsdbError("vgId:%d table %s tid %d uid %" PRIu64 " timestamp is out of range! now %" PRId64 " minKey %" PRId64
-                " maxKey %" PRId64,
-                REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), now, minKey, maxKey);
-      terrno = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;
-      return -1;
-    }
-
-    if (tsdbUpdateRowInMem(pRepo, row, pTable) < 0) return -1;
-
-    (*affectedrows)++;
-    points++;
-  }
-
-  STSchema *pSchema = tsdbGetTableSchemaByVersion(pTable, pBlock->sversion);
-  pRepo->stat.pointsWritten += points * schemaNCols(pSchema);
-  pRepo->stat.totalStorage += points * schemaVLen(pSchema);
-
-  return 0;
-}
-
-static int tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
-  if (pIter->len == 0) {
-    pIter->len += TSDB_SUBMIT_MSG_HEAD_SIZE;
-  } else {
-    SSubmitBlk *pSubmitBlk = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
-    pIter->len += (sizeof(SSubmitBlk) + pSubmitBlk->dataLen + pSubmitBlk->schemaLen);
-  }
-
-  if (pIter->len > pIter->totalLen) {
-    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
-    *pPBlock = NULL;
-    return -1;
-  }
-
-  *pPBlock = (pIter->len == pIter->totalLen) ? NULL : (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
-
-  return 0;
-}
-
-static SDataRow tsdbGetSubmitBlkNext(SSubmitBlkIter *pIter) {
-  SDataRow row = pIter->row;
-  if (row == NULL) return NULL;
-
-  pIter->len += dataRowLen(row);
-  if (pIter->len >= pIter->totalLen) {
-    pIter->row = NULL;
-  } else {
-    pIter->row = (char *)row + dataRowLen(row);
-  }
-
-  return row;
-}
-
 static int tsdbRestoreInfo(STsdbRepo *pRepo) {
   STsdbMeta * pMeta = pRepo->tsdbMeta;
   STsdbFileH *pFileH = pRepo->tsdbFileH;
@@ -855,14 +715,6 @@ _err:
   return -1;
 }
 
-static int tsdbInitSubmitBlkIter(SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
-  if (pBlock->dataLen <= 0) return -1;
-  pIter->totalLen = pBlock->dataLen;
-  pIter->len = 0;
-  pIter->row = (SDataRow)(pBlock->data+pBlock->schemaLen);
-  return 0;
-}
-
 static void tsdbAlterCompression(STsdbRepo *pRepo, int8_t compression) {
   int8_t ocompression = pRepo->config.compression;
   pRepo->config.compression = compression;
@@ -959,134 +811,6 @@ static void *tsdbDecodeCfg(void *buf, STsdbCfg *pCfg) {
   return buf;
 }
 
-static int tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable) {
-  ASSERT(pTable != NULL);
-
-  STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);
-  int       sversion = schemaVersion(pSchema);
-
-  if (pBlock->sversion == sversion) {
-    return 0;
-  } else {
-    if (TABLE_TYPE(pTable) == TSDB_STREAM_TABLE) {  // stream table is not allowed to change schema
-      terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
-      return -1;
-    }
-  }
-
-  if (pBlock->sversion > sversion) {  // may need to update table schema
-    if (pBlock->schemaLen > 0) {
-      tsdbDebug(
-          "vgId:%d table %s tid %d uid %" PRIu64 " schema version %d is out of data, client version %d, update...",
-          REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), sversion, pBlock->sversion);
-      ASSERT(pBlock->schemaLen % sizeof(STColumn) == 0);
-      int       numOfCols = pBlock->schemaLen / sizeof(STColumn);
-      STColumn *pTCol = (STColumn *)pBlock->data;
-
-      STSchemaBuilder schemaBuilder = {0};
-      if (tdInitTSchemaBuilder(&schemaBuilder, pBlock->sversion) < 0) {
-        terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-        tsdbError("vgId:%d failed to update schema of table %s since %s", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable),
-                  tstrerror(terrno));
-        return -1;
-      }
-
-      for (int i = 0; i < numOfCols; i++) {
-        if (tdAddColToSchema(&schemaBuilder, pTCol[i].type, htons(pTCol[i].colId), htons(pTCol[i].bytes)) < 0) {
-          terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-          tsdbError("vgId:%d failed to update schema of table %s since %s", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable),
-                    tstrerror(terrno));
-          tdDestroyTSchemaBuilder(&schemaBuilder);
-          return -1;
-        }
-      }
-
-      STSchema *pNSchema = tdGetSchemaFromBuilder(&schemaBuilder);
-      if (pNSchema == NULL) {
-        terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
-        tdDestroyTSchemaBuilder(&schemaBuilder);
-        return -1;
-      }
-
-      tdDestroyTSchemaBuilder(&schemaBuilder);
-      tsdbUpdateTableSchema(pRepo, pTable, pNSchema, true);
-    } else {
-      tsdbDebug(
-          "vgId:%d table %s tid %d uid %" PRIu64 " schema version %d is out of data, client version %d, reconfigure...",
-          REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), TABLE_TID(pTable), TABLE_UID(pTable), sversion, pBlock->sversion);
-      terrno = TSDB_CODE_TDB_TABLE_RECONFIGURE;
-      return -1;
-    }
-  } else {
-    ASSERT(pBlock->sversion >= 0);
-    if (tsdbGetTableSchemaImpl(pTable, false, false, pBlock->sversion) == NULL) {
-      tsdbError("vgId:%d invalid submit schema version %d to table %s tid %d from client", REPO_ID(pRepo),
-                pBlock->sversion, TABLE_CHAR_NAME(pTable), TABLE_TID(pTable));
-    }
-    terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
-    return -1;
-  }
-
-  return 0;
-}
-
-static int tsdbScanAndConvertSubmitMsg(STsdbRepo *pRepo, SSubmitMsg *pMsg) {
-  ASSERT(pMsg != NULL);
-  STsdbMeta *    pMeta = pRepo->tsdbMeta;
-  SSubmitMsgIter msgIter = {0};
-  SSubmitBlk *   pBlock = NULL;
-
-  terrno = TSDB_CODE_SUCCESS;
-  pMsg->length = htonl(pMsg->length);
-  pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
-
-  if (tsdbInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
-  while (true) {
-    if (tsdbGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
-    if (pBlock == NULL) break;
-
-    pBlock->uid = htobe64(pBlock->uid);
-    pBlock->tid = htonl(pBlock->tid);
-    pBlock->sversion = htonl(pBlock->sversion);
-    pBlock->dataLen = htonl(pBlock->dataLen);
-    pBlock->schemaLen = htonl(pBlock->schemaLen);
-    pBlock->numOfRows = htons(pBlock->numOfRows);
-
-    if (pBlock->tid <= 0 || pBlock->tid >= pMeta->maxTables) {
-      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pRepo), pBlock->uid,
-                pBlock->tid);
-      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
-      return -1;
-    }
-
-    STable *pTable = pMeta->tables[pBlock->tid];
-    if (pTable == NULL || TABLE_UID(pTable) != pBlock->uid) {
-      tsdbError("vgId:%d failed to get table to insert data, uid %" PRIu64 " tid %d", REPO_ID(pRepo), pBlock->uid,
-                pBlock->tid);
-      terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
-      return -1;
-    }
-
-    if (TABLE_TYPE(pTable) == TSDB_SUPER_TABLE) {
-      tsdbError("vgId:%d invalid action trying to insert a super table %s", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable));
-      terrno = TSDB_CODE_TDB_INVALID_ACTION;
-      return -1;
-    }
-
-    // Check schema version and update schema if needed
-    if (tsdbCheckTableSchema(pRepo, pBlock, pTable) < 0) {
-      if (terrno == TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-        continue;
-      } else {
-        return -1;
-      }
-    }
-  }
-
-  if (terrno != TSDB_CODE_SUCCESS) return -1;
-  return 0;
-}
-
 static int tsdbAlterCacheTotalBlocks(STsdbRepo *pRepo, int totalBlocks) {
   // TODO
   // STsdbCache *pCache = pRepo->tsdbCache;

src/tsdb/src/tsdbRWHelper.c

@@ -1595,7 +1595,7 @@ static int tsdbProcessMergeCommit(SRWHelper *pHelper, SCommitIter *pCommitIter,
       tblkIdx++;
     } else if (oBlock.numOfRows + pMergeInfo->rowsInserted - pMergeInfo->rowsDeleteSucceed == 0) {
       // Delete the block and do some stuff
-      ASSERT(pMergeInfo->keyFirst == INT64_MAX && pMergeInfo->keyFirst == INT64_MIN);
+      // ASSERT(pMergeInfo->keyFirst == INT64_MAX && pMergeInfo->keyFirst == INT64_MIN);
       if (tsdbDeleteSuperBlock(pHelper, tblkIdx) < 0) return -1;
       *pCommitIter->pIter = slIter;
       if (oBlock.last && pHelper->hasOldLastBlock) pHelper->hasOldLastBlock = false;

src/util/inc/tskiplist.h

@@ -131,6 +131,7 @@ SSkipList *tSkipListCreate(uint8_t maxLevel, uint8_t keyType, uint16_t keyLen, _
                            __sl_key_fn_t fn);
 void       tSkipListDestroy(SSkipList *pSkipList);
 SSkipListNode *    tSkipListPut(SSkipList *pSkipList, void *pData);
+void               tSkipListPutBatch(SSkipList *pSkipList, void **ppData, int ndata);
 SArray *           tSkipListGet(SSkipList *pSkipList, SSkipListKey pKey);
 void               tSkipListPrint(SSkipList *pSkipList, int16_t nlevel);
 SSkipListIterator *tSkipListCreateIter(SSkipList *pSkipList);

src/util/src/tlog.c

@@ -287,17 +287,17 @@ static int32_t taosOpenLogFile(char *fn, int32_t maxLines, int32_t maxFileNum) {
   tsLogObj.fileNum = maxFileNum;
   taosGetLogFileName(fn);
 
+
   if (strlen(fn) < LOG_FILE_NAME_LEN + 50 - 2) {
     strcpy(name, fn);
     strcat(name, ".0");
   }
+  bool log0Exist = stat(name, &logstat0) >= 0;
 
   if (strlen(fn) < LOG_FILE_NAME_LEN + 50 - 2) {
     strcpy(name, fn);
     strcat(name, ".1");
   }
-
-  bool log0Exist = stat(name, &logstat0) >= 0;
   bool log1Exist = stat(name, &logstat1) >= 0;
   
   // if none of the log files exist, open 0, if both exists, open the old one

src/util/src/tref.c

@@ -426,11 +426,11 @@ static int taosDecRefCount(int rsetId, int64_t rid, int remove) {
 		
       (*pSet->fp)(pNode->p); 
 
-      uTrace("rsetId:%d p:%p rid:%" PRId64 "is removed, count:%d, free mem: %p", rsetId, pNode->p, rid, pSet->count, pNode);
+      uTrace("rsetId:%d p:%p rid:%" PRId64 " is removed, count:%d, free mem: %p", rsetId, pNode->p, rid, pSet->count, pNode);
       free(pNode);
       released = 1;
     } else {
-      uTrace("rsetId:%d p:%p rid:%" PRId64 "is released, count:%d", rsetId, pNode->p, rid, pNode->count);
+      uTrace("rsetId:%d p:%p rid:%" PRId64 " is released, count:%d", rsetId, pNode->p, rid, pNode->count);
     }
   } else {
     uTrace("rsetId:%d rid:%" PRId64 " is not there, failed to release/remove", rsetId, rid);

src/util/src/tskiplist.c

@@ -24,10 +24,12 @@ static SSkipListNode *    getPriorNode(SSkipList *pSkipList, const char *val, in
 static void               tSkipListRemoveNodeImpl(SSkipList *pSkipList, SSkipListNode *pNode);
 static void               tSkipListCorrectLevel(SSkipList *pSkipList);
 static SSkipListIterator *doCreateSkipListIterator(SSkipList *pSkipList, int32_t order);
-static void               tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **backward, SSkipListNode *pNode);
-static bool               tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward, void *pData);
-static SSkipListNode *    tSkipListNewNode(uint8_t level);
+static void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **direction, SSkipListNode *pNode, bool isForward);
+static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward, void *pData);
+static SSkipListNode *tSkipListNewNode(uint8_t level);
 #define tSkipListFreeNode(n) tfree((n))
+static SSkipListNode *tSkipListPutImpl(SSkipList *pSkipList, void *pData, SSkipListNode **direction, bool isForward,
+                                       bool hasDup);
 
 static FORCE_INLINE int     tSkipListWLock(SSkipList *pSkipList);
 static FORCE_INLINE int     tSkipListRLock(SSkipList *pSkipList);
@@ -109,30 +111,85 @@ SSkipListNode *tSkipListPut(SSkipList *pSkipList, void *pData) {
   if (pSkipList == NULL || pData == NULL) return NULL;
 
   SSkipListNode *backward[MAX_SKIP_LIST_LEVEL] = {0};
-  uint8_t        dupMode = SL_DUP_MODE(pSkipList);
   SSkipListNode *pNode = NULL;
 
   tSkipListWLock(pSkipList);
 
   bool hasDup = tSkipListGetPosToPut(pSkipList, backward, pData);
+  pNode = tSkipListPutImpl(pSkipList, pData, backward, false, hasDup);
 
-  if (hasDup && (dupMode == SL_DISCARD_DUP_KEY || dupMode == SL_UPDATE_DUP_KEY)) {
-    if (dupMode == SL_UPDATE_DUP_KEY) {
-      pNode = SL_NODE_GET_BACKWARD_POINTER(backward[0], 0);
-      atomic_store_ptr(&(pNode->pData), pData);
-    }
-  } else {
-    pNode = tSkipListNewNode(getSkipListRandLevel(pSkipList));
-    if (pNode != NULL) {
-      pNode->pData = pData;
+  tSkipListUnlock(pSkipList);
+
+  return pNode;
+}
+
+// Put a batch of data into skiplist. The batch of data must be in ascending order
+void tSkipListPutBatch(SSkipList *pSkipList, void **ppData, int ndata) {
+  SSkipListNode *backward[MAX_SKIP_LIST_LEVEL] = {0};
+  SSkipListNode *forward[MAX_SKIP_LIST_LEVEL] = {0};
+  bool           hasDup = false;
+  char *         pKey = NULL;
+  char *         pDataKey = NULL;
+  int            compare = 0;
+
+  tSkipListWLock(pSkipList);
 
-      tSkipListDoInsert(pSkipList, backward, pNode);
+  // backward to put the first data
+  hasDup = tSkipListGetPosToPut(pSkipList, backward, ppData[0]);
+  tSkipListPutImpl(pSkipList, ppData[0], backward, false, hasDup);
+
+  for (int level = 0; level < pSkipList->maxLevel; level++) {
+    forward[level] = SL_NODE_GET_BACKWARD_POINTER(backward[level], level);
+  }
+
+  // forward to put the rest of data
+  for (int idata = 1; idata < ndata; idata++) {
+    pDataKey = pSkipList->keyFn(ppData[idata]);
+
+    // Compare max key
+    pKey = SL_GET_MAX_KEY(pSkipList);
+    compare = pSkipList->comparFn(pDataKey, pKey);
+    if (compare > 0) {
+      for (int i = 0; i < pSkipList->maxLevel; i++) {
+        forward[i] = SL_NODE_GET_BACKWARD_POINTER(pSkipList->pTail, i);
+      }
+
+      hasDup = false;
+    } else {
+      SSkipListNode *px = pSkipList->pHead;
+      for (int i = pSkipList->maxLevel - 1; i >= 0; --i) {
+        if (i < pSkipList->level) {
+          // set new px
+          if (forward[i] != pSkipList->pHead) {
+            if (px == pSkipList->pHead ||
+                pSkipList->comparFn(SL_GET_NODE_KEY(pSkipList, forward[i]), SL_GET_NODE_KEY(pSkipList, px)) > 0) {
+              px = forward[i];
+            }
+          }
+
+          SSkipListNode *p = SL_NODE_GET_FORWARD_POINTER(px, i);
+          while (p != pSkipList->pTail) {
+            pKey = SL_GET_NODE_KEY(pSkipList, p);
+
+            compare = pSkipList->comparFn(pKey, pDataKey);
+            if (compare >= 0) {
+              if (compare == 0) hasDup = true;
+              break;
+            } else {
+              px = p;
+              p = SL_NODE_GET_FORWARD_POINTER(px, i);
+            }
+          }
+        }
+
+        forward[i] = px;
+      }
     }
+
+    tSkipListPutImpl(pSkipList, ppData[idata], forward, true, hasDup);
   }
 
   tSkipListUnlock(pSkipList);
-
-  return pNode;
 }
 
 uint32_t tSkipListRemove(SSkipList *pSkipList, SSkipListKey key) {
@@ -310,22 +367,25 @@ void tSkipListPrint(SSkipList *pSkipList, int16_t nlevel) {
   }
 }
 
-static void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **backward, SSkipListNode *pNode) {
+static void tSkipListDoInsert(SSkipList *pSkipList, SSkipListNode **direction, SSkipListNode *pNode, bool isForward) {
   for (int32_t i = 0; i < pNode->level; ++i) {
-    if (i >= pSkipList->level) {
-      SL_NODE_GET_FORWARD_POINTER(pNode, i) = pSkipList->pTail;
-      SL_NODE_GET_BACKWARD_POINTER(pNode, i) = pSkipList->pHead;
-      SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, i) = pNode;
-      SL_NODE_GET_BACKWARD_POINTER(pSkipList->pTail, i) = pNode;
+    SSkipListNode *x = direction[i];
+    if (isForward) {
+      SL_NODE_GET_BACKWARD_POINTER(pNode, i) = x;
+
+      SSkipListNode *next = SL_NODE_GET_FORWARD_POINTER(x, i);
+      SL_NODE_GET_BACKWARD_POINTER(next, i) = pNode;
+
+      SL_NODE_GET_FORWARD_POINTER(pNode, i) = next;
+      SL_NODE_GET_FORWARD_POINTER(x, i) = pNode;
     } else {
-      SSkipListNode *x = backward[i];
       SL_NODE_GET_FORWARD_POINTER(pNode, i) = x;
 
       SSkipListNode *prev = SL_NODE_GET_BACKWARD_POINTER(x, i);
       SL_NODE_GET_FORWARD_POINTER(prev, i) = pNode;
 
-      SL_NODE_GET_BACKWARD_POINTER(x, i) = pNode;
       SL_NODE_GET_BACKWARD_POINTER(pNode, i) = prev;
+      SL_NODE_GET_BACKWARD_POINTER(x, i) = pNode;
     }
   }
 
@@ -377,7 +437,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
   char *  pDataKey = pSkipList->keyFn(pData);
 
   if (pSkipList->size == 0) {
-    for (int i = 0; i < pSkipList->level; i++) {
+    for (int i = 0; i < pSkipList->maxLevel; i++) {
       backward[i] = pSkipList->pTail;
     }
   } else {
@@ -387,7 +447,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
     pKey = SL_GET_MAX_KEY(pSkipList);
     compare = pSkipList->comparFn(pDataKey, pKey);
     if (compare >= 0) {
-      for (int i = 0; i < pSkipList->level; i++) {
+      for (int i = 0; i < pSkipList->maxLevel; i++) {
         backward[i] = pSkipList->pTail;
       }
 
@@ -398,7 +458,7 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
     pKey = SL_GET_MIN_KEY(pSkipList);
     compare = pSkipList->comparFn(pDataKey, pKey);
     if (compare < 0) {
-      for (int i = 0; i < pSkipList->level; i++) {
+      for (int i = 0; i < pSkipList->maxLevel; i++) {
         backward[i] = SL_NODE_GET_FORWARD_POINTER(pSkipList->pHead, i);
       }
 
@@ -406,18 +466,20 @@ static bool tSkipListGetPosToPut(SSkipList *pSkipList, SSkipListNode **backward,
     }
 
     SSkipListNode *px = pSkipList->pTail;
-    for (int i = pSkipList->level - 1; i >= 0; --i) {
-      SSkipListNode *p = SL_NODE_GET_BACKWARD_POINTER(px, i);
-      while (p != pSkipList->pHead) {
-        pKey = SL_GET_NODE_KEY(pSkipList, p);
-
-        compare = pSkipList->comparFn(pKey, pDataKey);
-        if (compare <= 0) {
-          if (compare == 0 && !hasDupKey) hasDupKey = true;
-          break;
-        } else {
-          px = p;
-          p = SL_NODE_GET_BACKWARD_POINTER(px, i);
+    for (int i = pSkipList->maxLevel - 1; i >= 0; --i) {
+      if (i < pSkipList->level) {
+        SSkipListNode *p = SL_NODE_GET_BACKWARD_POINTER(px, i);
+        while (p != pSkipList->pHead) {
+          pKey = SL_GET_NODE_KEY(pSkipList, p);
+
+          compare = pSkipList->comparFn(pKey, pDataKey);
+          if (compare <= 0) {
+            if (compare == 0 && !hasDupKey) hasDupKey = true;
+            break;
+          } else {
+            px = p;
+            p = SL_NODE_GET_BACKWARD_POINTER(px, i);
+          }
         }
       }
 
@@ -579,6 +641,32 @@ static SSkipListNode *tSkipListNewNode(uint8_t level) {
   return pNode;
 }
 
+static SSkipListNode *tSkipListPutImpl(SSkipList *pSkipList, void *pData, SSkipListNode **direction, bool isForward,
+                                       bool hasDup) {
+  uint8_t        dupMode = SL_DUP_MODE(pSkipList);
+  SSkipListNode *pNode = NULL;
+
+  if (hasDup && (dupMode == SL_DISCARD_DUP_KEY || dupMode == SL_UPDATE_DUP_KEY)) {
+    if (dupMode == SL_UPDATE_DUP_KEY) {
+      if (isForward) {
+        pNode = SL_NODE_GET_FORWARD_POINTER(direction[0], 0);
+      } else {
+        pNode = SL_NODE_GET_BACKWARD_POINTER(direction[0], 0);
+      }
+      atomic_store_ptr(&(pNode->pData), pData);
+    }
+  } else {
+    pNode = tSkipListNewNode(getSkipListRandLevel(pSkipList));
+    if (pNode != NULL) {
+      pNode->pData = pData;
+
+      tSkipListDoInsert(pSkipList, direction, pNode, isForward);
+    }
+  }
+
+  return pNode;
+}
+
 // static int32_t tSkipListEndParQuery(SSkipList *pSkipList, SSkipListNode *pStartNode, SSkipListKey *pEndKey,
 //                                    int32_t cond, SSkipListNode ***pRes) {
 //  pthread_rwlock_rdlock(&pSkipList->lock);

src/util/src/tsocket.c

@@ -107,7 +107,7 @@ int32_t taosWriteMsg(SOCKET fd, void *buf, int32_t nbytes) {
   while (nleft > 0) {
     nwritten = (int32_t)taosWriteSocket(fd, (char *)ptr, (size_t)nleft);
     if (nwritten <= 0) {
-      if (errno == EINTR)
+      if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
         continue;
       else
         return -1;
@@ -133,7 +133,7 @@ int32_t taosReadMsg(SOCKET fd, void *buf, int32_t nbytes) {
     if (nread == 0) {
       break;
     } else if (nread < 0) {
-      if (errno == EINTR) {
+      if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
         continue;
       } else {
         return -1;

src/vnode/src/vnodeMain.c

@@ -315,7 +315,13 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
     pVnode->version = walGetVersion(pVnode->wal);
   }
 
-  tsdbSyncCommit(pVnode->tsdb);
+  code = tsdbSyncCommit(pVnode->tsdb);
+  if (code != 0) {
+    vError("vgId:%d, failed to commit after restore from wal since %s", pVnode->vgId, tstrerror(code));
+    vnodeCleanUp(pVnode);
+    return code;
+  }
+
   walRemoveAllOldFiles(pVnode->wal);
   walRenew(pVnode->wal);
 
@@ -412,6 +418,7 @@ void vnodeRelease(void *pVnodeRaw) {
   }
 
   if (pVnode->wal) {
+    walRemoveAllOldFiles(pVnode->wal);
     walClose(pVnode->wal);
     pVnode->wal = NULL;
   }
@@ -589,7 +596,11 @@ static int vnodeProcessTsdbStatus(void *arg, int status) {
   if (status == TSDB_STATUS_COMMIT_START) {
     pVnode->fversion = pVnode->version;
     vDebug("vgId:%d, start commit, fver:%" PRIu64 " vver:%" PRIu64, pVnode->vgId, pVnode->fversion, pVnode->version);
-    return walRenew(pVnode->wal);
+    if (pVnode->status == TAOS_VN_STATUS_INIT) {
+      return 0;
+    } else {
+      return walRenew(pVnode->wal);
+    }
   }
 
   if (status == TSDB_STATUS_COMMIT_OVER) {

src/wal/src/walMgmt.c

@@ -124,15 +124,7 @@ void walClose(void *handle) {
 
   SWal *pWal = handle;
   pthread_mutex_lock(&pWal->mutex);
-
   taosClose(pWal->fd);
-
-  if (pWal->keep != TAOS_WAL_KEEP) {
-    walRemoveAllOldFiles(pWal);
-  } else {
-    wDebug("vgId:%d, wal:%p file:%s, it is closed and kept", pWal->vgId, pWal, pWal->name);
-  }
-
   pthread_mutex_unlock(&pWal->mutex);
   taosRemoveRef(tsWal.refId, pWal->rid);
 }

src/wal/src/walWrite.c

@@ -67,6 +67,7 @@ void walRemoveOneOldFile(void *handle) {
   SWal *pWal = handle;
   if (pWal == NULL) return;
   if (pWal->keep == TAOS_WAL_KEEP) return;
+  if (pWal->fd <= 0) return;
 
   pthread_mutex_lock(&pWal->mutex);
 
@@ -91,6 +92,8 @@ void walRemoveAllOldFiles(void *handle) {
 
   SWal *  pWal = handle;
   int64_t fileId = -1;
+
+  pthread_mutex_lock(&pWal->mutex);
   while (walGetNextFile(pWal, &fileId) >= 0) {
     snprintf(pWal->name, sizeof(pWal->name), "%s/%s%" PRId64, pWal->path, WAL_PREFIX, fileId);
 
@@ -100,6 +103,7 @@ void walRemoveAllOldFiles(void *handle) {
       wInfo("vgId:%d, wal:%p file:%s, it is removed", pWal->vgId, pWal, pWal->name);
     }
   }
+  pthread_mutex_unlock(&pWal->mutex);
 }
 
 int32_t walWrite(void *handle, SWalHead *pHead) {

tests/pytest/fulltest.sh

@@ -155,6 +155,9 @@ python3 ./test.py -f query/queryCountCSVData.py
 python3 ./test.py -f query/natualInterval.py
 python3 ./test.py -f query/bug1471.py
 #python3 ./test.py -f query/dataLossTest.py
+python3 ./test.py -f query/bug1874.py
+python3 ./test.py -f query/bug1875.py
+python3 ./test.py -f query/bug1876.py
 
 #stream
 python3 ./test.py -f stream/metric_1.py