Merge branch '3.0' of https://github.com/taosdata/TDengine into TD-20251

include/libs/qcom/query.h

@@ -262,24 +262,23 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
   (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \
    NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))
 
-#define SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR || (_code) == TSDB_CODE_VND_STOPPED)
+#define SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR || (_code) == TSDB_CODE_VND_STOPPED || (_code) == TSDB_CODE_APP_IS_STARTING || (_code) == TSDB_CODE_APP_IS_STOPPING)
 #define SYNC_SELF_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_RESTORING || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR)
-#define SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) (false)  // used later
+#define SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_MNODE_NOT_FOUND) 
+
+#define NO_RET_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL)
 
 #define NEED_REDIRECT_ERROR(_code)                                                      \
-  ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL ||  \
-   (_code) == TSDB_CODE_MNODE_NOT_FOUND || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \
-   SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || \
-   (_code) == TSDB_CODE_APP_IS_STARTING || (_code) == TSDB_CODE_APP_IS_STOPPING)
+  (NO_RET_REDIRECT_ERROR(_code) || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \
+   SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code))
+   
 
 #define NEED_CLIENT_RM_TBLMETA_REQ(_type)                                                                  \
   ((_type) == TDMT_VND_CREATE_TABLE || (_type) == TDMT_MND_CREATE_STB || (_type) == TDMT_VND_DROP_TABLE || \
    (_type) == TDMT_MND_DROP_STB)
 
 #define NEED_SCHEDULER_REDIRECT_ERROR(_code)                                            \
-  ((_code) == TSDB_CODE_MNODE_NOT_FOUND || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \
-   SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || \
-   (_code) == TSDB_CODE_APP_IS_STARTING || (_code) == TSDB_CODE_APP_IS_STOPPING)
+  (SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code))
 
 #define REQUEST_TOTAL_EXEC_TIMES 2
 

include/libs/wal/wal.h

@@ -107,6 +107,8 @@ typedef struct SWal {
   TdFilePtr pIdxFile;
   int32_t   writeCur;
   SArray   *fileInfoSet;  // SArray<SWalFileInfo>
+  // gc
+  SArray *toDeleteFiles;  // SArray<SWalFileInfo>
   // status
   int64_t totSize;
   int64_t lastRollSeq;

include/util/taoserror.h

@@ -72,7 +72,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_OUT_OF_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0102)
 // #define TSDB_CODE_COM_INVALID_CFG_MSG        TAOS_DEF_ERROR_CODE(0, 0x0103) // 2.x
 #define TSDB_CODE_FILE_CORRUPTED                TAOS_DEF_ERROR_CODE(0, 0x0104) //
-#define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0105) // internal
+// #define TSDB_CODE_REF_NO_MEMORY              TAOS_DEF_ERROR_CODE(0, 0x0105) // 2.x
 #define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0106) // internal
 #define TSDB_CODE_REF_ID_REMOVED                TAOS_DEF_ERROR_CODE(0, 0x0107) // internal
 #define TSDB_CODE_REF_INVALID_ID                TAOS_DEF_ERROR_CODE(0, 0x0108) // internal
@@ -163,7 +163,7 @@ int32_t* taosGetErrno();
 // #define TSDB_CODE_MND_ACTION_NEED_REPROCESSEDTAOS_DEF_ERROR_CODE(0, 0x0302) // 2.x
 #define TSDB_CODE_MND_NO_RIGHTS                 TAOS_DEF_ERROR_CODE(0, 0x0303)
 // #define TSDB_CODE_MND_APP_ERROR              TAOS_DEF_ERROR_CODE(0, 0x0304) // 2.x
-#define TSDB_CODE_MND_INVALID_CONNECTION        TAOS_DEF_ERROR_CODE(0, 0x0305)
+// #define TSDB_CODE_MND_INVALID_CONNECTION     TAOS_DEF_ERROR_CODE(0, 0x0305) // 2.x
 // #define TSDB_CODE_MND_INVALID_MSG_VERSION    TAOS_DEF_ERROR_CODE(0, 0x0306) // 2.x
 // #define TSDB_CODE_MND_INVALID_MSG_LEN        TAOS_DEF_ERROR_CODE(0, 0x0307) // 2.x
 // #define TSDB_CODE_MND_INVALID_MSG_TYPE       TAOS_DEF_ERROR_CODE(0, 0x0308) // 2.x
