TDengine/source/dnode/mnode/impl/src/mndStreamUtil.c

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

#include "mndDb.h"
#include "mndStb.h"
#include "mndStream.h"
#include "mndTrans.h"
#include "mndVgroup.h"
#include "taoserror.h"
#include "tmisce.h"

bool mstWaitLock(SRWLatch* pLock, bool readLock) {
  if (readLock) {
    while (taosRTryLockLatch(pLock)) {
      taosMsleep(1);
    }

    return true;
  }

  while (taosWTryLockLatch(pLock)) {
    taosMsleep(1);
  }

  return true;
}

void mndStreamDestroySStreamMgmtRsp(SStreamMgmtRsp* p) {
  taosArrayDestroy(p->cont.vgIds);
  taosArrayDestroy(p->cont.readerList);
}

void mstDestroySStmVgStreamStatus(void* p) { 
  SStmVgStreamStatus* pStatus = (SStmVgStreamStatus*)p;
  taosArrayDestroy(pStatus->trigReaders); 
  taosArrayDestroy(pStatus->calcReaders); 
}

void mstDestroySStmSnodeStreamStatus(void* p) { 
  SStmSnodeStreamStatus* pStatus = (SStmSnodeStreamStatus*)p;
  taosMemoryFreeClear(pStatus->trigger); 
  for (int32_t i = 0; i < MND_STREAM_RUNNER_DEPLOY_NUM; ++i) {
    taosArrayDestroy(pStatus->runners[i]);
    pStatus->runners[i] = NULL;
  }
}


void mstDestroyVgroupStatus(SStmVgroupStatus* pVgStatus) {
  taosHashCleanup(pVgStatus->streamTasks);
  pVgStatus->streamTasks = NULL;
}

int32_t mstIsStreamDropped(SMnode *pMnode, int64_t streamId, bool* dropped) {
  SSdb   *pSdb = pMnode->pSdb;
  void   *pIter = NULL;
  
  while (1) {
    SStreamObj *pStream = NULL;
    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
    if (pIter == NULL) break;

    if (pStream->pCreate->streamId == streamId) {
      *dropped = pStream->userDropped ? true : false;
      sdbRelease(pSdb, pStream);
      return TSDB_CODE_SUCCESS;
    }
    
    sdbRelease(pSdb, pStream);
  }

  *dropped = true;

  return TSDB_CODE_SUCCESS;
}

typedef struct SStmCheckDbInUseCtx {
  bool* dbStream;
  bool* vtableStream;
  bool  ignoreCurrDb;
} SStmCheckDbInUseCtx;

static bool mstChkSetDbInUse(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
  SStreamObj *pStream = pObj;
  if (atomic_load_8(&pStream->userDropped)) {
    return true;
  }

  SStmCheckDbInUseCtx* pCtx = (SStmCheckDbInUseCtx*)p2;
  if (pCtx->ignoreCurrDb && 0 == strcmp(pStream->pCreate->streamDB, p1)) {
    return true;
  }
  
  if (pStream->pCreate->triggerDB && 0 == strcmp(pStream->pCreate->triggerDB, p1)) {
    *pCtx->dbStream = true;
    return false;
  }

  int32_t calcDBNum = taosArrayGetSize(pStream->pCreate->calcDB);
  for (int32_t i = 0; i < calcDBNum; ++i) {
    char* calcDB = taosArrayGetP(pStream->pCreate->calcDB, i);
    if (0 == strcmp(calcDB, p1)) {
      *pCtx->dbStream = true;
      return false;
    }
  }

  if (pStream->pCreate->vtableCalc || STREAM_IS_VIRTUAL_TABLE(pStream->pCreate->triggerTblType, pStream->pCreate->flags)) {
    *pCtx->vtableStream = true;
    return true;
  }
  
  return true;
}

void mstCheckDbInUse(SMnode *pMnode, char *dbFName, bool *dbStream, bool *vtableStream, bool ignoreCurrDb) {
  int32_t streamNum = sdbGetSize(pMnode->pSdb, SDB_STREAM);
  if (streamNum <= 0) {
    return;
  }

  SStmCheckDbInUseCtx ctx = {dbStream, vtableStream, ignoreCurrDb};
  sdbTraverse(pMnode->pSdb, SDB_STREAM, mstChkSetDbInUse, dbFName, &ctx, NULL);
}

