taosdata / TDengine / #4897

coverage: 65.717% (-0.2%) from 65.929%
#4897

push

travis-ci

web-flow 
fix: [6622889291] Fix invalid rowSize. (#34043)

186011 of 283047 relevant lines covered (65.72%)

113853896.64 hits per line

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

91.04
/source/dnode/vnode/src/tq/tqScan.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 "tq.h"










17



static int32_t tqAddRawDataToRsp(const void* rawData, SMqDataRsp* pRsp, int8_t precision) {




50✔





18



  int32_t    code = TDB_CODE_SUCCESS;




50✔





19



  int32_t    lino = 0;




50✔





20



  void*      buf = NULL;




50✔





21












22



  int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + *(uint32_t *)rawData + INT_BYTES;




50✔





23



  buf = taosMemoryCalloc(1, dataStrLen);




50✔





24



  TSDB_CHECK_NULL(buf, code, lino, END, terrno);




50✔





25












26



  SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;




50✔





27



  pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION;




50✔





28



  pRetrieve->precision = precision;




50✔





29



  pRetrieve->compressed = 0;




50✔





30












31



  memcpy(pRetrieve->data, rawData, *(uint32_t *)rawData + INT_BYTES);




50✔





32



  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &dataStrLen), code, lino, END, terrno);




100✔





33



  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);




100✔





34



  pRsp->blockDataElementFree = true;




50✔





35












36



  tqTrace("tqAddRawDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);




50✔





37



  END:




50✔





38



  if (code != TSDB_CODE_SUCCESS) {




50✔





39



    taosMemoryFree(buf);




×







40



    tqError("%s failed at %d, failed to add block data to response:%s", __FUNCTION__, lino, tstrerror(code));




×







41



  }










42



  return code;




50✔





43



}










44












45



static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, const SSchemaWrapper* pSW, int8_t precision) {




56,152,760✔





46



  int32_t code = 0;




56,152,760✔





47



  int32_t lino = 0;




56,152,760✔





48



  SSchemaWrapper* pSchema = NULL;




56,152,760✔





49



  










50



  size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);




56,169,193✔





51



  int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;




56,183,834✔





52



  void*   buf = taosMemoryCalloc(1, dataStrLen);




56,183,834✔





53



  TSDB_CHECK_NULL(buf, code, lino, END, terrno);




56,166,919✔





54












55



  SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;




56,166,919✔





56



  pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_VERSION;




56,166,919✔





57



  pRetrieve->precision = precision;




56,143,588✔





58



  pRetrieve->compressed = 0;




56,161,032✔





59



  pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);




56,150,377✔





60












61



  int32_t actualLen = blockEncode(pBlock, pRetrieve->data, dataEncodeBufSize, pSW->nCols);




56,169,488✔





62



  TSDB_CHECK_CONDITION(actualLen >= 0, code, lino, END, terrno);




56,167,028✔





63












64



  actualLen += sizeof(SRetrieveTableRspForTmq);




56,167,028✔





65



  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &actualLen), code, lino, END, terrno);




112,347,151✔





66



  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);




112,359,261✔





67



  pSchema = tCloneSSchemaWrapper(pSW);




56,176,717✔





68



  TSDB_CHECK_NULL(pSchema, code, lino, END, terrno);




56,176,717✔





69



  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockSchema, &pSchema), code, lino, END, terrno);




112,357,510✔





70



  pSchema = NULL;




56,180,793✔





71



  pRsp->blockDataElementFree = true;




56,180,793✔





72



  tqTrace("tqAddBlockDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);




56,178,161✔





73












74



END:




56,178,161✔





75



  tDeleteSchemaWrapper(pSchema);




56,179,257✔





76



  if (code != TSDB_CODE_SUCCESS){




56,149,711✔





77



    taosMemoryFree(buf);




×







78



    tqError("%s failed at line %d with msg:%s", __func__, lino, tstrerror(code));




×







79



  }










80



  return code;




56,149,711✔





81



}










82












83



static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {




14,644,526✔





84



  int32_t    code = TDB_CODE_SUCCESS;




14,644,526✔





85



  int32_t    lino = 0;




14,644,526✔





86



  SMetaReader mr = {0};




14,644,526✔





87












88



  TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);




14,647,756✔





89



  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);




14,647,756✔





90