@@ -171,27 +171,28 @@ int32_t* taosGetErrno();
 // #define TSDB_CODE_MND_OUT_OF_MEMORY          TAOS_DEF_ERROR_CODE(0, 0x030A) // 2.x
 #define TSDB_CODE_MND_INVALID_SHOWOBJ           TAOS_DEF_ERROR_CODE(0, 0x030B)
 #define TSDB_CODE_MND_INVALID_QUERY_ID          TAOS_DEF_ERROR_CODE(0, 0x030C)
-#define TSDB_CODE_MND_INVALID_STREAM_ID         TAOS_DEF_ERROR_CODE(0, 0x030D)
+// #define TSDB_CODE_MND_INVALID_STREAM_ID      TAOS_DEF_ERROR_CODE(0, 0x030D) // 2.x
 #define TSDB_CODE_MND_INVALID_CONN_ID           TAOS_DEF_ERROR_CODE(0, 0x030E)
-#define TSDB_CODE_MND_MNODE_IS_RUNNING          TAOS_DEF_ERROR_CODE(0, 0x0310)
-#define TSDB_CODE_MND_FAILED_TO_CONFIG_SYNC     TAOS_DEF_ERROR_CODE(0, 0x0311)
-#define TSDB_CODE_MND_FAILED_TO_START_SYNC      TAOS_DEF_ERROR_CODE(0, 0x0312)
-#define TSDB_CODE_MND_FAILED_TO_CREATE_DIR      TAOS_DEF_ERROR_CODE(0, 0x0313)
-#define TSDB_CODE_MND_FAILED_TO_INIT_STEP       TAOS_DEF_ERROR_CODE(0, 0x0314)
+// #define TSDB_CODE_MND_MNODE_IS_RUNNING       TAOS_DEF_ERROR_CODE(0, 0x0310) // 2.x
+// #define TSDB_CODE_MND_FAILED_TO_CONFIG_SYNC  TAOS_DEF_ERROR_CODE(0, 0x0311) // 2.x
+// #define TSDB_CODE_MND_FAILED_TO_START_SYNC   TAOS_DEF_ERROR_CODE(0, 0x0312) // 2.x
+// #define TSDB_CODE_MND_FAILED_TO_CREATE_DIR   TAOS_DEF_ERROR_CODE(0, 0x0313) // 2.x
+// #define TSDB_CODE_MND_FAILED_TO_INIT_STEP    TAOS_DEF_ERROR_CODE(0, 0x0314) // 2.x
 #define TSDB_CODE_MND_USER_DISABLED             TAOS_DEF_ERROR_CODE(0, 0x0315)
 
 // mnode-sdb
-#define TSDB_CODE_SDB_OBJ_ALREADY_THERE         TAOS_DEF_ERROR_CODE(0, 0x0320)
-// #define TSDB_CODE_APP_ERROR                  TAOS_DEF_ERROR_CODE(0, 0x0321) // 2.x
-#define TSDB_CODE_SDB_INVALID_TABLE_TYPE        TAOS_DEF_ERROR_CODE(0, 0x0322)
+#define TSDB_CODE_SDB_OBJ_ALREADY_THERE         TAOS_DEF_ERROR_CODE(0, 0x0320) // internal
+// #define TSDB_CODE_MND_SDB_ERROR              TAOS_DEF_ERROR_CODE(0, 0x0321) // 2.x
+#define TSDB_CODE_SDB_INVALID_TABLE_TYPE        TAOS_DEF_ERROR_CODE(0, 0x0322) // internal
 #define TSDB_CODE_SDB_OBJ_NOT_THERE             TAOS_DEF_ERROR_CODE(0, 0x0323)