static void mstShowStreamStatus(char *dst, int8_t status, int32_t bufLen) {
  if (status == STREAM_STATUS_INIT) {
    tstrncpy(dst, "init", bufLen);
  } else if (status == STREAM_STATUS_RUNNING) {
    tstrncpy(dst, "running", bufLen);
  } else if (status == STREAM_STATUS_STOPPED) {
    tstrncpy(dst, "stopped", bufLen);
  } else if (status == STREAM_STATUS_FAILED) {
    tstrncpy(dst, "failed", bufLen);
  }
}

int32_t mstGenerateResBlock(SStreamObj *pStream, SSDataBlock *pBlock, int32_t numOfRows) {
  int32_t code = 0;
  int32_t cols = 0;
  int32_t lino = 0;

/* STREAMTODO
  char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetDbStr(pStream->name), sizeof(streamName));
  SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  // create time
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char *)&pStream->createTime, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // stream id
  char buf[128] = {0};
  int64ToHexStr(pStream->uid, buf, tListLen(buf));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, buf, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // related fill-history stream id
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  if (pStream->hTaskUid != 0) {
    int64ToHexStr(pStream->hTaskUid, buf, tListLen(buf));
    code = colDataSetVal(pColInfo, numOfRows, buf, false);
  } else {
    code = colDataSetVal(pColInfo, numOfRows, buf, true);
  }
  TSDB_CHECK_CODE(code, lino, _end);

  // related fill-history stream id
  char sql[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(sql, pStream->sql, sizeof(sql));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char *)sql, false);
  TSDB_CHECK_CODE(code, lino, _end);

  char status[20 + VARSTR_HEADER_SIZE] = {0};
  char status2[MND_STREAM_TRIGGER_NAME_SIZE] = {0};
  bool isPaused = false;
  //code = isAllTaskPaused(pStream, &isPaused);
  TSDB_CHECK_CODE(code, lino, _end);

  int8_t streamStatus = atomic_load_8(&pStream->status);
  if (isPaused && pStream->pTaskList != NULL) {
    streamStatus = STREAM_STATUS__PAUSE;
  }
  mndShowStreamStatus(status2, streamStatus);
  STR_WITH_MAXSIZE_TO_VARSTR(status, status2, sizeof(status));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  char sourceDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(sourceDB, mndGetDbStr(pStream->sourceDb), sizeof(sourceDB));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  char targetDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(targetDB, mndGetDbStr(pStream->targetDb), sizeof(targetDB));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  if (pStream->targetSTbName[0] == 0) {
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

    code = colDataSetVal(pColInfo, numOfRows, NULL, true);
  } else {
    char targetSTB[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(targetSTB, mndGetStbStr(pStream->targetSTbName), sizeof(targetSTB));
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

    code = colDataSetVal(pColInfo, numOfRows, (const char *)&targetSTB, false);
  }
  TSDB_CHECK_CODE(code, lino, _end);

  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  char trigger[20 + VARSTR_HEADER_SIZE] = {0};
  char trigger2[MND_STREAM_TRIGGER_NAME_SIZE] = {0};
  mndShowStreamTrigger(trigger2, pStream);
  STR_WITH_MAXSIZE_TO_VARSTR(trigger, trigger2, sizeof(trigger));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  // sink_quota
  char sinkQuota[20 + VARSTR_HEADER_SIZE] = {0};
  sinkQuota[0] = '0';
  char dstStr[20] = {0};
  STR_TO_VARSTR(dstStr, sinkQuota)
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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


  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

  // checkpoint backup type
  char backup[20 + VARSTR_HEADER_SIZE] = {0};
  STR_TO_VARSTR(backup, "none")
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  // history scan idle
  char scanHistoryIdle[20 + VARSTR_HEADER_SIZE] = {0};
  tstrncpy(scanHistoryIdle, "100a", sizeof(scanHistoryIdle));

  memset(dstStr, 0, tListLen(dstStr));
  STR_TO_VARSTR(dstStr, scanHistoryIdle)
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  char msg[TSDB_RESERVE_VALUE_LEN + VARSTR_HEADER_SIZE] = {0};
  if (streamStatus == STREAM_STATUS__FAILED){
    STR_TO_VARSTR(msg, pStream->reserve)
  } else {
    STR_TO_VARSTR(msg, " ")
  }
  code = colDataSetVal(pColInfo, numOfRows, (const char *)msg, false);

_end:
  if (code) {
    mError("error happens when build stream attr result block, lino:%d, code:%s", lino, tstrerror(code));
  }
*/

  return code;
}