91



  metaReaderDoInit(&mr, pTq->pVnode->pMeta, META_READER_LOCK);




14,647,756✔





92












93



  code = metaReaderGetTableEntryByUidCache(&mr, uid);




14,637,316✔





94



  TSDB_CHECK_CODE(code, lino, END);




14,625,515✔





95












96



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




29,275,492✔





97



    char* tbName = taosStrdup(mr.me.name);




14,620,127✔





98



    TSDB_CHECK_NULL(tbName, code, lino, END, terrno);




14,621,910✔





99



    if(taosArrayPush(pRsp->blockTbName, &tbName) == NULL){




29,271,021✔





100



      tqError("failed to push tbName to blockTbName:%s, uid:%"PRId64, tbName, uid);




×







101



      continue;




×







102



    }










103



    tqTrace("add tbName to response success tbname:%s, uid:%"PRId64, tbName, uid);




14,649,111✔





104



  }










105












106



END:




14,654,908✔





107



  if (code != TSDB_CODE_SUCCESS) {




14,641,048✔





108



    tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);




×







109



  }










110



  metaReaderClear(&mr);




14,641,048✔





111



  return code;




14,628,891✔





112



}










113












114



int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {




45,393,621✔





115



  if (task == NULL || pHandle == NULL || res == NULL) {




45,393,621✔





116



    return TSDB_CODE_INVALID_PARA;




×







117



  }










118



  uint64_t ts = 0;




45,397,361✔





119



  qStreamSetOpen(task);




45,396,684✔





120












121



  tqDebug("consumer:0x%" PRIx64 " vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);




45,396,136✔





122



  int32_t code = qExecTask(task, res, &ts);




45,398,963✔





123



  if (code != TSDB_CODE_SUCCESS) {




45,399,401✔





124



    tqError("consumer:0x%" PRIx64 " vgId:%d, task exec error since %s", pHandle->consumerId, vgId, tstrerror(code));




×







125



    return code;




×







126



  }










127












128



  tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq one task end executed, pDataBlock:%p", pHandle->consumerId, vgId, *res);




45,399,401✔





129



  return 0;




45,399,528✔





130



}










131












132



static int32_t tqProcessReplayRsp(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, const SMqPollReq* pRequest, SSDataBlock* pDataBlock, qTaskInfo_t task){




957✔





133



  int32_t code = 0;




957✔





134



  int32_t lino = 0;




957✔





135












136



  if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type) && pHandle->block != NULL) {




957✔





137



    blockDataDestroy(pHandle->block);




29✔





138



    pHandle->block = NULL;




29✔





139



  }










140



  if (pHandle->block == NULL) {




957✔





141



    if (pDataBlock == NULL) {




319✔





142



      goto END;




174✔





143



    }










144












145



    STqOffsetVal offset = {0};




145✔





146



    code = qStreamExtractOffset(task, &offset);




145✔





147



    TSDB_CHECK_CODE(code, lino, END);




145✔





148












149



    pHandle->block = NULL;




145✔





150












151



    code = createOneDataBlock(pDataBlock, true, &pHandle->block);




145✔





152



    TSDB_CHECK_CODE(code, lino, END);




145✔





153












154



    pHandle->blockTime = offset.ts;




145✔





155



    tOffsetDestroy(&offset);




145✔





156



    int32_t vgId = TD_VID(pTq->pVnode);




145✔





157



    code = getDataBlock(task, pHandle, vgId, &pDataBlock);




145✔





158



    TSDB_CHECK_CODE(code, lino, END);




145✔





159



  }










160












161



  const STqExecHandle* pExec = &pHandle->execHandle;




783✔





162



  code = tqAddBlockDataToRsp(pHandle->block, pRsp, &pExec->execCol.pSW, pTq->pVnode->config.tsdbCfg.precision);




783✔





163



  TSDB_CHECK_CODE(code, lino, END);




783✔





164












165



  pRsp->blockNum++;




783✔





166



  if (pDataBlock == NULL) {




783✔





167



    blockDataDestroy(pHandle->block);




116✔





168



    pHandle->block = NULL;




116✔





169



  } else {










170



    code = copyDataBlock(pHandle->block, pDataBlock);




667✔





171



    TSDB_CHECK_CODE(code, lino, END);




667✔





172












173



    STqOffsetVal offset = {0};




667✔





174



    code = qStreamExtractOffset(task, &offset);




667✔





175



    TSDB_CHECK_CODE(code, lino, END);




667✔





176












177



    pRsp->sleepTime = offset.ts - pHandle->blockTime;




667✔





178



    pHandle->blockTime = offset.ts;




667✔





179



    tOffsetDestroy(&offset);




667✔





180



  }










181












182



END:




957✔





183



  if (code != TSDB_CODE_SUCCESS) {




957✔





184



    tqError("%s failed at %d, failed to process replay response:%s", __FUNCTION__, lino, tstrerror(code));




×







185



  }










186



  return code;




957✔





187



}










188












189



int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {




4,433,755✔





190



  int32_t code = 0;




4,433,755✔





191



  int32_t lino = 0;




4,433,755✔





192



  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);




4,433,755✔





193



  TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);




4,433,755✔





194



  TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);




4,433,755✔





195



  TSDB_CHECK_NULL(pOffset, code, lino, END, TSDB_CODE_INVALID_PARA);




4,433,755✔





196



  TSDB_CHECK_NULL(pRequest, code, lino, END, TSDB_CODE_INVALID_PARA);




4,433,755✔





197












198



  int32_t vgId = TD_VID(pTq->pVnode);




4,433,755✔





199



  int32_t totalRows = 0;




4,434,022✔





200












201



  const STqExecHandle* pExec = &pHandle->execHandle;




4,434,022✔





202



  qTaskInfo_t          task = pExec->task;




4,434,444✔





203












204



  code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);




4,434,007✔





205



  TSDB_CHECK_CODE(code, lino, END);




4,433,067✔





206












207



  qStreamSetSourceExcluded(task, pRequest->sourceExcluded);




4,391,231✔





208



  int64_t st = taosGetTimestampMs();




4,390,407✔





209



  while (1) {




41,005,260✔





210



    SSDataBlock* pDataBlock = NULL;




45,395,667✔





211



    code = getDataBlock(task, pHandle, vgId, &pDataBlock);




45,396,719✔





212



    TSDB_CHECK_CODE(code, lino, END);




45,398,808✔





213












214



    if (pRequest->enableReplay) {




45,398,808✔





215



      code = tqProcessReplayRsp(pTq, pHandle, pRsp, pRequest, pDataBlock, task);




957✔





216



      TSDB_CHECK_CODE(code, lino, END);




957✔





217



      break;




957✔





218



    }










219



    if (pDataBlock == NULL) {




45,399,054✔





220



      break;




3,988,934✔





221



    }










222



    code = tqAddBlockDataToRsp(pDataBlock, pRsp, &pExec->execCol.pSW, pTq->pVnode->config.tsdbCfg.precision);




41,410,120✔





223



    TSDB_CHECK_CODE(code, lino, END);




41,405,285✔





224












225



    pRsp->blockNum++;




41,405,285✔





226



    totalRows += pDataBlock->info.rows;




41,405,076✔





227



    if (totalRows >= pRequest->minPollRows || (taosGetTimestampMs() - st > pRequest->timeout)) {




82,500,647✔





228



      break;










229



    }










230



  }










231












232



  tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq task executed finished, total blocks:%d, totalRows:%d", pHandle->consumerId, vgId, pRsp->blockNum, totalRows);




4,392,684✔





233



  code = qStreamExtractOffset(task, &pRsp->rspOffset);




4,392,729✔





234












235



END:




4,434,520✔





236



  if (code != 0) {




4,434,520✔





237



    tqError("%s failed at %d, tmq task executed error msg:%s", __FUNCTION__, lino, tstrerror(code));




41,836✔





238



  }










239



  return code;




4,434,520✔





240



}










241












242



int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {




10,319✔





243



  int32_t code = 0;




10,319✔





244



  int32_t lino = 0;




10,319✔





245



  char* tbName = NULL;




10,319✔





246



  const STqExecHandle* pExec = &pHandle->execHandle;




10,319✔





247



  qTaskInfo_t          task = pExec->task;




10,319✔





248



  code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);




10,319✔





249



  TSDB_CHECK_CODE(code, lino, END);




10,319✔





250












251



  int32_t rowCnt = 0;




10,319✔





252



  int64_t st = taosGetTimestampMs();




10,319✔