-#define TSDB_CODE_SDB_INVALID_KEY_TYPE          TAOS_DEF_ERROR_CODE(0, 0x0325)
-#define TSDB_CODE_SDB_INVALID_ACTION_TYPE       TAOS_DEF_ERROR_CODE(0, 0x0326)
-#define TSDB_CODE_SDB_INVALID_STATUS_TYPE       TAOS_DEF_ERROR_CODE(0, 0x0327)
-#define TSDB_CODE_SDB_INVALID_DATA_VER          TAOS_DEF_ERROR_CODE(0, 0x0328)
-#define TSDB_CODE_SDB_INVALID_DATA_LEN          TAOS_DEF_ERROR_CODE(0, 0x0329)
+// #define TSDB_CODE_MND_SDB_INVAID_META_ROW    TAOS_DEF_ERROR_CODE(0, 0x0324) // 2.x
+// #define TSDB_CODE_SDB_INVALID_KEY_TYPE       TAOS_DEF_ERROR_CODE(0, 0x0325) // 2.x
+#define TSDB_CODE_SDB_INVALID_ACTION_TYPE       TAOS_DEF_ERROR_CODE(0, 0x0326) // internal
+// #define TSDB_CODE_SDB_INVALID_STATUS_TYPE    TAOS_DEF_ERROR_CODE(0, 0x0327) // unused
+#define TSDB_CODE_SDB_INVALID_DATA_VER          TAOS_DEF_ERROR_CODE(0, 0x0328) // internal
+#define TSDB_CODE_SDB_INVALID_DATA_LEN          TAOS_DEF_ERROR_CODE(0, 0x0329) // internal
 #define TSDB_CODE_SDB_INVALID_DATA_CONTENT      TAOS_DEF_ERROR_CODE(0, 0x032A)
-#define TSDB_CODE_SDB_INVALID_WAl_VER           TAOS_DEF_ERROR_CODE(0, 0x032B)
+// #define TSDB_CODE_SDB_INVALID_WAl_VER        TAOS_DEF_ERROR_CODE(0, 0x032B) // unused
 #define TSDB_CODE_SDB_OBJ_CREATING              TAOS_DEF_ERROR_CODE(0, 0x032C)
 #define TSDB_CODE_SDB_OBJ_DROPPING              TAOS_DEF_ERROR_CODE(0, 0x032D)
 
@@ -512,7 +513,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TQ_INVALID_CONFIG             TAOS_DEF_ERROR_CODE(0, 0x0A00)
 #define TSDB_CODE_TQ_INIT_FAILED                TAOS_DEF_ERROR_CODE(0, 0x0A01)
 #define TSDB_CODE_TQ_NO_DISK_PERMISSIONS        TAOS_DEF_ERROR_CODE(0, 0x0A03)
-#define TSDB_CODE_TQ_FILE_CORRUPTED             TAOS_DEF_ERROR_CODE(0, 0x0A04)
+// #define TSDB_CODE_TQ_FILE_CORRUPTED          TAOS_DEF_ERROR_CODE(0, 0x0A04)
 // #define TSDB_CODE_TQ_OUT_OF_MEMORY           TAOS_DEF_ERROR_CODE(0, 0x0A05)
 #define TSDB_CODE_TQ_FILE_ALREADY_EXISTS        TAOS_DEF_ERROR_CODE(0, 0x0A06)
 #define TSDB_CODE_TQ_FAILED_TO_CREATE_DIR       TAOS_DEF_ERROR_CODE(0, 0x0A07)
@@ -698,7 +699,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_RSMA_INVALID_ENV               TAOS_DEF_ERROR_CODE(0, 0x3150)
 #define TSDB_CODE_RSMA_INVALID_STAT              TAOS_DEF_ERROR_CODE(0, 0x3151)
 #define TSDB_CODE_RSMA_QTASKINFO_CREATE          TAOS_DEF_ERROR_CODE(0, 0x3152)
-#define TSDB_CODE_RSMA_FILE_CORRUPTED            TAOS_DEF_ERROR_CODE(0, 0x3153)
+// #define TSDB_CODE_RSMA_FILE_CORRUPTED         TAOS_DEF_ERROR_CODE(0, 0x3153)
 #define TSDB_CODE_RSMA_REMOVE_EXISTS             TAOS_DEF_ERROR_CODE(0, 0x3154)
 #define TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP        TAOS_DEF_ERROR_CODE(0, 0x3155)
 #define TSDB_CODE_RSMA_EMPTY_INFO                TAOS_DEF_ERROR_CODE(0, 0x3156)