int32_t mstCheckSnodeExists(SMnode *pMnode) {
  SSdb      *pSdb = pMnode->pSdb;
  void      *pIter = NULL;
  SSnodeObj *pObj = NULL;

  while (1) {
    pIter = sdbFetch(pSdb, SDB_SNODE, pIter, (void **)&pObj);
    if (pIter == NULL) {
      break;
    }

    sdbRelease(pSdb, pObj);
    sdbCancelFetch(pSdb, pIter);
    return TSDB_CODE_SUCCESS;
  }

  return TSDB_CODE_SNODE_NOT_DEPLOYED;
}

void mstSetTaskStatusFromMsg(SStmGrpCtx* pCtx, SStmTaskStatus* pTask, SStmTaskStatusMsg* pMsg) {
  pTask->id.taskId = pMsg->taskId;
  pTask->id.deployId = pMsg->deployId;
  pTask->id.seriousId = pMsg->seriousId;
  pTask->id.nodeId = pMsg->nodeId;
  pTask->id.taskIdx = pMsg->taskIdx;

  pTask->type = pMsg->type;
  pTask->flags = pMsg->flags;
  pTask->status = pMsg->status;
  pTask->lastUpTs = pCtx->currTs;
}

bool mndStreamActionDequeue(SStmActionQ* pQueue, SStmQNode **param) {
  while (0 == atomic_load_64(&pQueue->qRemainNum)) {
    return false;
  }

  SStmQNode *orig = pQueue->head;

  SStmQNode *node = pQueue->head->next;
  pQueue->head = pQueue->head->next;

  *param = node;

  atomic_sub_fetch_64(&pQueue->qRemainNum, 1);

  return true;
}

void mndStreamActionEnqueue(SStmActionQ* pQueue, SStmQNode* param) {
  taosWLockLatch(&pQueue->lock);
  pQueue->tail->next = param;
  pQueue->tail = param;
  taosWUnLockLatch(&pQueue->lock);

  atomic_add_fetch_64(&pQueue->qRemainNum, 1);
}


void mndStreamPostAction(SStmActionQ*       actionQ, int64_t streamId, char* streamName, int32_t action) {
  SStmQNode *pNode = taosMemoryMalloc(sizeof(SStmQNode));
  if (NULL == pNode) {
    mstsError("%s failed at line %d, error:%s", __FUNCTION__, __LINE__, tstrerror(terrno));
    return;
  }

  pNode->type = action;
  pNode->streamAct = true;
  pNode->action.stream.streamId = streamId;
  TAOS_STRCPY(pNode->action.stream.streamName, streamName);
  
  pNode->next = NULL;

  mndStreamActionEnqueue(actionQ, pNode);

  mstsDebug("stream action %s posted enqueue", gMndStreamAction[action]);
}

void mndStreamPostTaskAction(SStmActionQ*        actionQ, SStmTaskAction* pAction, int32_t action) {
  SStmQNode *pNode = taosMemoryMalloc(sizeof(SStmQNode));
  if (NULL == pNode) {
    int64_t streamId = pAction->streamId;
    mstsError("%s failed at line %d, error:%s", __FUNCTION__, __LINE__, tstrerror(terrno));
    return;
  }

  pNode->type = action;
  pNode->streamAct = false;
  pNode->action.task = *pAction;
  
  pNode->next = NULL;

  mndStreamActionEnqueue(actionQ, pNode);
}

void mndStreamDestroyDbVgroupsHash(SSHashObj *pDbVgs) {
  int32_t iter = 0;
  SDBVgHashInfo* pVg = NULL;
  void* p = NULL;
  while (NULL != (p = tSimpleHashIterate(pDbVgs, p, &iter))) {
    pVg = (SDBVgHashInfo*)p;
    taosArrayDestroy(pVg->vgArray);
  }
  
  tSimpleHashCleanup(pDbVgs);
}


int32_t mndStreamBuildDBVgroupsMap(SMnode* pMnode, SSHashObj** ppRes) {
  void*   pIter = NULL;
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  SArray* pTarget = NULL;
  SArray* pNew = NULL;
  SDbObj* pDb = NULL;
  SDBVgHashInfo dbInfo = {0}, *pDbInfo = NULL;
  SVgObj* pVgroup = NULL;

  SSHashObj* pDbVgroup = tSimpleHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY));
  TSDB_CHECK_NULL(pDbVgroup, code, lino, _exit, terrno);

  while (1) {
    pIter = sdbFetch(pMnode->pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
    if (pIter == NULL) {
      break;
    }

    pDbInfo = (SDBVgHashInfo*)tSimpleHashGet(pDbVgroup, pVgroup->dbName, strlen(pVgroup->dbName) + 1);
    if (NULL == pDbInfo) {
      pNew = taosArrayInit(20, sizeof(SVGroupHashInfo));
      TSDB_CHECK_NULL(pNew, code, lino, _exit, terrno);

      pDb = mndAcquireDb(pMnode, pVgroup->dbName);
      TSDB_CHECK_NULL(pNew, code, lino, _exit, terrno);

      dbInfo.vgSorted = false;
      dbInfo.hashMethod = pDb->cfg.hashMethod;
      dbInfo.hashPrefix = pDb->cfg.hashPrefix;
      dbInfo.hashSuffix = pDb->cfg.hashSuffix;
      dbInfo.vgArray = pNew;
      
      mndReleaseDb(pMnode, pDb);

      pTarget = pNew;
    } else {
      pTarget = pDbInfo->vgArray;
    }

    SVGroupHashInfo vgInfo = {.vgId = pVgroup->vgId, .hashBegin = pVgroup->hashBegin, .hashEnd = pVgroup->hashEnd};
    TSDB_CHECK_NULL(taosArrayPush(pTarget, &vgInfo), code, lino, _exit, terrno);

    if (NULL == pDbInfo) {
      TAOS_CHECK_EXIT(tSimpleHashPut(pDbVgroup, pVgroup->dbName, strlen(pVgroup->dbName) + 1, &dbInfo, sizeof(dbInfo)));
      pNew = NULL;
    }

    sdbRelease(pMnode->pSdb, pVgroup);
    pVgroup = NULL;
  }

  *ppRes = pDbVgroup;
  
_exit:

  taosArrayDestroy(pNew);
  sdbRelease(pMnode->pSdb, pVgroup);

  if (code) {
    mstError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
  }

  return code;
}

int mstDbVgInfoComp(const void* lp, const void* rp) {
  SVGroupHashInfo* pLeft = (SVGroupHashInfo*)lp;
  SVGroupHashInfo* pRight = (SVGroupHashInfo*)rp;
  if (pLeft->hashBegin < pRight->hashBegin) {
    return -1;
  } else if (pLeft->hashBegin > pRight->hashBegin) {
    return 1;
  }

  return 0;
}

int32_t mstTableHashValueComp(void const* lp, void const* rp) {
  uint32_t*    key = (uint32_t*)lp;
  SVgroupInfo* pVg = (SVgroupInfo*)rp;

  if (*key < pVg->hashBegin) {
    return -1;
  } else if (*key > pVg->hashEnd) {
    return 1;
  }

  return 0;
}