253



  while (1) {




240,110✔





254



    SSDataBlock* pDataBlock = NULL;




250,429✔





255



    uint64_t     ts = 0;




250,429✔





256



    tqDebug("tmqsnap task start to execute");




250,365✔





257



    code = qExecTask(task, &pDataBlock, &ts);




250,429✔





258



    TSDB_CHECK_CODE(code, lino, END);




250,015✔





259



    tqDebug("tmqsnap task execute end, get %p", pDataBlock);




250,015✔





260












261



    if (pDataBlock != NULL && pDataBlock->info.rows > 0) {




250,015✔





262



      if (pRsp->withTbName) {




121,107✔





263



        tbName = taosStrdup(qExtractTbnameFromTask(task));




121,107✔





264



        TSDB_CHECK_NULL(tbName, code, lino, END, terrno);




121,061✔





265



        TSDB_CHECK_NULL(taosArrayPush(pRsp->blockTbName, &tbName), code, lino, END, terrno);




242,168✔





266



        tqDebug("vgId:%d, add tbname:%s to rsp msg", pTq->pVnode->config.vgId, tbName);




121,107✔





267



        tbName = NULL;




121,107✔





268



      }










269












270



      code = tqAddBlockDataToRsp(pDataBlock, pRsp, qExtractSchemaFromTask(task), pTq->pVnode->config.tsdbCfg.precision);




121,107✔





271



      TSDB_CHECK_CODE(code, lino, END);




121,061✔





272












273



      pRsp->blockNum++;




121,061✔





274



      rowCnt += pDataBlock->info.rows;




120,973✔





275



      if (rowCnt <= pRequest->minPollRows && (taosGetTimestampMs() - st <= pRequest->timeout)) {




235,211✔





276



        continue;




114,128✔





277



      }










278



    }










279












280



    // get meta










281



    SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);




135,777✔





282



    if (taosArrayGetSize(tmp->batchMetaReq) > 0) {




136,191✔





283



      code = qStreamExtractOffset(task, &tmp->rspOffset);




1,149✔





284



      TSDB_CHECK_CODE(code, lino, END);




1,149✔





285



      *pBatchMetaRsp = *tmp;




1,149✔





286



      tqDebug("tmqsnap task get meta");




1,149✔





287



      break;




1,149✔





288



    }










289












290



    if (pDataBlock == NULL) {




135,042✔





291



      code = qStreamExtractOffset(task, pOffset);




128,173✔





292



      TSDB_CHECK_CODE(code, lino, END);




128,173✔





293












294



      if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {




128,173✔





295



        continue;




125,872✔





296



      }










297












298



      tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);




2,301✔





299



      code = qStreamExtractOffset(task, &pRsp->rspOffset);




2,301✔





300



      break;




2,301✔





301



    }










302












303



    if (pRsp->blockNum > 0) {




6,869✔





304



      tqDebug("tmqsnap task exec exited, get data");




6,869✔





305



      code = qStreamExtractOffset(task, &pRsp->rspOffset);




6,869✔





306



      break;




6,869✔





307



    }










308



  }










309



  tqDebug("%s:%d success", __FUNCTION__, lino);




10,319✔





310



END:




10,273✔





311



  if (code != 0){




10,319✔





312



    tqError("%s failed at %d, vgId:%d, task exec error since %s", __FUNCTION__ , lino, pTq->pVnode->config.vgId, tstrerror(code));




×







313



  }










314



  taosMemoryFree(tbName);




10,319✔





315



  return code;




10,319✔





316



}










317












318



static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows,




14,694,492✔





319



                             const SMqPollReq* pRequest, SArray* rawList){










320



  int32_t code = 0;




14,694,492✔





321



  int32_t lino = 0;




14,694,492✔





322



  SArray* pBlocks = NULL;




14,694,492✔





323



  SArray* pSchemas = NULL;




14,694,492✔





324












325



  STqExecHandle* pExec = &pHandle->execHandle;




14,694,492✔





326



  STqReader* pReader = pExec->pTqReader;




14,696,427✔





327












328



  pBlocks = taosArrayInit(0, sizeof(SSDataBlock));




14,696,623✔





329



  TSDB_CHECK_NULL(pBlocks, code, lino, END, terrno);




14,696,108✔





330



  pSchemas = taosArrayInit(0, sizeof(void*));




14,696,108✔





331



  TSDB_CHECK_NULL(pSchemas, code, lino, END, terrno);




14,696,655✔





332












333



  SSubmitTbData* pSubmitTbData = NULL;




14,696,655✔





334



  code = tqRetrieveTaosxBlock(pReader, pRsp, pBlocks, pSchemas, &pSubmitTbData, rawList, pHandle->fetchMeta);




14,696,745✔





335



  TSDB_CHECK_CODE(code, lino, END);




14,673,915✔





336



  bool tmp = (pSubmitTbData->flags & pRequest->sourceExcluded) != 0;




14,634,063✔





337



  TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);