source/dnode/mnode/impl/src/mndShow.c

@@ -198,9 +198,9 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
   }
 
   if (retrieveReq.showId == 0) {
-    STableMetaRsp *pMeta = (STableMetaRsp *)taosHashGet(pMnode->infosMeta, retrieveReq.tb, strlen(retrieveReq.tb));
+    STableMetaRsp *pMeta = taosHashGet(pMnode->infosMeta, retrieveReq.tb, strlen(retrieveReq.tb));
     if (pMeta == NULL) {
-      pMeta = (STableMetaRsp *)taosHashGet(pMnode->perfsMeta, retrieveReq.tb, strlen(retrieveReq.tb));
+      pMeta = taosHashGet(pMnode->perfsMeta, retrieveReq.tb, strlen(retrieveReq.tb));
       if (pMeta == NULL) {
         terrno = TSDB_CODE_MND_INVALID_SYS_TABLENAME;
         mError("failed to process show-retrieve req:%p since %s", pShow, terrstr());

source/dnode/vnode/src/vnd/vnodeSync.c

@@ -606,16 +606,7 @@ void vnodeSyncPreClose(SVnode *pVnode) {
   vInfo("vgId:%d, pre close sync", pVnode->config.vgId);
   syncLeaderTransfer(pVnode->sync);
   syncPreStop(pVnode->sync);
-#if 0
-  while (syncSnapshotRecving(pVnode->sync)) {
-    vInfo("vgId:%d, snapshot is recving", pVnode->config.vgId);
-    taosMsleep(300);
-  }
-  while (syncSnapshotSending(pVnode->sync)) {
-    vInfo("vgId:%d, snapshot is sending", pVnode->config.vgId);
-    taosMsleep(300);
-  }
-#endif
+
   taosThreadMutexLock(&pVnode->lock);
   if (pVnode->blocked) {
     vInfo("vgId:%d, post block after close sync", pVnode->config.vgId);

source/libs/scheduler/inc/schInt.h

@@ -293,6 +293,7 @@ typedef struct SSchJob {
   void                *chkKillParam;
   SSchTask            *fetchTask;
   int32_t              errCode;
+  int32_t              redirectCode;
   SRWLatch             resLock;
   SExecResult          execRes;
   void                *fetchRes;  // TODO free it or not
@@ -331,6 +332,9 @@ extern SSchedulerMgmt schMgmt;
   ((_job)->attr.localExec && SCH_IS_QUERY_JOB(_job) && (!SCH_IS_INSERT_JOB(_job)) && \
    (!SCH_IS_DATA_BIND_QRY_TASK(_task)))
 
+#define SCH_UPDATE_REDICT_CODE(job, _code) atomic_val_compare_exchange_32(&((job)->redirectCode), 0, _code)
+#define SCH_GET_REDICT_CODE(job, _code) (((!NO_RET_REDIRECT_ERROR(_code)) || (job)->redirectCode == 0) ? (_code) : (job)->redirectCode)
+
 #define SCH_SET_TASK_STATUS(task, st) atomic_store_8(&(task)->status, st)
 #define SCH_GET_TASK_STATUS(task)     atomic_load_8(&(task)->status)
 #define SCH_GET_TASK_STATUS_STR(task) jobTaskStatusStr(SCH_GET_TASK_STATUS(task))

source/libs/scheduler/src/schJob.c

@@ -481,6 +481,10 @@ _return:
 }
 
 int32_t schProcessOnJobFailure(SSchJob *pJob, int32_t errCode) {
+  if (TSDB_CODE_SCH_IGNORE_ERROR == errCode) {
+    return TSDB_CODE_SCH_IGNORE_ERROR;
+  }
+  
   schUpdateJobErrCode(pJob, errCode);
 
   int32_t code = atomic_load_32(&pJob->errCode);

source/libs/scheduler/src/schTask.c

@@ -340,7 +340,7 @@ int32_t schRescheduleTask(SSchJob *pJob, SSchTask *pTask) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t schChkUpdateRedirectCtx(SSchJob *pJob, SSchTask *pTask, SEpSet *pEpSet) {
+int32_t schChkUpdateRedirectCtx(SSchJob *pJob, SSchTask *pTask, SEpSet *pEpSet, int32_t rspCode) {
   SSchRedirectCtx *pCtx = &pTask->redirectCtx;
   if (!pCtx->inRedirect) {
     pCtx->inRedirect = true;
@@ -380,7 +380,7 @@ int32_t schChkUpdateRedirectCtx(SSchJob *pJob, SSchTask *pTask, SEpSet *pEpSet)
     if (lastTime > tsMaxRetryWaitTime) {
       SCH_TASK_DLOG("task no more redirect retry since timeout, now:%" PRId64 ", start:%" PRId64 ", max:%d, total:%d",
                     nowTs, pCtx->startTs, tsMaxRetryWaitTime, pCtx->totalTimes);
-      SCH_ERR_RET(TSDB_CODE_TIMEOUT_ERROR);
+      SCH_ERR_RET(SCH_GET_REDICT_CODE(pJob, rspCode));
     }
 
     pCtx->periodMs *= tsRedirectFactor;
@@ -415,7 +415,11 @@ int32_t schDoTaskRedirect(SSchJob *pJob, SSchTask *pTask, SDataBuf *pData, int32
     pTask->retryTimes = 0;
   }
 
-  SCH_ERR_JRET(schChkUpdateRedirectCtx(pJob, pTask, pData ? pData->pEpSet : NULL));
+  if (!NO_RET_REDIRECT_ERROR(rspCode)) {
+    SCH_UPDATE_REDICT_CODE(pJob, rspCode);
+  }
+
+  SCH_ERR_JRET(schChkUpdateRedirectCtx(pJob, pTask, pData ? pData->pEpSet : NULL, rspCode));
 
   pTask->waitRetry = true;
 

source/libs/sync/src/syncMain.c

@@ -1221,6 +1221,26 @@ int32_t syncNodeStartStandBy(SSyncNode* pSyncNode) {
 }
 
 void syncNodePreClose(SSyncNode* pSyncNode) {
+  if (pSyncNode != NULL && pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpApplyQueueItems != NULL) {
+    while (1) {
+      int32_t aqItems = pSyncNode->pFsm->FpApplyQueueItems(pSyncNode->pFsm);
+      sTrace("vgId:%d, pre close, %d items in apply queue", pSyncNode->vgId, aqItems);
+      if (aqItems == 0 || aqItems == -1) {
+        break;
+      }
+      taosMsleep(20);
+    }
+  }
+
+  if (pSyncNode->pNewNodeReceiver != NULL) {
+    if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
+      snapshotReceiverForceStop(pSyncNode->pNewNodeReceiver);
+    }
+
+    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
+    pSyncNode->pNewNodeReceiver = NULL;
+  }
+
   // stop elect timer
   syncNodeStopElectTimer(pSyncNode);
 

source/libs/wal/src/walMeta.c

@@ -24,7 +24,9 @@ bool FORCE_INLINE walLogExist(SWal* pWal, int64_t ver) {
   return !walIsEmpty(pWal) && walGetFirstVer(pWal) <= ver && walGetLastVer(pWal) >= ver;
 }
 
-bool FORCE_INLINE walIsEmpty(SWal* pWal) { return pWal->vers.firstVer == -1; }
+bool FORCE_INLINE walIsEmpty(SWal* pWal) {
+  return (pWal->vers.firstVer == -1 || pWal->vers.lastVer < pWal->vers.firstVer); // [firstVer, lastVer + 1)
+}
 
 int64_t FORCE_INLINE walGetFirstVer(SWal* pWal) { return pWal->vers.firstVer; }
 

source/libs/wal/src/walMgmt.c

@@ -121,7 +121,16 @@ SWal *walOpen(const char *path, SWalCfg *pCfg) {
   pWal->writeCur = -1;
   pWal->fileInfoSet = taosArrayInit(8, sizeof(SWalFileInfo));
   if (pWal->fileInfoSet == NULL) {
-    wError("vgId:%d, path:%s, failed to init taosArray %s", pWal->cfg.vgId, pWal->path, strerror(errno));
+    wError("vgId:%d, failed to init taosArray of fileInfoSet due to %s. path:%s", pWal->cfg.vgId, strerror(errno),
+           pWal->path);
+    goto _err;
+  }
+
+  // init gc
+  pWal->toDeleteFiles = taosArrayInit(8, sizeof(SWalFileInfo));
+  if (pWal->toDeleteFiles == NULL) {
+    wError("vgId:%d, failed to init taosArray of toDeleteFiles due to %s. path:%s", pWal->cfg.vgId, strerror(errno),
+           pWal->path);
     goto _err;
   }
 
@@ -203,6 +212,8 @@ void walClose(SWal *pWal) {
   pWal->pIdxFile = NULL;
   taosArrayDestroy(pWal->fileInfoSet);
   pWal->fileInfoSet = NULL;
+  taosArrayDestroy(pWal->toDeleteFiles);
+  pWal->toDeleteFiles = NULL;
 
   void *pIter = NULL;
   while (1) {

source/libs/wal/src/walRead.c

@@ -489,7 +489,7 @@ int32_t walReadVer(SWalReader *pReader, int64_t ver) {
   int32_t code;
   bool    seeked = false;
 
-  if (pReader->pWal->vers.firstVer == -1) {
+  if (walIsEmpty(pReader->pWal)) {
     terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
     return -1;
   }

source/libs/wal/src/walWrite.c

@@ -120,9 +120,9 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
       return -1;
     }
 
-    // delete files
+    // delete files in descending order
     int fileSetSize = taosArrayGetSize(pWal->fileInfoSet);
-    for (int i = pWal->writeCur + 1; i < fileSetSize; i++) {
+    for (int i = fileSetSize - 1; i >= pWal->writeCur + 1; i--) {
       walBuildLogName(pWal, ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, i))->firstVer, fnameStr);
       wDebug("vgId:%d, wal remove file %s for rollback", pWal->cfg.vgId, fnameStr);
       taosRemoveFile(fnameStr);
@@ -217,14 +217,9 @@ int32_t walRollback(SWal *pWal, int64_t ver) {
   pWal->vers.lastVer = ver - 1;
   if (pWal->vers.lastVer < pWal->vers.firstVer) {
     ASSERT(pWal->vers.lastVer == pWal->vers.firstVer - 1);
-    pWal->vers.firstVer = -1;
   }
   ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer = ver - 1;
   ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize = entry.offset;
-  if (((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->lastVer < ver - 1) {
-    ASSERT(((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->fileSize == 0);
-    ((SWalFileInfo *)taosArrayGetLast(pWal->fileInfoSet))->firstVer = -1;
-  }
   taosCloseFile(&pIdxFile);
   taosCloseFile(&pLogFile);
 
@@ -338,6 +333,7 @@ int32_t walEndSnapshot(SWal *pWal) {
     } else {
       wDebug("vgId:%d, wal no remove", pWal->cfg.vgId);
     }
+
     // iterate files, until the searched result
     for (SWalFileInfo *iter = pWal->fileInfoSet->pData; iter < pInfo; iter++) {
       wDebug("vgId:%d, wal check remove file %" PRId64 "(file size %" PRId64 " close ts %" PRId64
@@ -350,34 +346,17 @@ int32_t walEndSnapshot(SWal *pWal) {
         wDebug("vgId:%d, check pass", pWal->cfg.vgId);
         deleteCnt++;
         newTotSize -= iter->fileSize;
+        taosArrayPush(pWal->toDeleteFiles, iter);
       }
       wDebug("vgId:%d, check not pass", pWal->cfg.vgId);
     }
-    wDebug("vgId:%d, wal should delete %d files", pWal->cfg.vgId, deleteCnt);
-    int32_t actualDelete = 0;
-    char    fnameStr[WAL_FILE_LEN];
-    // remove file
-    for (int i = 0; i < deleteCnt; i++) {
-      pInfo = taosArrayGet(pWal->fileInfoSet, i);
-      walBuildLogName(pWal, pInfo->firstVer, fnameStr);
-      wDebug("vgId:%d, wal remove file %s", pWal->cfg.vgId, fnameStr);
-      if (taosRemoveFile(fnameStr) < 0) {
-        goto UPDATE_META;
-      }
-      walBuildIdxName(pWal, pInfo->firstVer, fnameStr);
-      wDebug("vgId:%d, wal remove file %s", pWal->cfg.vgId, fnameStr);
-      if (taosRemoveFile(fnameStr) < 0) {
-        ASSERT(0);
-      }
-      actualDelete++;
-    }
 
   UPDATE_META:
     // make new array, remove files
-    taosArrayPopFrontBatch(pWal->fileInfoSet, actualDelete);
+    taosArrayPopFrontBatch(pWal->fileInfoSet, deleteCnt);
     if (taosArrayGetSize(pWal->fileInfoSet) == 0) {
       pWal->writeCur = -1;
-      pWal->vers.firstVer = -1;
+      pWal->vers.firstVer = pWal->vers.lastVer + 1;
     } else {
       pWal->vers.firstVer = ((SWalFileInfo *)taosArrayGet(pWal->fileInfoSet, 0))->firstVer;
     }
@@ -392,6 +371,26 @@ int32_t walEndSnapshot(SWal *pWal) {
     goto END;
   }
 
+  // delete files
+  deleteCnt = taosArrayGetSize(pWal->toDeleteFiles);
+  wDebug("vgId:%d, wal should delete %d files", pWal->cfg.vgId, deleteCnt);
+  char fnameStr[WAL_FILE_LEN];
+  for (int i = 0; i < deleteCnt; i++) {
+    pInfo = taosArrayGet(pWal->toDeleteFiles, i);
+    walBuildLogName(pWal, pInfo->firstVer, fnameStr);
+    wDebug("vgId:%d, wal remove file %s", pWal->cfg.vgId, fnameStr);
+    if (taosRemoveFile(fnameStr) < 0 && errno != ENOENT) {
+      wError("vgId:%d, failed to remove log file %s due to %s", pWal->cfg.vgId, fnameStr, strerror(errno));
+      goto END;
+    }
+    walBuildIdxName(pWal, pInfo->firstVer, fnameStr);
+    wDebug("vgId:%d, wal remove file %s", pWal->cfg.vgId, fnameStr);
+    if (taosRemoveFile(fnameStr) < 0 && errno != ENOENT) {
+      ASSERT(0);
+    }
+  }
+  taosArrayClear(pWal->toDeleteFiles);
+
 END:
   taosThreadMutexUnlock(&pWal->mutex);
   return code;
@@ -489,9 +488,7 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
   SWalFileInfo *pFileInfo = walGetCurFileInfo(pWal);
   ASSERT(pFileInfo != NULL);
 
-  if (pFileInfo->firstVer == -1) {
-    pFileInfo->firstVer = index;
-  }
+  ASSERT(pFileInfo->firstVer != -1);
   pWal->writeHead.head.version = index;
   pWal->writeHead.head.bodyLen = bodyLen;
   pWal->writeHead.head.msgType = msgType;
@@ -527,7 +524,10 @@ static FORCE_INLINE int32_t walWriteImpl(SWal *pWal, int64_t index, tmsg_t msgTy
   }
 
   // set status
-  if (pWal->vers.firstVer == -1) pWal->vers.firstVer = index;
+  if (pWal->vers.firstVer == -1) {
+    ASSERT(index == 0);
+    pWal->vers.firstVer = 0;
+  }
   pWal->vers.lastVer = index;
   pWal->totSize += sizeof(SWalCkHead) + bodyLen;
   pFileInfo->lastVer = index;

source/util/src/terror.c

@@ -57,7 +57,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TIME_UNSYNCED,                "Client and server's t
 TAOS_DEFINE_ERROR(TSDB_CODE_OPS_NOT_SUPPORT,              "Operation not supported")
 TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_MEMORY,                "Out of Memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_FILE_CORRUPTED,               "Data file corrupted")
-TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY,                "Ref out of memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_REF_FULL,                     "too many Ref Objs")
 TAOS_DEFINE_ERROR(TSDB_CODE_REF_ID_REMOVED,               "Ref ID is removed")
 TAOS_DEFINE_ERROR(TSDB_CODE_REF_INVALID_ID,               "Invalid Ref ID")
@@ -140,29 +139,19 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSC_NOT_STABLE_ERROR,         "Table is not a super
 
 // mnode-common
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_NO_RIGHTS,                "Insufficient privilege for operation")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CONNECTION,       "Invalid message connection")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SHOWOBJ,          "Data expired")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_QUERY_ID,         "Invalid query id")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_STREAM_ID,        "Invalid stream id")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_CONN_ID,          "Invalid connection id")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_MNODE_IS_RUNNING,         "mnode is alreay running")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_FAILED_TO_CONFIG_SYNC,    "failed to config sync")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_FAILED_TO_START_SYNC,     "failed to start sync")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_FAILED_TO_CREATE_DIR,     "failed to create mnode dir")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_FAILED_TO_INIT_STEP,      "failed to init components")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_USER_DISABLED,            "User is disabled")
 
 // mnode-sdb
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_OBJ_ALREADY_THERE,        "Object already there")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_TABLE_TYPE,       "Invalid table type")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_OBJ_NOT_THERE,            "Object not there")
-TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_KEY_TYPE,         "Invalid key type")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_ACTION_TYPE,      "Invalid action type")
-TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_STATUS_TYPE,      "Invalid status type")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_DATA_VER,         "Invalid raw data version")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_DATA_LEN,         "Invalid raw data len")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_DATA_CONTENT,     "Invalid raw data content")
-TAOS_DEFINE_ERROR(TSDB_CODE_SDB_INVALID_WAl_VER,          "Invalid wal version")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_OBJ_CREATING,             "Object is creating")
 TAOS_DEFINE_ERROR(TSDB_CODE_SDB_OBJ_DROPPING,             "Object is dropping")
 