int32_t mndStreamGetTableVgId(SSHashObj* pDbVgroups, char* dbFName, char *tbName, int32_t* vgId) {
  int32_t code = 0;
  int32_t lino = 0;
  SVgroupInfo* vgInfo = NULL;
  char         tbFullName[TSDB_TABLE_FNAME_LEN];

  SDBVgHashInfo* dbInfo = (SDBVgHashInfo*)tSimpleHashGet(pDbVgroups, dbFName, strlen(dbFName) + 1);
  if (NULL == dbInfo) {
    mstError("db %s does not exist", dbFName);
    TAOS_CHECK_EXIT(TSDB_CODE_MND_DB_NOT_EXIST);
  }
  
  (void)snprintf(tbFullName, sizeof(tbFullName), "%s.%s", dbFName, tbName);
  uint32_t hashValue = taosGetTbHashVal(tbFullName, (uint32_t)strlen(tbFullName), dbInfo->hashMethod,
                                        dbInfo->hashPrefix, dbInfo->hashSuffix);

  if (!dbInfo->vgSorted) {
    taosArraySort(dbInfo->vgArray, mstDbVgInfoComp);
    dbInfo->vgSorted = true;
  }

  vgInfo = taosArraySearch(dbInfo->vgArray, &hashValue, mstTableHashValueComp, TD_EQ);
  if (NULL == vgInfo) {
    mstError("no hash range found for hash value [%u], dbFName:%s, numOfVgId:%d", hashValue, dbFName,
             (int32_t)taosArrayGetSize(dbInfo->vgArray));
    TAOS_CHECK_EXIT(TSDB_CODE_MND_STREAM_INTERNAL_ERROR);
  }

  *vgId = vgInfo->vgId;

_exit:

  if (code) {
    mstError("%s failed at line %d, error:%s", __FUNCTION__, lino, tstrerror(code));
  }

  return code;
}


