taosdata / TDengine / #3599

coverage: 1.77% (-61.6%) from 63.396%
#3599

push

travis-ci

web-flow 
Merge pull request #29712 from taosdata/fix/TD-33652-3.0

fix: reduce write rows from 30w to 3w

3776 of 278949 branches covered (1.35%)

Branch coverage included in aggregate %.

6012 of 274147 relevant lines covered (2.19%)

1642.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0
/source/dnode/mnode/impl/src/mndStreamUtil.c
1 
/*
2 
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3 
 *
4 
 * This program is free software: you can use, redistribute, and/or modify
5 
 * it under the terms of the GNU Affero General Public License, version 3
6 
 * or later ("AGPL"), as published by the Free Software Foundation.
7 
 *
8 
 * This program is distributed in the hope that it will be useful, but WITHOUT
9 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 
 * FITNESS FOR A PARTICULAR PURPOSE.
11 
 *
12 
 * You should have received a copy of the GNU Affero General Public License
13 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14 
 */
15 









16



#include "mndDb.h"










17



#include "mndStb.h"










18



#include "mndStream.h"










19



#include "mndTrans.h"










20



#include "mndVgroup.h"










21



#include "taoserror.h"










22



#include "tmisce.h"










23












24



struct SStreamTaskIter {










25



  SStreamObj  *pStream;










26



  int32_t      level;










27



  int32_t      ordinalIndex;










28



  int32_t      totalLevel;










29



  SStreamTask *pTask;










30



};










31












32



int32_t doRemoveTasks(SStreamExecInfo *pExecNode, STaskId *pRemovedId);










33












34



int32_t createStreamTaskIter(SStreamObj *pStream, SStreamTaskIter **pIter) {




×







35



  *pIter = taosMemoryCalloc(1, sizeof(SStreamTaskIter));





×







36



  if (*pIter == NULL) {





×







37



    return terrno;




×







38



  }










39












40



  (*pIter)->level = -1;




×







41



  (*pIter)->ordinalIndex = 0;




×







42



  (*pIter)->pStream = pStream;




×







43



  (*pIter)->totalLevel = taosArrayGetSize(pStream->tasks);




×







44



  (*pIter)->pTask = NULL;




×







45












46



  return 0;




×







47



}










48












49



bool streamTaskIterNextTask(SStreamTaskIter *pIter) {




×







50



  if (pIter->level >= pIter->totalLevel) {





×







51



    pIter->pTask = NULL;




×







52



    return false;




×







53



  }










54












55



  if (pIter->level == -1) {





×







56



    pIter->level += 1;




×







57



  }










58












59



  while (pIter->level < pIter->totalLevel) {





×







60



    SArray *pList = taosArrayGetP(pIter->pStream->tasks, pIter->level);




×







61



    if (pIter->ordinalIndex >= taosArrayGetSize(pList)) {





×







62



      pIter->level += 1;




×







63



      pIter->ordinalIndex = 0;




×







64



      pIter->pTask = NULL;




×







65



      continue;




×







66



    }










67












68



    pIter->pTask = taosArrayGetP(pList, pIter->ordinalIndex);




×







69



    pIter->ordinalIndex += 1;




×







70



    return true;




×







71



  }










72












73



  pIter->pTask = NULL;




×







74



  return false;




×







75



}










76












77



int32_t streamTaskIterGetCurrent(SStreamTaskIter *pIter, SStreamTask **pTask) {




×







78



  if (pTask) {





×







79



    *pTask = pIter->pTask;




×







80



    if (*pTask != NULL) {





×







81



      return TSDB_CODE_SUCCESS;




×







82



    }










83



  }










84












85



  return TSDB_CODE_INVALID_PARA;




×







86



}










87












88



void destroyStreamTaskIter(SStreamTaskIter *pIter) { taosMemoryFree(pIter); }





×







89












90



static bool checkStatusForEachReplica(SVgObj *pVgroup) {




×







91



  for (int32_t i = 0; i < pVgroup->replica; ++i) {





×







92



    if (!pVgroup->vnodeGid[i].syncRestore) {





×







93



      mInfo("vgId:%d not restored, not ready for checkpoint or other operations", pVgroup->vgId);





×







94



      return false;




×







95



    }










96












97



    ESyncState state = pVgroup->vnodeGid[i].syncState;




×







98



    if (state == TAOS_SYNC_STATE_OFFLINE || state == TAOS_SYNC_STATE_ERROR || state == TAOS_SYNC_STATE_LEARNER ||





×







99



        state == TAOS_SYNC_STATE_CANDIDATE) {










100



      mInfo("vgId:%d state:%d , not ready for checkpoint or other operations, not check other vgroups", pVgroup->vgId,





×







101



            state);










102



      return false;




×







103



    }










104



  }










105












106



  return true;




×







107



}










108












109



static int32_t mndAddSnodeInfo(SMnode *pMnode, SArray *pVgroupList) {




×







110



  SSnodeObj *pObj = NULL;




×







111



  void      *pIter = NULL;




×







112



  int32_t    code = 0;




×







113












114



  while (1) {




×







115



    pIter = sdbFetch(pMnode->pSdb, SDB_SNODE, pIter, (void **)&pObj);




×







116



    if (pIter == NULL) {





×







117



      break;




×







118



    }










119












120



    SNodeEntry entry = {.nodeId = SNODE_HANDLE};




×







121



    code = addEpIntoEpSet(&entry.epset, pObj->pDnode->fqdn, pObj->pDnode->port);




×







122



    if (code) {





×







123



      sdbRelease(pMnode->pSdb, pObj);




×







124



      sdbCancelFetch(pMnode->pSdb, pIter);




×







125



      mError("failed to extract epset for fqdn:%s during task vgroup snapshot", pObj->pDnode->fqdn);





×







126



      return code;




×







127



    }










128












129



    char buf[256] = {0};




×







130



    code = epsetToStr(&entry.epset, buf, tListLen(buf));




×







131



    if (code != 0) {  // print error and continue





×







132



      mError("failed to convert epset to str, code:%s", tstrerror(code));





×







133



    }










134












135



    void *p = taosArrayPush(pVgroupList, &entry);




×







136



    if (p == NULL) {





×







137



      code = terrno;




×







138



      sdbRelease(pMnode->pSdb, pObj);




×







139



      sdbCancelFetch(pMnode->pSdb, pIter);




×







140



      mError("failed to put entry in vgroup list, nodeId:%d code:%s", entry.nodeId, tstrerror(code));





×







141



      return code;




×







142



    } else {










143



      mDebug("take snode snapshot, nodeId:%d %s", entry.nodeId, buf);





×







144



    }










145












146



    sdbRelease(pMnode->pSdb, pObj);




×







147



  }










148












149



  return code;




×







150



}










151












152



static int32_t mndCheckMnodeStatus(SMnode* pMnode) {




×







153



  int32_t    code = 0;




×







154



  ESdbStatus objStatus;










155



  void      *pIter = NULL;




×







156



  SMnodeObj *pObj = NULL;




×







157












158



  while (1) {










159



    pIter = sdbFetchAll(pMnode->pSdb, SDB_MNODE, pIter, (void **)&pObj, &objStatus, true);




×







160



    if (pIter == NULL) {





×







161



      break;




×







162



    }










163












164



    if (pObj->syncState != TAOS_SYNC_STATE_LEADER && pObj->syncState != TAOS_SYNC_STATE_FOLLOWER) {





×







165



      mDebug("mnode sync state:%d not leader/follower", pObj->syncState);





×







166



      sdbRelease(pMnode->pSdb, pObj);




×







167



      sdbCancelFetch(pMnode->pSdb, pIter);




×







168



      return TSDB_CODE_FAILED;




×







169



    }










170












171



    if (objStatus != SDB_STATUS_READY) {





×







172



      mWarn("mnode status:%d not ready", objStatus);





×







173



      sdbRelease(pMnode->pSdb, pObj);




×







174



      sdbCancelFetch(pMnode->pSdb, pIter);




×







175



      return TSDB_CODE_FAILED;




×







176



    }










177












178



    sdbRelease(pMnode->pSdb, pObj);




×







179



  }










180












181



  return TSDB_CODE_SUCCESS;




×







182



}










183












184