@@ -418,7 +407,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR,           "Sync internal error")
 TAOS_DEFINE_ERROR(TSDB_CODE_TQ_INVALID_CONFIG,            "TQ invalid config")
 TAOS_DEFINE_ERROR(TSDB_CODE_TQ_INIT_FAILED,               "TQ init falied")
 TAOS_DEFINE_ERROR(TSDB_CODE_TQ_NO_DISK_PERMISSIONS,       "TQ no disk permissions")
-TAOS_DEFINE_ERROR(TSDB_CODE_TQ_FILE_CORRUPTED,            "TQ file corrupted")
 TAOS_DEFINE_ERROR(TSDB_CODE_TQ_FILE_ALREADY_EXISTS,       "TQ file already exists")
 TAOS_DEFINE_ERROR(TSDB_CODE_TQ_FAILED_TO_CREATE_DIR,      "TQ failed to create dir")
 TAOS_DEFINE_ERROR(TSDB_CODE_TQ_META_NO_SUCH_KEY,          "TQ meta no such key")
@@ -596,7 +584,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSMA_NO_INDEX_IN_CACHE,         "No tsma index in ca
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_ENV,               "Invalid rsma env")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_STAT,              "Invalid rsma state")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_QTASKINFO_CREATE,          "Rsma qtaskinfo creation error")
-TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FILE_CORRUPTED,            "Rsma file corrupted")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_REMOVE_EXISTS,             "Rsma remove exists")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP,        "Rsma fetch msg is messed up")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_EMPTY_INFO,                "Rsma info is empty")

source/util/src/tref.c

@@ -67,14 +67,14 @@ int32_t taosOpenRef(int32_t max, RefFp fp) {
 
   nodeList = taosMemoryCalloc(sizeof(SRefNode *), (size_t)max);
   if (nodeList == NULL) {
-    terrno = TSDB_CODE_REF_NO_MEMORY;
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
 
   lockedBy = taosMemoryCalloc(sizeof(int64_t), (size_t)max);
   if (lockedBy == NULL) {
     taosMemoryFree(nodeList);
-    terrno = TSDB_CODE_REF_NO_MEMORY;
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
 
@@ -164,7 +164,7 @@ int64_t taosAddRef(int32_t rsetId, void *p) {
 
   pNode = taosMemoryCalloc(sizeof(SRefNode), 1);
   if (pNode == NULL) {
-    terrno = TSDB_CODE_REF_NO_MEMORY;
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }