#3620

Committed 21 Feb 2025 09:00AM UTC coverage: 63.573% (+0.2%) from 63.423%

Build # #3620

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

ci: taosBenchmark add coverage cases branch 3.0 (#29788)

* fix: add unit test for taos-tools

* fix: only .cpp include

* fix: remove no use function

* fix: restore toolsSys.c

* fix: add toolsSys case

* fix: rebuild error fixed

* fix: fix build error

* fix: support get vgroups with core and memory limit

* fix: build error for strcasecmp

* fix: add insertBasic.py case

* fix: add command line set vgroups=3

* fix: change with ns database

* toolscJson read with int replace float and add insertPrecison.py

* fix: add insertBindVGroup.json case

* fix: remove public fun removeQuotation

* fix: vgroups change method

* fix: memory leak for runInsertLimitThread slot

* insertPrecision.py word write wrong

* fix: check isFloat number

* fix: vgroups change logic error

* fix: insertBasic.py real and expect error

* fix: adjust default vgroups

* fix: adjust default vgroups modify comment

Run Details

148962 of 300203 branches covered (49.62%)

Branch coverage included in aggregate %.

15 of 16 new or added lines in 1 file covered. (93.75%)

2018 existing lines in 133 files now uncovered.

233201 of 300933 relevant lines covered (77.49%)

18174406.98 hits per line

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

73.71

/source/dnode/vnode/src/tq/tqScan.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 "tq.h"
static int32_t tqAddRawDataToRsp(const void* rawData, SMqDataRsp* pRsp, int8_t precision) {
  int32_t    code = TDB_CODE_SUCCESS;
  int32_t    lino = 0;
  void*      buf = NULL;

  int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + *(uint32_t *)rawData + INT_BYTES;
  buf = taosMemoryCalloc(1, dataStrLen);
  TSDB_CHECK_NULL(buf, code, lino, END, terrno);

  SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;
  pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION;
  pRetrieve->precision = precision;
  pRetrieve->compressed = 0;

  memcpy(pRetrieve->data, rawData, *(uint32_t *)rawData + INT_BYTES);
  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &dataStrLen), code, lino, END, terrno);
  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);
  pRsp->blockDataElementFree = true;

  tqTrace("tqAddRawDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);
  END:
  if (code != TSDB_CODE_SUCCESS) {
    taosMemoryFree(buf);
    tqError("%s failed at %d, failed to add block data to response:%s", __FUNCTION__, lino, tstrerror(code));
  }
  return code;
}

static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision) {
  int32_t code = 0;
  int32_t lino = 0;

  size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);
  int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;
  void*   buf = taosMemoryCalloc(1, dataStrLen);
  TSDB_CHECK_NULL(buf, code, lino, END, terrno);

  SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;
  pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_VERSION;
  pRetrieve->precision = precision;
  pRetrieve->compressed = 0;
  pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);

  int32_t actualLen = blockEncode(pBlock, pRetrieve->data, dataEncodeBufSize, numOfCols);
  TSDB_CHECK_CONDITION(actualLen >= 0, code, lino, END, terrno);

  actualLen += sizeof(SRetrieveTableRspForTmq);
  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &actualLen), code, lino, END, terrno);
  TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);
  pRsp->blockDataElementFree = true;
  tqTrace("tqAddBlockDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);

END:
  if (code != TSDB_CODE_SUCCESS){
    taosMemoryFree(buf);
    tqError("%s failed at line %d with msg:%s", __func__, lino, tstrerror(code));
  }
  return code;
}

static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {
  int32_t    code = TDB_CODE_SUCCESS;
  int32_t    lino = 0;
  SMetaReader mr = {0};

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

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

  code = metaReaderGetTableEntryByUidCache(&mr, uid);
  TSDB_CHECK_CODE(code, lino, END);

  for (int32_t i = 0; i < n; i++) {
    char* tbName = taosStrdup(mr.me.name);
    TSDB_CHECK_NULL(tbName, code, lino, END, terrno);
    if(taosArrayPush(pRsp->blockTbName, &tbName) == NULL){
      tqError("failed to push tbName to blockTbName:%s, uid:%"PRId64, tbName, uid);
      continue;
    }
    tqTrace("add tbName to response success tbname:%s, uid:%"PRId64, tbName, uid);
  }

END:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);
  }
  metaReaderClear(&mr);
  return code;
}

int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {
  if (task == NULL || pHandle == NULL || res == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  uint64_t ts = 0;
  qStreamSetOpen(task);

  tqDebug("consumer:0x%" PRIx64 " vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);
  int32_t code = qExecTask(task, res, &ts);
  if (code != TSDB_CODE_SUCCESS) {
    tqError("consumer:0x%" PRIx64 " vgId:%d, task exec error since %s", pHandle->consumerId, vgId, tstrerror(code));
    return code;
  }

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

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

  if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type) && pHandle->block != NULL) {
    blockDataDestroy(pHandle->block);
    pHandle->block = NULL;
  }
  if (pHandle->block == NULL) {
    if (pDataBlock == NULL) {
      goto END;
    }

    STqOffsetVal offset = {0};
    code = qStreamExtractOffset(task, &offset);
    TSDB_CHECK_CODE(code, lino, END);

    pHandle->block = NULL;

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

    pHandle->blockTime = offset.ts;
    tOffsetDestroy(&offset);
    int32_t vgId = TD_VID(pTq->pVnode);
    code = getDataBlock(task, pHandle, vgId, &pDataBlock);
    TSDB_CHECK_CODE(code, lino, END);
  }

  const STqExecHandle* pExec = &pHandle->execHandle;
  code = tqAddBlockDataToRsp(pHandle->block, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);
  TSDB_CHECK_CODE(code, lino, END);

  pRsp->blockNum++;
  if (pDataBlock == NULL) {
    blockDataDestroy(pHandle->block);
    pHandle->block = NULL;
  } else {
    code = copyDataBlock(pHandle->block, pDataBlock);
    TSDB_CHECK_CODE(code, lino, END);

    STqOffsetVal offset = {0};
    code = qStreamExtractOffset(task, &offset);
    TSDB_CHECK_CODE(code, lino, END);

    pRsp->sleepTime = offset.ts - pHandle->blockTime;
    pHandle->blockTime = offset.ts;
    tOffsetDestroy(&offset);
  }

END:
  if (code != TSDB_CODE_SUCCESS) {
    tqError("%s failed at %d, failed to process replay response:%s", __FUNCTION__, lino, tstrerror(code));
  }
  return code;
}

int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {
  int32_t code = 0;
  int32_t lino = 0;
  TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(pOffset, code, lino, END, TSDB_CODE_INVALID_PARA);
  TSDB_CHECK_NULL(pRequest, code, lino, END, TSDB_CODE_INVALID_PARA);

  int32_t vgId = TD_VID(pTq->pVnode);
  int32_t totalRows = 0;

  const STqExecHandle* pExec = &pHandle->execHandle;
  qTaskInfo_t          task = pExec->task;

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

  qStreamSetSourceExcluded(task, pRequest->sourceExcluded);
  int64_t st = taosGetTimestampMs();
  while (1) {
    SSDataBlock* pDataBlock = NULL;
    code = getDataBlock(task, pHandle, vgId, &pDataBlock);
    TSDB_CHECK_CODE(code, lino, END);

    if (pRequest->enableReplay) {
      code = tqProcessReplayRsp(pTq, pHandle, pRsp, pRequest, pDataBlock, task);
      TSDB_CHECK_CODE(code, lino, END);
      break;
    }
    if (pDataBlock == NULL) {
      break;
    }
    code = tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);
    TSDB_CHECK_CODE(code, lino, END);

    pRsp->blockNum++;
    totalRows += pDataBlock->info.rows;
    if (totalRows >= pRequest->minPollRows || (taosGetTimestampMs() - st > pRequest->timeout)) {
      break;
    }
  }

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

END:
  if (code != 0) {
    tqError("%s failed at %d, tmq task executed error msg:%s", __FUNCTION__, lino, tstrerror(code));
  }
  return code;
}

int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {
  int32_t code = 0;
  int32_t lino = 0;
  char* tbName = NULL;
  SSchemaWrapper* pSW = NULL;
  const STqExecHandle* pExec = &pHandle->execHandle;
  qTaskInfo_t          task = pExec->task;
  code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);
  TSDB_CHECK_CODE(code, lino, END);

  int32_t rowCnt = 0;
  int64_t st = taosGetTimestampMs();
  while (1) {
    SSDataBlock* pDataBlock = NULL;
    uint64_t     ts = 0;
    tqDebug("tmqsnap task start to execute");
    code = qExecTask(task, &pDataBlock, &ts);
    TSDB_CHECK_CODE(code, lino, END);
    tqDebug("tmqsnap task execute end, get %p", pDataBlock);

    if (pDataBlock != NULL && pDataBlock->info.rows > 0) {
      if (pRsp->withTbName) {
        tbName = taosStrdup(qExtractTbnameFromTask(task));
        TSDB_CHECK_NULL(tbName, code, lino, END, terrno);
        TSDB_CHECK_NULL(taosArrayPush(pRsp->blockTbName, &tbName), code, lino, END, terrno);
        tqDebug("vgId:%d, add tbname:%s to rsp msg", pTq->pVnode->config.vgId, tbName);
        tbName = NULL;
      }
      if (pRsp->withSchema) {
        pSW = tCloneSSchemaWrapper(qExtractSchemaFromTask(task));
        TSDB_CHECK_NULL(pSW, code, lino, END, terrno);
        TSDB_CHECK_NULL(taosArrayPush(pRsp->blockSchema, &pSW), code, lino, END, terrno);
        pSW = NULL;
      }

      code = tqAddBlockDataToRsp(pDataBlock, pRsp, taosArrayGetSize(pDataBlock->pDataBlock),
                                 pTq->pVnode->config.tsdbCfg.precision);
      TSDB_CHECK_CODE(code, lino, END);

      pRsp->blockNum++;
      rowCnt += pDataBlock->info.rows;
      if (rowCnt <= pRequest->minPollRows && (taosGetTimestampMs() - st <= pRequest->timeout)) {
        continue;
      }
    }

    // get meta
    SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);
    if (taosArrayGetSize(tmp->batchMetaReq) > 0) {
      code = qStreamExtractOffset(task, &tmp->rspOffset);
      TSDB_CHECK_CODE(code, lino, END);
      *pBatchMetaRsp = *tmp;
      tqDebug("tmqsnap task get meta");
      break;
    }

    if (pDataBlock == NULL) {
      code = qStreamExtractOffset(task, pOffset);
      TSDB_CHECK_CODE(code, lino, END);

      if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
        continue;
      }

      tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);
      code = qStreamExtractOffset(task, &pRsp->rspOffset);
      break;
    }

    if (pRsp->blockNum > 0) {
      tqDebug("tmqsnap task exec exited, get data");
      code = qStreamExtractOffset(task, &pRsp->rspOffset);
      break;
    }
  }
  tqDebug("%s:%d success", __FUNCTION__, lino);
END:
  if (code != 0){
    tqError("%s failed at %d, vgId:%d, task exec error since %s", __FUNCTION__ , lino, pTq->pVnode->config.vgId, tstrerror(code));
  }
  taosMemoryFree(pSW);
  taosMemoryFree(tbName);
  return code;
}

static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows,
                             const SMqPollReq* pRequest, SArray* rawList){
  int32_t code = 0;
  int32_t lino = 0;
  SArray* pBlocks = NULL;
  SArray* pSchemas = NULL;

  STqExecHandle* pExec = &pHandle->execHandle;
  STqReader* pReader = pExec->pTqReader;

  pBlocks = taosArrayInit(0, sizeof(SSDataBlock));
  TSDB_CHECK_NULL(pBlocks, code, lino, END, terrno);
  pSchemas = taosArrayInit(0, sizeof(void*));
  TSDB_CHECK_NULL(pSchemas, code, lino, END, terrno);

  SSubmitTbData* pSubmitTbData = NULL;
  code = tqRetrieveTaosxBlock(pReader, pRsp, pBlocks, pSchemas, &pSubmitTbData, rawList, pHandle->fetchMeta);
  TSDB_CHECK_CODE(code, lino, END);
  bool tmp = (pSubmitTbData->flags & pRequest->sourceExcluded) != 0;
  TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);


  int32_t blockNum = taosArrayGetSize(pBlocks) == 0 ? 1 : taosArrayGetSize(pBlocks);
  if (pRsp->withTbName) {
    int64_t uid = pExec->pTqReader->lastBlkUid;
    code = tqAddTbNameToRsp(pTq, uid, pRsp, blockNum);
    TSDB_CHECK_CODE(code, lino, END);
  }

  tmp = (pHandle->fetchMeta == ONLY_META && pSubmitTbData->pCreateTbReq == NULL);
  TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);
  for (int32_t i = 0; i < blockNum; i++) {
    if (taosArrayGetSize(pBlocks) == 0){
      void* rawData = taosArrayGetP(rawList, pReader->nextBlk - 1);
      if (rawData == NULL) {
        continue;
      }
      if (tqAddRawDataToRsp(rawData, pRsp, pTq->pVnode->config.tsdbCfg.precision) != 0){
        tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);
        continue;
      }
      *totalRows += *(uint32_t *)rawData + INT_BYTES; // bytes actually
    } else {
      SSDataBlock* pBlock = taosArrayGet(pBlocks, i);
      if (pBlock == NULL) {
        continue;
      }

      if (tqAddBlockDataToRsp(pBlock, pRsp, taosArrayGetSize(pBlock->pDataBlock), pTq->pVnode->config.tsdbCfg.precision) != 0){
        tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);
        continue;
      }
      *totalRows += pBlock->info.rows;
    }

    void** pSW = taosArrayGet(pSchemas, i);
    if (taosArrayPush(pRsp->blockSchema, pSW) == NULL){
      tqError("vgId:%d, failed to add schema to rsp msg", pTq->pVnode->config.vgId);
      continue;
    }
    *pSW = NULL;
    pRsp->blockNum++;
  }
  tqTrace("vgId:%d, process sub data success, response blocknum:%d, rows:%d", pTq->pVnode->config.vgId, pRsp->blockNum, *totalRows);
END:
  if (code != 0) {
    tqError("%s failed at %d, failed to process sub data:%s", __FUNCTION__, lino, tstrerror(code));
  }
  taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);
  taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);
}

static void preProcessSubmitMsg(STqHandle* pHandle, const SMqPollReq* pRequest, SArray** rawList){
  STqExecHandle* pExec = &pHandle->execHandle;
  STqReader* pReader = pExec->pTqReader;
  int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);
  for (int32_t i = 0; i < blockSz; i++){
    SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, i);
    if (pSubmitTbData== NULL){
      taosArrayDestroy(*rawList);
      *rawList = NULL;
      return;
    }

    int64_t uid = pSubmitTbData->uid;
    if (taosHashGet(pRequest->uidHash, &uid, LONG_BYTES) != NULL) {
      tqDebug("poll rawdata split,uid:%" PRId64 " is already exists", uid);
      terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;
      return;
    } else {
      int32_t code = taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES);
      if (code != 0){
        tqError("failed to add table uid to hash, code:%d, uid:%"PRId64, code, uid);
      }
    }

    if (pSubmitTbData->pCreateTbReq == NULL){
      continue;
    }

    int64_t createTime = INT64_MAX;
    int64_t *cTime = (int64_t*)taosHashGet(pHandle->tableCreateTimeHash, &uid, LONG_BYTES);
    if (cTime != NULL){
      createTime = *cTime;
    } else{
      createTime = metaGetTableCreateTime(pReader->pVnodeMeta, uid, 1);
      if (createTime != INT64_MAX){
        int32_t code = taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES);
        if (code != 0){
          tqError("failed to add table create time to hash,code:%d, uid:%"PRId64, code, uid);
        }
      }
    }
    if (pHandle->fetchMeta == WITH_DATA || pSubmitTbData->ctimeMs > createTime){
      tDestroySVCreateTbReq(pSubmitTbData->pCreateTbReq, TSDB_MSG_FLG_DECODE);
      pSubmitTbData->pCreateTbReq = NULL;
    } else{
      taosArrayDestroy(*rawList);
      *rawList = NULL;
    }
  }
}

int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, const SMqPollReq* pRequest) {
  int32_t code = 0;
  int32_t lino = 0;
  STqExecHandle* pExec = &pHandle->execHandle;
  STqReader* pReader = pExec->pTqReader;
  SArray *rawList = NULL;
  if (pRequest->rawData){
    rawList = taosArrayInit(0, POINTER_BYTES);
    TSDB_CHECK_NULL(rawList, code, lino, END, terrno);
  }
  code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver, rawList);
  TSDB_CHECK_CODE(code, lino, END);
  if (pRequest->rawData) {
    preProcessSubmitMsg(pHandle, pRequest, &rawList);
  }
  // data could not contains same uid data in rawdata mode
  if (pRequest->rawData != 0 && terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){
    goto END;
  }

  // this submit data is metadata and previous data is rawdata
  if (pRequest->rawData != 0 && *totalRows > 0 && pRsp->createTableNum == 0 && rawList == NULL){
    tqDebug("poll rawdata split,vgId:%d, this wal submit data contains metadata and previous data is data", pTq->pVnode->config.vgId);
    terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;
    goto END;
  }

  // this submit data is rawdata and previous data is metadata
  if (pRequest->rawData != 0 && pRsp->createTableNum > 0 && rawList != NULL){
    tqDebug("poll rawdata split,vgId:%d, this wal submit data is data and previous data is metadata", pTq->pVnode->config.vgId);
    terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;
    goto END;
  }

  if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
    while (tqNextBlockImpl(pReader, NULL)) {
      tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);
    }
  } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
    while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {
      tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);
    }
  }