void mndStreamLogSStreamObj(char* tips, SStreamObj* p) {
  if (!(stDebugFlag & DEBUG_DEBUG)) {
    return;
  }
  
  if (NULL == p) {
    mstDebug("%s: stream is NULL", tips);
    return;
  }

  mstDebug("%s: stream obj", tips);
  mstDebug("name:%s mainSnodeId:%d userDropped:%d userStopped:%d createTime:%" PRId64 " updateTime:%" PRId64,
      p->name, p->mainSnodeId, p->userDropped, p->userStopped, p->createTime, p->updateTime);

  SCMCreateStreamReq* q = p->pCreate;
  if (NULL == q) {
    mstDebug("stream pCreate is NULL");
    return;
  }

  int64_t streamId = q->streamId;
  int32_t calcDBNum = taosArrayGetSize(q->calcDB);
  int32_t calcScanNum = taosArrayGetSize(q->calcScanPlanList);
  int32_t notifyUrlNum = taosArrayGetSize(q->pNotifyAddrUrls);
  int32_t outColNum = taosArrayGetSize(q->outCols);
  int32_t outTagNum = taosArrayGetSize(q->outTags);
  int32_t forceOutColNum = taosArrayGetSize(q->forceOutCols);

  mstsDebugL("create_info: name:%s sql:%s streamDB:%s triggerDB:%s outDB:%s calcDBNum:%d triggerTblName:%s outTblName:%s "
      "igExists:%d triggerType:%d igDisorder:%d deleteReCalc:%d deleteOutTbl:%d fillHistory:%d fillHistroyFirst:%d "
      "calcNotifyOnly:%d lowLatencyCalc:%d notifyUrlNum:%d notifyEventTypes:%d notifyErrorHandle:%d notifyHistory:%d "
      "outColsNum:%d outTagsNum:%d maxDelay:%" PRId64 " fillHistoryStartTs:%" PRId64 " watermark:%" PRId64 " expiredTime:%" PRId64 " "
      "triggerTblType:%d triggerTblUid:%" PRIx64 " triggerTblSuid:%" PRIx64 " vtableCalc:%d outTblType:%d outStbExists:%d outStbUid:%" PRIu64 " outStbSversion:%d "
      "eventTypes:0x%" PRIx64 " flags:0x%" PRIx64 " tsmaId:0x%" PRIx64 " placeHolderBitmap:0x%" PRIx64 " tsSlotId:%d "
      "triggerTblVgId:%d outTblVgId:%d calcScanPlanNum:%d forceOutCols:%d",
      q->name, q->sql, q->streamDB, q->triggerDB, q->outDB, calcDBNum, q->triggerTblName, q->outTblName,
      q->igExists, q->triggerType, q->igDisorder, q->deleteReCalc, q->deleteOutTbl, q->fillHistory, q->fillHistoryFirst,
      q->calcNotifyOnly, q->lowLatencyCalc, notifyUrlNum, q->notifyEventTypes, q->notifyErrorHandle, q->notifyHistory,
      outColNum, outTagNum, q->maxDelay, q->fillHistoryStartTime, q->watermark, q->expiredTime,
      q->triggerTblType, q->triggerTblUid, q->triggerTblSuid, q->vtableCalc, q->outTblType, q->outStbExists, q->outStbUid, q->outStbSversion,
      q->eventTypes, q->flags, q->tsmaId, q->placeHolderBitmap, q->tsSlotId,
      q->triggerTblVgId, q->outTblVgId, calcScanNum, forceOutColNum);

  mstsDebugL("create_info: triggerCols:[%s]", (char*)q->triggerCols);

  mstsDebugL("create_info: partitionCols:[%s]", (char*)q->partitionCols);

  mstsDebugL("create_info: triggerScanPlan:[%s]", (char*)q->triggerScanPlan);

  mstsDebugL("create_info: calcPlan:[%s]", (char*)q->calcPlan);

  mstsDebugL("create_info: subTblNameExpr:[%s]", (char*)q->subTblNameExpr);

  mstsDebugL("create_info: tagValueExpr:[%s]", (char*)q->tagValueExpr);


  for (int32_t i = 0; i < calcDBNum; ++i) {
    char* dbName = taosArrayGetP(q->calcDB, i);
    mstsDebug("create_info: calcDB[%d] - %s", i, dbName);
  }

  for (int32_t i = 0; i < calcScanNum; ++i) {
    SStreamCalcScan* pScan = taosArrayGet(q->calcScanPlanList, i);
    int32_t vgNum = taosArrayGetSize(pScan->vgList);
    mstsDebugL("create_info: calcScanPlan[%d] - readFromCache:%d vgNum:%d scanPlan:[%s]", i, pScan->readFromCache, vgNum, (char*)pScan->scanPlan);
    for (int32_t v = 0; v < vgNum; ++v) {
      mstsDebug("create_info: calcScanPlan[%d] vg[%d] - vgId:%d", i, v, *(int32_t*)taosArrayGet(pScan->vgList, v));
    }
  }

  for (int32_t i = 0; i < notifyUrlNum; ++i) {
    char* url = taosArrayGetP(q->pNotifyAddrUrls, i);
    mstsDebug("create_info: notifyUrl[%d] - %s", i, url);
  }

  for (int32_t i = 0; i < outColNum; ++i) {
    SFieldWithOptions* o = taosArrayGet(q->outCols, i);
    mstsDebug("create_info: outCol[%d] - name:%s type:%d flags:%d bytes:%d compress:%u typeMod:%d", 
        i, o->name, o->type, o->flags, o->bytes, o->compress, o->typeMod);
  }
      
}

void mndStreamLogSStmTaskStatus(char* name, int64_t streamId, SStmTaskStatus* pTask, int32_t idx) {
  mstsDebug("%s[%d]: task %" PRIx64 " deployId:%d SID:%" PRId64 " nodeId:%d tidx:%d type:%s flags:%" PRIx64 " status:%s lastUpTs:%" PRId64, 
      name, idx, pTask->id.taskId, pTask->id.deployId, pTask->id.seriousId, pTask->id.nodeId, pTask->id.taskIdx,
      gStreamTaskTypeStr[pTask->type], pTask->flags, gStreamStatusStr[pTask->status], pTask->lastUpTs);
}