static int32_t mndCheckAndAddVgroupsInfo(SMnode *pMnode, SArray *pVgroupList, bool* allReady) {




×







185



  SSdb     *pSdb = pMnode->pSdb;




×







186



  void     *pIter = NULL;




×







187



  SVgObj   *pVgroup = NULL;




×







188



  int32_t   code = 0;




×







189



  SHashObj *pHash = NULL;




×







190












191



  pHash = taosHashInit(10, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);




×







192



  if (pHash == NULL) {





×







193



    mError("failed to prepare hashmap during take vgroup snapshot, code:%s", tstrerror(terrno));





×







194



    return terrno;




×







195



  }










196












197



  while (1) {




×







198



    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);




×







199



    if (pIter == NULL) {





×







200



      break;




×







201



    }










202












203



    SNodeEntry entry = {.nodeId = pVgroup->vgId, .hbTimestamp = pVgroup->updateTime};




×







204



    entry.epset = mndGetVgroupEpset(pMnode, pVgroup);




×







205












206



    int8_t *pReplica = taosHashGet(pHash, &pVgroup->dbUid, sizeof(pVgroup->dbUid));




×







207



    if (pReplica == NULL) {  // not exist, add it into hash map





×







208



      code = taosHashPut(pHash, &pVgroup->dbUid, sizeof(pVgroup->dbUid), &pVgroup->replica, sizeof(pVgroup->replica));




×







209



      if (code) {





×







210



        mError("failed to put info into hashmap during task vgroup snapshot, code:%s", tstrerror(code));





×







211



        sdbRelease(pSdb, pVgroup);




×







212



        sdbCancelFetch(pSdb, pIter);




×







213



        goto _end;  // take snapshot failed, and not all ready




×







214



      }










215



    } else {










216



      if (*pReplica != pVgroup->replica) {





×







217



        mInfo("vgId:%d replica:%d inconsistent with other vgroups replica:%d, not ready for stream operations",





×







218



              pVgroup->vgId, pVgroup->replica, *pReplica);










219



        *allReady = false;  // task snap success, but not all ready




×







220



      }










221



    }










222












223



    // if not all ready till now, no need to check the remaining vgroups,










224



    // but still we need to put the info of the existed vgroups into the snapshot list










225



    if (*allReady) {





×







226



      *allReady = checkStatusForEachReplica(pVgroup);




×







227



    }










228












229



    char buf[256] = {0};




×







230



    code = epsetToStr(&entry.epset, buf, tListLen(buf));




×







231



    if (code != 0) {  // print error and continue





×







232



      mError("failed to convert epset to str, code:%s", tstrerror(code));





×







233



    }










234












235



    void *p = taosArrayPush(pVgroupList, &entry);




×







236



    if (p == NULL) {





×







237



      mError("failed to put entry in vgroup list, nodeId:%d code:out of memory", entry.nodeId);





×







238



      code = terrno;




×







239



      sdbRelease(pSdb, pVgroup);




×







240



      sdbCancelFetch(pSdb, pIter);




×







241



      goto _end;




×







242



    } else {










243



      mDebug("take node snapshot, nodeId:%d %s", entry.nodeId, buf);





×







244



    }










245












246



    sdbRelease(pSdb, pVgroup);




×







247



  }










248












249



_end:




×







250



  taosHashCleanup(pHash);




×







251



  return code;




×







252



}










253












254



int32_t mndTakeVgroupSnapshot(SMnode *pMnode, bool *allReady, SArray **pList) {




×







255



  int32_t   code = 0;




×







256



  SArray   *pVgroupList = NULL;




×







257












258



  *pList = NULL;




×







259



  *allReady = true;




×







260












261



  pVgroupList = taosArrayInit(4, sizeof(SNodeEntry));




×







262



  if (pVgroupList == NULL) {





×







263



    mError("failed to prepare arraylist during take vgroup snapshot, code:%s", tstrerror(terrno));





×







264



    code = terrno;




×







265



    goto _err;




×







266



  }










267












268



  // 1. check for all vnodes status










269



  code = mndCheckAndAddVgroupsInfo(pMnode, pVgroupList, allReady);




×







270



  if (code) {





×







271



    goto _err;




×







272



  }










273












274



  // 2. add snode info










275



  code = mndAddSnodeInfo(pMnode, pVgroupList);




×







276



  if (code) {





×







277



    goto _err;




×







278



  }










279












280



  // 3. check for mnode status










281



  code = mndCheckMnodeStatus(pMnode);




×







282



  if (code != TSDB_CODE_SUCCESS) {





×







283



    *allReady = false;




×







284



  }










285












286



  *pList = pVgroupList;




×







287



  return code;




×







288












289



_err:




×







290



  *allReady = false;




×







291



  taosArrayDestroy(pVgroupList);




×







292



  return code;




×







293



}










294












295



int32_t mndGetStreamObj(SMnode *pMnode, int64_t streamId, SStreamObj **pStream) {




×







296



  void *pIter = NULL;




×







297



  SSdb *pSdb = pMnode->pSdb;




×







298



  *pStream = NULL;




×







299












300



  SStreamObj *p = NULL;




×







301



  while ((pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&p)) != NULL) {





×







302



    if (p->uid == streamId) {





×







303



      sdbCancelFetch(pSdb, pIter);




×







304



      *pStream = p;




×







305



      return TSDB_CODE_SUCCESS;




×







306



    }










307



    sdbRelease(pSdb, p);




×







308



  }










309












310



  return TSDB_CODE_STREAM_TASK_NOT_EXIST;




×







311



}










312












313



void mndKillTransImpl(SMnode *pMnode, int32_t transId, const char *pDbName) {




×







314



  STrans *pTrans = mndAcquireTrans(pMnode, transId);




×







315



  if (pTrans != NULL) {





×







316



    mInfo("kill active transId:%d in Db:%s", transId, pDbName);





×







317



    int32_t code = mndKillTrans(pMnode, pTrans);




×







318



    mndReleaseTrans(pMnode, pTrans);




×







319



    if (code) {





×







320



      mError("failed to kill transId:%d, code:%s", pTrans->id, tstrerror(code));





×







321



    }










322



  } else {










323



    mError("failed to acquire trans in Db:%s, transId:%d", pDbName, transId);





×







324



  }










325



}




×







326












327



int32_t extractNodeEpset(SMnode *pMnode, SEpSet *pEpSet, bool *hasEpset, int32_t taskId, int32_t nodeId) {




×







328



  *hasEpset = false;




×







329












330



  pEpSet->numOfEps = 0;




×







331



  if (nodeId == SNODE_HANDLE) {





×







332



    SSnodeObj *pObj = NULL;




×







333



    void      *pIter = NULL;




×







334












335



    pIter = sdbFetch(pMnode->pSdb, SDB_SNODE, pIter, (void **)&pObj);




×







336



    if (pIter != NULL) {





×







337



      int32_t code = addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, pObj->pDnode->port);




×







338



      sdbRelease(pMnode->pSdb, pObj);




×







339



      sdbCancelFetch(pMnode->pSdb, pIter);




×







340



      if (code) {





×







341



        *hasEpset = false;




×







342



        mError("failed to set epset");





×







343



      } else {










344



        *hasEpset = true;




×







345



      }










346



      return code;




×







347



    } else {










348



      mError("failed to acquire snode epset");





×







349



      return TSDB_CODE_INVALID_PARA;




×







350



    }










351



  } else {










352



    SVgObj *pVgObj = mndAcquireVgroup(pMnode, nodeId);




×







353



    if (pVgObj != NULL) {





×







354



      SEpSet epset = mndGetVgroupEpset(pMnode, pVgObj);




×







355



      mndReleaseVgroup(pMnode, pVgObj);




×







356












357



      epsetAssign(pEpSet, &epset);




×







358



      *hasEpset = true;




×







359



      return TSDB_CODE_SUCCESS;




×







360



    } else {










361



      mDebug("orphaned task:0x%x need to be dropped, nodeId:%d, no redo action", taskId, nodeId);





×







362



      return TSDB_CODE_SUCCESS;




×







363



    }










364



  }










365



}










366












367



int32_t mndGetStreamTask(STaskId *pId, SStreamObj *pStream, SStreamTask **pTask) {




×







368



  *pTask = NULL;




×







369












370



  SStreamTask     *p = NULL;




×







371



  SStreamTaskIter *pIter = NULL;




×







372



  int32_t          code = createStreamTaskIter(pStream, &pIter);




×







373



  if (code) {





×







374



    mError("failed to create stream task iter:%s", pStream->name);





×







375



    return code;




×







376



  }










377












378



  while (streamTaskIterNextTask(pIter)) {





×







379



    code = streamTaskIterGetCurrent(pIter, &p);




×







380



    if (code) {





×







381



      continue;




×







382



    }










383












384



    if (p->id.taskId == pId->taskId) {





×







385



      destroyStreamTaskIter(pIter);




×







386



      *pTask = p;




×







387



      return 0;




×







388



    }










389



  }










390












391



  destroyStreamTaskIter(pIter);




×







392



  return TSDB_CODE_FAILED;




×







393



}










394












395



int32_t mndGetNumOfStreamTasks(const SStreamObj *pStream) {




×







396



  int32_t num = 0;




×







397



  for (int32_t i = 0; i < taosArrayGetSize(pStream->tasks); ++i) {





×







398



    SArray *pLevel = taosArrayGetP(pStream->tasks, i);




×







399



    num += taosArrayGetSize(pLevel);




×







400



  }










401












402



  return num;




×







403



}










404












405



int32_t mndGetNumOfStreams(SMnode *pMnode, char *dbName, int32_t *pNumOfStreams) {




×







406



  SSdb   *pSdb = pMnode->pSdb;




×







407



  SDbObj *pDb = mndAcquireDb(pMnode, dbName);




×







408



  if (pDb == NULL) {





×







409



    TAOS_RETURN(TSDB_CODE_MND_DB_NOT_SELECTED);




×







410



  }










411












412



  int32_t numOfStreams = 0;




×







413



  void   *pIter = NULL;




×







414



  while (1) {




×







415



    SStreamObj *pStream = NULL;




×







416



    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);




×







417



    if (pIter == NULL) break;





×







418












419



    if (pStream->sourceDbUid == pDb->uid) {





×







420



      numOfStreams++;




×







421



    }










422












423



    sdbRelease(pSdb, pStream);




×







424



  }










425












426



  *pNumOfStreams = numOfStreams;




×







427



  mndReleaseDb(pMnode, pDb);




×







428



  return 0;




×







429



}










430












431



static void freeTaskList(void *param) {




×







432



  SArray **pList = (SArray **)param;




×







433



  taosArrayDestroy(*pList);




×







434



}




×







435












436



int32_t mndInitExecInfo() {




×







437



  int32_t code = taosThreadMutexInit(&execInfo.lock, NULL);




×







438



  if (code) {





×







439



    return code;




×







440



  }










441












442



  _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR);




×







443












444



  execInfo.pTaskList = taosArrayInit(4, sizeof(STaskId));




×







445



  execInfo.pTaskMap = taosHashInit(64, fn, true, HASH_NO_LOCK);




×







446



  execInfo.transMgmt.pDBTrans = taosHashInit(32, fn, true, HASH_NO_LOCK);




×







447



  execInfo.pTransferStateStreams = taosHashInit(32, fn, true, HASH_NO_LOCK);




×







448



  execInfo.pChkptStreams = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




×







449



  execInfo.pStreamConsensus = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




×







450



  execInfo.pNodeList = taosArrayInit(4, sizeof(SNodeEntry));




×







451



  execInfo.pKilledChkptTrans = taosArrayInit(4, sizeof(SStreamTaskResetMsg));




×







452












453



  if (execInfo.pTaskList == NULL || execInfo.pTaskMap == NULL || execInfo.transMgmt.pDBTrans == NULL ||





×







454



      execInfo.pTransferStateStreams == NULL || execInfo.pChkptStreams == NULL || execInfo.pStreamConsensus == NULL ||





×







455



      execInfo.pNodeList == NULL || execInfo.pKilledChkptTrans == NULL) {





×







456



    mError("failed to initialize the stream runtime env, code:%s", tstrerror(terrno));





×







457



    return terrno;




×







458



  }










459












460



  execInfo.role = NODE_ROLE_UNINIT;




×







461



  execInfo.switchFromFollower = false;




×







462












463



  taosHashSetFreeFp(execInfo.pTransferStateStreams, freeTaskList);




×







464



  taosHashSetFreeFp(execInfo.pChkptStreams, freeTaskList);




×







465



  taosHashSetFreeFp(execInfo.pStreamConsensus, freeTaskList);




×







466



  return 0;




×







467



}










468












469



void removeExpiredNodeInfo(const SArray *pNodeSnapshot) {




×







470



  SArray *pValidList = taosArrayInit(4, sizeof(SNodeEntry));




×







471



  if (pValidList == NULL) {  // not continue





×







472



    return;




×







473



  }










474












475



  int32_t size = taosArrayGetSize(pNodeSnapshot);




×







476



  int32_t oldSize = taosArrayGetSize(execInfo.pNodeList);




×







477












478



  for (int32_t i = 0; i < oldSize; ++i) {





×







479



    SNodeEntry *p = taosArrayGet(execInfo.pNodeList, i);




×







480



    if (p == NULL) {





×







481



      continue;




×







482



    }










483












484



    for (int32_t j = 0; j < size; ++j) {





×







485



      SNodeEntry *pEntry = taosArrayGet(pNodeSnapshot, j);




×







486



      if (pEntry == NULL) {





×







487



        continue;




×







488



      }










489












490



      if (pEntry->nodeId == p->nodeId) {





×







491



        p->hbTimestamp = pEntry->hbTimestamp;




×







492












493



        void *px = taosArrayPush(pValidList, p);




×







494



        if (px == NULL) {





×







495



          mError("failed to put node into list, nodeId:%d", p->nodeId);





×







496



        } else {










497



          mDebug("vgId:%d ts:%" PRId64 " HbMsgId:%d is valid", p->nodeId, p->hbTimestamp, p->lastHbMsgId);





×







498



        }










499



        break;




×







500



      }










501



    }










502



  }










503












504



  taosArrayDestroy(execInfo.pNodeList);




×







505



  execInfo.pNodeList = pValidList;




×







506












507



  mDebug("remain %d valid node entries after clean expired nodes info, prev size:%d",





×







508



         (int32_t)taosArrayGetSize(pValidList), oldSize);










509



}










510












511



int32_t doRemoveTasks(SStreamExecInfo *pExecNode, STaskId *pRemovedId) {




×







512



  void *p = taosHashGet(pExecNode->pTaskMap, pRemovedId, sizeof(*pRemovedId));




×







513



  if (p == NULL) {





×







514



    return TSDB_CODE_SUCCESS;




×







515



  }










516












517



  int32_t code = taosHashRemove(pExecNode->pTaskMap, pRemovedId, sizeof(*pRemovedId));




×







518



  if (code) {





×







519



    return code;




×







520



  }










521












522



  for (int32_t k = 0; k < taosArrayGetSize(pExecNode->pTaskList); ++k) {





×







523



    STaskId *pId = taosArrayGet(pExecNode->pTaskList, k);




×







524



    if (pId == NULL) {





×







525



      continue;




×







526



    }










527












528



    if (pId->taskId == pRemovedId->taskId && pId->streamId == pRemovedId->streamId) {





×







529



      taosArrayRemove(pExecNode->pTaskList, k);




×







530












531



      int32_t num = taosArrayGetSize(pExecNode->pTaskList);




×







532



      mInfo("s-task:0x%x removed from buffer, remain:%d in buffer list", (int32_t)pRemovedId->taskId, num);





×







533



      break;




×







534



    }










535



  }










536












537



  return TSDB_CODE_SUCCESS;




×







538



}










539












540



void removeTasksInBuf(SArray *pTaskIds, SStreamExecInfo *pExecInfo) {




×







541



  for (int32_t i = 0; i < taosArrayGetSize(pTaskIds); ++i) {





×







542



    STaskId *pId = taosArrayGet(pTaskIds, i);




×







543



    if (pId == NULL) {





×







544



      continue;




×







545



    }










546












547



    int32_t code = doRemoveTasks(pExecInfo, pId);




×







548



    if (code) {





×







549



      mError("failed to remove task in buffer list, 0x%" PRIx64, pId->taskId);





×







550



    }










551



  }










552



}




×







553












554



void removeStreamTasksInBuf(SStreamObj *pStream, SStreamExecInfo *pExecNode) {




×







555



  SStreamTaskIter *pIter = NULL;




×







556



  streamMutexLock(&pExecNode->lock);




×







557












558



  // 1. remove task entries










559



  int32_t code = createStreamTaskIter(pStream, &pIter);




×







560



  if (code) {





×







561



    streamMutexUnlock(&pExecNode->lock);




×







562



    mError("failed to create stream task iter:%s", pStream->name);





×







563



    return;




×







564



  }










565












566



  while (streamTaskIterNextTask(pIter)) {





×







567



    SStreamTask *pTask = NULL;




×







568



    code = streamTaskIterGetCurrent(pIter, &pTask);




×







569



    if (code) {





×







570



      continue;




×







571



    }










572












573



    STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId};




×







574



    code = doRemoveTasks(pExecNode, &id);




×







575



    if (code) {





×







576



      mError("failed to remove task in buffer list, 0x%" PRIx64, id.taskId);





×







577



    }










578



  }










579












580



  if (taosHashGetSize(pExecNode->pTaskMap) != taosArrayGetSize(pExecNode->pTaskList)) {





×







581



    streamMutexUnlock(&pExecNode->lock);




×







582



    destroyStreamTaskIter(pIter);




×







583



    mError("task map size, task list size, not equal");





×







584



    return;




×







585



  }










586












587



  // 2. remove stream entry in consensus hash table and checkpoint-report hash table










588



  code = mndClearConsensusCheckpointId(execInfo.pStreamConsensus, pStream->uid);




×







589



  if (code) {





×







590



    mError("failed to clear consensus checkpointId, code:%s", tstrerror(code));





×







591



  }










592












593



  code = mndClearChkptReportInfo(execInfo.pChkptStreams, pStream->uid);




×







594



  if (code) {





×







595



    mError("failed to clear the checkpoint report info, code:%s", tstrerror(code));





×







596



  }










597












598



  streamMutexUnlock(&pExecNode->lock);




×







599



  destroyStreamTaskIter(pIter);




×







600



}










601












602



static bool taskNodeExists(SArray *pList, int32_t nodeId) {




×







603



  size_t num = taosArrayGetSize(pList);




×







604












605



  for (int32_t i = 0; i < num; ++i) {





×







606



    SNodeEntry *pEntry = taosArrayGet(pList, i);




×







607



    if (pEntry == NULL) {





×







608



      continue;




×







609



    }










610












611



    if (pEntry->nodeId == nodeId) {





×







612



      return true;




×







613



    }










614



  }










615












616



  return false;




×







617



}










618












619



int32_t removeExpiredNodeEntryAndTaskInBuf(SArray *pNodeSnapshot) {




×







620



  SArray *pRemovedTasks = taosArrayInit(4, sizeof(STaskId));




×







621



  if (pRemovedTasks == NULL) {





×







622



    return terrno;




×







623



  }










624












625



  int32_t numOfTask = taosArrayGetSize(execInfo.pTaskList);




×







626



  for (int32_t i = 0; i < numOfTask; ++i) {





×







627



    STaskId *pId = taosArrayGet(execInfo.pTaskList, i);




×







628



    if (pId == NULL) {





×







629



      continue;




×







630



    }










631












632



    STaskStatusEntry *pEntry = taosHashGet(execInfo.pTaskMap, pId, sizeof(*pId));




×







633



    if (pEntry == NULL) {





×







634



      continue;




×







635



    }










636












637



    if (pEntry->nodeId == SNODE_HANDLE) {





×







638



      continue;




×







639



    }










640












641



    bool existed = taskNodeExists(pNodeSnapshot, pEntry->nodeId);




×







642



    if (!existed) {





×







643



      void *p = taosArrayPush(pRemovedTasks, pId);




×







644



      if (p == NULL) {





×







645



        mError("failed to put task entry into remove list, taskId:0x%" PRIx64, pId->taskId);





×







646



      }










647



    }










648



  }










649












650



  removeTasksInBuf(pRemovedTasks, &execInfo);




×







651












652



  mDebug("remove invalid stream tasks:%d, remain:%d", (int32_t)taosArrayGetSize(pRemovedTasks),





×







653



         (int32_t)taosArrayGetSize(execInfo.pTaskList));










654












655



  removeExpiredNodeInfo(pNodeSnapshot);




×







656












657



  taosArrayDestroy(pRemovedTasks);




×







658



  return 0;




×







659



}










660












661



static int32_t allTasksSendChkptReport(SChkptReportInfo* pReportInfo, int32_t numOfTasks, const char* pName) {




×







662



  int64_t checkpointId = -1;




×







663



  int32_t transId = -1;




×







664



  int32_t taskId = -1;




×







665












666



  int32_t existed = (int32_t)taosArrayGetSize(pReportInfo->pTaskList);




×







667



  if (existed != numOfTasks) {





×







668



    mDebug("stream:0x%" PRIx64 " %s %d/%d tasks send checkpoint-report, %d not send", pReportInfo->streamId, pName,





×







669



           existed, numOfTasks, numOfTasks - existed);










670



    return -1;




×







671



  }










672












673



  // acquire current active checkpointId, and do cross-check checkpointId info in exec.pTaskList










674



  for(int32_t i = 0; i < numOfTasks; ++i) {





×







675



    STaskChkptInfo *pInfo = taosArrayGet(pReportInfo->pTaskList, i);




×







676



    if (pInfo == NULL) {





×







677



      continue;




×







678



    }










679












680



    if (checkpointId == -1) {





×







681



      checkpointId = pInfo->checkpointId;




×







682



      transId = pInfo->transId;




×







683



      taskId = pInfo->taskId;




×







684



    } else if (checkpointId != pInfo->checkpointId) {





×







685



      mError("stream:0x%" PRIx64





×







686



             " checkpointId in checkpoint-report list are not identical, type 1 taskId:0x%x checkpointId:%" PRId64










687



             ", type 2 taskId:0x%x checkpointId:%" PRId64,










688



             pReportInfo->streamId, taskId, checkpointId, pInfo->taskId, pInfo->checkpointId);










689



      return -1;




×







690



    }










691



  }










692












693



  // check for the correct checkpointId for current task info in STaskChkptInfo










694



  STaskChkptInfo  *p = taosArrayGet(pReportInfo->pTaskList, 0);




×







695



  STaskId id = {.streamId = p->streamId, .taskId = p->taskId};




×







696



  STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id));




×







697












698



  // cross-check failed, there must be something unknown wrong










699



  SStreamTransInfo *pTransInfo = taosHashGet(execInfo.transMgmt.pDBTrans, &id.streamId, sizeof(id.streamId));




×







700



  if (pTransInfo == NULL) {





×







701



    mWarn("stream:0x%" PRIx64 " no active trans exists for checkpoint transId:%d, it may have been cleared already",





×







702



           id.streamId, transId);










703












704



    if (pe->checkpointInfo.activeId != 0 && pe->checkpointInfo.activeId != checkpointId) {





×







705



      mWarn("stream:0x%" PRIx64 " active checkpointId is not equalled to the required, current:%" PRId64





×







706



            ", req:%" PRId64 " recheck next time",










707



            id.streamId, pe->checkpointInfo.activeId, checkpointId);










708



      return -1;




×







709



    } else {










710



      //  do nothing










711



    }










712



  } else {










713



    if (pTransInfo->transId != transId) {





×







714



      mError("stream:0x%" PRIx64





×







715



             " checkpoint-report list info are expired, active transId:%d trans in list:%d, recheck next time",










716



             id.streamId, pTransInfo->transId, transId);










717



      return -1;




×







718



    }










719



  }










720












721



  mDebug("stream:0x%" PRIx64 " %s all %d tasks send checkpoint-report, start to update checkpoint-info", id.streamId,





×







722



         pName, numOfTasks);










723












724



  return TSDB_CODE_SUCCESS;




×







725



}










726












727



int32_t mndScanCheckpointReportInfo(SRpcMsg *pReq) {




×







728



  SMnode *pMnode = pReq->info.node;




×







729



  void   *pIter = NULL;




×







730



  int32_t code = 0;




×







731



  SArray *pDropped = taosArrayInit(4, sizeof(int64_t));




×







732



  if (pDropped == NULL) {





×







733



    return terrno;




×







734



  }










735












736



  mDebug("start to scan checkpoint report info");





×







737












738



  streamMutexLock(&execInfo.lock);




×







739












740



  while ((pIter = taosHashIterate(execInfo.pChkptStreams, pIter)) != NULL) {





×







741



    SChkptReportInfo *px = (SChkptReportInfo *)pIter;




×







742



    if (taosArrayGetSize(px->pTaskList) == 0) {





×







743



      continue;




×







744



    }










745












746



    STaskChkptInfo *pInfo = taosArrayGet(px->pTaskList, 0);




×







747



    if (pInfo == NULL) {





×







748



      continue;




×







749



    }










750












751



    SStreamObj *pStream = NULL;




×







752



    code = mndGetStreamObj(pMnode, pInfo->streamId, &pStream);




×







753



    if (pStream == NULL || code != 0) {





×







754



      mDebug("failed to acquire stream:0x%" PRIx64 " remove it from checkpoint-report list", pInfo->streamId);





×







755



      void *p = taosArrayPush(pDropped, &pInfo->streamId);




×







756



      if (p == NULL) {





×







757



        mError("failed to put stream into drop list:0x%" PRIx64, pInfo->streamId);





×







758



      }










759



      continue;




×







760



    }










761












762



    int32_t total = mndGetNumOfStreamTasks(pStream);




×







763



    int32_t ret = allTasksSendChkptReport(px, total, pStream->name);




×







764



    if (ret == 0) {





×







765



      code = mndStreamTransConflictCheck(pMnode, pStream->uid, MND_STREAM_CHKPT_UPDATE_NAME, false);




×







766



      if (code == 0) {





×







767



        code = mndCreateStreamChkptInfoUpdateTrans(pMnode, pStream, px->pTaskList);




×







768



        if (code == TSDB_CODE_SUCCESS || code == TSDB_CODE_ACTION_IN_PROGRESS) {  // remove this entry





×







769



          taosArrayClear(px->pTaskList);




×







770



          mInfo("stream:0x%" PRIx64 " clear checkpoint-report list and update the report checkpointId from:%" PRId64





×







771



                " to %" PRId64,










772



                pInfo->streamId, px->reportChkpt, pInfo->checkpointId);










773



          px->reportChkpt = pInfo->checkpointId;




×







774



        } else {










775



          mDebug("stream:0x%" PRIx64 " not launch chkpt-info update trans, due to checkpoint not finished yet",





×







776



                 pInfo->streamId);










777



        }










778












779



        sdbRelease(pMnode->pSdb, pStream);




×







780



        break;




×







781



      } else {










782



        mDebug("stream:0x%" PRIx64 " active checkpoint trans not finished yet, wait", pInfo->streamId);





×







783



      }










784



    }










785












786



    sdbRelease(pMnode->pSdb, pStream);




×







787



  }










788












789



  int32_t size = taosArrayGetSize(pDropped);




×







790



  if (size > 0) {





×







791



    for (int32_t i = 0; i < size; ++i) {





×







792



      int64_t *pStreamId = (int64_t *)taosArrayGet(pDropped, i);




×







793



      if (pStreamId == NULL) {





×







794



        continue;




×







795



      }










796












797



      code = taosHashRemove(execInfo.pChkptStreams, pStreamId, sizeof(*pStreamId));




×







798



      if (code) {





×







799



        mError("failed to remove stream in buf:0x%" PRIx64, *pStreamId);





×







800



      }










801



    }










802












803



    int32_t numOfStreams = taosHashGetSize(execInfo.pChkptStreams);




×







804



    mDebug("drop %d stream(s) in checkpoint-report list, remain:%d", size, numOfStreams);





×







805



  }










806












807



  streamMutexUnlock(&execInfo.lock);




×







808












809



  taosArrayDestroy(pDropped);




×







810












811



  mDebug("end to scan checkpoint report info")





×







812



  return TSDB_CODE_SUCCESS;




×







813



}










814












815



int32_t mndCreateSetConsensusChkptIdTrans(SMnode *pMnode, SStreamObj *pStream, int32_t taskId, int64_t checkpointId,




×







816



                                          int64_t ts) {










817



  char msg[128] = {0};




×







818



  snprintf(msg, tListLen(msg), "set consen-chkpt-id for task:0x%x", taskId);




×







819












820



  STrans *pTrans = NULL;




×







821



  int32_t code = doCreateTrans(pMnode, pStream, NULL, TRN_CONFLICT_NOTHING, MND_STREAM_CHKPT_CONSEN_NAME, msg, &pTrans);




×







822



  if (pTrans == NULL || code != 0) {





×







823



    return terrno;




×







824



  }










825












826



  STaskId      id = {.streamId = pStream->uid, .taskId = taskId};




×







827



  SStreamTask *pTask = NULL;




×







828



  code = mndGetStreamTask(&id, pStream, &pTask);




×







829



  if (code) {





×







830



    mError("failed to get task:0x%x in stream:%s, failed to create consensus-checkpointId", taskId, pStream->name);





×







831



    sdbRelease(pMnode->pSdb, pStream);




×







832



    return code;




×







833



  }










834












835



  code = mndStreamRegisterTrans(pTrans, MND_STREAM_CHKPT_CONSEN_NAME, pStream->uid);




×







836



  if (code) {





×







837



    sdbRelease(pMnode->pSdb, pStream);




×







838



    return code;




×







839



  }










840












841



  code = mndStreamSetChkptIdAction(pMnode, pTrans, pTask, checkpointId, ts);




×







842



  if (code != 0) {





×







843



    sdbRelease(pMnode->pSdb, pStream);




×







844



    mndTransDrop(pTrans);




×







845



    return code;




×







846



  }










847












848



  code = mndPersistTransLog(pStream, pTrans, SDB_STATUS_READY);




×







849



  if (code) {





×







850



    sdbRelease(pMnode->pSdb, pStream);




×







851



    mndTransDrop(pTrans);




×







852



    return code;




×







853



  }










854












855



  code = mndTransPrepare(pMnode, pTrans);




×







856



  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_ACTION_IN_PROGRESS) {





×







857



    mError("trans:%d, failed to prepare set consensus-chkptId trans since %s", pTrans->id, terrstr());





×







858



    sdbRelease(pMnode->pSdb, pStream);




×







859



    mndTransDrop(pTrans);




×







860



    return code;




×







861



  }










862












863



  sdbRelease(pMnode->pSdb, pStream);




×







864



  mndTransDrop(pTrans);




×







865












866



  return TSDB_CODE_ACTION_IN_PROGRESS;




×







867



}










868












869



int32_t mndGetConsensusInfo(SHashObj *pHash, int64_t streamId, int32_t numOfTasks, SCheckpointConsensusInfo **pInfo) {




×







870



  *pInfo = NULL;




×







871












872



  void *px = taosHashGet(pHash, &streamId, sizeof(streamId));




×







873



  if (px != NULL) {





×







874



    *pInfo = px;




×







875



    return 0;




×







876



  }










877












878



  SCheckpointConsensusInfo p = {




×







879



      .pTaskList = taosArrayInit(4, sizeof(SCheckpointConsensusEntry)),




×







880



      .numOfTasks = numOfTasks,










881



      .streamId = streamId,










882



  };










883












884



  if (p.pTaskList == NULL) {





×







885



    return terrno;




×







886



  }










887












888



  int32_t code = taosHashPut(pHash, &streamId, sizeof(streamId), &p, sizeof(p));




×







889



  if (code == 0) {





×







890



    void *pChkptInfo = (SCheckpointConsensusInfo *)taosHashGet(pHash, &streamId, sizeof(streamId));




×







891



    *pInfo = pChkptInfo;




×







892



  } else {










893



    *pInfo = NULL;




×







894



  }










895












896



  return code;




×







897



}










898












899



// no matter existed or not, add the request into info list anyway, since we need to send rsp mannually










900



// discard the msg may lead to the lost of connections.










901



void mndAddConsensusTasks(SCheckpointConsensusInfo *pInfo, const SRestoreCheckpointInfo *pRestoreInfo) {




×







902



  SCheckpointConsensusEntry info = {.ts = taosGetTimestampMs()};




×







903



  memcpy(&info.req, pRestoreInfo, sizeof(info.req));




×







904












905



  int32_t num = (int32_t) taosArrayGetSize(pInfo->pTaskList);




×







906



  for (int32_t i = 0; i < num; ++i) {





×







907



    SCheckpointConsensusEntry *p = taosArrayGet(pInfo->pTaskList, i);




×







908



    if (p == NULL) {





×







909



      continue;




×







910



    }










911












912



    if (p->req.taskId == info.req.taskId) {





×







913



      mDebug("s-task:0x%x already in consensus-checkpointId list for stream:0x%" PRIx64 ", update ts %" PRId64





×







914



             "->%" PRId64 " checkpointId:%" PRId64 " -> %" PRId64 " total existed:%d",










915



             pRestoreInfo->taskId, pRestoreInfo->streamId, p->req.startTs, info.req.startTs, p->req.checkpointId,










916



             info.req.checkpointId, num);










917



      p->req.startTs = info.req.startTs;




×







918



      p->req.checkpointId = info.req.checkpointId;




×







919



      p->req.transId = info.req.transId;




×







920



      return;




×







921



    }










922



  }










923












924



  void *p = taosArrayPush(pInfo->pTaskList, &info);




×







925



  if (p == NULL) {





×







926



    mError("s-task:0x%x failed to put task into consensus-checkpointId list, code: out of memory", info.req.taskId);





×







927



  } else {










928



    num = taosArrayGetSize(pInfo->pTaskList);




×







929



    mDebug("s-task:0x%x checkpointId:%" PRId64 " added into consensus-checkpointId list, stream:0x%" PRIx64





×







930



           " waiting tasks:%d",










931



           pRestoreInfo->taskId, pRestoreInfo->checkpointId, pRestoreInfo->streamId, num);










932



  }










933



}










934












935



void mndClearConsensusRspEntry(SCheckpointConsensusInfo *pInfo) {




×







936



  taosArrayDestroy(pInfo->pTaskList);




×







937



  pInfo->pTaskList = NULL;




×







938



}




×







939












940



int32_t mndClearConsensusCheckpointId(SHashObj *pHash, int64_t streamId) {




×







941



  int32_t code = 0;




×







942



  int32_t numOfStreams = taosHashGetSize(pHash);




×







943



  if (numOfStreams == 0) {





×







944



    return code;




×







945



  }










946












947



  code = taosHashRemove(pHash, &streamId, sizeof(streamId));




×







948



  if (code == 0) {





×







949



    mDebug("drop stream:0x%" PRIx64 " in consensus-checkpointId list, remain:%d", streamId, numOfStreams);





×







950



  } else {










951



    mError("failed to remove stream:0x%" PRIx64 " in consensus-checkpointId list, remain:%d", streamId, numOfStreams);





×







952



  }










953












954



  return code;




×







955



}










956












957



int32_t mndClearChkptReportInfo(SHashObj *pHash, int64_t streamId) {




×







958



  int32_t code = 0;




×







959



  int32_t numOfStreams = taosHashGetSize(pHash);




×







960



  if (numOfStreams == 0) {





×







961



    return code;




×







962



  }










963












964



  code = taosHashRemove(pHash, &streamId, sizeof(streamId));




×







965



  if (code == 0) {





×







966



    mDebug("drop stream:0x%" PRIx64 " in chkpt-report list, remain:%d", streamId, numOfStreams);





×







967



  } else {










968



    mError("failed to remove stream:0x%" PRIx64 " in chkpt-report list, remain:%d", streamId, numOfStreams);





×







969



  }










970












971



  return code;




×







972



}










973












974



int32_t mndResetChkptReportInfo(SHashObj *pHash, int64_t streamId) {




×







975



  SChkptReportInfo *pInfo = taosHashGet(pHash, &streamId, sizeof(streamId));




×







976



  if (pInfo != NULL) {





×







977



    taosArrayClear(pInfo->pTaskList);




×







978



    mDebug("stream:0x%" PRIx64 " checkpoint-report list cleared, prev report checkpointId:%" PRId64, streamId,





×







979



           pInfo->reportChkpt);










980



    return 0;




×







981



  }










982












983



  return TSDB_CODE_MND_STREAM_NOT_EXIST;




×







984



}










985












986



static void mndShowStreamStatus(char *dst, int8_t status) {




×







987



  if (status == STREAM_STATUS__NORMAL) {





×







988



    tstrncpy(dst, "ready", MND_STREAM_TRIGGER_NAME_SIZE);




×







989



  } else if (status == STREAM_STATUS__STOP) {





×







990



    tstrncpy(dst, "stop", MND_STREAM_TRIGGER_NAME_SIZE);




×







991



  } else if (status == STREAM_STATUS__FAILED) {





×







992



    tstrncpy(dst, "failed", MND_STREAM_TRIGGER_NAME_SIZE);




×







993



  } else if (status == STREAM_STATUS__RECOVER) {





×







994



    tstrncpy(dst, "recover", MND_STREAM_TRIGGER_NAME_SIZE);




×







995



  } else if (status == STREAM_STATUS__PAUSE) {





×







996



    tstrncpy(dst, "paused", MND_STREAM_TRIGGER_NAME_SIZE);




×







997



  }










998



}




×







999












1000



static void mndShowStreamTrigger(char *dst, SStreamObj *pStream) {




×







1001



  int8_t trigger = pStream->conf.trigger;




×







1002



  if (trigger == STREAM_TRIGGER_AT_ONCE) {





×







1003



    tstrncpy(dst, "at once", MND_STREAM_TRIGGER_NAME_SIZE);




×







1004



  } else if (trigger == STREAM_TRIGGER_WINDOW_CLOSE) {





×







1005



    tstrncpy(dst, "window close", MND_STREAM_TRIGGER_NAME_SIZE);




×







1006



  } else if (trigger == STREAM_TRIGGER_MAX_DELAY) {





×







1007



    tstrncpy(dst, "max delay", MND_STREAM_TRIGGER_NAME_SIZE);




×







1008



  } else if (trigger == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {





×







1009



    tstrncpy(dst, "force window close", MND_STREAM_TRIGGER_NAME_SIZE);




×







1010



  }










1011



}




×







1012












1013



static void int64ToHexStr(int64_t id, char *pBuf, int32_t bufLen) {




×







1014



  memset(pBuf, 0, bufLen);




×







1015



  pBuf[2] = '0';




×







1016



  pBuf[3] = 'x';




×







1017












1018



  int32_t len = tintToHex(id, &pBuf[4]);




×







1019



  varDataSetLen(pBuf, len + 2);




×







1020



}




×







1021












1022



static int32_t isAllTaskPaused(SStreamObj *pStream, bool *pRes) {




×







1023



  int32_t          code = TSDB_CODE_SUCCESS;




×







1024



  int32_t          lino = 0;




×







1025



  SStreamTaskIter *pIter = NULL;




×







1026



  bool             isPaused =  true;




×







1027












1028



  taosRLockLatch(&pStream->lock);




×







1029



  code = createStreamTaskIter(pStream, &pIter);




×







1030



  TSDB_CHECK_CODE(code, lino, _end);





×







1031












1032



  while (streamTaskIterNextTask(pIter)) {





×







1033



    SStreamTask *pTask = NULL;




×







1034



    code = streamTaskIterGetCurrent(pIter, &pTask);




×







1035



    TSDB_CHECK_CODE(code, lino, _end);





×







1036












1037



    STaskId           id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId};




×







1038



    STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id));




×







1039



    if (pe == NULL) {





×







1040



      continue;




×







1041



    }










1042



    if (pe->status != TASK_STATUS__PAUSE) {





×







1043



      isPaused = false;




×







1044



    }










1045



  }










1046



  (*pRes) = isPaused;




×







1047












1048



_end:




×







1049



  destroyStreamTaskIter(pIter);




×







1050



  taosRUnLockLatch(&pStream->lock);




×







1051



  if (code != TSDB_CODE_SUCCESS) {





×







1052



    mError("error happens when get stream status, lino:%d, code:%s", lino, tstrerror(code));





×







1053



  }










1054



  return code;




×







1055



}










1056












1057



int32_t setStreamAttrInResBlock(SStreamObj *pStream, SSDataBlock *pBlock, int32_t numOfRows) {




×







1058



  int32_t code = 0;




×







1059



  int32_t cols = 0;




×







1060



  int32_t lino = 0;




×







1061












1062



  char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};




×







1063



  STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetDbStr(pStream->name), sizeof(streamName));




×







1064



  SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1065



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1066












1067



  code = colDataSetVal(pColInfo, numOfRows, (const char *)streamName, false);




×







1068



  TSDB_CHECK_CODE(code, lino, _end);





×







1069












1070



  // create time










1071



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1072



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1073



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pStream->createTime, false);




×







1074



  TSDB_CHECK_CODE(code, lino, _end);





×







1075












1076



  // stream id










1077



  char buf[128] = {0};




×







1078



  int64ToHexStr(pStream->uid, buf, tListLen(buf));




×







1079



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1080



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1081



  code = colDataSetVal(pColInfo, numOfRows, buf, false);




×







1082



  TSDB_CHECK_CODE(code, lino, _end);





×







1083












1084



  // related fill-history stream id










1085



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1086



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1087



  if (pStream->hTaskUid != 0) {





×







1088



    int64ToHexStr(pStream->hTaskUid, buf, tListLen(buf));




×







1089



    code = colDataSetVal(pColInfo, numOfRows, buf, false);




×







1090



  } else {










1091



    code = colDataSetVal(pColInfo, numOfRows, buf, true);




×







1092



  }










1093



  TSDB_CHECK_CODE(code, lino, _end);





×







1094












1095



  // related fill-history stream id










1096



  char sql[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};




×







1097



  STR_WITH_MAXSIZE_TO_VARSTR(sql, pStream->sql, sizeof(sql));




×







1098



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1099



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1100



  code = colDataSetVal(pColInfo, numOfRows, (const char *)sql, false);




×







1101



  TSDB_CHECK_CODE(code, lino, _end);





×







1102












1103



  char status[20 + VARSTR_HEADER_SIZE] = {0};




×







1104



  char status2[MND_STREAM_TRIGGER_NAME_SIZE] = {0};




×







1105



  bool isPaused = false;




×







1106



  code = isAllTaskPaused(pStream, &isPaused);




×







1107



  TSDB_CHECK_CODE(code, lino, _end);





×







1108












1109



  int8_t streamStatus = atomic_load_8(&pStream->status);




×







1110



  if (isPaused) {





×







1111



    streamStatus = STREAM_STATUS__PAUSE;




×







1112



  }










1113



  mndShowStreamStatus(status2, streamStatus);




×







1114



  STR_WITH_MAXSIZE_TO_VARSTR(status, status2, sizeof(status));




×







1115



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1116



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1117












1118



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&status, false);




×







1119



  TSDB_CHECK_CODE(code, lino, _end);





×







1120












1121



  char sourceDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};




×







1122



  STR_WITH_MAXSIZE_TO_VARSTR(sourceDB, mndGetDbStr(pStream->sourceDb), sizeof(sourceDB));




×







1123



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1124



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1125












1126



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&sourceDB, false);




×







1127



  TSDB_CHECK_CODE(code, lino, _end);





×







1128












1129



  char targetDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};




×







1130



  STR_WITH_MAXSIZE_TO_VARSTR(targetDB, mndGetDbStr(pStream->targetDb), sizeof(targetDB));




×







1131



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1132



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1133












1134



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&targetDB, false);




×







1135



  TSDB_CHECK_CODE(code, lino, _end);





×







1136












1137



  if (pStream->targetSTbName[0] == 0) {





×







1138



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1139



    TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1140












1141



    code = colDataSetVal(pColInfo, numOfRows, NULL, true);




×







1142



  } else {










1143



    char targetSTB[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};




×







1144



    STR_WITH_MAXSIZE_TO_VARSTR(targetSTB, mndGetStbStr(pStream->targetSTbName), sizeof(targetSTB));




×







1145



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1146



    TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1147












1148



    code = colDataSetVal(pColInfo, numOfRows, (const char *)&targetSTB, false);




×







1149



  }










1150



  TSDB_CHECK_CODE(code, lino, _end);





×







1151












1152



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1153



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1154












1155



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pStream->conf.watermark, false);




×







1156



  TSDB_CHECK_CODE(code, lino, _end);





×







1157












1158



  char trigger[20 + VARSTR_HEADER_SIZE] = {0};




×







1159



  char trigger2[MND_STREAM_TRIGGER_NAME_SIZE] = {0};




×







1160



  mndShowStreamTrigger(trigger2, pStream);




×







1161



  STR_WITH_MAXSIZE_TO_VARSTR(trigger, trigger2, sizeof(trigger));




×







1162



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1163



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1164












1165



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&trigger, false);




×







1166



  TSDB_CHECK_CODE(code, lino, _end);





×







1167












1168



  // sink_quota










1169



  char sinkQuota[20 + VARSTR_HEADER_SIZE] = {0};




×







1170



  sinkQuota[0] = '0';




×







1171



  char dstStr[20] = {0};




×







1172



  STR_TO_VARSTR(dstStr, sinkQuota)




×







1173



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1174



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1175












1176



  code = colDataSetVal(pColInfo, numOfRows, (const char *)dstStr, false);




×







1177



  TSDB_CHECK_CODE(code, lino, _end);





×







1178












1179



  // checkpoint interval










1180



  char tmp[20 + VARSTR_HEADER_SIZE] = {0};




×







1181



  (void)tsnprintf(varDataVal(tmp), sizeof(tmp) - VARSTR_HEADER_SIZE, "%d sec", tsStreamCheckpointInterval);




×







1182



  varDataSetLen(tmp, strlen(varDataVal(tmp)));




×







1183












1184



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1185



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1186












1187



  code = colDataSetVal(pColInfo, numOfRows, (const char *)tmp, false);




×







1188



  TSDB_CHECK_CODE(code, lino, _end);





×







1189












1190



  // checkpoint backup type










1191



  char backup[20 + VARSTR_HEADER_SIZE] = {0};




×







1192



  STR_TO_VARSTR(backup, "none")




×







1193



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1194



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1195












1196



  code = colDataSetVal(pColInfo, numOfRows, (const char *)backup, false);




×







1197



  TSDB_CHECK_CODE(code, lino, _end);





×







1198












1199



  // history scan idle










1200



  char scanHistoryIdle[20 + VARSTR_HEADER_SIZE] = {0};




×







1201



  tstrncpy(scanHistoryIdle, "100a", sizeof(scanHistoryIdle));




×







1202












1203



  memset(dstStr, 0, tListLen(dstStr));




×







1204



  STR_TO_VARSTR(dstStr, scanHistoryIdle)




×







1205



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1206



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1207












1208



  code = colDataSetVal(pColInfo, numOfRows, (const char *)dstStr, false);




×







1209












1210



_end:




×







1211



  if (code) {





×







1212



    mError("error happens when build stream attr result block, lino:%d, code:%s", lino, tstrerror(code));





×







1213



  }










1214



  return code;




×







1215



}










1216












1217



int32_t setTaskAttrInResBlock(SStreamObj *pStream, SStreamTask *pTask, SSDataBlock *pBlock, int32_t numOfRows,




×







1218



                              int32_t precision) {










1219



  SColumnInfoData *pColInfo = NULL;




×







1220



  int32_t          cols = 0;




×







1221



  int32_t          code = 0;




×







1222



  int32_t          lino = 0;




×







1223












1224



  STaskId id = {.streamId = pTask->id.streamId, .taskId = pTask->id.taskId};




×







1225












1226



  STaskStatusEntry *pe = taosHashGet(execInfo.pTaskMap, &id, sizeof(id));




×







1227



  if (pe == NULL) {





×







1228



    mError("task:0x%" PRIx64 " not exists in any vnodes, streamName:%s, streamId:0x%" PRIx64 " createTs:%" PRId64





×







1229



           " no valid status/stage info",










1230



           id.taskId, pStream->name, pStream->uid, pStream->createTime);










1231



    return TSDB_CODE_STREAM_TASK_NOT_EXIST;




×







1232



  }










1233












1234



  // stream name










1235



  char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};




×







1236



  STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetDbStr(pStream->name), sizeof(streamName));




×







1237












1238



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1239



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1240












1241



  code = colDataSetVal(pColInfo, numOfRows, (const char *)streamName, false);




×







1242



  TSDB_CHECK_CODE(code, lino, _end);





×







1243












1244



  // task id










1245



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1246



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1247












1248



  char idstr[128] = {0};




×







1249



  int64ToHexStr(pTask->id.taskId, idstr, tListLen(idstr));




×







1250



  code = colDataSetVal(pColInfo, numOfRows, idstr, false);




×







1251



  TSDB_CHECK_CODE(code, lino, _end);





×







1252












1253



  // node type










1254



  char nodeType[20 + VARSTR_HEADER_SIZE] = {0};




×







1255



  varDataSetLen(nodeType, 5);




×







1256



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1257



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1258












1259



  if (pTask->info.nodeId > 0) {





×







1260



    memcpy(varDataVal(nodeType), "vnode", 5);




×







1261



  } else {










1262



    memcpy(varDataVal(nodeType), "snode", 5);




×







1263



  }










1264



  code = colDataSetVal(pColInfo, numOfRows, nodeType, false);




×







1265



  TSDB_CHECK_CODE(code, lino, _end);





×







1266












1267



  // node id










1268



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1269



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1270












1271



  int64_t nodeId = TMAX(pTask->info.nodeId, 0);




×







1272



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&nodeId, false);




×







1273



  TSDB_CHECK_CODE(code, lino, _end);





×







1274












1275



  // level










1276



  char level[20 + VARSTR_HEADER_SIZE] = {0};




×







1277



  if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {





×







1278



    STR_WITH_SIZE_TO_VARSTR(level, "source", 6);




×







1279



  } else if (pTask->info.taskLevel == TASK_LEVEL__AGG) {





×







1280



    STR_WITH_SIZE_TO_VARSTR(level, "agg", 3);




×







1281



  } else if (pTask->info.taskLevel == TASK_LEVEL__SINK) {





×







1282



    STR_WITH_SIZE_TO_VARSTR(level, "sink", 4);




×







1283



  }










1284












1285



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1286



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1287












1288



  code = colDataSetVal(pColInfo, numOfRows, (const char *)level, false);




×







1289



  TSDB_CHECK_CODE(code, lino, _end);





×







1290












1291



  // status










1292



  char status[20 + VARSTR_HEADER_SIZE] = {0};




×







1293












1294



  const char *pStatus = streamTaskGetStatusStr(pe->status);




×







1295



  STR_TO_VARSTR(status, pStatus);




×







1296












1297



  // status










1298



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1299



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1300












1301



  code = colDataSetVal(pColInfo, numOfRows, (const char *)status, false);




×







1302



  TSDB_CHECK_CODE(code, lino, _end);





×







1303












1304



  // stage










1305



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1306



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1307












1308



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pe->stage, false);




×







1309



  TSDB_CHECK_CODE(code, lino, _end);





×







1310












1311



  // input queue










1312



  char        vbuf[40] = {0};




×







1313



  char        buf[38] = {0};




×







1314



  const char *queueInfoStr = "%4.2f MiB (%6.2f%)";




×







1315



  snprintf(buf, tListLen(buf), queueInfoStr, pe->inputQUsed, pe->inputRate);




×







1316



  STR_TO_VARSTR(vbuf, buf);




×







1317












1318



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1319



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1320












1321



  code = colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false);




×







1322



  TSDB_CHECK_CODE(code, lino, _end);





×







1323












1324



  // input total










1325



  const char *formatTotalMb = "%7.2f MiB";




×







1326



  const char *formatTotalGb = "%7.2f GiB";




×







1327



  if (pe->procsTotal < 1024) {





×







1328



    snprintf(buf, tListLen(buf), formatTotalMb, pe->procsTotal);




×







1329



  } else {










1330



    snprintf(buf, tListLen(buf), formatTotalGb, pe->procsTotal / 1024);




×







1331



  }










1332












1333



  memset(vbuf, 0, tListLen(vbuf));




×







1334



  STR_TO_VARSTR(vbuf, buf);




×







1335












1336



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1337



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1338












1339



  code = colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false);




×







1340



  TSDB_CHECK_CODE(code, lino, _end);





×







1341












1342



  // process throughput










1343



  const char *formatKb = "%7.2f KiB/s";




×







1344



  const char *formatMb = "%7.2f MiB/s";




×







1345



  if (pe->procsThroughput < 1024) {





×







1346



    snprintf(buf, tListLen(buf), formatKb, pe->procsThroughput);




×







1347



  } else {










1348



    snprintf(buf, tListLen(buf), formatMb, pe->procsThroughput / 1024);




×







1349



  }










1350












1351



  memset(vbuf, 0, tListLen(vbuf));




×







1352



  STR_TO_VARSTR(vbuf, buf);




×







1353












1354



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1355



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1356












1357



  code = colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false);




×







1358



  TSDB_CHECK_CODE(code, lino, _end);





×







1359












1360



  // output total










1361



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1362



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1363












1364



  if (pTask->info.taskLevel == TASK_LEVEL__SINK) {





×







1365



    colDataSetNULL(pColInfo, numOfRows);





×







1366



  } else {










1367



    (void)tsnprintf(buf, sizeof(buf), formatTotalMb, pe->outputTotal);




×







1368



    memset(vbuf, 0, tListLen(vbuf));




×







1369



    STR_TO_VARSTR(vbuf, buf);




×







1370












1371



    code = colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false);




×







1372



    TSDB_CHECK_CODE(code, lino, _end);





×







1373



  }










1374












1375



  // output throughput










1376



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1377



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1378












1379



  if (pTask->info.taskLevel == TASK_LEVEL__SINK) {





×







1380



    colDataSetNULL(pColInfo, numOfRows);





×







1381



  } else {










1382



    if (pe->outputThroughput < 1024) {





×







1383



      snprintf(buf, tListLen(buf), formatKb, pe->outputThroughput);




×







1384



    } else {










1385



      snprintf(buf, tListLen(buf), formatMb, pe->outputThroughput / 1024);




×







1386



    }










1387












1388



    memset(vbuf, 0, tListLen(vbuf));




×







1389



    STR_TO_VARSTR(vbuf, buf);




×







1390












1391



    code = colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false);




×







1392



    TSDB_CHECK_CODE(code, lino, _end);





×







1393



  }










1394



  // info










1395



  if (pTask->info.taskLevel == TASK_LEVEL__SINK) {





×







1396



    const char *sinkStr = "%.2f MiB";




×







1397



    snprintf(buf, tListLen(buf), sinkStr, pe->sinkDataSize);




×







1398



  } else if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {  // offset info





×







1399



    if (pTask->info.trigger == STREAM_TRIGGER_FORCE_WINDOW_CLOSE) {





×







1400



      int32_t ret = taosFormatUtcTime(buf, tListLen(buf), pe->processedVer, precision);




×







1401



      if (ret != 0) {





×







1402



        mError("failed to format processed timewindow, skey:%" PRId64, pe->processedVer);





×







1403



        memset(buf, 0, tListLen(buf));




×







1404



      }










1405



    } else {










1406



      const char *offsetStr = "%" PRId64 " [%" PRId64 ", %" PRId64 "]";




×







1407



      snprintf(buf, tListLen(buf), offsetStr, pe->processedVer, pe->verRange.minVer, pe->verRange.maxVer);




×







1408



    }










1409



  } else {










1410



    memset(buf, 0, tListLen(buf));




×







1411



  }










1412












1413



  STR_TO_VARSTR(vbuf, buf);




×







1414












1415



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1416



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1417












1418



  code = colDataSetVal(pColInfo, numOfRows, (const char *)vbuf, false);




×







1419



  TSDB_CHECK_CODE(code, lino, _end);





×







1420












1421



  // start_time










1422



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1423



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1424












1425



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pe->startTime, false);




×







1426



  TSDB_CHECK_CODE(code, lino, _end);





×







1427












1428



  // start id










1429



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1430



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1431












1432



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pe->startCheckpointId, false);




×







1433



  TSDB_CHECK_CODE(code, lino, _end);





×







1434












1435



  // start ver










1436



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1437



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1438












1439



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pe->startCheckpointVer, false);




×







1440



  TSDB_CHECK_CODE(code, lino, _end);





×







1441












1442



  // checkpoint time










1443



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1444



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1445












1446



  if (pe->checkpointInfo.latestTime != 0) {





×







1447



    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pe->checkpointInfo.latestTime, false);




×







1448



  } else {










1449



    code = colDataSetVal(pColInfo, numOfRows, 0, true);




×







1450



  }










1451



  TSDB_CHECK_CODE(code, lino, _end);





×







1452












1453



  // checkpoint_id










1454



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1455



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1456












1457



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pe->checkpointInfo.latestId, false);




×







1458



  TSDB_CHECK_CODE(code, lino, _end);





×







1459












1460



  // checkpoint version










1461



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1462



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1463












1464



  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pe->checkpointInfo.latestVer, false);




×







1465



  TSDB_CHECK_CODE(code, lino, _end);





×







1466












1467



  // checkpoint size










1468



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1469



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1470












1471



  colDataSetNULL(pColInfo, numOfRows);





×







1472












1473



  // checkpoint backup status










1474



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1475



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1476












1477



  code = colDataSetVal(pColInfo, numOfRows, 0, true);




×







1478



  TSDB_CHECK_CODE(code, lino, _end);





×







1479












1480



  // ds_err_info










1481



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1482



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1483












1484



  code = colDataSetVal(pColInfo, numOfRows, 0, true);




×







1485



  TSDB_CHECK_CODE(code, lino, _end);





×







1486












1487



  // history_task_id










1488



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1489



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1490












1491



  if (pe->hTaskId != 0) {





×







1492



    int64ToHexStr(pe->hTaskId, idstr, tListLen(idstr));




×







1493



    code = colDataSetVal(pColInfo, numOfRows, idstr, false);




×







1494



  } else {










1495



    code = colDataSetVal(pColInfo, numOfRows, 0, true);




×







1496



  }