14,655,033✔





338












339



  if (pHandle->fetchMeta == ONLY_META){




14,652,489✔





340



    goto END;




1,058✔





341



  }










342












343



  int32_t blockNum = taosArrayGetSize(pBlocks) == 0 ? 1 : taosArrayGetSize(pBlocks);




14,649,706✔





344



  if (pRsp->withTbName) {




14,651,517✔





345



    int64_t uid = pExec->pTqReader->lastBlkUid;




14,652,475✔





346



    code = tqAddTbNameToRsp(pTq, uid, pRsp, blockNum);




14,652,244✔





347



    TSDB_CHECK_CODE(code, lino, END);




14,634,597✔





348



  }










349












350



  TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);




14,634,226✔





351



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




29,281,530✔





352



    if (taosArrayGetSize(pBlocks) == 0){




14,613,067✔





353



      void* rawData = taosArrayGetP(rawList, pReader->nextBlk - 1);




50✔





354



      if (rawData == NULL) {




50✔





355



        continue;




×







356



      }










357



      if (tqAddRawDataToRsp(rawData, pRsp, pTq->pVnode->config.tsdbCfg.precision) != 0){




50✔





358



        tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);




×







359



        continue;




×







360



      }










361



      *totalRows += *(uint32_t *)rawData + INT_BYTES; // bytes actually




50✔





362



    } else {










363



      SSDataBlock* pBlock = taosArrayGet(pBlocks, i);




14,642,487✔





364



      if (pBlock == NULL) {




14,632,767✔





365



        continue;




×







366



      }










367












368



      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pSchemas, i);




14,632,767✔





369



      if (tqAddBlockDataToRsp(pBlock, pRsp, pSW, pTq->pVnode->config.tsdbCfg.precision) != 0){




14,651,881✔





370



        tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);




×







371



        continue;




×







372



      }










373



      *totalRows += pBlock->info.rows;




14,616,973✔





374



    }










375












376



    pRsp->blockNum++;




14,650,347✔





377



  }










378



  tqTrace("vgId:%d, process sub data success, response blocknum:%d, rows:%d", pTq->pVnode->config.vgId, pRsp->blockNum, *totalRows);




14,668,463✔





379



END:




14,711,917✔





380



  if (code != 0) {




14,662,469✔





381



    tqError("%s failed at %d, failed to process sub data:%s", __FUNCTION__, lino, tstrerror(code));




39,852✔





382



  }










383



  taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);




14,662,469✔





384



  taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);




14,667,939✔





385



}




14,681,827✔





386












387



static void preProcessSubmitMsg(STqHandle* pHandle, const SMqPollReq* pRequest, SArray** rawList){




15,018,026✔





388



  STqExecHandle* pExec = &pHandle->execHandle;




15,018,026✔





389



  STqReader* pReader = pExec->pTqReader;




15,019,324✔





390



  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);




15,019,414✔





391



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




30,040,847✔





392



    SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, i);




15,022,168✔





393



    if (pSubmitTbData== NULL){




15,022,161✔





394



      taosArrayDestroy(*rawList);




×







395



      *rawList = NULL;




×







396



      return;




×







397



    }










398












399



    int64_t uid = pSubmitTbData->uid;




15,022,161✔





400



    if (pRequest->rawData) {




15,023,254✔





401



      if (taosHashGet(pRequest->uidHash, &uid, LONG_BYTES) != NULL) {




500✔





402



        tqDebug("poll rawdata split,uid:%" PRId64 " is already exists", uid);




100✔





403



        terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;




100✔





404



        return;




100✔





405



      } else {










406



        int32_t code = taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES);




400✔





407



        if (code != 0) {




400✔





408



          tqError("failed to add table uid to hash, code:%d, uid:%" PRId64, code, uid);




×







409



        }










410



      }