END:
  tqReaderClearSubmitMsg(pReader);
  taosArrayDestroy(rawList);
  if (code != 0){
    tqError("%s failed at %d, failed to scan log:%s", __FUNCTION__, lino, tstrerror(code));
  }
  return code;
}

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) {	18✔
18	int32_t code = TDB_CODE_SUCCESS;	18✔
19	int32_t lino = 0;	18✔
20	void* buf = NULL;	18✔
21
22	int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + (uint32_t )rawData + INT_BYTES;	18✔
23	buf = taosMemoryCalloc(1, dataStrLen);	18!
24	TSDB_CHECK_NULL(buf, code, lino, END, terrno);	18!
25
26	SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;	18✔
27	pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION;	18✔
28	pRetrieve->precision = precision;	18✔
29	pRetrieve->compressed = 0;	18✔
30
31	memcpy(pRetrieve->data, rawData, (uint32_t )rawData + INT_BYTES);	18✔
32	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &dataStrLen), code, lino, END, terrno);	36!
33	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);	36!
34	pRsp->blockDataElementFree = true;	18✔
35
36	tqTrace("tqAddRawDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);	18!
37	END:	18✔
38	if (code != TSDB_CODE_SUCCESS) {	18!
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;	18✔
43	}
44
45	static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision) {	641,831✔
46	int32_t code = 0;	641,831✔
47	int32_t lino = 0;	641,831✔
48
49	size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);	641,831✔
50	int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;	641,670✔
51	void* buf = taosMemoryCalloc(1, dataStrLen);	641,670!
52	TSDB_CHECK_NULL(buf, code, lino, END, terrno);	642,046!
53
54	SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;	642,046✔
55	pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_VERSION;	642,046✔
56	pRetrieve->precision = precision;	642,046✔
57	pRetrieve->compressed = 0;	642,046✔
58	pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);	642,046✔
59
60	int32_t actualLen = blockEncode(pBlock, pRetrieve->data, dataEncodeBufSize, numOfCols);	641,974✔
61	TSDB_CHECK_CONDITION(actualLen >= 0, code, lino, END, terrno);	641,989!
62
63	actualLen += sizeof(SRetrieveTableRspForTmq);	641,989✔
64	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &actualLen), code, lino, END, terrno);	1,283,960!
65	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);	1,283,881!
66	pRsp->blockDataElementFree = true;	641,910✔
67	tqTrace("tqAddBlockDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);	641,910✔
68
69	END:	641,904✔
70	if (code != TSDB_CODE_SUCCESS){	641,910!
71	taosMemoryFree(buf);	×
72	tqError("%s failed at line %d with msg:%s", __func__, lino, tstrerror(code));	×
73	}
74	return code;	641,812✔
75	}
76
77	static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {	117,906✔
78	int32_t code = TDB_CODE_SUCCESS;	117,906✔
79	int32_t lino = 0;	117,906✔
80	SMetaReader mr = {0};	117,906✔
81
82	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	117,906!
83	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	117,906!
84
85	metaReaderDoInit(&mr, pTq->pVnode->pMeta, META_READER_LOCK);	117,906✔
86
87	code = metaReaderGetTableEntryByUidCache(&mr, uid);	118,078✔
88	TSDB_CHECK_CODE(code, lino, END);	117,760!
89
90	for (int32_t i = 0; i < n; i++) {	235,524✔
91	char* tbName = taosStrdup(mr.me.name);	117,732!
92	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	117,710!
93	if(taosArrayPush(pRsp->blockTbName, &tbName) == NULL){	235,462!
94	tqError("failed to push tbName to blockTbName:%s, uid:%"PRId64, tbName, uid);	×
95	continue;	×
96	}
97	tqTrace("add tbName to response success tbname:%s, uid:%"PRId64, tbName, uid);	117,752!
98	}
99
100	END:	117,792✔
101	if (code != TSDB_CODE_SUCCESS) {	117,792!
UNCOV 102	tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);	×
103	}
104	metaReaderClear(&mr);	117,792✔
105	return code;	117,956✔
106	}
107
108	int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {	574,958✔
109	if (task == NULL \|\| pHandle == NULL \|\| res == NULL) {	574,958!
110	return TSDB_CODE_INVALID_PARA;	×
111	}
112	uint64_t ts = 0;	574,961✔
113	qStreamSetOpen(task);	574,961✔
114
115	tqDebug("consumer:0x%" PRIx64 " vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);	574,950!
116	int32_t code = qExecTask(task, res, &ts);	575,005✔
117	if (code != TSDB_CODE_SUCCESS) {	575,001!
118	tqError("consumer:0x%" PRIx64 " vgId:%d, task exec error since %s", pHandle->consumerId, vgId, tstrerror(code));	×
119	return code;	×
120	}
121
122	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq one task end executed, pDataBlock:%p", pHandle->consumerId, vgId, *res);	575,001!
123	return 0;	575,011✔
124	}
125
126	static int32_t tqProcessReplayRsp(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, const SMqPollReq* pRequest, SSDataBlock* pDataBlock, qTaskInfo_t task){	33✔
127	int32_t code = 0;	33✔
128	int32_t lino = 0;	33✔
129
130	if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type) && pHandle->block != NULL) {	33✔
131	blockDataDestroy(pHandle->block);	1✔
132	pHandle->block = NULL;	1✔
133	}
134	if (pHandle->block == NULL) {	33✔
135	if (pDataBlock == NULL) {	11✔
136	goto END;	6✔
137	}
138
139	STqOffsetVal offset = {0};	5✔
140	code = qStreamExtractOffset(task, &offset);	5✔
141	TSDB_CHECK_CODE(code, lino, END);	5!
142
143	pHandle->block = NULL;	5✔
144
145	code = createOneDataBlock(pDataBlock, true, &pHandle->block);	5✔
146	TSDB_CHECK_CODE(code, lino, END);	5!
147
148	pHandle->blockTime = offset.ts;	5✔
149	tOffsetDestroy(&offset);	5✔
150	int32_t vgId = TD_VID(pTq->pVnode);	5✔
151	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	5✔
152	TSDB_CHECK_CODE(code, lino, END);	5!
153	}
154
155	const STqExecHandle* pExec = &pHandle->execHandle;	27✔
156	code = tqAddBlockDataToRsp(pHandle->block, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);	27✔
157	TSDB_CHECK_CODE(code, lino, END);	27!
158
159	pRsp->blockNum++;	27✔
160	if (pDataBlock == NULL) {	27✔
161	blockDataDestroy(pHandle->block);	4✔
162	pHandle->block = NULL;	4✔
163	} else {
164	code = copyDataBlock(pHandle->block, pDataBlock);	23✔
165	TSDB_CHECK_CODE(code, lino, END);	23!
166
167	STqOffsetVal offset = {0};	23✔
168	code = qStreamExtractOffset(task, &offset);	23✔
169	TSDB_CHECK_CODE(code, lino, END);	23!
170
171	pRsp->sleepTime = offset.ts - pHandle->blockTime;	23✔
172	pHandle->blockTime = offset.ts;	23✔
173	tOffsetDestroy(&offset);	23✔
174	}
175
176	END:	33✔
177	if (code != TSDB_CODE_SUCCESS) {	33!
178	tqError("%s failed at %d, failed to process replay response:%s", __FUNCTION__, lino, tstrerror(code));	×
179	}
180	return code;	33✔
181	}
182
183	int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {	56,570✔
184	int32_t code = 0;	56,570✔
185	int32_t lino = 0;	56,570✔
186	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	56,570!
187	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	56,570!
188	TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);	56,570!
189	TSDB_CHECK_NULL(pOffset, code, lino, END, TSDB_CODE_INVALID_PARA);	56,570!
190	TSDB_CHECK_NULL(pRequest, code, lino, END, TSDB_CODE_INVALID_PARA);	56,570!
191
192	int32_t vgId = TD_VID(pTq->pVnode);	56,570✔
193	int32_t totalRows = 0;	56,570✔
194
195	const STqExecHandle* pExec = &pHandle->execHandle;	56,570✔
196	qTaskInfo_t task = pExec->task;	56,570✔
197
198	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	56,570✔
199	TSDB_CHECK_CODE(code, lino, END);	56,573✔
200
201	qStreamSetSourceExcluded(task, pRequest->sourceExcluded);	55,885✔
202	int64_t st = taosGetTimestampMs();	55,885✔
203	while (1) {	519,072✔
204	SSDataBlock* pDataBlock = NULL;	574,957✔
205	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	574,957✔
206	TSDB_CHECK_CODE(code, lino, END);	575,006!
207
208	if (pRequest->enableReplay) {	575,006✔
209	code = tqProcessReplayRsp(pTq, pHandle, pRsp, pRequest, pDataBlock, task);	33✔
210	TSDB_CHECK_CODE(code, lino, END);	33!
211	break;	33✔
212	}
213	if (pDataBlock == NULL) {	574,973✔
214	break;	52,967✔
215	}
216	code = tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);	522,006✔
217	TSDB_CHECK_CODE(code, lino, END);	521,957!
218
219	pRsp->blockNum++;	521,957✔
220	totalRows += pDataBlock->info.rows;	521,957✔
221	if (totalRows >= pRequest->minPollRows \|\| (taosGetTimestampMs() - st > pRequest->timeout)) {	1,041,667✔
222	break;
223	}
224	}
225
226	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq task executed finished, total blocks:%d, totalRows:%d", pHandle->consumerId, vgId, pRsp->blockNum, totalRows);	55,883!
227	code = qStreamExtractOffset(task, &pRsp->rspOffset);	55,883✔
228
229	END:	56,574✔
230	if (code != 0) {	56,574✔
231	tqError("%s failed at %d, tmq task executed error msg:%s", __FUNCTION__, lino, tstrerror(code));	688!
232	}
233	return code;	56,574✔
234	}
235
236	int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {	69✔
237	int32_t code = 0;	69✔
238	int32_t lino = 0;	69✔
239	char* tbName = NULL;	69✔
240	SSchemaWrapper* pSW = NULL;	69✔
241	const STqExecHandle* pExec = &pHandle->execHandle;	69✔
242	qTaskInfo_t task = pExec->task;	69✔
243	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	69✔
244	TSDB_CHECK_CODE(code, lino, END);	69!
245
246	int32_t rowCnt = 0;	69✔
247	int64_t st = taosGetTimestampMs();	69✔
248	while (1) {	4,184✔
249	SSDataBlock* pDataBlock = NULL;	4,253✔
250	uint64_t ts = 0;	4,253✔
251	tqDebug("tmqsnap task start to execute");	4,253!
252	code = qExecTask(task, &pDataBlock, &ts);	4,255✔
253	TSDB_CHECK_CODE(code, lino, END);	4,255!
254	tqDebug("tmqsnap task execute end, get %p", pDataBlock);	4,255!
255
256	if (pDataBlock != NULL && pDataBlock->info.rows > 0) {	4,255!
257	if (pRsp->withTbName) {	2,127!
258	tbName = taosStrdup(qExtractTbnameFromTask(task));	2,127!
259	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	2,127!
260	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockTbName, &tbName), code, lino, END, terrno);	4,253!
261	tqDebug("vgId:%d, add tbname:%s to rsp msg", pTq->pVnode->config.vgId, tbName);	2,126!
262	tbName = NULL;	2,127✔
263	}
264	if (pRsp->withSchema) {	2,127!
265	pSW = tCloneSSchemaWrapper(qExtractSchemaFromTask(task));	2,127✔
266	TSDB_CHECK_NULL(pSW, code, lino, END, terrno);	2,127!
267	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockSchema, &pSW), code, lino, END, terrno);	4,253!
268	pSW = NULL;	2,126✔
269	}
270
271	code = tqAddBlockDataToRsp(pDataBlock, pRsp, taosArrayGetSize(pDataBlock->pDataBlock),	2,126✔
272	pTq->pVnode->config.tsdbCfg.precision);	2,126✔
273	TSDB_CHECK_CODE(code, lino, END);	2,125!
274
275	pRsp->blockNum++;	2,125✔
276	rowCnt += pDataBlock->info.rows;	2,125✔
277	if (rowCnt <= pRequest->minPollRows && (taosGetTimestampMs() - st <= pRequest->timeout)) {	4,221!
278	continue;	4,184✔
279	}
280	}
281
282	// get meta
283	SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);	2,157✔
284	if (taosArrayGetSize(tmp->batchMetaReq) > 0) {	2,157✔
285	code = qStreamExtractOffset(task, &tmp->rspOffset);	14✔
286	TSDB_CHECK_CODE(code, lino, END);	14!
287	pBatchMetaRsp = tmp;	14✔
288	tqDebug("tmqsnap task get meta");	14!
289	break;	14✔
290	}
291
292	if (pDataBlock == NULL) {	2,143✔
293	code = qStreamExtractOffset(task, pOffset);	2,114✔
294	TSDB_CHECK_CODE(code, lino, END);	2,114!
295
296	if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {	2,114✔
297	continue;	2,088✔
298	}
299
300	tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);	26!
301	code = qStreamExtractOffset(task, &pRsp->rspOffset);	26✔
302	break;	26✔
303	}
304
305	if (pRsp->blockNum > 0) {	29!
306	tqDebug("tmqsnap task exec exited, get data");	29!
307	code = qStreamExtractOffset(task, &pRsp->rspOffset);	29✔
308	break;	29✔
309	}
310	}
311	tqDebug("%s:%d success", __FUNCTION__, lino);	69!
312	END:	×
313	if (code != 0){	69!
314	tqError("%s failed at %d, vgId:%d, task exec error since %s", __FUNCTION__ , lino, pTq->pVnode->config.vgId, tstrerror(code));	×
315	}
316	taosMemoryFree(pSW);	69!
317	taosMemoryFree(tbName);	69!
318	return code;	69✔
319	}
320
321	static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows,	118,641✔
322	const SMqPollReq* pRequest, SArray* rawList){
323	int32_t code = 0;	118,641✔
324	int32_t lino = 0;	118,641✔
325	SArray* pBlocks = NULL;	118,641✔
326	SArray* pSchemas = NULL;	118,641✔
327
328	STqExecHandle* pExec = &pHandle->execHandle;	118,641✔
329	STqReader* pReader = pExec->pTqReader;	118,641✔
330
331	pBlocks = taosArrayInit(0, sizeof(SSDataBlock));	118,641✔
332	TSDB_CHECK_NULL(pBlocks, code, lino, END, terrno);	118,662!
333	pSchemas = taosArrayInit(0, sizeof(void*));	118,662✔
334	TSDB_CHECK_NULL(pSchemas, code, lino, END, terrno);	118,666!
335
336	SSubmitTbData* pSubmitTbData = NULL;	118,666✔
337	code = tqRetrieveTaosxBlock(pReader, pRsp, pBlocks, pSchemas, &pSubmitTbData, rawList, pHandle->fetchMeta);	118,666✔
338	TSDB_CHECK_CODE(code, lino, END);	118,488✔
339	bool tmp = (pSubmitTbData->flags & pRequest->sourceExcluded) != 0;	118,061✔
340	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	118,061✔
341
342
343	int32_t blockNum = taosArrayGetSize(pBlocks) == 0 ? 1 : taosArrayGetSize(pBlocks);	117,930✔
344	if (pRsp->withTbName) {	117,936!
345	int64_t uid = pExec->pTqReader->lastBlkUid;	117,951✔
346	code = tqAddTbNameToRsp(pTq, uid, pRsp, blockNum);	117,951✔
347	TSDB_CHECK_CODE(code, lino, END);	117,946!
348	}
349
350	tmp = (pHandle->fetchMeta == ONLY_META && pSubmitTbData->pCreateTbReq == NULL);	117,931✔
351	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	117,931✔
352	for (int32_t i = 0; i < blockNum; i++) {	235,710✔
353	if (taosArrayGetSize(pBlocks) == 0){	117,814✔
354	void* rawData = taosArrayGetP(rawList, pReader->nextBlk - 1);	18✔
355	if (rawData == NULL) {	18!
356	continue;	×
357	}
358	if (tqAddRawDataToRsp(rawData, pRsp, pTq->pVnode->config.tsdbCfg.precision) != 0){	18!
359	tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);	×
360	continue;	×
361	}
362	totalRows += (uint32_t *)rawData + INT_BYTES; // bytes actually	18✔
363	} else {
364	SSDataBlock* pBlock = taosArrayGet(pBlocks, i);	117,796✔
365	if (pBlock == NULL) {	117,740!
366	continue;	×
367	}
368
369	if (tqAddBlockDataToRsp(pBlock, pRsp, taosArrayGetSize(pBlock->pDataBlock), pTq->pVnode->config.tsdbCfg.precision) != 0){	117,740!
370	tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);	×
371	continue;	×
372	}
373	*totalRows += pBlock->info.rows;	117,765✔
374	}
375
376	void** pSW = taosArrayGet(pSchemas, i);	117,783✔
377	if (taosArrayPush(pRsp->blockSchema, pSW) == NULL){	235,570!
378	tqError("vgId:%d, failed to add schema to rsp msg", pTq->pVnode->config.vgId);	×
379	continue;	×
380	}
381	*pSW = NULL;	117,786✔
382	pRsp->blockNum++;	117,786✔
383	}
384	tqTrace("vgId:%d, process sub data success, response blocknum:%d, rows:%d", pTq->pVnode->config.vgId, pRsp->blockNum, *totalRows);	117,896!
385	END:	117,896✔
386	if (code != 0) {	118,461✔
387	tqError("%s failed at %d, failed to process sub data:%s", __FUNCTION__, lino, tstrerror(code));	427!
388	}
389	taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);	118,461✔
390	taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);	118,552✔
391	}	118,629✔
392
393	static void preProcessSubmitMsg(STqHandle* pHandle, const SMqPollReq* pRequest, SArray** rawList){	30✔
394	STqExecHandle* pExec = &pHandle->execHandle;	30✔
395	STqReader* pReader = pExec->pTqReader;	30✔
396	int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);	30✔
397	for (int32_t i = 0; i < blockSz; i++){	60✔
398	SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, i);	38✔
399	if (pSubmitTbData== NULL){	38!
400	taosArrayDestroy(*rawList);	×
401	*rawList = NULL;	×
402	return;	8✔
403	}
404
405	int64_t uid = pSubmitTbData->uid;	38✔
406	if (taosHashGet(pRequest->uidHash, &uid, LONG_BYTES) != NULL) {	38✔
407	tqDebug("poll rawdata split,uid:%" PRId64 " is already exists", uid);	8!
408	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	8✔
409	return;	8✔
410	} else {
411	int32_t code = taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES);	30✔
412	if (code != 0){	30!
413	tqError("failed to add table uid to hash, code:%d, uid:%"PRId64, code, uid);	×
414	}
415	}
416
417	if (pSubmitTbData->pCreateTbReq == NULL){	30✔
418	continue;	20✔
419	}
420
421	int64_t createTime = INT64_MAX;	10✔
422	int64_t cTime = (int64_t)taosHashGet(pHandle->tableCreateTimeHash, &uid, LONG_BYTES);	10✔
423	if (cTime != NULL){	10✔
424	createTime = *cTime;	2✔
425	} else{
426	createTime = metaGetTableCreateTime(pReader->pVnodeMeta, uid, 1);	8✔
427	if (createTime != INT64_MAX){	8!
428	int32_t code = taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES);	8✔
429	if (code != 0){	8!
430	tqError("failed to add table create time to hash,code:%d, uid:%"PRId64, code, uid);	×
431	}
432	}
433	}
434	if (pHandle->fetchMeta == WITH_DATA \|\| pSubmitTbData->ctimeMs > createTime){	10!
435	tDestroySVCreateTbReq(pSubmitTbData->pCreateTbReq, TSDB_MSG_FLG_DECODE);	1✔
436	pSubmitTbData->pCreateTbReq = NULL;	1✔
437	} else{
438	taosArrayDestroy(*rawList);	9✔
439	*rawList = NULL;	9✔
440	}
441	}
442	}
443
444	int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, const SMqPollReq* pRequest) {	122,106✔
445	int32_t code = 0;	122,106✔
446	int32_t lino = 0;	122,106✔
447	STqExecHandle* pExec = &pHandle->execHandle;	122,106✔
448	STqReader* pReader = pExec->pTqReader;	122,106✔
449	SArray *rawList = NULL;	122,106✔
450	if (pRequest->rawData){	122,106✔
451	rawList = taosArrayInit(0, POINTER_BYTES);	30✔
452	TSDB_CHECK_NULL(rawList, code, lino, END, terrno);	30!
453	}
454	code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver, rawList);	122,106✔
455	TSDB_CHECK_CODE(code, lino, END);	122,083!
456	if (pRequest->rawData) {	122,083✔
457	preProcessSubmitMsg(pHandle, pRequest, &rawList);	30✔
458	}
459	// data could not contains same uid data in rawdata mode
460	if (pRequest->rawData != 0 && terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){	122,090✔
461	goto END;	8✔
462	}
463
464	// this submit data is metadata and previous data is rawdata
465	if (pRequest->rawData != 0 && *totalRows > 0 && pRsp->createTableNum == 0 && rawList == NULL){	122,082!
466	tqDebug("poll rawdata split,vgId:%d, this wal submit data contains metadata and previous data is data", pTq->pVnode->config.vgId);	×
467	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
468	goto END;	×
469	}
470
471	// this submit data is rawdata and previous data is metadata
472	if (pRequest->rawData != 0 && pRsp->createTableNum > 0 && rawList != NULL){	122,082!
473	tqDebug("poll rawdata split,vgId:%d, this wal submit data is data and previous data is metadata", pTq->pVnode->config.vgId);	×
474	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
475	goto END;	×
476	}
477
478	if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {	122,082✔
479	while (tqNextBlockImpl(pReader, NULL)) {	69,838✔
480	tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);	33,171✔
481	}
482	} else if (pExec->subType == TOPIC_SUB_TYPE__DB) {	85,407!
483	while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {	170,811✔
484	tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);	85,481✔
485	}
486	}
487
488	END:	85,175✔
489	tqReaderClearSubmitMsg(pReader);	121,843✔
490	taosArrayDestroy(rawList);	122,010✔
491	if (code != 0){	121,995!
492	tqError("%s failed at %d, failed to scan log:%s", __FUNCTION__, lino, tstrerror(code));	×
493	}
494	return code;	121,889✔
495	}

taosdata / TDengine / #3620

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

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