1497



  TSDB_CHECK_CODE(code, lino, _end);





×







1498












1499



  // history_task_status










1500



  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×







1501



  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);





×







1502












1503



  code = colDataSetVal(pColInfo, numOfRows, 0, true);




×







1504



  TSDB_CHECK_CODE(code, lino, _end);





×







1505












1506



_end:




×







1507



  if (code) {





×







1508



    mError("error happens during build task attr result blocks, lino:%d, code:%s", lino, tstrerror(code));





×







1509



  }










1510



  return code;




×







1511



}










1512












1513



static bool isNodeEpsetChanged(const SEpSet *pPrevEpset, const SEpSet *pCurrent) {




×







1514



  const SEp *pEp = GET_ACTIVE_EP(pPrevEpset);




×







1515



  const SEp *p = GET_ACTIVE_EP(pCurrent);




×







1516












1517



  if (pEp->port == p->port && strncmp(pEp->fqdn, p->fqdn, TSDB_FQDN_LEN) == 0) {





×







1518



    return false;




×







1519



  }










1520



  return true;




×







1521



}










1522












1523



void mndDestroyVgroupChangeInfo(SVgroupChangeInfo *pInfo) {




×







1524



  if (pInfo != NULL) {





×







1525



    taosArrayDestroy(pInfo->pUpdateNodeList);




×







1526



    taosHashCleanup(pInfo->pDBMap);




×







1527



  }










1528



}




×







1529












1530



// 1. increase the replica does not affect the stream process.










1531



// 2. decreasing the replica may affect the stream task execution in the way that there is one or more running stream










1532



// tasks on the will be removed replica.










1533



// 3. vgroup redistribution is an combination operation of first increase replica and then decrease replica. So we










1534



// will handle it as mentioned in 1 & 2 items.










1535



int32_t mndFindChangedNodeInfo(SMnode *pMnode, const SArray *pPrevNodeList, const SArray *pNodeList,




×







1536



                               SVgroupChangeInfo *pInfo) {










1537



  int32_t code = 0;




×







1538



  int32_t lino = 0;




×







1539












1540



  if (pInfo == NULL) {





×







1541



    return TSDB_CODE_INVALID_PARA;




×







1542



  }










1543












1544



  pInfo->pUpdateNodeList = taosArrayInit(4, sizeof(SNodeUpdateInfo));




×







1545



  pInfo->pDBMap = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK);




×







1546












1547



  if (pInfo->pUpdateNodeList == NULL || pInfo->pDBMap == NULL) {





×







1548



    mndDestroyVgroupChangeInfo(pInfo);




×







1549



    TSDB_CHECK_NULL(NULL, code, lino, _err, terrno);




×







1550



  }










1551












1552



  int32_t numOfNodes = taosArrayGetSize(pPrevNodeList);




×







1553



  for (int32_t i = 0; i < numOfNodes; ++i) {





×







1554



    SNodeEntry *pPrevEntry = taosArrayGet(pPrevNodeList, i);




×







1555



    if (pPrevEntry == NULL) {





×







1556



      continue;




×







1557



    }










1558












1559



    int32_t num = taosArrayGetSize(pNodeList);




×







1560



    for (int32_t j = 0; j < num; ++j) {





×







1561



      SNodeEntry *pCurrent = taosArrayGet(pNodeList, j);




×







1562



      if (pCurrent == NULL) {





×







1563



        continue;




×







1564



      }










1565












1566



      if (pCurrent->nodeId == pPrevEntry->nodeId) {





×







1567



        if (pPrevEntry->stageUpdated || isNodeEpsetChanged(&pPrevEntry->epset, &pCurrent->epset)) {





×







1568



          const SEp *pPrevEp = GET_ACTIVE_EP(&pPrevEntry->epset);




×







1569












1570



          char buf[256] = {0};




×







1571



          code = epsetToStr(&pCurrent->epset, buf, tListLen(buf));  // ignore this error




×







1572



          if (code) {





×







1573



            mError("failed to convert epset string, code:%s", tstrerror(code));





×







1574



            TSDB_CHECK_CODE(code, lino, _err);





×







1575



          }










1576












1577



          mDebug("nodeId:%d restart/epset changed detected, old:%s:%d -> new:%s, stageUpdate:%d", pCurrent->nodeId,





×







1578



                 pPrevEp->fqdn, pPrevEp->port, buf, pPrevEntry->stageUpdated);










1579












1580



          SNodeUpdateInfo updateInfo = {.nodeId = pPrevEntry->nodeId};




×







1581



          epsetAssign(&updateInfo.prevEp, &pPrevEntry->epset);




×







1582



          epsetAssign(&updateInfo.newEp, &pCurrent->epset);




×







1583












1584



          void *p = taosArrayPush(pInfo->pUpdateNodeList, &updateInfo);




×







1585



          TSDB_CHECK_NULL(p, code, lino, _err, terrno);





×







1586



        }










1587












1588



        // todo handle the snode info










1589



        if (pCurrent->nodeId != SNODE_HANDLE) {





×







1590



          SVgObj *pVgroup = mndAcquireVgroup(pMnode, pCurrent->nodeId);




×







1591



          code = taosHashPut(pInfo->pDBMap, pVgroup->dbName, strlen(pVgroup->dbName), NULL, 0);




×







1592



          mndReleaseVgroup(pMnode, pVgroup);




×







1593



          TSDB_CHECK_CODE(code, lino, _err);





×







1594



        }










1595












1596



        break;




×







1597



      }










1598



    }










1599



  }










1600












1601



  return code;




×







1602












1603



_err:




×







1604



  mError("failed to find node change info, code:%s at %s line:%d", tstrerror(code), __func__, lino);





×







1605



  mndDestroyVgroupChangeInfo(pInfo);




×







1606



  return code;




×







1607



}










1608












1609



static int32_t doCheckForUpdated(SMnode *pMnode, SArray **ppNodeSnapshot) {




×







1610



  bool              allReady = false;




×







1611



  bool              nodeUpdated = false;




×







1612



  SVgroupChangeInfo changeInfo = {0};




×







1613












1614



  int32_t numOfNodes = extractStreamNodeList(pMnode);




×







1615












1616



  if (numOfNodes == 0) {





×







1617



    mDebug("stream task node change checking done, no vgroups exist, do nothing");





×







1618



    execInfo.ts = taosGetTimestampSec();




×







1619



    return false;




×







1620



  }










1621












1622



  for (int32_t i = 0; i < numOfNodes; ++i) {





×







1623



    SNodeEntry *pNodeEntry = taosArrayGet(execInfo.pNodeList, i);




×







1624



    if (pNodeEntry == NULL) {





×







1625



      continue;




×







1626



    }










1627












1628



    if (pNodeEntry->stageUpdated) {





×







1629



      mDebug("stream task not ready due to node update detected, checkpoint not issued");





×







1630



      return true;




×







1631



    }










1632



  }










1633












1634



  int32_t code = mndTakeVgroupSnapshot(pMnode, &allReady, ppNodeSnapshot);




×







1635



  if (code) {





×







1636



    mError("failed to get the vgroup snapshot, ignore it and continue");





×







1637



  }










1638












1639



  if (!allReady) {





×







1640



    mWarn("not all vnodes ready, quit from vnodes status check");





×







1641



    return true;




×







1642



  }










1643












1644



  code = mndFindChangedNodeInfo(pMnode, execInfo.pNodeList, *ppNodeSnapshot, &changeInfo);




×







1645



  if (code) {





×







1646



    nodeUpdated = false;




×







1647



  } else {










1648



    nodeUpdated = (taosArrayGetSize(changeInfo.pUpdateNodeList) > 0);




×







1649



    if (nodeUpdated) {





×







1650



      mDebug("stream tasks not ready due to node update");





×







1651



    }










1652



  }










1653












1654



  mndDestroyVgroupChangeInfo(&changeInfo);




×







1655



  return nodeUpdated;




×







1656



}










1657












1658



// check if the node update happens or not










1659



bool mndStreamNodeIsUpdated(SMnode *pMnode) {




×







1660



  SArray *pNodeSnapshot = NULL;




×







1661












1662



  streamMutexLock(&execInfo.lock);




×







1663



  bool updated = doCheckForUpdated(pMnode, &pNodeSnapshot);




×







1664



  streamMutexUnlock(&execInfo.lock);




×







1665












1666



  taosArrayDestroy(pNodeSnapshot);




×







1667



  return updated;




×







1668



}










1669












1670



int32_t mndCheckForSnode(SMnode *pMnode, SDbObj *pSrcDb) {




×







1671



  SSdb      *pSdb = pMnode->pSdb;




×







1672



  void      *pIter = NULL;




×







1673



  SSnodeObj *pObj = NULL;




×







1674












1675



  if (pSrcDb->cfg.replications == 1) {





×







1676



    return TSDB_CODE_SUCCESS;




×







1677



  } else {










1678



    while (1) {










1679



      pIter = sdbFetch(pSdb, SDB_SNODE, pIter, (void **)&pObj);




×







1680



      if (pIter == NULL) {





×







1681



        break;




×







1682



      }










1683












1684



      sdbRelease(pSdb, pObj);




×







1685



      sdbCancelFetch(pSdb, pIter);




×







1686



      return TSDB_CODE_SUCCESS;




×







1687



    }










1688












1689



    mError("snode not existed when trying to create stream in db with multiple replica");





×







1690



    return TSDB_CODE_SNODE_NOT_DEPLOYED;




×







1691



  }










1692



}























    

  • 

    •