411



    }










412












413



    if (pSubmitTbData->pCreateTbReq == NULL){




15,023,289✔





414



      continue;




15,007,875✔





415



    }










416












417



    int64_t createTime = INT64_MAX;




14,367✔





418



    int64_t *cTime = (int64_t*)taosHashGet(pHandle->tableCreateTimeHash, &uid, LONG_BYTES);




14,396✔





419



    if (cTime != NULL){




14,367✔





420



      createTime = *cTime;




554✔





421



    } else{










422



      createTime = metaGetTableCreateTime(pReader->pVnodeMeta, uid, 1);




13,813✔





423



      if (createTime != INT64_MAX){




13,813✔





424



        int32_t code = taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES);




13,813✔





425



        if (code != 0){




13,842✔





426



          tqError("failed to add table create time to hash,code:%d, uid:%"PRId64, code, uid);




×







427



        }










428



      }










429



    }










430



    if (pHandle->fetchMeta == WITH_DATA || pSubmitTbData->ctimeMs > createTime){




14,396✔





431



      tDestroySVSubmitCreateTbReq(pSubmitTbData->pCreateTbReq, TSDB_MSG_FLG_DECODE);




12,821✔





432



      taosMemoryFreeClear(pSubmitTbData->pCreateTbReq);




12,792✔





433



    } else{










434



      taosArrayDestroy(*rawList);




1,575✔





435



      *rawList = NULL;




1,575✔





436



    }










437



  }










438



}










439












440



int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, const SMqPollReq* pRequest) {




15,021,186✔





441



  int32_t code = 0;




15,021,186✔





442



  int32_t lino = 0;




15,021,186✔





443



  SDecoder decoder = {0};




15,021,186✔





444



  STqExecHandle* pExec = &pHandle->execHandle;




15,021,186✔





445



  STqReader* pReader = pExec->pTqReader;




15,021,186✔





446



  SArray *rawList = NULL;




15,021,186✔





447



  if (pRequest->rawData){




15,021,231✔





448



    rawList = taosArrayInit(0, POINTER_BYTES);




250✔





449



    TSDB_CHECK_NULL(rawList, code, lino, END, terrno);




250✔





450



  }










451



  code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver, rawList, &decoder);




15,021,141✔





452



  TSDB_CHECK_CODE(code, lino, END);




15,018,936✔





453



  preProcessSubmitMsg(pHandle, pRequest, &rawList);




15,018,936✔





454



  // data could not contains same uid data in rawdata mode










455



  if (pRequest->rawData != 0 && terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){




15,017,848✔





456



    goto END;




100✔





457



  }










458












459



  // this submit data is metadata and previous data is rawdata










460



  if (pRequest->rawData != 0 && *totalRows > 0 && pRsp->createTableNum == 0 && rawList == NULL){




15,019,330✔





461



    tqDebug("poll rawdata split,vgId:%d, this wal submit data contains metadata and previous data is data", pTq->pVnode->config.vgId);




×







462



    terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;




×







463



    goto END;




×







464



  }










465












466



  // this submit data is rawdata and previous data is metadata










467



  if (pRequest->rawData != 0 && pRsp->createTableNum > 0 && rawList != NULL){




15,017,478✔





468



    tqDebug("poll rawdata split,vgId:%d, this wal submit data is data and previous data is metadata", pTq->pVnode->config.vgId);




×







469



    terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;




×







470



    goto END;




×







471



  }










472












473



  if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {




15,018,391✔





474



    while (tqNextBlockImpl(pReader, NULL)) {




5,858,355✔





475



      tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);




2,766,655✔





476



    }










477



  } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {




11,925,490✔





478



    while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {




23,825,664✔





479



      tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);




11,928,746✔





480



    }










481



  }










482












483



END:




14,976,774✔





484



  tDecoderClear(&decoder);




14,998,157✔





485



  tqReaderClearSubmitMsg(pReader);




15,015,060✔





486



  taosArrayDestroy(rawList);




15,019,503✔





487



  if (code != 0){




15,019,008✔





488



    tqError("%s failed at %d, failed to scan log:%s", __FUNCTION__, lino, tstrerror(code));




×







489



  }










490



  return code;




15,019,008✔





491



}























    

  • 

    •