void mndStreamLogSStmStatus(char* tips, int64_t streamId, SStmStatus* p) {
  if (!(stDebugFlag & DEBUG_DEBUG)) {
    return;
  }
  
  if (NULL == p) {
    mstsDebug("%s: stream status is NULL", tips);
    return;
  }

  int32_t trigReaderNum = taosArrayGetSize(p->trigReaders);
  int32_t calcReaderNum = taosArrayGetSize(p->calcReaders);
  int32_t triggerNum = p->triggerTask ? 1 : 0;
  int32_t runnerNum = 0;

  for (int32_t i = 0; i < p->runnerDeploys; ++i) {
    runnerNum += taosArrayGetSize(p->runners[i]);
  }

  mstsDebug("%s: stream status", tips);
  mstsDebug("name:%s runnerNum:%d runnerDeploys:%d runnerReplica:%d lastActionTs:%" PRId64
           " trigReaders:%d calcReaders:%d trigger:%d runners:%d",
      p->streamName, p->runnerNum, p->runnerDeploys, p->runnerReplica, p->lastActionTs,
      trigReaderNum, calcReaderNum, triggerNum, runnerNum);

  SStmTaskStatus* pTask = NULL;
  for (int32_t i = 0; i < trigReaderNum; ++i) {
    pTask = taosArrayGet(p->trigReaders, i);
    mndStreamLogSStmTaskStatus("trigReader task", streamId, pTask, i);
  }

  for (int32_t i = 0; i < calcReaderNum; ++i) {
    pTask = taosArrayGet(p->calcReaders, i);
    mndStreamLogSStmTaskStatus("calcReader task", streamId, pTask, i);
  }

  if (triggerNum > 0) {
    mndStreamLogSStmTaskStatus("trigger task", streamId, p->triggerTask, 0);
  }

  for (int32_t i = 0; i < p->runnerDeploys; ++i) {
    int32_t num = taosArrayGetSize(p->runners[i]);
    if (num <= 0) {
      continue;
    }
    
    mstsDebug("the %dth deploy runners status", i);
    for (int32_t m = 0; m < num; ++m) {
      pTask = taosArrayGet(p->runners[i], m);
      mndStreamLogSStmTaskStatus("runner task", streamId, pTask, m);
    }
  }
      
}

int32_t setStreamAttrInResBlock(SStreamObj* pStream, SSDataBlock* pBlock, int32_t numOfRows) {
  int32_t code = 0;
  int32_t cols = 0;
  int32_t lino = 0;

  char streamName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(streamName, mndGetStableStr(pStream->name), sizeof(streamName));
  SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);

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

  // db_name
  char streamDB[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(streamDB, mndGetDbStr(pStream->pCreate->streamDB), sizeof(streamDB));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char*)&streamDB, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // create time
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char*)&pStream->createTime, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // stream id
  char streamId2[19] = {0};
  char streamId[19 + VARSTR_HEADER_SIZE] = {0};
  snprintf(streamId2, sizeof(streamId2), "0X%" PRIx64, pStream->pCreate->streamId);
  STR_WITH_MAXSIZE_TO_VARSTR(streamId, streamId2, sizeof(streamId));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char*)streamId, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // sql
  char sql[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_WITH_MAXSIZE_TO_VARSTR(sql, pStream->pCreate->sql, sizeof(sql));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char*)sql, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // todo status
  char   status[20 + VARSTR_HEADER_SIZE] = {0};
  char   status2[MND_STREAM_TRIGGER_NAME_SIZE] = {0};
  //mndShowStreamStatus(status2, streamStatus);
  STR_WITH_MAXSIZE_TO_VARSTR(status, status2, sizeof(status));
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char*)&status, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // snodeLeader
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char*)&pStream->mainSnodeId, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // snodeReplica
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  code = colDataSetVal(pColInfo, numOfRows, (const char*)&pStream->mainSnodeId, false);
  TSDB_CHECK_CODE(code, lino, _end);

  // todo msg
  pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
  TSDB_CHECK_NULL(pColInfo, code, lino, _end, terrno);
  char msg[TSDB_RESERVE_VALUE_LEN + VARSTR_HEADER_SIZE] = {0};
  if (true) {
    STR_TO_VARSTR(msg, " ")
  } else {
    STR_TO_VARSTR(msg, " ")
  }
  code = colDataSetVal(pColInfo, numOfRows, (const char*)msg, false);

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